Finalize tests and opensdk implementation (876beaf9) · Commits · OOP / code / transformation-function-sdk

src/sunrise6g_opensdk/network/core/network_interface.py

+12 −11

Original line number	Diff line number	Diff line
		@@ -14,7 +14,7 @@ import uuid
		from abc import ABC
		from itertools import product
		from typing import Dict
		from datetime import datetime, timedelta
		from datetime import datetime, timedelta, timezone

		from sunrise6g_opensdk import logger
		from sunrise6g_opensdk.network.clients.errors import NetworkPlatformError
		@@ -224,19 +224,18 @@ class NetworkManagementInterface(ABC):

		return subscription_3gpp

		def _compute_camara_last_location_time(self, event_time_str: datetime, age_of_location_info_min : int = None) -> datetime:
		def _compute_camara_last_location_time(self, event_time: 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
		"""
		event_time = datetime.fromisoformat(event_time_str.replace("Z", "+00:00"))
		if age_of_location_info_min is not None:
		last_location_time = event_time - timedelta(minutes=age_of_location_info_min)
		return last_location_time
		return last_location_time.replace(tzinfo=timezone.utc)
		else:
		return event_time_str
		return event_time.replace(tzinfo=timezone.utc)

		def create_monitoring_event_subscription(self, retrieve_location_request: schemas.RetrievalLocationRequest) -> dict:
		def create_monitoring_event_subscription(self, retrieve_location_request: schemas.RetrievalLocationRequest) -> schemas.Location:
		"""
		Creates a Monitoring Event subscription based on CAMARA Location API input.

		@@ -262,13 +261,15 @@ class NetworkManagementInterface(ABC):
		if monitoring_event_report.locationInfo.ageOfLocationInfo is not None:
		age_of_location_info = monitoring_event_report.locationInfo.ageOfLocationInfo.duration
		last_location_time = self._compute_camara_last_location_time(report_event_time,age_of_location_info)
		print(f"Last Location time is {last_location_time}")
		camara_point_list : list[schemas.Point] = []
		for point in geo_area.polygon.point_list.geographical_coords:
		camara_point_list.append(schemas.Point(latitude=point.lat,longitude=point.lon))
		camara_polygon = schemas.Polygon(areaType=schemas.AreaType.polygon,boundary=schemas.PointList(camara_point_list))

		camara_polygon = schemas.Polygon(areaType=schemas.AreaType.polygon,boundary=geo_area.polygon.point_list)
		camara_loc_area = schemas.Area.model_validate(camara_polygon)
		camara_location = schemas.Location(area=camara_polygon,lastLocationTime=last_location_time)

		camara_location = schemas.Location(area=camara_loc_area,lastLocationTime=last_location_time)

		return camara_location.model_dump(mode="json")
		return camara_location

		def create_qod_session(self, session_info: Dict) -> Dict:
		"""

tests/network/conftest.py

+18 −15

Original line number	Diff line number	Diff line
		import pytest
		import datetime

		from sunrise6g_opensdk.common.sdk import Sdk as sdkclient
		from sunrise6g_opensdk.network.core.schemas import RetrievalLocationRequest, Device, MonitoringEventSubscriptionRequest, Location, AreaType, PointList, Point,Polygon
		@@ -13,7 +12,7 @@ def instantiate_network_client():
		client_specs = {
		"network": {
		"client_name": "open5gs",
		"base_url": "http://127.0.0.1:8082/",
		"base_url": "http://127.0.0.1:8000/",
		"scs_as_id": "af_1",
		}
		}
		@@ -40,7 +39,7 @@ def camara_payload_input_data() -> RetrievalLocationRequest:
		This data is used in tests that require a specific payload structure.
		"""

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

		# Sample output test data 3GPP MonitoringEventSubscription Request Payload
		# {
		@@ -71,33 +70,37 @@ def monitoring_request_camara_payload_output_data(camara_payload_input_data: Ret
		Example:

		{
		"lastLocationTime": "2023-10-27T15:30:00Z",
		"lastLocationTime": "2025-06-23T20:47:22Z",
		"area": {
		"areaType": "POLYGON",
		"boundary": [
		{
		"latitude": 34.0522,
		"longitude": -118.2437
		"latitude": 37.9838,
		"longitude": 23.7275
		},
		{
		"latitude": 34.0535,
		"longitude": -118.2500
		"latitude": 37.98,
		"longitude": 23.75
		},
		{
		"latitude": 34.0480,
		"longitude": -118.2520
		"latitude": 37.97,
		"longitude": 23.73
		},
		{ "latitude": 37.975,
		"longtitude": 23.71
		}
		]
		}
		}
		"""
		point1 = Point(latitude=34.0522, longitude=-118.2437)
		point2 = Point(latitude=34.0535, longitude=-118.2500)
		point3 = Point(latitude=34.0480, longitude=-118.2520)
		point1 = Point(latitude=37.9838, longitude=23.7275)
		point2 = Point(latitude=37.98, longitude=23.75)
		point3 = Point(latitude=37.97, longitude=23.73)
		point4 = Point(latitude=37.975, longitude=23.71)

		point_list = PointList(root=[point1, point2, point3])
		point_list = PointList(root=[point1, point2, point3,point4])

		polygon_area = Polygon(areaType=AreaType.polygon,boundary=point_list)

		location = Location(lastLocationTime="2023-10-27T15:30:00Z",area=polygon_area)
		location = Location(lastLocationTime="2025-06-23T20:47:22Z",area=polygon_area)
		return location

tests/network/test_create_monitoring_event.py

+1 −1

Original line number	Diff line number	Diff line
		@@ -18,6 +18,6 @@ def test_camara_tf_3gpp_event(network_client : NetworkManagementInterface ,camar
		def test_create_monitoring_event(network_client : NetworkManagementInterface, camara_payload_input_data: RetrievalLocationRequest,expected_camara_output_data: Location ):
		try:
		actual_result = network_client.create_monitoring_event_subscription(retrieve_location_request=camara_payload_input_data)
		assert actual_result == expected_camara_output_data
		assert actual_result == expected_camara_output_data, f"Expected actual_result ({actual_result}) \n to be equal to expected_result ({expected_camara_output_data}), but they were not."
		except CoreHttpError as e:
		pytest.fail(f"Failed to retrieve event report: {e}")
		No newline at end of file

Admin message