Loading src/sunrise6g_opensdk/network/core/common.py +12 −0 Original line number Diff line number Diff line Loading @@ -82,6 +82,18 @@ def traffic_influence_put( return _make_request("PUT", url, data=data) def traffic_influence_get(base_url: str, scs_as_id: str, sessionId: str = None) -> dict: url = traffic_influence_build_url(base_url, scs_as_id, sessionId) return _make_request("GET", url) def traffic_influence_get_all( base_url: str, scs_as_id: str, sessionId: str = None ) -> list[dict]: url = traffic_influence_build_url(base_url, scs_as_id) return _make_request("GET", url) def traffic_influence_build_url(base_url: str, scs_as_id: str, session_id: str = None): url = f"{base_url}/3gpp-traffic-influence/v1/{scs_as_id}/subscriptions" if session_id is not None and len(session_id) > 0: Loading src/sunrise6g_opensdk/network/core/network_interface.py +32 −0 Original line number Diff line number Diff line Loading @@ -187,6 +187,27 @@ class NetworkManagementInterface(ABC): self.add_core_specific_ti_parameters(traffic_influence_data, subscription) return subscription def _build_camara_ti(self, trafficInflSub: Dict): traffic_influence_data = schemas.TrafficInfluSub.model_validate(trafficInflSub) flowDesc = traffic_influence_data.trafficFilters[0].flowDescriptions[0] serverIp = flowDesc.split("to ")[1].split("/32")[0] edgeId = traffic_influence_data.trafficRoutes[0].dnai camara_ti = schemas.CreateTrafficInfluence( appId=traffic_influence_data.afAppId, appInstanceId=serverIp, edgeCloudZoneId=edgeId, notificationUri=traffic_influence_data.notificationDestination, device=schemas.Device( ipv4Address=schemas.DeviceIpv4Addr1( publicAddress=traffic_influence_data.ipv4Addr, privateAddress=traffic_influence_data.ipv4Addr, ) ), ) return camara_ti def create_qod_session(self, session_info: Dict) -> Dict: """ Creates a QoS session based on CAMARA QoD API input. Loading Loading @@ -307,5 +328,16 @@ class NetworkManagementInterface(ABC): common.traffic_influence_delete(self.base_url, self.scs_as_id, resource_id) return def get_individual_traffic_influence_resource(self, resource_id: str) -> Dict: nef_response = common.traffic_influence_get( self.base_url, self.scs_as_id, resource_id ) camara_ti = self._build_camara_ti(nef_response) return camara_ti def get_all_traffic_influence_resource(self) -> list[Dict]: r = common.traffic_influence_get(self.base_url, self.scs_as_id) return [self._build_camara_ti(item) for item in r] # Placeholder for other CAMARA APIs (e.g., Traffic Influence, # Location-retrieval, etc.) tests/network/test_create_traffic_influence.py +110 −36 Original line number Diff line number Diff line Loading @@ -8,6 +8,36 @@ from sunrise6g_opensdk.network.core.common import CoreHttpError from sunrise6g_opensdk.network.core.network_interface import NetworkManagementInterface from tests.network.test_cases import test_cases ti_session1 = { "device": { "ipv4Address": {"publicAddress": "12.1.2.31", "privateAddress": "12.1.2.31"} }, "edgeCloudZoneId": "edge", "appId": "testSdk-ffff-aaaa-c0ffe", "appInstanceId": "172.21.18.3", "notificationUri": "https://endpoint.example.com/sink", } ti_session1_put = { "device": { "ipv4Address": {"publicAddress": "12.1.2.31", "privateAddress": "12.1.2.31"} }, "edgeCloudZoneId": "edge2", "appId": "testSdk-ffff-aaaa-c0ffe", "appInstanceId": "172.21.18.3", "notificationUri": "https://endpoint.example.com/sink", } ti_session2 = { "device": { "ipv4Address": {"publicAddress": "12.1.2.31", "privateAddress": "12.1.2.31"} }, "edgeCloudZoneId": "edge", "appId": "testSdk-ffff-aaaa-c0ffe", "appInstanceId": "172.21.18.65", "notificationUri": "https://endpoint.example.com/sink", } @pytest.fixture(scope="module", name="network_client") def instantiate_network_client(request): Loading @@ -29,32 +59,30 @@ def id_func(val): ) def test_valid_input(network_client: NetworkManagementInterface): ti_session = { "device": { "ipv4Address": {"publicAddress": "12.1.2.31", "privateAddress": "12.1.2.31"} }, "edgeCloudZoneId": "edge", "appId": "testSdk-ffff-aaaa-c0ffe", "appInstanceId": "172.21.18.3", "notificationUri": "https://endpoint.example.com/sink", } network_client._build_ti_subscription(ti_session) network_client._build_ti_subscription(ti_session1) network_client._build_ti_subscription(ti_session1_put) network_client._build_ti_subscription(ti_session2) @pytest.fixture(scope="module") def traffic_influence_id(network_client: NetworkManagementInterface): try: response = network_client.create_traffic_influence_resource(ti_session1) assert response is not None, "Response should not be None" assert isinstance(response, dict), "Response should be a dictionary" assert ( "trafficInfluenceID" in response ), "Response should contain 'trafficInfluenceID'" yield str(response["trafficInfluenceID"]) finally: pass ti_session = { "device": { "ipv4Address": {"publicAddress": "12.1.2.31", "privateAddress": "12.1.2.31"} }, "edgeCloudZoneId": "edge", "appId": "testSdk-ffff-aaaa-c0ffe", "appInstanceId": "172.21.18.3", "notificationUri": "https://endpoint.example.com/sink", } @pytest.fixture(scope="module") def traffic_influence_id2(network_client: NetworkManagementInterface): try: response = network_client.create_traffic_influence_resource(ti_session) response = network_client.create_traffic_influence_resource(ti_session2) assert response is not None, "Response should not be None" assert isinstance(response, dict), "Response should be a dictionary" assert ( Loading @@ -71,42 +99,88 @@ def traffic_influence_id(network_client: NetworkManagementInterface): ids=id_func, indirect=True, ) def test_create_traffic_influence(traffic_influence_id): def test_create_traffic_influence_1(traffic_influence_id): assert traffic_influence_id is not None @pytest.mark.parametrize( "network_client", test_cases, ids=id_func, indirect=True, ) def test_create_traffic_influence_2(traffic_influence_id2): assert traffic_influence_id2 is not None @pytest.mark.parametrize("network_client", test_cases, ids=id_func, indirect=True) def test_timer_wait_5_seconds(network_client): time.sleep(5) @pytest.mark.parametrize("network_client", test_cases, ids=id_func, indirect=True) def test_put_traffic_influence_session( def test_get_traffic_influence_session_1( network_client: NetworkManagementInterface, traffic_influence_id ): try: ti_session = { "device": { "ipv4Address": { "publicAddress": "12.1.2.31", "privateAddress": "12.1.2.31", } }, "edgeCloudZoneId": "edge", "appId": "testSdk-ffff-aaaa-c0ffe", "appInstanceId": "172.21.18.5", "notificationUri": "https://endpoint.example.com/sink", } network_client.put_traffic_influence_resource(traffic_influence_id, ti_session) response = network_client.get_individual_traffic_influence_resource( traffic_influence_id ) assert response is not None, "response should not be None" except CoreHttpError as e: pytest.fail(f"Failed to get traffic influence: {e}") @pytest.mark.parametrize("network_client", test_cases, ids=id_func, indirect=True) def test_put_traffic_influence_session_1( network_client: NetworkManagementInterface, traffic_influence_id ): try: network_client.put_traffic_influence_resource( traffic_influence_id, ti_session1_put ) except CoreHttpError as e: pytest.fail(f"Failed to update traffic influence session: {e}") @pytest.mark.parametrize("network_client", test_cases, ids=id_func, indirect=True) def test_delete_traffic_influence_session( def test_get_traffic_influence_session_after_put_1( network_client: NetworkManagementInterface, traffic_influence_id ): try: response = network_client.get_individual_traffic_influence_resource( traffic_influence_id ) assert response is not None, "response should not be None" except CoreHttpError as e: pytest.fail(f"Failed to get traffic influence: {e}") @pytest.mark.parametrize("network_client", test_cases, ids=id_func, indirect=True) def test_get_all_traffic_influence_sessions(network_client: NetworkManagementInterface): try: response = network_client.get_all_traffic_influence_resource() assert response is not None, "response should not be None" assert len(response) == 2, "response must containt 2 elements" except CoreHttpError as e: pytest.fail(f"Failed to get traffic influence: {e}") @pytest.mark.parametrize("network_client", test_cases, ids=id_func, indirect=True) def test_delete_traffic_influence_session_1( network_client: NetworkManagementInterface, traffic_influence_id ): try: network_client.delete_traffic_influence_resource(traffic_influence_id) except CoreHttpError as e: pytest.fail(f"Failed to delete traffic influence: {e}") @pytest.mark.parametrize("network_client", test_cases, ids=id_func, indirect=True) def test_delete_traffic_influence_session_2( network_client: NetworkManagementInterface, traffic_influence_id2 ): try: network_client.delete_traffic_influence_resource(traffic_influence_id2) except CoreHttpError as e: pytest.fail(f"Failed to delete traffic influence: {e}") Loading
src/sunrise6g_opensdk/network/core/common.py +12 −0 Original line number Diff line number Diff line Loading @@ -82,6 +82,18 @@ def traffic_influence_put( return _make_request("PUT", url, data=data) def traffic_influence_get(base_url: str, scs_as_id: str, sessionId: str = None) -> dict: url = traffic_influence_build_url(base_url, scs_as_id, sessionId) return _make_request("GET", url) def traffic_influence_get_all( base_url: str, scs_as_id: str, sessionId: str = None ) -> list[dict]: url = traffic_influence_build_url(base_url, scs_as_id) return _make_request("GET", url) def traffic_influence_build_url(base_url: str, scs_as_id: str, session_id: str = None): url = f"{base_url}/3gpp-traffic-influence/v1/{scs_as_id}/subscriptions" if session_id is not None and len(session_id) > 0: Loading
src/sunrise6g_opensdk/network/core/network_interface.py +32 −0 Original line number Diff line number Diff line Loading @@ -187,6 +187,27 @@ class NetworkManagementInterface(ABC): self.add_core_specific_ti_parameters(traffic_influence_data, subscription) return subscription def _build_camara_ti(self, trafficInflSub: Dict): traffic_influence_data = schemas.TrafficInfluSub.model_validate(trafficInflSub) flowDesc = traffic_influence_data.trafficFilters[0].flowDescriptions[0] serverIp = flowDesc.split("to ")[1].split("/32")[0] edgeId = traffic_influence_data.trafficRoutes[0].dnai camara_ti = schemas.CreateTrafficInfluence( appId=traffic_influence_data.afAppId, appInstanceId=serverIp, edgeCloudZoneId=edgeId, notificationUri=traffic_influence_data.notificationDestination, device=schemas.Device( ipv4Address=schemas.DeviceIpv4Addr1( publicAddress=traffic_influence_data.ipv4Addr, privateAddress=traffic_influence_data.ipv4Addr, ) ), ) return camara_ti def create_qod_session(self, session_info: Dict) -> Dict: """ Creates a QoS session based on CAMARA QoD API input. Loading Loading @@ -307,5 +328,16 @@ class NetworkManagementInterface(ABC): common.traffic_influence_delete(self.base_url, self.scs_as_id, resource_id) return def get_individual_traffic_influence_resource(self, resource_id: str) -> Dict: nef_response = common.traffic_influence_get( self.base_url, self.scs_as_id, resource_id ) camara_ti = self._build_camara_ti(nef_response) return camara_ti def get_all_traffic_influence_resource(self) -> list[Dict]: r = common.traffic_influence_get(self.base_url, self.scs_as_id) return [self._build_camara_ti(item) for item in r] # Placeholder for other CAMARA APIs (e.g., Traffic Influence, # Location-retrieval, etc.)
tests/network/test_create_traffic_influence.py +110 −36 Original line number Diff line number Diff line Loading @@ -8,6 +8,36 @@ from sunrise6g_opensdk.network.core.common import CoreHttpError from sunrise6g_opensdk.network.core.network_interface import NetworkManagementInterface from tests.network.test_cases import test_cases ti_session1 = { "device": { "ipv4Address": {"publicAddress": "12.1.2.31", "privateAddress": "12.1.2.31"} }, "edgeCloudZoneId": "edge", "appId": "testSdk-ffff-aaaa-c0ffe", "appInstanceId": "172.21.18.3", "notificationUri": "https://endpoint.example.com/sink", } ti_session1_put = { "device": { "ipv4Address": {"publicAddress": "12.1.2.31", "privateAddress": "12.1.2.31"} }, "edgeCloudZoneId": "edge2", "appId": "testSdk-ffff-aaaa-c0ffe", "appInstanceId": "172.21.18.3", "notificationUri": "https://endpoint.example.com/sink", } ti_session2 = { "device": { "ipv4Address": {"publicAddress": "12.1.2.31", "privateAddress": "12.1.2.31"} }, "edgeCloudZoneId": "edge", "appId": "testSdk-ffff-aaaa-c0ffe", "appInstanceId": "172.21.18.65", "notificationUri": "https://endpoint.example.com/sink", } @pytest.fixture(scope="module", name="network_client") def instantiate_network_client(request): Loading @@ -29,32 +59,30 @@ def id_func(val): ) def test_valid_input(network_client: NetworkManagementInterface): ti_session = { "device": { "ipv4Address": {"publicAddress": "12.1.2.31", "privateAddress": "12.1.2.31"} }, "edgeCloudZoneId": "edge", "appId": "testSdk-ffff-aaaa-c0ffe", "appInstanceId": "172.21.18.3", "notificationUri": "https://endpoint.example.com/sink", } network_client._build_ti_subscription(ti_session) network_client._build_ti_subscription(ti_session1) network_client._build_ti_subscription(ti_session1_put) network_client._build_ti_subscription(ti_session2) @pytest.fixture(scope="module") def traffic_influence_id(network_client: NetworkManagementInterface): try: response = network_client.create_traffic_influence_resource(ti_session1) assert response is not None, "Response should not be None" assert isinstance(response, dict), "Response should be a dictionary" assert ( "trafficInfluenceID" in response ), "Response should contain 'trafficInfluenceID'" yield str(response["trafficInfluenceID"]) finally: pass ti_session = { "device": { "ipv4Address": {"publicAddress": "12.1.2.31", "privateAddress": "12.1.2.31"} }, "edgeCloudZoneId": "edge", "appId": "testSdk-ffff-aaaa-c0ffe", "appInstanceId": "172.21.18.3", "notificationUri": "https://endpoint.example.com/sink", } @pytest.fixture(scope="module") def traffic_influence_id2(network_client: NetworkManagementInterface): try: response = network_client.create_traffic_influence_resource(ti_session) response = network_client.create_traffic_influence_resource(ti_session2) assert response is not None, "Response should not be None" assert isinstance(response, dict), "Response should be a dictionary" assert ( Loading @@ -71,42 +99,88 @@ def traffic_influence_id(network_client: NetworkManagementInterface): ids=id_func, indirect=True, ) def test_create_traffic_influence(traffic_influence_id): def test_create_traffic_influence_1(traffic_influence_id): assert traffic_influence_id is not None @pytest.mark.parametrize( "network_client", test_cases, ids=id_func, indirect=True, ) def test_create_traffic_influence_2(traffic_influence_id2): assert traffic_influence_id2 is not None @pytest.mark.parametrize("network_client", test_cases, ids=id_func, indirect=True) def test_timer_wait_5_seconds(network_client): time.sleep(5) @pytest.mark.parametrize("network_client", test_cases, ids=id_func, indirect=True) def test_put_traffic_influence_session( def test_get_traffic_influence_session_1( network_client: NetworkManagementInterface, traffic_influence_id ): try: ti_session = { "device": { "ipv4Address": { "publicAddress": "12.1.2.31", "privateAddress": "12.1.2.31", } }, "edgeCloudZoneId": "edge", "appId": "testSdk-ffff-aaaa-c0ffe", "appInstanceId": "172.21.18.5", "notificationUri": "https://endpoint.example.com/sink", } network_client.put_traffic_influence_resource(traffic_influence_id, ti_session) response = network_client.get_individual_traffic_influence_resource( traffic_influence_id ) assert response is not None, "response should not be None" except CoreHttpError as e: pytest.fail(f"Failed to get traffic influence: {e}") @pytest.mark.parametrize("network_client", test_cases, ids=id_func, indirect=True) def test_put_traffic_influence_session_1( network_client: NetworkManagementInterface, traffic_influence_id ): try: network_client.put_traffic_influence_resource( traffic_influence_id, ti_session1_put ) except CoreHttpError as e: pytest.fail(f"Failed to update traffic influence session: {e}") @pytest.mark.parametrize("network_client", test_cases, ids=id_func, indirect=True) def test_delete_traffic_influence_session( def test_get_traffic_influence_session_after_put_1( network_client: NetworkManagementInterface, traffic_influence_id ): try: response = network_client.get_individual_traffic_influence_resource( traffic_influence_id ) assert response is not None, "response should not be None" except CoreHttpError as e: pytest.fail(f"Failed to get traffic influence: {e}") @pytest.mark.parametrize("network_client", test_cases, ids=id_func, indirect=True) def test_get_all_traffic_influence_sessions(network_client: NetworkManagementInterface): try: response = network_client.get_all_traffic_influence_resource() assert response is not None, "response should not be None" assert len(response) == 2, "response must containt 2 elements" except CoreHttpError as e: pytest.fail(f"Failed to get traffic influence: {e}") @pytest.mark.parametrize("network_client", test_cases, ids=id_func, indirect=True) def test_delete_traffic_influence_session_1( network_client: NetworkManagementInterface, traffic_influence_id ): try: network_client.delete_traffic_influence_resource(traffic_influence_id) except CoreHttpError as e: pytest.fail(f"Failed to delete traffic influence: {e}") @pytest.mark.parametrize("network_client", test_cases, ids=id_func, indirect=True) def test_delete_traffic_influence_session_2( network_client: NetworkManagementInterface, traffic_influence_id2 ): try: network_client.delete_traffic_influence_resource(traffic_influence_id2) except CoreHttpError as e: pytest.fail(f"Failed to delete traffic influence: {e}")
