Implement TF, test framework setup and unit test (90e293d9) · Commits · OOP / code / transformation-function-sdk

src/sunrise6g_opensdk/network/clients/open5gs/client.py

+13 −10

Original line number	Diff line number	Diff line
		@@ -61,16 +61,19 @@ class NetworkManager(NetworkManagementInterface):
		raise ValidationError(
		"Open5GS requires a device to be specified for location retrieval in NEF."
		)
		def add_core_specific_location_parameters(self, retrieve_location_request: schemas.RetrievalLocationRequest, subscription: schemas.MonitoringEventSubscriptionRequest) -> None:
		subscription.msisdn = retrieve_location_request.device.phoneNumber
		subscription.monitoringType = schemas.MonitoringType.LOCATION_REPORTING
		subscription.locationType = schemas.LocationType.CURRENT_LOCATION
		# subscription.locationType = schemas.LocationType.LAST_KNOWN
		# subscription.maximumNumberOfReports = 1
		# subscription.repPeriod = schemas.DurationSec(root=20)

		return subscription

		def add_core_specific_location_parameters(self, retrieve_location_request: schemas.RetrievalLocationRequest) -> schemas.MonitoringEventSubscriptionRequest:
		return schemas.MonitoringEventSubscriptionRequest(
		msisdn=retrieve_location_request.device.phoneNumber.root.lstrip('+'),
		notificationDestination="http://127.0.0.1:8001",
		monitoringType=schemas.MonitoringType.LOCATION_REPORTING,
		locationType=schemas.LocationType.LAST_KNOWN
		)
		# subscription.msisdn = retrieve_location_request.device.phoneNumber.root.lstrip('+')
		# monitoringType = schemas.MonitoringType.LOCATION_REPORTING
		# locationType = schemas.LocationType.LAST_KNOWN
		# locationType = schemas.LocationType.CURRENT_LOCATION
		# maximumNumberOfReports = 1
		# repPeriod = schemas.DurationSec(root=20)


		# Note:

src/sunrise6g_opensdk/network/core/network_interface.py

+11 −12

Original line number	Diff line number	Diff line
		@@ -114,8 +114,7 @@ class NetworkManagementInterface(ABC):

		def add_core_specific_location_parameters(
		self,
		retrieve_location_request: schemas.RetrievalLocationRequest,
		subscription: schemas.MonitoringEventSubscriptionRequest,
		retrieve_location_request: schemas.RetrievalLocationRequest
		)-> schemas.MonitoringEventSubscriptionRequest:
		"""
		Placeholder for adding core-specific parameters to the location subscription.
		@@ -215,17 +214,17 @@ class NetworkManagementInterface(ABC):

		def _build_monitoring_event_subscription(self, retrieve_location_request: schemas.RetrievalLocationRequest) -> schemas.MonitoringEventSubscriptionRequest:
		self.core_specific_monitoring_event_validation(retrieve_location_request)
		subscription_3gpp = self.add_core_specific_location_parameters(retrieve_location_request)
		device = retrieve_location_request.device
		subscription = schemas.MonitoringEventSubscriptionRequest(
		externalId=device.networkAccessIdentifier,
		ipv4Address=device.ipv4Address,
		ipv6Addr=device.ipv6Address,
		msisdn=device.phoneNumber,
		notificationDestination= "http://test_server:8001")
		mapped_3gpp_subscription = self.add_core_specific_location_parameters(retrieve_location_request,subscription)
		return mapped_3gpp_subscription

		def _compute_camara_last_location_time(self, event_time_str: datetime, age_of_location_info_min : int = None) -> :
		subscription_3gpp.externalId = device.networkAccessIdentifier
		subscription_3gpp.ipv4Addr = device.ipv4Address
		subscription_3gpp.ipv6Addr = device.ipv6Address
		# subscription.msisdn = device.phoneNumber.root.lstrip('+')
		# subscription.notificationDestination = "http://127.0.0.1:8001"

		return subscription_3gpp

		def _compute_camara_last_location_time(self, event_time_str: datetime, age_of_location_info_min : int = None) -> datetime:
		"""
		event_time_str: ISO 8601 string, e.g. "2025-06-18T12:30:00Z"
		age_of_location_info_min: unsigned int, age of location info in minutes

tests/network/conftest.py

0 → 100644

+62 −0

Original line number	Diff line number	Diff line
		import pytest

		from sunrise6g_opensdk.common.sdk import Sdk as sdkclient
		from sunrise6g_opensdk.network.core.schemas import RetrievalLocationRequest, Device, MonitoringEventSubscriptionRequest




		@pytest.fixture(scope="session", name="network_client")
		def instantiate_network_client():
		"""Fixture to create and share a network client across tests"""
		client_specs = {
		"network": {
		"client_name": "open5gs",
		"base_url": "http://127.0.0.1:8082/",
		"scs_as_id": "af_1",
		}
		}
		clients = sdkclient.create_clients_from(client_specs)
		return clients.get("network")


		# Test full input data from Camara Payload
		# {
		# "phoneNumber": "+1234567890",
		# "networkAccessIdentifier": "user123@example.com",
		# "ipv4Address": {
		# "publicAddress": "198.51.100.10",
		# "privateAddress": "10.0.0.1",
		# "publicPort": 12345
		# },
		# "ipv6Address": "2001:0db8:85a3:0000:0000:8a2e:0370:7334"
		# }

		@pytest.fixture(scope="module")
		def camara_payload_input_data() -> RetrievalLocationRequest:
		"""
		Fixture to provide input data for CAMARA payload.
		This data is used in tests that require a specific payload structure.
		"""

		return RetrievalLocationRequest(device=Device(phoneNumber="+1234567890"))

		# Sample output test data 3GPP MonitoringEventSubscription Request Payload
		# {
		# "msisdn": "+306912345678",
		# "notificationDestination": "https://af.example.com/location_notifications",
		# "monitoringType": "LOCATION_REPORTING",
		# "locationType": "CURRENT_LOCATION"
		# }
		@pytest.fixture(scope="module", name="expected_output_data")
		def monitoring_request_3gpp_payload_output_data(camara_payload_input_data: RetrievalLocationRequest) -> MonitoringEventSubscriptionRequest:
		"""
		Fixture to provide output data for 3GPP monitoring event request payload.
		"""
		output_msisdn = camara_payload_input_data.device.phoneNumber.root.lstrip('+')
		return MonitoringEventSubscriptionRequest(
		msisdn=output_msisdn,
		notificationDestination="http://127.0.0.1:8001",
		monitoringType="LOCATION_REPORTING",
		locationType="LAST_KNOWN_LOCATION"
		)
		No newline at end of file

tests/network/test_create_monitoring_event.py

+19 −0

Original line number	Diff line number	Diff line


		import pytest

		from sunrise6g_opensdk.common.sdk import Sdk as sdkclient
		from sunrise6g_opensdk.network.core.common import CoreHttpError
		from sunrise6g_opensdk.network.core.network_interface import NetworkManagementInterface
		from sunrise6g_opensdk.network.core.schemas import RetrievalLocationRequest


		from tests.network.conftest import instantiate_network_client, camara_payload_input_data


		def test_camara_tf_3gpp_event(network_client : NetworkManagementInterface ,camara_payload_input_data: RetrievalLocationRequest, expected_output_data) -> None:
		actual_result = network_client._build_monitoring_event_subscription(retrieve_location_request=camara_payload_input_data)
		assert actual_result == expected_output_data


		# def test_create_monitoring_event():
		No newline at end of file

Admin message