Merge branch 'feature/location-retrieval-oai' into 'main' (df9e272b) · Commits · OOP / code / service-resource-manager

service-resource-manager-implementation/pytest.ini

0 → 100644

+5 −0

Original line number	Diff line number	Diff line
		[pytest]
		testpaths = tests
		pythonpath = .
		log_cli = true
		log_cli_level = WARNING

service-resource-manager-implementation/src/controllers/network_functions_controller.py

+37 −3

Original line number	Diff line number	Diff line
		@@ -2,6 +2,7 @@ from os import environ
		import logging
		import connexion
		from sunrise6g_opensdk.common.sdk import Sdk as sdkclient
		from sunrise6g_opensdk.network.core.schemas import RetrievalLocationRequest

		logger = logging.getLogger(__name__)
		logging.basicConfig(level=logging.DEBUG)
		@@ -11,9 +12,12 @@ if environ.get('NETWORK_ADAPTER_NAME') and environ.get('NETWORK_ADAPTER_BASE_URL
		network_adapter_name = environ.get('NETWORK_ADAPTER_NAME')
		adapter_base_url = environ.get('NETWORK_ADAPTER_BASE_URL')
		scs_as_id = environ.get('SCS_AS_ID')
		network_adapter_specs = {'client_name': network_adapter_name, 'base_url': adapter_base_url, 'scs_as_id': scs_as_id}
		# network_adapter_specs.update(environ)
		print('Creating network adapter with env: ', network_adapter_specs)
		network_adapter_specs = {
		'client_name': network_adapter_name,
		'base_url': adapter_base_url,
		'scs_as_id': scs_as_id,
		}
		print('Creating network adapter with specs: ', network_adapter_specs)
		adapters = sdkclient.create_adapters_from(adapter_specs={'network': network_adapter_specs})
		network_adapter = adapters.get("network")

		@@ -26,6 +30,7 @@ def _safe_http_json_response(response):
		"""
		Normalize adapter responses:
		- list / dict → return directly (200)
		- Pydantic model → serialize via model_dump (200)
		- requests.Response → parse JSON body and status code safely
		"""
		if isinstance(response, (list, dict)):
		@@ -34,6 +39,9 @@ def _safe_http_json_response(response):
		if response is None:
		return {"error": "Adapter returned no response"}, 502

		if hasattr(response, 'model_dump'):
		return response.model_dump(mode='json', exclude_none=True), 200

		try:
		return response.json(), response.status_code
		except Exception as e:
		@@ -152,3 +160,29 @@ def get_all_traffic_influence_resources():
		except Exception as ce_:
		logger.error(ce_)
		return {"error": str(ce_)}, 500


		def retrieve_location():
		"""Retrieve the location of a device via the CAMARA Location Retrieval API."""

		if connexion.request.is_json:
		try:
		if network_adapter is not None:
		body = connexion.request.get_json()
		retrieve_location_request = RetrievalLocationRequest.model_validate(body)
		response = network_adapter.create_monitoring_event_subscription(retrieve_location_request)
		return _safe_http_json_response(response)
		else:
		return {
		"lastLocationTime": "2024-01-01T00:00:00Z",
		"area": {
		"areaType": "CIRCLE",
		"center": {"latitude": 0.0, "longitude": 0.0},
		"radius": 50000
		}
		}, 200
		except Exception as ce_:
		logger.error(ce_)
		return {"error": str(ce_)}, 500
		else:
		return {"error": "Could not read JSON payload."}, 400

service-resource-manager-implementation/src/swagger/swagger.yaml

+113 −0

Original line number	Diff line number	Diff line
		@@ -431,6 +431,32 @@ paths:
		"404":
		description: Session not found
		x-openapi-router-controller: src.controllers.network_functions_controller
		/location/retrieve:
		post:
		tags:
		- Location Retrieval Functions
		summary: Retrieve the location of a device
		operationId: retrieve_location
		requestBody:
		description: Location retrieval request following CAMARA Device Location API.
		content:
		application/json:
		schema:
		$ref: '#/components/schemas/RetrievalLocationRequest'
		responses:
		"200":
		description: Device location retrieved successfully
		content:
		application/json:
		schema:
		$ref: '#/components/schemas/LocationResponse'
		"400":
		description: Invalid request
		"404":
		description: Device not found
		"500":
		description: Internal server error
		x-openapi-router-controller: src.controllers.network_functions_controller
		components:
		schemas:
		serviceFunctionNodeMigration:
		@@ -1249,6 +1275,93 @@ components:
		type: string
		example: Test

		RetrievalLocationRequest:
		type: object
		properties:
		device:
		$ref: '#/components/schemas/Device'
		maxAge:
		type: integer
		description: Maximum age of the location information accepted (in seconds).
		example: 60
		maxSurface:
		type: integer
		description: Maximum surface in square meters accepted for the location retrieval.
		minimum: 1
		example: 10000
		Device:
		type: object
		properties:
		phoneNumber:
		type: string
		description: Phone number of the device in E.164 format.
		example: "+123456789"
		networkAccessIdentifier:
		type: string
		description: External identifier of the device (e.g. GPSI).
		example: "device@testnet.net"
		ipv4Address:
		$ref: '#/components/schemas/DeviceIpv4Addr'
		ipv6Address:
		type: string
		description: IPv6 address of the device.
		example: "2001:db8::1"
		DeviceIpv4Addr:
		type: object
		properties:
		publicAddress:
		type: string
		example: "198.51.100.1"
		privateAddress:
		type: string
		example: "10.0.0.1"
		publicPort:
		type: integer
		example: 59765
		LocationResponse:
		type: object
		properties:
		lastLocationTime:
		type: string
		format: date-time
		description: Timestamp of the last known location.
		example: "2024-06-01T12:00:00Z"
		area:
		$ref: '#/components/schemas/LocationArea'
		LocationArea:
		type: object
		description: Geographic area of the location (Circle or Polygon).
		properties:
		areaType:
		type: string
		enum:
		- CIRCLE
		- POLYGON
		example: CIRCLE
		center:
		$ref: '#/components/schemas/GeoPoint'
		radius:
		type: number
		description: Radius in meters (when areaType is CIRCLE).
		example: 800
		boundary:
		type: array
		description: List of points forming the polygon boundary (when areaType is POLYGON).
		items:
		$ref: '#/components/schemas/GeoPoint'
		GeoPoint:
		type: object
		properties:
		latitude:
		type: number
		minimum: -90
		maximum: 90
		example: 37.9553
		longitude:
		type: number
		minimum: -180
		maximum: 180
		example: 23.8522
		securitySchemes:
		registry_auth:
		type: oauth2

service-resource-manager-implementation/tests/conftest.py

0 → 100644

+137 −0

Original line number	Diff line number	Diff line
		"""
		Shared fixtures for SRM controller unit tests.

		Sets up:
		- A Flask test app (needed for connexion.request context)
		- A mock network adapter pre-wired with a standard Location response
		- Helper to POST JSON to the retrieve_location controller
		"""

		import json
		import sys
		import os
		from datetime import datetime, timezone
		from unittest.mock import MagicMock, patch

		import connexion
		import pytest
		from flask import Flask

		sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

		from sunrise6g_opensdk.network.core.schemas import (
		AreaType,
		Circle,
		Location,
		Point,
		)


		# ---------------------------------------------------------------------------
		# Constants reused across tests
		# ---------------------------------------------------------------------------

		MOCK_LATITUDE = 48.8566
		MOCK_LONGITUDE = 2.3522
		MOCK_RADIUS = 150.0
		MOCK_TIMESTAMP = datetime(2026, 2, 10, 12, 0, 0, tzinfo=timezone.utc)

		PAYLOAD_BY_NAI = {
		"device": {"networkAccessIdentifier": "imsi-001010000000001"},
		"maxAge": 60,
		}

		PAYLOAD_BY_IP = {
		"device": {
		"ipv4Address": {
		"publicAddress": "12.1.0.1",
		"privateAddress": "12.1.0.1",
		}
		}
		}

		PAYLOAD_BY_PHONE = {
		"device": {"phoneNumber": "+10010100001"}
		}


		# ---------------------------------------------------------------------------
		# Fixtures
		# ---------------------------------------------------------------------------

		@pytest.fixture()
		def flask_app():
		"""Minimal Flask app for request context."""
		return Flask(__name__)


		@pytest.fixture()
		def mock_location():
		"""A CAMARA Location object with arbitrary coordinates for testing."""
		return Location(
		lastLocationTime=MOCK_TIMESTAMP,
		area=Circle(
		areaType=AreaType.circle,
		center=Point(latitude=MOCK_LATITUDE, longitude=MOCK_LONGITUDE),
		radius=MOCK_RADIUS,
		),
		)


		@pytest.fixture()
		def mock_adapter(mock_location):
		"""A mock network adapter that returns a fixed Location by default."""
		adapter = MagicMock()
		adapter.create_monitoring_event_subscription.return_value = mock_location
		return adapter


		@pytest.fixture(autouse=True)
		def reset_network_adapter():
		"""
		Reset the module-level network_adapter to None before each test so tests
		are fully isolated from each other and from any real env vars.
		"""
		import src.controllers.network_functions_controller as ctrl
		original = ctrl.network_adapter
		ctrl.network_adapter = None
		yield
		ctrl.network_adapter = original


		def call_retrieve_location(flask_app, payload):
		"""
		Helper: call retrieve_location() inside a proper Flask+connexion request
		context with the given JSON payload.
		Returns the controller's return value. For success (body, 200) returns just
		the body for convenience; for errors returns the full (body, status) tuple.
		"""
		import src.controllers.network_functions_controller as ctrl

		with flask_app.test_request_context(
		"/location/retrieve",
		method="POST",
		content_type="application/json",
		data=json.dumps(payload),
		):
		from flask import request as flask_request
		with patch.object(connexion, "request", flask_request):
		ret = ctrl.retrieve_location()
		if isinstance(ret, tuple) and len(ret) == 2 and isinstance(ret[1], int) and 200 <= ret[1] < 300:
		return ret[0]
		return ret


		def call_retrieve_location_plain_text(flask_app):
		"""Helper: call retrieve_location() with non-JSON content."""
		import src.controllers.network_functions_controller as ctrl

		with flask_app.test_request_context(
		"/location/retrieve",
		method="POST",
		content_type="text/plain",
		data="not json",
		):
		from flask import request as flask_request
		with patch.object(connexion, "request", flask_request):
		return ctrl.retrieve_location()

service-resource-manager-implementation/tests/test_retrieve_location.py

0 → 100644

+218 −0

Original line number	Diff line number	Diff line
		"""
		Unit tests for the retrieve_location controller.

		Structure:
		TestStubMode – no adapter configured (NETWORK_ADAPTER_NAME not set)
		TestWithAdapter – adapter is configured, SDK returns a Location object
		TestInputValidation – bad / edge-case request bodies
		TestErrorHandling – adapter raises exceptions
		"""

		import pytest
		from conftest import (
		MOCK_LATITUDE,
		MOCK_LONGITUDE,
		MOCK_RADIUS,
		PAYLOAD_BY_IP,
		PAYLOAD_BY_NAI,
		PAYLOAD_BY_PHONE,
		call_retrieve_location,
		call_retrieve_location_plain_text,
		)


		# ---------------------------------------------------------------------------
		# Stub mode (no adapter)
		# ---------------------------------------------------------------------------

		class TestStubMode:
		"""When NETWORK_ADAPTER_NAME is not set the controller returns a static stub."""

		def test_returns_200_shape(self, flask_app):
		result = call_retrieve_location(flask_app, PAYLOAD_BY_NAI)
		assert isinstance(result, dict)
		assert "lastLocationTime" in result
		assert "area" in result

		def test_stub_area_is_circle(self, flask_app):
		result = call_retrieve_location(flask_app, PAYLOAD_BY_NAI)
		assert result["area"]["areaType"] == "CIRCLE"

		def test_stub_has_center_and_radius(self, flask_app):
		area = call_retrieve_location(flask_app, PAYLOAD_BY_NAI)["area"]
		assert "center" in area
		assert "latitude" in area["center"]
		assert "longitude" in area["center"]
		assert "radius" in area

		def test_non_json_returns_400(self, flask_app):
		result = call_retrieve_location_plain_text(flask_app)
		assert result == ({"error": "Could not read JSON payload."}, 400)


		# ---------------------------------------------------------------------------
		# With adapter (mocked SDK)
		# ---------------------------------------------------------------------------

		class TestWithAdapter:
		"""When an adapter is set the controller must delegate to it correctly."""

		@pytest.fixture(autouse=True)
		def inject_adapter(self, mock_adapter):
		import src.controllers.network_functions_controller as ctrl
		ctrl.network_adapter = mock_adapter

		# --- Response structure ---

		def test_returns_dict(self, flask_app):
		result = call_retrieve_location(flask_app, PAYLOAD_BY_NAI)
		assert isinstance(result, dict)

		def test_response_has_last_location_time(self, flask_app):
		result = call_retrieve_location(flask_app, PAYLOAD_BY_NAI)
		assert "lastLocationTime" in result

		def test_response_last_location_time_is_iso_string(self, flask_app):
		result = call_retrieve_location(flask_app, PAYLOAD_BY_NAI)
		# model_dump(mode='json') must have converted datetime → ISO string
		assert isinstance(result["lastLocationTime"], str)
		assert "2026-02-10" in result["lastLocationTime"]

		def test_response_has_area(self, flask_app):
		result = call_retrieve_location(flask_app, PAYLOAD_BY_NAI)
		assert "area" in result

		def test_response_area_type_is_circle(self, flask_app):
		result = call_retrieve_location(flask_app, PAYLOAD_BY_NAI)
		assert result["area"]["areaType"] == "CIRCLE"

		def test_response_coordinates(self, flask_app):
		area = call_retrieve_location(flask_app, PAYLOAD_BY_NAI)["area"]
		assert area["center"]["latitude"] == MOCK_LATITUDE
		assert area["center"]["longitude"] == MOCK_LONGITUDE
		assert area["radius"] == MOCK_RADIUS

		def test_no_none_values_in_response(self, flask_app):
		"""model_dump(exclude_none=True) must strip None fields."""
		result = call_retrieve_location(flask_app, PAYLOAD_BY_NAI)
		def has_none(obj):
		if isinstance(obj, dict):
		return None in obj.values() or any(has_none(v) for v in obj.values())
		return False
		assert not has_none(result)

		# --- Adapter is called with the right model ---

		def test_adapter_called_once(self, flask_app, mock_adapter):
		call_retrieve_location(flask_app, PAYLOAD_BY_NAI)
		mock_adapter.create_monitoring_event_subscription.assert_called_once()

		def test_adapter_receives_retrieval_location_request(self, flask_app, mock_adapter):
		from sunrise6g_opensdk.network.core.schemas import RetrievalLocationRequest
		call_retrieve_location(flask_app, PAYLOAD_BY_NAI)
		arg = mock_adapter.create_monitoring_event_subscription.call_args.args[0]
		assert isinstance(arg, RetrievalLocationRequest)

		def test_adapter_receives_correct_nai(self, flask_app, mock_adapter):
		call_retrieve_location(flask_app, PAYLOAD_BY_NAI)
		arg = mock_adapter.create_monitoring_event_subscription.call_args.args[0]
		assert arg.device.networkAccessIdentifier is not None

		def test_adapter_receives_correct_max_age(self, flask_app, mock_adapter):
		call_retrieve_location(flask_app, PAYLOAD_BY_NAI)
		arg = mock_adapter.create_monitoring_event_subscription.call_args.args[0]
		assert arg.maxAge == 60

		# --- Different device identifier types ---

		def test_identify_by_ip(self, flask_app, mock_adapter):
		result = call_retrieve_location(flask_app, PAYLOAD_BY_IP)
		assert "area" in result
		arg = mock_adapter.create_monitoring_event_subscription.call_args.args[0]
		assert arg.device.ipv4Address is not None

		def test_identify_by_phone(self, flask_app, mock_adapter):
		result = call_retrieve_location(flask_app, PAYLOAD_BY_PHONE)
		assert "area" in result
		arg = mock_adapter.create_monitoring_event_subscription.call_args.args[0]
		assert arg.device.phoneNumber is not None

		def test_non_json_still_returns_400_with_adapter(self, flask_app):
		result = call_retrieve_location_plain_text(flask_app)
		assert result == ({"error": "Could not read JSON payload."}, 400)


		# ---------------------------------------------------------------------------
		# Input validation
		# ---------------------------------------------------------------------------

		class TestInputValidation:
		"""Invalid request bodies must be handled gracefully, never crash."""

		@pytest.fixture(autouse=True)
		def inject_adapter(self, mock_adapter):
		import src.controllers.network_functions_controller as ctrl
		ctrl.network_adapter = mock_adapter

		def test_missing_device_field(self, flask_app, mock_adapter):
		"""No device key → Pydantic accepts it (device is Optional) but adapter may reject."""
		# Controller must not crash — it should either return a result or a 500
		result = call_retrieve_location(flask_app, {"maxAge": 30})
		assert isinstance(result, dict) or (isinstance(result, tuple) and result[1] == 500)

		def test_empty_body(self, flask_app, mock_adapter):
		result = call_retrieve_location(flask_app, {})
		assert isinstance(result, dict) or (isinstance(result, tuple) and result[1] == 500)

		def test_max_age_is_passed_when_provided(self, flask_app, mock_adapter):
		call_retrieve_location(flask_app, {"device": {"networkAccessIdentifier": "imsi-x"}, "maxAge": 120})
		arg = mock_adapter.create_monitoring_event_subscription.call_args.args[0]
		assert arg.maxAge == 120

		def test_max_age_is_none_when_omitted(self, flask_app, mock_adapter):
		call_retrieve_location(flask_app, {"device": {"networkAccessIdentifier": "imsi-x"}})
		arg = mock_adapter.create_monitoring_event_subscription.call_args.args[0]
		assert arg.maxAge is None

		def test_max_surface_is_passed_when_provided(self, flask_app, mock_adapter):
		payload = {"device": {"networkAccessIdentifier": "imsi-x"}, "maxSurface": 5000}
		call_retrieve_location(flask_app, payload)
		arg = mock_adapter.create_monitoring_event_subscription.call_args.args[0]
		assert arg.maxSurface == 5000


		# ---------------------------------------------------------------------------
		# Error handling
		# ---------------------------------------------------------------------------

		class TestErrorHandling:
		"""Adapter errors must be caught and returned as HTTP 500, never re-raised."""

		@pytest.fixture(autouse=True)
		def inject_adapter(self, mock_adapter):
		import src.controllers.network_functions_controller as ctrl
		ctrl.network_adapter = mock_adapter

		def test_adapter_generic_exception_returns_500(self, flask_app, mock_adapter):
		mock_adapter.create_monitoring_event_subscription.side_effect = Exception("NEF down")
		body, status = call_retrieve_location(flask_app, PAYLOAD_BY_NAI)
		assert status == 500

		def test_adapter_exception_message_in_body(self, flask_app, mock_adapter):
		mock_adapter.create_monitoring_event_subscription.side_effect = Exception("connection refused")
		body, status = call_retrieve_location(flask_app, PAYLOAD_BY_NAI)
		assert body.get("error") == "connection refused"

		def test_adapter_value_error_returns_500(self, flask_app, mock_adapter):
		mock_adapter.create_monitoring_event_subscription.side_effect = ValueError("bad response from NEF")
		body, status = call_retrieve_location(flask_app, PAYLOAD_BY_NAI)
		assert status == 500

		def test_controller_does_not_reraise(self, flask_app, mock_adapter):
		"""The controller must never let an exception bubble up to the caller."""
		mock_adapter.create_monitoring_event_subscription.side_effect = RuntimeError("unexpected")
		try:
		result = call_retrieve_location(flask_app, PAYLOAD_BY_NAI)
		assert result is not None
		except RuntimeError:
		pytest.fail("Controller re-raised the exception instead of catching it")

Admin message