add tests for edge cloud zones controller. add base implementation for edge... (098dde3f) · Commits · OOP / code / open-exposure-gateway

edge_cloud_management_api/main.py

+3 −2

Original line number	Diff line number	Diff line
		from edge_cloud_management_api.app import main
		from edge_cloud_management_api.app import get_app_instance

		if __name__ == "__main__":
		main()
		app = get_app_instance()
		app.run(host="127.0.0.1", port=8080)

edge_cloud_management_api/app.py

+4 −3

Original line number	Diff line number	Diff line
		@@ -3,7 +3,7 @@ from connexion import FlaskApp
		from connexion.options import SwaggerUIOptions


		def main():
		def get_app_instance() -> FlaskApp:
		file_path = Path(__file__).resolve().parent
		swagger_options = SwaggerUIOptions(swagger_ui_path="/docs")
		app = FlaskApp(__name__, specification_dir=file_path / "specification")
		@@ -12,8 +12,9 @@ def main():
		swagger_ui_options=swagger_options,
		strict_validation=True,
		)
		app.run(host="127.0.0.1", port=8080)
		return app


		if __name__ == "__main__":
		main()
		app = get_app_instance()
		app.run(host="127.0.0.1", port=8080)

edge_cloud_management_api/controllers/edge_cloud_controller.py

+68 −1

Original line number	Diff line number	Diff line
		from flask import jsonify
		from pydantic import BaseModel, Field, ValidationError
		from typing import List


		class EdgeCloudZone(BaseModel):
		edgeCloudZoneId: str = Field(
		..., description="Unique identifier of the Edge Cloud Zone"
		)
		edgeCloudZoneName: str = Field(..., description="Name of the Edge Cloud Zone")
		edgeCloudZoneStatus: str = Field(
		...,
		description="Status of the Edge Cloud Zone",
		pattern="^(active\|inactive\|unknown)$",
		)
		edgeCloudProvider: str = Field(..., description="Name of the Edge Cloud Provider")
		edgeCloudRegion: str = Field(..., description="Region of the Edge Cloud Zone")


		class EdgeCloudQueryParams(BaseModel):
		x_correlator: str \| None
		region: str \| None
		status: str \| None = Field(
		...,
		description="Status of the Edge Cloud Zone",
		pattern="^(active\|inactive\|unknown)$",
		)


		def get_local_zones() -> List[EdgeCloudZone]:
		"""get local Operator Platform available zones from Service Resource Manager"""
		return []


		def get_federated_zones() -> List[EdgeCloudZone]:
		"""get partner/federated Operator Platform available zones from Federation Manager"""
		return []


		def get_all_cloud_zones() -> List[EdgeCloudZone]:
		"""get all available zones from local and federated Operator Platforms"""
		return get_local_zones() + get_federated_zones()


		def get_edge_cloud_zones(x_correlator=None, region=None, status=None): # noqa: E501
		@@ -16,7 +57,33 @@ def get_edge_cloud_zones(x_correlator=None, region=None, status=None): # noqa:
		:rtype: EdgeCloudZones
		"""
		try:
		return "edge_cloud_zones", 200
		query_params = EdgeCloudQueryParams(
		x_correlator=x_correlator,
		region=region,
		status=status,
		)

		filtered_zones = [
		zone
		for zone in get_all_cloud_zones()
		if (
		(query_params.region is None)
		or (zone["edgeCloudRegion"] == query_params.region)
		)
		and (
		(query_params.status is None)
		or (zone["edgeCloudZoneStatus"] == query_params.status)
		)
		]
		response = [EdgeCloudZone(**zone).model_dump() for zone in filtered_zones]
		return jsonify(response), 200

		except ValidationError as e:
		return (
		jsonify({"status": 400, "code": "VALIDATION_ERROR", "message": e.errors()}),
		400,
		)

		except Exception as e:
		error_info = {
		"status": 500,

tests/controllers/init.py→tests/e2e/init.py

+0 −0

File moved.

tests/managers/init.py→tests/unit/init.py

+0 −0

File moved.