Update FM integration test cleanup (85802c52) · Commits · OOP / code / federation-manager

src/test/init.py

+157 −6

Original line number	Diff line number	Diff line
		import logging
		import uuid

		import connexion
		import pymongo
		import requests
		from flask_testing import TestCase

		from encoder import JSONEncoder

		from configparser import ConfigParser
		from clients import tf_sdk
		from clients import srm
		import util

		CONFIG = ConfigParser()
		@@ -21,7 +23,7 @@ HOST_KEYCLOAK = CONFIG.get("keycloak", "host")
		PORT_KEYCLOAK = int(CONFIG.get("keycloak", "port"))

		# Dual role testing configuration
		PARTNER_API_ROOT = "http://127.0.0.1:8990"
		PARTNER_API_ROOT = "http://federation-manager-remote:8989"


		class BaseTestCase(TestCase):
		@@ -40,17 +42,28 @@ class BaseTestCase(TestCase):
		# Base url application server
		self.base_url = f"{HOST}:{PORT}{SERVER}"

		if not hasattr(BaseTestCase, "run_suffix"):
		BaseTestCase.run_suffix = uuid.uuid4().hex[:8]
		BaseTestCase.artefact_id_pop_run = str(uuid.uuid5(uuid.NAMESPACE_DNS,
		f"fm-test-pop-{BaseTestCase.run_suffix}"))
		BaseTestCase.artefact_id_oop_run = str(uuid.uuid5(uuid.NAMESPACE_DNS,
		f"fm-test-oop-{BaseTestCase.run_suffix}"))
		BaseTestCase.app_id_pop_run = f"test_app_pop_{BaseTestCase.run_suffix}"
		BaseTestCase.app_id_oop_run = f"test_app_oop_{BaseTestCase.run_suffix}"

		# Partner OP (POP) configuration
		self.artefact_id_pop = "3fa85f64-5717-4562-b3fc-222222222222"
		self.app_id_pop = "test_app_pop"
		self.artefact_id_pop = BaseTestCase.artefact_id_pop_run
		self.app_id_pop = BaseTestCase.app_id_pop_run
		self.artefact_name_pop = "ollama"
		self.artefact_repo_pop = "https://otwld.github.io/ollama-helm/"

		# Originating OP (OOP) configuration
		self.artefact_id_oop = "3fa85f64-5717-4562-b3fc-222222222223"
		self.app_id_oop = "test_app_oop"
		self.artefact_id_oop = BaseTestCase.artefact_id_oop_run
		self.app_id_oop = BaseTestCase.app_id_oop_run
		self.artefact_name_oop = "kubernetes-dashboard"
		self.artefact_repo_oop = "https://kubernetes.github.io/dashboard/"
		self.deployable_artefact_name = self.artefact_name_pop
		self.deployable_artefact_repo = self.artefact_repo_pop

		# Common artifact configuration
		self.app_provider = "test_app_provider"
		@@ -59,6 +72,144 @@ class BaseTestCase(TestCase):

		return app.app

		def cleanup_fm_test_state(self):
		"""Remove leftover FM documents for the fixed integration test IDs."""
		app_ids = [self.app_id_pop, self.app_id_oop]
		artefact_ids = [self.artefact_id_pop, self.artefact_id_oop]

		client = pymongo.MongoClient(f"mongodb://{CONFIG.get('mongodb', 'host')}:{CONFIG.get('mongodb', 'port')}")
		database = client["federation-manager"]

		database["originating_application_deployment_management"].delete_many({"orig_ad_app_id": {"$in": app_ids}})
		database["originating_application_deployment_management_originating_o_p"].delete_many(
		{"orig_ad_app_id": {"$in": app_ids}}
		)
		database["originating_application_onboarding_management"].delete_many({"orig_ao_app_id": {"$in": app_ids}})
		database["originating_application_onboarding_management_originating_o_p"].delete_many(
		{"orig_ao_app_id": {"$in": app_ids}}
		)
		database["originating_application_onboarding_management_update"].delete_many({})
		database["originating_application_onboarding_management_update_originating_o_p"].delete_many({})
		database["originating_artefact_management"].delete_many({"orig_am_artefact_id": {"$in": artefact_ids}})
		database["originating_artefact_management_originating_o_p"].delete_many(
		{"orig_am_artefact_id": {"$in": artefact_ids}}
		)
		database["originating_operator_platform"].delete_many({})
		database["originating_operator_platform_originating_o_p"].delete_many({})
		database["originating_operator_platform_update"].delete_many({})
		database["originating_operator_platform_update_originating_o_p"].delete_many({})
		client.close()

		BaseTestCase.federation_context_id_partner = ""
		BaseTestCase.federation_context_id_originating = ""
		BaseTestCase.zone_partner = ""
		BaseTestCase.zone_originating = ""
		BaseTestCase.instances_partner = []
		BaseTestCase.instances_originating = []

		def cleanup_ecp_test_state(self):
		"""Remove leftover SRM/ECP artefacts and onboardings for test fixtures."""
		for app_id in [self.app_id_pop, self.app_id_oop]:
		try:
		srm.delete_onboarding(app_id)
		except Exception:
		pass

		def build_api_url(self, api_root, path):
		return f"{api_root}{SERVER}/{path}"

		def get_federation_context_at_api_root(self, token, api_root):
		"""Get federation context directly from a specific FM API root."""
		url = self.build_api_url(api_root, "fed-context-id")
		return self.make_request_partner_op("GET", url, token=token)

		def get_federation_at_api_root(self, federation_id, token, api_root):
		"""Get federation details directly from a specific FM API root."""
		url = self.build_api_url(api_root, f"{federation_id}/partner")
		return self.make_request_partner_op("GET", url, token=token)

		def get_app_instances_at_api_root(self, federation_id, app_id, app_provider_id, token, api_root):
		"""Get application instances directly from a specific FM API root."""
		url = self.build_api_url(api_root, f"{federation_id}/application/lcm/app/{app_id}/appProvider/{app_provider_id}")
		return self.make_request_partner_op("GET", url, token=token)

		def cleanup_remote_partner_test_state(self, token, api_root=PARTNER_API_ROOT):
		"""Remove leftover partner-side FM resources created by originating-mode integration tests."""
		response = self.get_federation_context_at_api_root(token, api_root)
		if response.status_code != 200:
		return

		federation_context_id = response.json().get("federationContextId")
		if not federation_context_id:
		return

		instances_response = self.get_app_instances_at_api_root(
		federation_context_id,
		self.app_id_oop,
		self.app_provider,
		token,
		api_root,
		)
		if instances_response.status_code == 200:
		for zone_info in instances_response.json():
		zone_id = zone_info.get("zoneId")
		if not zone_id:
		continue
		for instance in zone_info.get("appInstanceInfo", []):
		instance_id = instance.get("appInstIdentifier")
		if not instance_id:
		continue
		try:
		url = self.build_api_url(
		api_root,
		f"{federation_context_id}/application/lcm/app/{self.app_id_oop}/instance/{instance_id}/zone/{zone_id}",
		)
		self.make_request_partner_op("DELETE", url, token=token)
		except Exception:
		pass

		for app_id in [self.app_id_oop]:
		try:
		url = self.build_api_url(api_root, f"{federation_context_id}/application/onboarding/app/{app_id}")
		self.make_request_partner_op("DELETE", url, token=token)
		except Exception:
		pass

		for artefact_id in [self.artefact_id_oop]:
		try:
		url = self.build_api_url(api_root, f"{federation_context_id}/artefact/{artefact_id}")
		self.make_request_partner_op("DELETE", url, token=token)
		except Exception:
		pass

		try:
		federation_response = self.get_federation_at_api_root(federation_context_id, token, api_root)
		if federation_response.status_code == 200:
		federation_data = federation_response.json()
		for zone in federation_data.get("offeredAvailabilityZones", []):
		zone_id = zone.get("zoneId") if isinstance(zone, dict) else None
		if not zone_id:
		continue
		try:
		url = self.build_api_url(api_root, f"{federation_context_id}/zones/{zone_id}")
		self.make_request_partner_op("DELETE", url, token=token)
		except Exception:
		pass
		except Exception:
		pass

		try:
		url = self.build_api_url(api_root, f"{federation_context_id}/partner")
		self.make_request_partner_op("DELETE", url, token=token)
		except Exception:
		pass

		for artefact_id in [self.artefact_id_pop, self.artefact_id_oop]:
		try:
		srm.delete_artefact(artefact_id)
		except Exception:
		pass

		def get_access_token(self):
		# URL to obtain access token from Keycloak
		token_url = f"http://{HOST_KEYCLOAK}:{PORT_KEYCLOAK}/realms/federation/protocol/openid-connect/token"

src/test/test_application_onboarding_management.py

+5 −23

Original line number	Diff line number	Diff line
		@@ -17,7 +17,7 @@
		from __future__ import absolute_import

		from test import BaseTestCase
		from clients import tf_sdk
		from clients import srm
		import unittest


		@@ -35,7 +35,7 @@ class TestApplicationOnboardingManagementController(BaseTestCase):
		try:
		# Check if there is connection with Edge Cloud Platform. Otherwise stop the test
		try:
		tf_sdk.get_zones()
		srm.get_zones()
		except Exception:
		self.skipTest("Edge Cloud Platform connection not available")

		@@ -63,27 +63,9 @@ class TestApplicationOnboardingManagementController(BaseTestCase):

		def test_00_setup(self):
		"""Setup federation contexts for both Partner OP and Originating OP roles"""
		try:
		# Check if there are artefact and profile created at Edge Cloud Platform and must be deleted because remains there due
		# to a possible issue during the test
		tf_sdk.delete_onboarding(self.app_id_pop)
		except Exception:
		pass

		try:
		tf_sdk.delete_onboarding(self.app_id_oop)
		except Exception:
		pass

		try:
		tf_sdk.delete_artefact(self.artefact_id_pop)
		except Exception:
		pass

		try:
		tf_sdk.delete_artefact(self.artefact_id_oop)
		except Exception:
		pass
		self.cleanup_fm_test_state()
		self.cleanup_ecp_test_state()
		self.cleanup_remote_partner_test_state(BaseTestCase.token)

		try:
		# Create federation context for Partner OP role