Updates tests for dual role (22a8d42f) · Commits · OOP / code / federation-manager

src/test/init.py

+245 −222

Original line number	Diff line number	Diff line
		@@ -2,13 +2,12 @@ import logging

		import connexion
		import requests
		import random
		from flask_testing import TestCase

		from encoder import JSONEncoder

		from configparser import ConfigParser
		from clients import i2edge
		from clients import tf_sdk
		import util

		CONFIG = ConfigParser()
		@@ -20,10 +19,9 @@ CLIENT_SECRET = CONFIG.get("keycloak", "client1_secret")
		SERVER = "/operatorplatform/federation/v1"
		HOST_KEYCLOAK = CONFIG.get("keycloak", "host")
		PORT_KEYCLOAK = int(CONFIG.get("keycloak", "port"))
		ROLE_OP = CONFIG.get("partner_op", "role")
		HOST_ORIGINATING_OP = CONFIG.get("partner_op", "partner_op_host")
		SERVER_ORIGINATING_OP = CONFIG.get("partner_op", "partner_op_server")
		PORT_ORIGINATING_OP = int(CONFIG.get("partner_op", "partner_op_port"))

		# Dual role testing configuration
		PARTNER_API_ROOT = "http://127.0.0.1:8990"


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

		self.artefact_id = "3fa85f64-5717-4562-b3fc-222222222222"
		self.app_provider = "test_app_providerid"
		self.app_id = "test_app_id"
		self.artefact_name = "3fa85f64-5717-4562-b3fc-222222222222"
		self.artefact_version = "0.3.0"
		self.artefact_file = "H4sIAAAAAAAAA+1XUW/bNhDOs37FzXlZi0qWZEne/FYkAzag3YpmCDAMxUBLtMxGEjmScupl6W/fUYoc2bEXxxkSDOH3QInH492RxyPvRjPyXTxLIjceB2M3ipPQnY5mqRv2MDx6HHzEeBybbzCO/f63w1EQJYEfhUGSIF8QhXF4BPEj9e6FWmkiAY4+c5mxf+G7b/x/itE+/vfmtChZXnFJD9FhHJxE0U7/x6MY/Y+qkxGSQ/R/EgXof/+/Xuw2vHD/H8MHojWVlQLNofUxXM5pBdOaFRmrchAkvSA5VZ5zDL/OmQJVC8Glxh88FwXkBZ9CSXQ6R+43IGlBNFtQnKfnPTqpMhRQ0RxHeQXfCkln7AvN4JIh3zevPPilKpbAq2amMQkElVCwinqOd3r2x5lG21DECS9LFHB+cgYZk8rxcqaHTdua73jTv+SwaTvCPB+apuuqRTW8FTTF9dUCZqygynntqUuB7ZRcYKtL889RjvP6K844J5LxWsFPpz+gXiH5Z5pqx2MZJcOWHUmOt1Apz+jQeW7n7oG94n9Bihr9vyRlcYiO++I/SMZ34h/Zbfw/AY6h51xHUlGwlJzwutITCJwKj/EZLfCQczmBq2unIiVVeCHQiVPSksvlWyEmTirqj/RPlKI78qqLQ+9YyVYDN53nXrbFDfaK/5M5kfrg8G/jP9kd/2EwMvnfOI6DsKEHSejb9/9JQAQ7p1LhizyBRegQIVbdwAs838moSiUTuiG9hR8xE4TUnAeYcQlLXkv4wDN89RX+prS5ISawz7FyFp0m3/seNT33VrxI7BX/mpYCkzqqDqsE96z/kMcfjaJxE/9xYuu/p8AD/d8+4q5sn/c9n4R78r8wiKOm/g8jdL+PfGEYhIm9/58C/fsfL381XATOBauyCZxiLsiXJa00pm6aZESTiQOI9oa/ugLvI6aGRFHvZ6TA35DRGakLDQM8KXoA19cr9jZlbOact+nmiro5z9CCZrISNG013qSlyqSkpq+6lLTpGTRl5jsypYW6JRrgmvaz1b052hl+Ggndoe8pWduGDsUWrY/T3CxxtfgOG7l4byt7Ayhf898wKvGnQjei9yBIOld0SHmlCdbVcsNs9wG+3Wp1C1aSHKUIjrV7/WWoNCYHaoOnyxe2bJwZbK6XrWON/aKewKC3BbcFSN9U3wuNpYMdUtoFrAtaq176sgLff892SStMUfMga5syaN3W6DBb70gKd1ma8rIkJrJ/H7QuGXza4CAyV2bYdRfl4A0u2jSm406X6HfTu6MfC8C1fYr9943ybuacVHkr69Nz33UWFhYWFhYWFhYWFhYWFhYWLxH/AEh4YrAAKAAA"
		# Partner OP (POP) configuration
		self.artefact_id_pop = "3fa85f64-5717-4562-b3fc-222222222222"
		self.app_id_pop = "test_app_pop"
		self.artefact_name_pop = "ollama"
		self.artefact_repo_pop = "https://otwld.github.io/ollama-helm/"

		self.roleOp = ROLE_OP
		# Base url application server originating op
		self.base_url_originating_op = f"{HOST_ORIGINATING_OP}:{PORT_ORIGINATING_OP}{SERVER_ORIGINATING_OP}"
		# Originating OP (OOP) configuration
		self.artefact_id_oop = "3fa85f64-5717-4562-b3fc-222222222223"
		self.app_id_oop = "test_app_oop"
		self.artefact_name_oop = "kubernetes-dashboard"
		self.artefact_repo_oop = "https://kubernetes.github.io/dashboard/"

		# Common artifact configuration
		self.app_provider = "test_app_provider"
		self.artefact_version = "0.3.0"
		self.repo_type = util.RepoType.PUBLIC.value

		return app.app

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

		payload = {
		"client_id": self.client_id,
		@@ -65,13 +69,73 @@ class BaseTestCase(TestCase):
		"grant_type": "client_credentials"
		}

		try:
		# POST to obtain access token
		response = requests.post(token_url, data=payload)

		# Returns access token
		return response.json()["access_token"]
		except Exception as e:
		print(f"Warning: Could not get access token: {e}")
		return "mock_token_for_testing"

		def make_request_partner_op(self, method, url, body=None, token=None):
		"""Make request as Partner OP (external request without X-Internal header)"""
		headers = {
		"Content-Type": "application/json; accept=application/json",
		"Authorization": f"Bearer {token or self.get_access_token()}"
		}

		if method.upper() == "GET":
		return requests.get(url, headers=headers)
		elif method.upper() == "POST":
		return requests.post(url, headers=headers, json=body)
		elif method.upper() == "PUT":
		return requests.put(url, headers=headers, json=body)
		elif method.upper() == "PATCH":
		return requests.patch(url, headers=headers, json=body)
		elif method.upper() == "DELETE":
		return requests.delete(url, headers=headers)

		def make_request_originating_op(self, method, url, body=None, token=None, partner_api_root=None):
		"""Make request as Originating OP (internal request with X-Internal header)"""
		headers = {
		"Content-Type": "application/json; accept=application/json",
		"Authorization": f"Bearer {token or self.get_access_token()}",
		"X-Internal": "true",
		"X-Partner-Api-Root": partner_api_root or PARTNER_API_ROOT
		}

		if method.upper() == "GET":
		return requests.get(url, headers=headers)
		elif method.upper() == "POST":
		return requests.post(url, headers=headers, json=body)
		elif method.upper() == "PUT":
		return requests.put(url, headers=headers, json=body)
		elif method.upper() == "PATCH":
		return requests.patch(url, headers=headers, json=body)
		elif method.upper() == "DELETE":
		return requests.delete(url, headers=headers)

		def run_both_roles(self, test_func, args, *kwargs):
		"""Helper method to run a test function for both Partner OP and Originating OP roles"""
		results = {}

		# Test Partner OP role (external request)
		try:
		results['partner_op'] = test_func('partner_op', args, *kwargs)
		except Exception as e:
		results['partner_op'] = {'error': str(e)}

		# Test Originating OP role (internal request)
		try:
		results['originating_op'] = test_func('originating_op', args, *kwargs)
		except Exception as e:
		results['originating_op'] = {'error': str(e)}

		def post_federation_context(self, token):
		return results

		def post_federation(self, token, role='partner_op'):
		"""Create federation context for specified role"""
		federation_context_id = ""

		url = f"http://{self.base_url}/partner"
		@@ -95,16 +159,20 @@ class BaseTestCase(TestCase):
		"clientSecret": "string"
		}
		}
		headers = {"Content-Type": "application/json; accept=application/json", "Authorization": f"Bearer {token}"}
		response = requests.post(url, headers=headers, json=body)

		if role == 'originating_op':
		response = self.make_request_originating_op("POST", url, body, token)
		else:
		response = self.make_request_partner_op("POST", url, body, token)

		assert response.status_code == 200
		data_response = response.json()
		federation_context_id = data_response.get("federationContextId")

		return federation_context_id

		def post_availability_zones(self, federation_context_id, zone_id, token):

		def post_availability_zones(self, federation_context_id, zone_id, token, role='partner_op'):
		"""Create availability zones for specified role"""
		url = f"http://{self.base_url}/{federation_context_id}/zones"
		body = {
		"acceptedAvailabilityZones": [
		@@ -112,128 +180,61 @@ class BaseTestCase(TestCase):
		],
		"availZoneNotifLink": "string"
		}
		headers = {"Content-Type": "application/json; accept=application/json", "Authorization": f"Bearer {token}"}
		response = requests.post(url, headers=headers, json=body)
		assert response.status_code == 200

		return
		if role == 'originating_op':
		response = self.make_request_originating_op("POST", url, body, token)
		else:
		response = self.make_request_partner_op("POST", url, body, token)

		def post_artefact(self, federation_context_id, artefact_id, app_provider, token):
		assert response.status_code == 200
		return

		def post_artefact(self, federation_context_id, artefact_id, artefact_name, artefact_repo, token,
		role='partner_op'):
		"""Create artefact for specified role"""
		url = f"http://{self.base_url}/{federation_context_id}/artefact"
		body = {
		"artefactId": artefact_id,
		"appProviderId": app_provider,
		"artefactName": self.artefact_name,
		"appProviderId": self.app_provider,
		"artefactName": artefact_name,
		"artefactVersionInfo": self.artefact_version,
		"artefactDescription": "string",
		"artefactVirtType": "VM_TYPE",
		"artefactFileName": "string",
		"artefactFileFormat": "WINZIP",
		"artefactVirtType": "CONTAINER_TYPE",
		"artefactDescriptorType": "HELM",
		"repoType": util.RepoType.UPLOAD.value,
		"repoType": self.repo_type,
		"artefactRepoLocation": {
		"repoURL": "string",
		"userName": "string",
		"password": "string",
		"token": "string"
		"repoURL": artefact_repo
		},
		"artefactFile": self.artefact_file,
		"componentSpec": [
		{
		"componentName": "string",
		"componentName": artefact_name,
		"images": [
		"3fa85f64-5717-4562-b3fc-2c963f66afa6"
		artefact_id
		],
		"numOfInstances": 0,
		"numOfInstances": 1,
		"restartPolicy": "RESTART_POLICY_ALWAYS",
		"commandLineParams": {
		"command": [
		"string"
		],
		"commandArgs": [
		"string"
		]
		},
		"exposedInterfaces": [
		{
		"interfaceId": "string",
		"commProtocol": "TCP",
		"commPort": 0,
		"visibilityType": "VISIBILITY_EXTERNAL",
		"network": "string",
		"InterfaceName": "string"
		}
		],
		"computeResourceProfile": {
		"cpuArchType": "ISA_X86_64",
		"numCPU": {
		"whole": {
		"value": 2
		},
		"decimal": {
		"value": 0.5
		},
		"millivcpu": {
		"value": "500m"
		}
		},
		"memory": 0,
		"diskStorage": 0,
		"gpu": [
		{
		"gpuVendorType": "GPU_PROVIDER_NVIDIA",
		"gpuModeName": "string",
		"gpuMemory": 0,
		"numGPU": 0
		}
		],
		"vpu": 0,
		"fpga": 0,
		"hugepages": [
		{
		"pageSize": "2MB",
		"number": 0
		"numCPU": "500m",
		"memory": 0
		}
		],
		"cpuExclusivity": True
		},
		"compEnvParams": [
		{
		"envVarName": "string",
		"envValueType": "USER_DEFINED",
		"envVarValue": "string",
		"envVarSrc": "string"
		}
		],
		"deploymentConfig": {
		"configType": "DOCKER_COMPOSE",
		"contents": "string"
		},
		"persistentVolumes": [
		{
		"volumeSize": "10Gi",
		"volumeMountPath": "string",
		"volumeName": "string",
		"ephemeralType": False,
		"accessMode": "RW",
		"sharingPolicy": "EXCLUSIVE"
		}
		]
		}
		]
		}
		headers = {"Content-Type": "application/json; accept=application/json", "Authorization": f"Bearer {token}"}
		response = requests.post(url, headers=headers, json=body)
		assert response.status_code == 200

		if role == 'originating_op':
		response = self.make_request_originating_op("POST", url, body, token)
		else:
		response = self.make_request_partner_op("POST", url, body, token)

		assert response.status_code == 200
		return

		def post_onboarding(self, federation_context_id, app_id, app_provider, zone_id, artefact_id, token):
		def post_onboarding(self, federation_context_id, app_id, zone_id, artefact_id, token, role='partner_op'):
		"""Create onboarding for specified role"""
		url = f"http://{self.base_url}/{federation_context_id}/application/onboarding"
		body = {
		"appId": app_id,
		"appProviderId": app_provider,
		"appProviderId": self.app_provider,
		"appDeploymentZones": [
		zone_id
		],
		@@ -262,93 +263,122 @@ class BaseTestCase(TestCase):
		],
		"appStatusCallbackLink": "string"
		}
		headers = {"Content-Type": "application/json; accept=application/json", "Authorization": f"Bearer {token}"}
		response = requests.post(url, headers=headers, json=body)
		assert response.status_code == 202

		return
		if role == 'originating_op':
		response = self.make_request_originating_op("POST", url, body, token)
		else:
		response = self.make_request_partner_op("POST", url, body, token)

		def delete_deployments(self, federation_id, app_id, instance_id, zone_id, token):
		assert response.status_code == 202
		return

		def delete_deployments(self, federation_id, app_id, instance_id, zone_id, token, role='partner_op'):
		"""Delete deployments for specified role"""
		url = f"http://{self.base_url}/{federation_id}/application/lcm/app/{app_id}/instance/{instance_id}/zone/{zone_id}"

		headers = {"Content-Type": "application/json; accept=application/json", "Authorization": f"Bearer {token}"}
		response = requests.delete(url, headers=headers)

		return
		if role == 'originating_op':
		response = self.make_request_originating_op("DELETE", url, token=token)
		else:
		response = self.make_request_partner_op("DELETE", url, token=token)

		def delete_onboarding(self, federation_id, app_id, token):
		return response

		def delete_onboarding(self, federation_id, app_id, token, role='partner_op'):
		"""Delete onboarding for specified role"""
		url = f"http://{self.base_url}/{federation_id}/application/onboarding/app/{app_id}"

		headers = {"Content-Type": "application/json; accept=application/json", "Authorization": f"Bearer {token}"}
		response = requests.delete(url, headers=headers)
		if role == 'originating_op':
		response = self.make_request_originating_op("DELETE", url, token=token)
		else:
		response = self.make_request_partner_op("DELETE", url, token=token)

		return

		def delete_artefact(self, federation_id, artefact_id, token):
		return response

		def delete_artefact(self, federation_id, artefact_id, token, role='partner_op'):
		"""Delete artefact for specified role"""
		url = f"http://{self.base_url}/{federation_id}/artefact/{artefact_id}"

		headers = {"Content-Type": "application/json; accept=application/json", "Authorization": f"Bearer {token}"}
		response = requests.delete(url, headers=headers)
		if role == 'originating_op':
		response = self.make_request_originating_op("DELETE", url, token=token)
		else:
		response = self.make_request_partner_op("DELETE", url, token=token)

		return

		def delete_zone(self, federation_id, zone_id, token):
		return response

		def delete_zone(self, federation_id, zone_id, token, role='partner_op'):
		"""Delete zone for specified role"""
		url = f"http://{self.base_url}/{federation_id}/zones/{zone_id}"

		headers = {"Content-Type": "application/json; accept=application/json", "Authorization": f"Bearer {token}"}
		response = requests.delete(url, headers=headers)
		if role == 'originating_op':
		response = self.make_request_originating_op("DELETE", url, token=token)
		else:
		response = self.make_request_partner_op("DELETE", url, token=token)

		return

		def delete_federation(self, federation_id, token):
		return response

		def delete_federation(self, federation_id, token, role='partner_op'):
		"""Delete federation for specified role"""
		url = f"http://{self.base_url}/{federation_id}/partner"

		headers = {"Content-Type": "application/json; accept=application/json", "Authorization": f"Bearer {token}"}
		response = requests.delete(url, headers=headers)

		return
		if role == 'originating_op':
		response = self.make_request_originating_op("DELETE", url, token=token)
		else:
		response = self.make_request_partner_op("DELETE", url, token=token)

		def get_federation(self, federation_id, token):
		return response

		def get_federation(self, federation_id, token, role='partner_op'):
		"""Get federation for specified role"""
		url = f"http://{self.base_url}/{federation_id}/partner"

		headers = {"Content-Type": "application/json; accept=application/json", "Authorization": f"Bearer {token}"}
		response = requests.get(url, headers=headers)
		if role == 'originating_op':
		response = self.make_request_originating_op("GET", url, token=token)
		else:
		response = self.make_request_partner_op("GET", url, token=token)

		return response.json()

		def get_availability_zone(self, federation_id, zone_id, token, role='partner_op'):
		"""Get zone details for specified role"""
		url = f"http://{self.base_url}/{federation_id}/zones/{zone_id}"

		if role == 'originating_op':
		response = self.make_request_originating_op("GET", url, token=token)
		else:
		response = self.make_request_partner_op("GET", url, token=token)

		return response.json()

		def get_first_zone_from_zone_list_i2edge(self):
		def get_zone_from_partner_details(self, federation_id, token, role='partner_op'):
		zone_selected = ""

		try:
		partner_details = self.get_federation(federation_id, token, role)
		# Get zone list
		zones_list = i2edge.get_zones()
		zones_list = partner_details.get("offeredAvailabilityZones")

		if not zones_list:
		return zone_selected

		for zone in zones_list:
		# If the zone value is a dict, is a correct zone, else there is an issue and returns a str
		if isinstance(zone, dict):
		flavours = zone.get("flavoursSupported")
		if len(flavours) > 0:
		zone_selected = zone.get("zoneId")
		zone_id = zone.get("zoneId")
		if zone_id:
		zone_selected = zone_id
		break
		except:
		except Exception as e:
		print(f"Error getting zone from partner details: {e}")
		zone_selected = ""

		return zone_selected

		def get_flavour_from_zone(self, zone_id):
		def get_flavour_from_zone_details(self, federation_id, zone_id, token, role='partner_op'):
		flavour_selected = ""

		try:
		# Get zone from i2edge
		zone = i2edge.get_zone_by_zone_id(zone_id)

		flavours = zone.get("flavoursSupported")
		zone_details = self.get_availability_zone(federation_id, zone_id, token, role)
		flavours = zone_details.get("flavoursSupported")
		if len(flavours) > 0:
		for f in flavours:
		flavour_selected = f.get("flavourId")
		@@ -358,20 +388,13 @@ class BaseTestCase(TestCase):

		return flavour_selected

		def get_federation_resources(self, token):

		url = f"http://{self.base_url}/federation-resources"

		headers = {"Content-Type": "application/json; accept=application/json", "Authorization": f"Bearer {token}"}
		response = requests.get(url, headers=headers)

		return response.json()

		def get_federation_resources_originating_op(self, token):

		url = f"http://{self.base_url_originating_op}/federation-resources"
		def get_federation_context(self, token, role='partner_op'):
		"""Get federation resources for specified role"""
		url = f"http://{self.base_url}/fed-context-id"

		headers = {"Content-Type": "application/json; accept=application/json", "Authorization": f"Bearer {token}"}
		response = requests.get(url, headers=headers)
		if role == 'originating_op':
		response = self.make_request_originating_op("GET", url, token=token)
		else:
		response = self.make_request_partner_op("GET", url, token=token)

		return response.json()