Loading service-resource-manager-implementation/pytest.inideleted 100644 → 0 +0 −5 Original line number Diff line number Diff line [pytest] testpaths = tests pythonpath = . log_cli = true log_cli_level = WARNING service-resource-manager-implementation/tests/conftest.pydeleted 100644 → 0 +0 −137 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.pydeleted 100644 → 0 +0 −214 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 503 UNAVAILABLE.""" def test_returns_503_when_no_adapter(self, flask_app): body, status = call_retrieve_location(flask_app, PAYLOAD_BY_NAI) assert status == 503 def test_503_body_has_code_unavailable(self, flask_app): body, status = call_retrieve_location(flask_app, PAYLOAD_BY_NAI) assert body.get("code") == "UNAVAILABLE" def test_503_body_has_message(self, flask_app): body, status = call_retrieve_location(flask_app, PAYLOAD_BY_NAI) assert "message" in body assert "Network Adapter" in body["message"] 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") Loading
service-resource-manager-implementation/pytest.inideleted 100644 → 0 +0 −5 Original line number Diff line number Diff line [pytest] testpaths = tests pythonpath = . log_cli = true log_cli_level = WARNING
service-resource-manager-implementation/tests/conftest.pydeleted 100644 → 0 +0 −137 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.pydeleted 100644 → 0 +0 −214 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 503 UNAVAILABLE.""" def test_returns_503_when_no_adapter(self, flask_app): body, status = call_retrieve_location(flask_app, PAYLOAD_BY_NAI) assert status == 503 def test_503_body_has_code_unavailable(self, flask_app): body, status = call_retrieve_location(flask_app, PAYLOAD_BY_NAI) assert body.get("code") == "UNAVAILABLE" def test_503_body_has_message(self, flask_app): body, status = call_retrieve_location(flask_app, PAYLOAD_BY_NAI) assert "message" in body assert "Network Adapter" in body["message"] 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")
