Merge pull request #151 from SunriseOpenOperatorPlatform/fix/review-comments-i2edge-150 (9bd5e6b6) · Commits · OOP / code / transformation-function-sdk

src/sunrise6g_opensdk/edgecloud/adapters/i2edge/client.py

+25 −13

Original line number	Original line	Diff line number	Diff line
	@@ -43,6 +43,25 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	self.content_type_gsma = "application/json"		self.content_type_gsma = "application/json"
	self.encoding_gsma = "utf-8"		self.encoding_gsma = "utf-8"

			def _transform_to_camara_zone(self, zone_data: dict) -> camara_schemas.EdgeCloudZone:
			"""
			Transform i2Edge zone data to CAMARA EdgeCloudZone format.

			:param zone_data: Raw zone data from i2Edge API
			:return: CAMARA-compliant EdgeCloudZone object
			"""
			return camara_schemas.EdgeCloudZone(
			edgeCloudZoneId=camara_schemas.EdgeCloudZoneId(zone_data.get("zoneId", "unknown")),
			edgeCloudZoneName=camara_schemas.EdgeCloudZoneName(
			zone_data.get("nodeName", "unknown")
			),
			edgeCloudProvider=camara_schemas.EdgeCloudProvider("i2edge"),
			edgeCloudRegion=camara_schemas.EdgeCloudRegion(
			zone_data.get("geographyDetails", "unknown")
			),
			edgeCloudZoneStatus=camara_schemas.EdgeCloudZoneStatus.unknown,
			)

	# ########################################################################		# ########################################################################
	# CAMARA EDGE CLOUD MANAGEMENT API		# CAMARA EDGE CLOUD MANAGEMENT API
	# ########################################################################		# ########################################################################
	@@ -71,17 +90,7 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	i2edge_response = response.json()		i2edge_response = response.json()
	log.info("Availability zones retrieved successfully")		log.info("Availability zones retrieved successfully")
	# Normalise to CAMARA format		# Normalise to CAMARA format
	camara_response = []		camara_response = [self._transform_to_camara_zone(z) for z in i2edge_response]
	for z in i2edge_response:
	zone = camara_schemas.EdgeCloudZone(
	# edgeCloudZoneId = camara_schemas.EdgeCloudZoneId(z["zoneId"]),
	edgeCloudZoneId=camara_schemas.EdgeCloudZoneId(z["zoneId"]),
	edgeCloudZoneName=camara_schemas.EdgeCloudZoneName(z["nodeName"]),
	edgeCloudProvider=camara_schemas.EdgeCloudProvider("i2edge"),
	edgeCloudRegion=camara_schemas.EdgeCloudRegion(z["geographyDetails"]),
	edgeCloudZoneStatus=camara_schemas.EdgeCloudZoneStatus.unknown,
	)
	camara_response.append(zone)
	# Wrap into a Response object		# Wrap into a Response object
	return build_custom_http_response(		return build_custom_http_response(
	status_code=response.status_code,		status_code=response.status_code,
	@@ -267,8 +276,11 @@ class EdgeApplicationManager(EdgeCloudManagementInterface):
	i2edge_response.raise_for_status()		i2edge_response.raise_for_status()
	# TODO: Implement CAMARA-compliant error handling for failed onboarding responses		# TODO: Implement CAMARA-compliant error handling for failed onboarding responses
	except ValidationError as e:		except ValidationError as e:
	log.error(f"Invalid CAMARA manifest: {e}")		error_details = "; ".join(
	raise ValueError(f"Invalid CAMARA manifest: {e}")		[f"Field '{err['loc'][0]}': {err['msg']}" for err in e.errors()]
			)
			log.error(f"Invalid CAMARA manifest: {error_details}")
			raise ValueError(f"Invalid CAMARA manifest: {error_details}")
	except I2EdgeError as e:		except I2EdgeError as e:
	log.error(f"Failed to onboard app to i2Edge: {e}")		log.error(f"Failed to onboard app to i2Edge: {e}")
	raise		raise

tests/edgecloud/test_e2e.py

+2 −2

Original line number	Original line	Diff line number	Diff line
	@@ -190,8 +190,8 @@ def test_deploy_app(app_instance_id):


	@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True)		@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True)
	def test_timer_wait_10_seconds(edgecloud_client):		def test_timer_wait_30_seconds(edgecloud_client):
	time.sleep(10)		time.sleep(30)


	@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True)		@pytest.mark.parametrize("edgecloud_client", test_cases, ids=id_func, indirect=True)