refactor: enforce strict model configuration in schemas and improve error... (b1caa8aa) · Commits · OOP / code / open-exposure-gateway

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

+35 −1

Original line number	Diff line number	Diff line
		@@ -3,7 +3,7 @@ from enum import StrEnum
		from typing import Any, Literal, Optional
		from uuid import UUID

		from pydantic import BaseModel, Field
		from pydantic import BaseModel, ConfigDict, Field


		class AppInstanceStatus(StrEnum):
		@@ -33,6 +33,8 @@ class SubmittedApp(BaseModel):


		class AppRepo(BaseModel):
		model_config = ConfigDict(extra="forbid")

		type: Literal["PRIVATEREPO", "PUBLICREPO"]
		imagePath: str = Field(max_length=2048)
		userName: Optional[str] = Field(default=None, max_length=64)
		@@ -42,6 +44,8 @@ class AppRepo(BaseModel):


		class OperatingSystem(BaseModel):
		model_config = ConfigDict(extra="forbid")

		architecture: Literal["x86_64", "x86"]
		family: Literal["RHEL", "UBUNTU", "COREOS", "WINDOWS", "OTHER"]
		version: Literal[
		@@ -54,6 +58,8 @@ class OperatingSystem(BaseModel):


		class NetworkInterface(BaseModel):
		model_config = ConfigDict(extra="forbid")

		interfaceId: str = Field(
		min_length=4,
		max_length=32,
		@@ -65,41 +71,55 @@ class NetworkInterface(BaseModel):


		class ComponentSpecItem(BaseModel):
		model_config = ConfigDict(extra="forbid")

		componentName: str = Field(max_length=64)
		networkInterfaces: list[NetworkInterface]


		class VmResources(BaseModel):
		model_config = ConfigDict(extra="forbid")

		infraKind: Literal["virtualMachine"]
		numCPU: int = Field(ge=1, le=256)
		memory: int = Field(ge=1, le=32768)


		class ContainerResources(BaseModel):
		model_config = ConfigDict(extra="forbid")

		infraKind: Literal["container"]
		numCPU: str = Field(pattern=r"^\d+((\.\d{1,3})\|(m))?$")
		memory: int = Field(ge=1, le=16384)


		class DockerComposeResources(BaseModel):
		model_config = ConfigDict(extra="forbid")

		infraKind: Literal["dockerCompose"]
		numCPU: int = Field(ge=1, le=256)
		memory: int = Field(ge=1, le=16384)


		class CpuPoolTopology(BaseModel):
		model_config = ConfigDict(extra="forbid")

		minNumberOfNodes: int = Field(ge=1, le=1000)
		minNodeCpu: int = Field(ge=1, le=256)
		minNodeMemory: int = Field(ge=1, le=16384)


		class CpuPool(BaseModel):
		model_config = ConfigDict(extra="forbid")

		numCPU: int = Field(ge=1, le=256)
		memory: int = Field(ge=1, le=16384)
		topology: CpuPoolTopology


		class GpuPoolTopology(BaseModel):
		model_config = ConfigDict(extra="forbid")

		minNumberOfNodes: int = Field(ge=1, le=1000)
		minNodeCpu: int = Field(ge=1, le=256)
		minNodeMemory: int = Field(ge=1, le=16384)
		@@ -107,6 +127,8 @@ class GpuPoolTopology(BaseModel):


		class GpuPool(BaseModel):
		model_config = ConfigDict(extra="forbid")

		numCPU: int = Field(ge=1, le=1024)
		memory: int = Field(ge=1, le=16384)
		gpuMemory: int = Field(ge=1, le=16)
		@@ -114,11 +136,15 @@ class GpuPool(BaseModel):


		class ApplicationResources(BaseModel):
		model_config = ConfigDict(extra="forbid")

		cpuPool: Optional[CpuPool] = None
		gpuPool: Optional[GpuPool] = None


		class KubernetesResources(BaseModel):
		model_config = ConfigDict(extra="forbid")

		infraKind: Literal["kubernetes"]
		applicationResources: ApplicationResources
		isStandalone: bool
		@@ -129,6 +155,8 @@ RequiredResources = VmResources \| ContainerResources \| DockerComposeResources \|


		class AppManifest(BaseModel):
		model_config = ConfigDict(extra="forbid")

		appId: Optional[UUID] = None
		name: str = Field(max_length=64, pattern=r"^[A-Za-z][A-Za-z0-9_]{1,63}$")
		appProvider: str = Field(max_length=64, pattern=r"^[A-Za-z][A-Za-z0-9_]{7,63}$")
		@@ -168,6 +196,8 @@ class AppInstanceInfo(BaseModel):


		class SubscriptionConfig(BaseModel):
		model_config = ConfigDict(extra="forbid")

		subscriptionDetail: Optional[dict[str, Any]] = None
		subscriptionExpireTime: Optional[datetime] = None
		subscriptionMaxEvents: Optional[int] = None
		@@ -175,6 +205,8 @@ class SubscriptionConfig(BaseModel):


		class SubscriptionRequest(BaseModel):
		model_config = ConfigDict(extra="forbid")

		sink: str
		sinkCredential: Optional[dict[str, Any]] = None
		types: list[str]
		@@ -182,6 +214,8 @@ class SubscriptionRequest(BaseModel):


		class CreateAppInstanceRequest(BaseModel):
		model_config = ConfigDict(extra="forbid")

		name: str = Field(max_length=64, pattern=r"^[A-Za-z][A-Za-z0-9_]{1,63}$")
		appId: UUID
		edgeCloudZoneId: UUID

src/open_exposure_gateway/application/mappers/edge_application_mapper.py

+8 −3

Original line number	Diff line number	Diff line
		@@ -149,9 +149,14 @@ def build_edge_cloud_zone(srm_zone: SRMZone) -> EdgeCloudZone:

		def build_app_manifest(catalog: SRMCatalogPayload) -> AppManifest:
		spec = catalog.service_specification
		unit = catalog.service_deployment_units[0]
		if unit.artifact_ref is None:
		raise ValueError(f"deployment unit {unit.ref!r} has no artifact_ref")
		unit = next(
		(u for u in catalog.service_deployment_units if u.artifact_ref is not None),
		None,
		)
		if unit is None:
		raise ValueError(
		f"no deployment unit in catalog entry {spec.ref!r} has an artifact_ref"
		)

		try:
		app_id: Optional[UUID] = UUID(spec.ref)

src/open_exposure_gateway/application/services/edge_application_management_service.py

+14 −26

Original line number	Diff line number	Diff line
		@@ -38,9 +38,7 @@ from open_exposure_gateway.core.exceptions import (
		NotImplementedException,
		)
		from open_exposure_gateway.domain.edge_application_management import (
		SRMCatalogPayload,
		SRMOperationCompleted,
		SRMZone,
		Subject,
		)
		from open_exposure_gateway.domain.models import (
		@@ -140,13 +138,13 @@ class EdgeApplicationManagementService:
		status=status,
		x_correlator=x_correlator,
		)
		zones = []
		for zone in srm_zones:
		try:
		zones.append(build_edge_cloud_zone(zone))
		return [build_edge_cloud_zone(zone) for zone in srm_zones]
		except (ValueError, TypeError) as exc:
		self._log_skipped_entry("zone", zone, exc)
		return zones
		raise DownstreamServiceException(
		message="SRM returned a malformed edge cloud zone",
		details=str(exc),
		) from exc

		async def get_apps(self, x_correlator: Optional[str] = None) -> list[AppManifest]:
		# TODO: scope by caller.tenant_id/app_provider_id once JWT auth is wired in
		@@ -155,23 +153,13 @@ class EdgeApplicationManagementService:
		# tenant_id isn't in SRM's catalog response, so filtering must happen via
		# app_registration_repo, not srm_client.get_apps.
		catalogs = await self.srm_client.get_apps(x_correlator=x_correlator)
		manifests = []
		for catalog in catalogs:
		try:
		manifests.append(build_app_manifest(catalog))
		return [build_app_manifest(catalog) for catalog in catalogs]
		except (ValueError, TypeError, IndexError) as exc:
		self._log_skipped_entry("catalog entry", catalog, exc)
		return manifests

		def _log_skipped_entry(
		self, kind: str, entry: SRMZone \| SRMCatalogPayload, exc: Exception
		) -> None:
		logger.warning(
		"unmappable_srm_entry_skipped",
		kind=kind,
		error=str(exc),
		entry=entry.model_dump(mode="json"),
		)
		raise DownstreamServiceException(
		message="SRM returned a malformed service specification",
		details=str(exc),
		) from exc

		async def get_app(
		self, app_id: UUID, x_correlator: Optional[str] = None

tests/unit/test_eam_flows.py

+15 −11

Original line number	Diff line number	Diff line
		@@ -484,11 +484,13 @@ class TestGetAppsFlow:
		assert response.status_code == 200
		assert len(response.json()) == 1

		def test_entry_without_deployment_units_does_not_break_listing(
		def test_entry_without_deployment_units_fails_the_whole_listing(
		self, api_client: TestClient, fake_srm: FakeSRMClient
		) -> None:
		"""One malformed catalog entry must not take down the whole listing:
		the healthy entries must still be returned."""
		"""CAMARA's 200 for getApps has no partial-success shape (bare array,
		no metadata slot) and its description promises the complete list, so
		a malformed catalog entry must surface as a downstream failure (503)
		rather than silently shrinking the response."""
		api_client.post(f"{EAM_BASE}/apps", json=_manifest_body(num_cpu=2))
		broken_id = str(uuid4())
		fake_srm.catalog[broken_id] = {
		@@ -505,8 +507,8 @@ class TestGetAppsFlow:
		}

		response = api_client.get(f"{EAM_BASE}/apps")
		assert response.status_code == 200
		assert "myvideoapp" in [m["name"] for m in response.json()]
		assert response.status_code == 503
		assert response.json()["code"] == "UNAVAILABLE"


		class TestGetAppFlow:
		@@ -557,12 +559,14 @@ class TestEdgeCloudZonesFlow:
		assert zone["edgeCloudZoneName"] == "berlin-edge-1"
		assert zone["edgeCloudZoneStatus"] == "active"

		def test_one_malformed_zone_id_does_not_break_listing(
		def test_one_malformed_zone_id_fails_the_whole_listing(
		self, api_client: TestClient, fake_srm: FakeSRMClient
		) -> None:
		"""SRM's SRMZone model allows free-form string ids; one non-UUID id
		must not turn the whole zone listing into a 500 — healthy zones must
		still be returned."""
		"""CAMARA's 200 for getEdgeCloudZones has no partial-success shape
		(bare array, no metadata slot) and promises the Available Edge Cloud
		Zones, so SRM's free-form string ids producing a non-UUID id must
		surface as a downstream failure (503) rather than silently dropping
		the zone from the list."""
		fake_srm.zones.append(
		SRMZone(
		id=str(ZONE_ID),
		@@ -580,8 +584,8 @@ class TestEdgeCloudZonesFlow:
		)
		)
		response = api_client.get(f"{EAM_BASE}/edge-cloud-zones")
		assert response.status_code == 200
		assert str(ZONE_ID) in [z["edgeCloudZoneId"] for z in response.json()]
		assert response.status_code == 503
		assert response.json()["code"] == "UNAVAILABLE"

		def test_x_correlator_is_echoed_on_success_responses(
		self, api_client: TestClient, fake_srm: FakeSRMClient

tests/unit/test_eam_mapper.py

+25 −0

Original line number	Diff line number	Diff line
		@@ -197,6 +197,31 @@ class TestBuildEdgeCloudZone:


		class TestBuildAppManifest:
		def test_skips_leading_units_without_artifact_ref(self) -> None:
		"""A catalog entry's artifact-bearing unit isn't always index 0;
		earlier units (e.g. init containers, sidecars) may have no
		artifact_ref of their own."""
		catalog = _make_srm_catalog()
		sidecar = SRMDeploymentUnit(
		ref="sidecar",
		name="Sidecar",
		runtime_kind="helm",
		artifact_ref=None,
		resource_requirements=SRMComputeIntent(),
		)
		catalog.service_deployment_units.insert(0, sidecar)

		result = build_app_manifest(catalog)
		assert result.appRepo.imagePath == "oci://registry.example.com/charts/app:1.0"

		def test_raises_when_no_unit_has_artifact_ref(self) -> None:
		catalog = _make_srm_catalog()
		for unit in catalog.service_deployment_units:
		unit.artifact_ref = None

		with pytest.raises(ValueError, match="artifact_ref"):
		build_app_manifest(catalog)

		def test_helm_maps_app_id_and_metadata(self) -> None:
		catalog = _make_srm_catalog()
		result = build_app_manifest(catalog)

Admin message