fix enforce CAMARA array limits on EAM AppManifest (8ac20992) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/api/camara/edge_application_management/v0_1_0_alpha_1/schemas.py

+2 −2

Changes for src/open_exposure_gateway/api/camara/edge_application_management/v0_1_0_alpha_1/schemas.py: 2 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -74,7 +74,7 @@ class ComponentSpecItem(BaseModel):
		model_config = ConfigDict(extra="forbid")

		componentName: str = Field(max_length=64)
		networkInterfaces: list[NetworkInterface]
		networkInterfaces: list[NetworkInterface] = Field(min_length=1, max_length=100)


		class GpuInfo(BaseModel):
		@@ -223,7 +223,7 @@ class AppManifest(BaseModel):
		operatingSystem: Optional[OperatingSystem] = None
		appRepo: AppRepo
		requiredResources: RequiredResources
		componentSpec: list[ComponentSpecItem]
		componentSpec: list[ComponentSpecItem] = Field(max_length=100)


		class AppManifestEnvelope(BaseModel):

tests/unit/test_eam_flows.py

+56 −0

Changes for tests/unit/test_eam_flows.py: 56 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -92,6 +92,20 @@ def _manifest_body(num_cpu: float \| int) -> dict[str, Any]:
		}


		def _component(index: int) -> dict[str, Any]:
		return {
		"componentName": f"comp{index}",
		"networkInterfaces": [
		{
		"interfaceId": f"if_{index:03d}",
		"protocol": "TCP",
		"port": 8080,
		"visibilityType": "VISIBILITY_EXTERNAL",
		}
		],
		}


		class TestCreateAppInstanceFlow:
		@pytest.fixture(autouse=True)
		def _registered_app(self, register_app: Callable[[Any], None]) -> None:
		@@ -690,6 +704,48 @@ class TestSubmitAppFlow:
		response = api_client.post(f"{EAM_BASE}/apps", json=body)
		assert response.status_code == 400

		def test_manifest_with_component_and_interface_is_accepted(
		self, api_client: TestClient
		) -> None:
		body = _manifest_body(num_cpu=2)
		body["componentSpec"] = [_component(0)]
		response = api_client.post(f"{EAM_BASE}/apps", json=body)
		assert response.status_code == 201

		def test_component_without_network_interfaces_is_rejected(
		self, api_client: TestClient, fake_srm: FakeSRMClient
		) -> None:
		"""componentSpec[].networkInterfaces has minItems 1 in the CAMARA EAM spec."""
		body = _manifest_body(num_cpu=2)
		body["componentSpec"] = [{"componentName": "web", "networkInterfaces": []}]
		response = api_client.post(f"{EAM_BASE}/apps", json=body)
		assert response.status_code == 400
		assert response.json()["code"] == "INVALID_ARGUMENT"
		assert fake_srm.catalog == {}

		def test_more_than_100_components_is_rejected(
		self, api_client: TestClient, fake_srm: FakeSRMClient
		) -> None:
		"""AppManifest.componentSpec has maxItems 100 in the CAMARA EAM spec."""
		body = _manifest_body(num_cpu=2)
		body["componentSpec"] = [_component(i) for i in range(101)]
		response = api_client.post(f"{EAM_BASE}/apps", json=body)
		assert response.status_code == 400
		assert response.json()["code"] == "INVALID_ARGUMENT"
		assert fake_srm.catalog == {}

		def test_more_than_100_network_interfaces_is_rejected(self, api_client: TestClient) -> None:
		"""componentSpec[].networkInterfaces has maxItems 100 in the CAMARA EAM spec."""
		body = _manifest_body(num_cpu=2)
		component = _component(0)
		component["networkInterfaces"] = [
		{**component["networkInterfaces"][0], "interfaceId": f"if_{i:03d}"} for i in range(101)
		]
		body["componentSpec"] = [component]
		response = api_client.post(f"{EAM_BASE}/apps", json=body)
		assert response.status_code == 400
		assert response.json()["code"] == "INVALID_ARGUMENT"

		def test_qcow2_package_type_is_rejected_as_not_implemented(
		self, api_client: TestClient, fake_srm: FakeSRMClient
		) -> None: