Loading src/open_exposure_gateway/application/mappers/quality_on_demand_mapper.py +0 −7 Original line number Diff line number Diff line Loading @@ -98,13 +98,6 @@ def build_session_info(qod_session: QodSession, capability: SRMNetworkCapability ipv4_address = None if device_target.ipv4: # Sessions created before ipv4_port/ipv4_private were sent still hold only the bare # address, so the reconstruction can still fall short of DeviceIpv4Addr's "publicAddress # + one of private/port" rule. model_construct bypasses that validator, which exists to # constrain client input, not this internal reconstruction. Device.model_construct is # required alongside it: pydantic revalidates nested model fields against their own # validators even when a pre-built instance is passed in, so the bypass has to hold at # every level enclosing a potentially incomplete address. ipv4_address = DeviceIpv4Addr.model_construct( publicAddress=device_target.ipv4, privateAddress=device_target.ipv4_private, Loading src/open_exposure_gateway/application/services/location_retrieval_service.py +0 −8 Original line number Diff line number Diff line Loading @@ -39,20 +39,12 @@ class LocationRetrievalService: app_provider_id: str, x_correlator: Optional[str] = None, ) -> Location: # CAMARA makes `device` optional only because a three-legged token identifies the # subject on its own. OEG has no JWT subject extraction yet (see # dependencies.get_caller_context), so every call is effectively two-legged and the # device must be supplied. UNNECESSARY_IDENTIFIER -- the mirror error, for a device # sent alongside a three-legged token -- is therefore unreachable until that lands. if request.device is None: raise UnprocessableEntityException( error_code=ErrorCode.MISSING_IDENTIFIER, message="The device cannot be identified.", ) # One value for both surfaces. The router always supplies the caller context's # correlator, but a direct call with None must not put a generated id in the body and # nothing in the header -- that is the divergence this fallback exists to close. correlation_id = x_correlator or str(uuid4()) query = build_location_query( request=request, Loading src/open_exposure_gateway/dependencies.py +0 −3 Original line number Diff line number Diff line Loading @@ -61,9 +61,6 @@ def get_caller_context( request: Request, x_correlator: XCorrelatorHeader = None, ) -> CallerContext: # `x_correlator` stays declared so FastAPI still validates the header against CAMARA's # pattern and documents it; the value comes from x_correlator_header, which generates and # caches one when the caller sent none, so every surface reports the same id. return CallerContext( x_correlator=x_correlator_header(request), tenant_id="placeholder", # TODO: extract from JWT Loading tests/unit/test_eam_endpoints.py +0 −2 Original line number Diff line number Diff line Loading @@ -95,8 +95,6 @@ class TestGetEdgeCloudZones: self, client: TestClient, mock_eam_service: AsyncMock ) -> None: client.get(f"{EAM_BASE}/edge-cloud-zones?region=eu-west&status=active") # x_correlator is never None: when the caller sends no header OEG generates one at the # edge and uses that single value downstream and in the response (REQ-OEG-13). kwargs = mock_eam_service.get_edge_cloud_zones.call_args.kwargs assert kwargs["region"] == "eu-west" assert kwargs["status"] == "active" Loading tests/unit/test_qod_flows.py +0 −5 Original line number Diff line number Diff line Loading @@ -52,9 +52,6 @@ class TestQodSessionFlow: def test_generated_correlation_id_is_the_one_the_caller_is_given( self, api_client: TestClient, fake_bus: FakeDataBus ) -> None: """Same invariant as EAM and Location Retrieval: when the caller sends no x-correlator, the id OEG generates must reach SRM and come back to the caller as one value (oeg/architecture.md REQ-OEG-13).""" response = api_client.post(f"{QOD_BASE}/sessions", json=SESSION_BODY) _, command = fake_bus.published[0] Loading @@ -78,8 +75,6 @@ class TestQodSessionFlow: assert command.network_capability.parameters.duration_seconds == 3600 assert command.network_capability.target.application_server.ipv4 == "192.168.0.1" assert command.network_capability.target.device.ipv4 == "84.125.93.10" # The NAT'd public address alone names thousands of subscribers; the port is what # picks out this device, so dropping it would leave SRM unable to resolve the target. assert command.network_capability.target.device.ipv4_port == 59765 assert command.source == "nbi_camara" Loading Loading
src/open_exposure_gateway/application/mappers/quality_on_demand_mapper.py +0 −7 Original line number Diff line number Diff line Loading @@ -98,13 +98,6 @@ def build_session_info(qod_session: QodSession, capability: SRMNetworkCapability ipv4_address = None if device_target.ipv4: # Sessions created before ipv4_port/ipv4_private were sent still hold only the bare # address, so the reconstruction can still fall short of DeviceIpv4Addr's "publicAddress # + one of private/port" rule. model_construct bypasses that validator, which exists to # constrain client input, not this internal reconstruction. Device.model_construct is # required alongside it: pydantic revalidates nested model fields against their own # validators even when a pre-built instance is passed in, so the bypass has to hold at # every level enclosing a potentially incomplete address. ipv4_address = DeviceIpv4Addr.model_construct( publicAddress=device_target.ipv4, privateAddress=device_target.ipv4_private, Loading
src/open_exposure_gateway/application/services/location_retrieval_service.py +0 −8 Original line number Diff line number Diff line Loading @@ -39,20 +39,12 @@ class LocationRetrievalService: app_provider_id: str, x_correlator: Optional[str] = None, ) -> Location: # CAMARA makes `device` optional only because a three-legged token identifies the # subject on its own. OEG has no JWT subject extraction yet (see # dependencies.get_caller_context), so every call is effectively two-legged and the # device must be supplied. UNNECESSARY_IDENTIFIER -- the mirror error, for a device # sent alongside a three-legged token -- is therefore unreachable until that lands. if request.device is None: raise UnprocessableEntityException( error_code=ErrorCode.MISSING_IDENTIFIER, message="The device cannot be identified.", ) # One value for both surfaces. The router always supplies the caller context's # correlator, but a direct call with None must not put a generated id in the body and # nothing in the header -- that is the divergence this fallback exists to close. correlation_id = x_correlator or str(uuid4()) query = build_location_query( request=request, Loading
src/open_exposure_gateway/dependencies.py +0 −3 Original line number Diff line number Diff line Loading @@ -61,9 +61,6 @@ def get_caller_context( request: Request, x_correlator: XCorrelatorHeader = None, ) -> CallerContext: # `x_correlator` stays declared so FastAPI still validates the header against CAMARA's # pattern and documents it; the value comes from x_correlator_header, which generates and # caches one when the caller sent none, so every surface reports the same id. return CallerContext( x_correlator=x_correlator_header(request), tenant_id="placeholder", # TODO: extract from JWT Loading
tests/unit/test_eam_endpoints.py +0 −2 Original line number Diff line number Diff line Loading @@ -95,8 +95,6 @@ class TestGetEdgeCloudZones: self, client: TestClient, mock_eam_service: AsyncMock ) -> None: client.get(f"{EAM_BASE}/edge-cloud-zones?region=eu-west&status=active") # x_correlator is never None: when the caller sends no header OEG generates one at the # edge and uses that single value downstream and in the response (REQ-OEG-13). kwargs = mock_eam_service.get_edge_cloud_zones.call_args.kwargs assert kwargs["region"] == "eu-west" assert kwargs["status"] == "active" Loading
tests/unit/test_qod_flows.py +0 −5 Original line number Diff line number Diff line Loading @@ -52,9 +52,6 @@ class TestQodSessionFlow: def test_generated_correlation_id_is_the_one_the_caller_is_given( self, api_client: TestClient, fake_bus: FakeDataBus ) -> None: """Same invariant as EAM and Location Retrieval: when the caller sends no x-correlator, the id OEG generates must reach SRM and come back to the caller as one value (oeg/architecture.md REQ-OEG-13).""" response = api_client.post(f"{QOD_BASE}/sessions", json=SESSION_BODY) _, command = fake_bus.published[0] Loading @@ -78,8 +75,6 @@ class TestQodSessionFlow: assert command.network_capability.parameters.duration_seconds == 3600 assert command.network_capability.target.application_server.ipv4 == "192.168.0.1" assert command.network_capability.target.device.ipv4 == "84.125.93.10" # The NAT'd public address alone names thousands of subscribers; the port is what # picks out this device, so dropping it would leave SRM unable to resolve the target. assert command.network_capability.target.device.ipv4_port == 59765 assert command.source == "nbi_camara" Loading