fix: relax SRM catalago GET response fileds to optional (3e81fd1b) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/application/mappers/edge_application_mapper.py

+2 −0

Original line number	Diff line number	Diff line
		@@ -135,6 +135,8 @@ def build_edge_cloud_zone(srm_zone: ResourceZone) -> 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")

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

src/open_exposure_gateway/domain/edge_application_management.py

+6 −4

Original line number	Diff line number	Diff line
		@@ -150,7 +150,7 @@ class SRMComputeIntent(BaseModel):


		class SRMServiceSpecDescriptor(BaseModel):
		artifact_type: str
		artifact_type: str \| None = None
		source_api: str = "edge-application-management"


		@@ -180,16 +180,18 @@ class SRMDeploymentUnit(BaseModel):
		ref: str
		name: str
		runtime_kind: str
		artifact_ref: str
		artifact_ref: str \| None = None
		metadata: SRMDeploymentUnitMetadata \| None = None
		resource_requirements: SRMComputeIntent


		class SRMCapabilityRequirement(BaseModel):
		ref: str
		deployment_unit_ref: str
		# Not present on SRM's GET response (ServiceCapabilityRequirementResponseSchema
		# drops it); only meaningful on the POST request we build ourselves.
		deployment_unit_ref: str \| None = None
		capability_kind: str
		domain_kind: str
		domain_kind: str \| None = None
		is_required: bool = True

tests/unit/test_eam_contract.py

+59 −0

Original line number	Diff line number	Diff line
		@@ -281,3 +281,62 @@ class TestInternalHttpPaths:
		await client.get_resource_zones()

		assert calls == [("GET", "/internal/zones")]

		async def test_get_app_parses_srm_catalog_read_shape(self) -> None:
		"""SRM's GET /internal/catalog/service-specifications/{id} response
		(ServiceCapabilityRequirementResponseSchema) never carries
		deployment_unit_ref or domain_kind, and artifact_ref is nullable — parsing
		must not require fields SRM's own response schema doesn't send."""
		client = SRMClient.__new__(SRMClient)
		client.base_url = "http://srm:8081"
		client.timeout = 1.0

		srm_response = {
		"service_specification": {
		"id": str(APP_ID),
		"app_provider_id": "VideoAppsCo",
		"ref": str(APP_ID),
		"name": "myvideoapp",
		"version": "1.0.0",
		"descriptor": {},
		"metadata": {},
		},
		"service_deployment_units": [
		{
		"ref": "main-runtime",
		"name": "Main Runtime",
		"runtime_kind": "helm",
		"artifact_ref": None,
		"resource_requirements": {},
		"parameters_schema": {},
		"metadata": {},
		}
		],
		"service_capability_requirements": [
		{
		"ref": "require-workload-deployment",
		"capability_kind": "deploy_workload",
		"domain_kind": None,
		"is_required": True,
		"selector": {},
		"policy": {},
		"metadata": {},
		}
		],
		}

		async def record(
		method: str,
		path: str,
		json: dict[str, Any] \| None = None,
		params: dict[str, Any] \| None = None,
		headers: dict[str, str] \| None = None,
		) -> Any:
		return srm_response

		client._request = record # type: ignore[method-assign]
		catalog = await client.get_app(APP_ID)

		assert catalog.service_deployment_units[0].artifact_ref is None
		assert catalog.service_capability_requirements[0].deployment_unit_ref is None
		assert catalog.service_capability_requirements[0].domain_kind is None

Admin message