#
# ============LICENSE_END=========================================================
-import base64
-import os
+from abc import ABC
+from onapsdk.so.so_element import SoElement
+from onapsdk.onap_service import OnapService
+from onapsdk.utils.headers_creator import headers_so_creator
-from kubernetes import config, client
-from kubernetes.stream import stream
+class SoDBUpdate(SoElement, ABC):
-class SoDBAdapter:
+ @classmethod
+ def add_region_to_so_db(cls,
+ cloud_region_id: str,
+ complex_id: str,
+ identity_service_id: str = None,
+ **kwargs
+ ):
+ """Method to add cloud_site data with identity_service to SO db.
- def __init__(self, cloud_region_id, complex_id, onap_kubeconfig_path):
- self.CLOUD_REGION_ID = cloud_region_id
- self.COMPLEX_ID = complex_id
- self.ONAP_KUBECONFIG_PATH = onap_kubeconfig_path
- self.MYPATH = os.path.dirname(os.path.realpath(__file__))
+ Args:
+ cloud_region_id: the name of cloud region
+ complex_id: name of complex
+ identity_service_id: optional - id of identity service
+ **kwargs: keyword arguments with parameters for identity service creation, like below
- config.load_kube_config(config_file=os.path.join(self.MYPATH, self.ONAP_KUBECONFIG_PATH))
- self.api_instance = client.CoreV1Api()
- self.pod_name = self.get_mariadb_pod_name()
- self.password = self.get_mariadb_root_username_password()
+ Important:
+ identity_services data will be overwrite, but in the same time
+ cloud_sites data will not (shouldn't) be overwrite!
- def get_mariadb_pod_name(self):
- pods = self.api_instance.list_namespaced_pod(namespace="onap")
- for pod in pods.items:
- if pod.metadata.name.find("mariadb-galera-0") != -1:
- return pod.metadata.name
+ Return:
+ response object
+ """
- def get_mariadb_root_username_password(self):
- secrets = self.api_instance.list_namespaced_secret(namespace="onap")
- for secret in secrets.items:
- if secret.metadata.name.find("mariadb-galera-db-root-password") != -1:
- base64_password = secret.data["password"]
- base64_bytes = base64_password.encode('ascii')
- password_bytes = base64.b64decode(base64_bytes)
+ if not identity_service_id:
+ identity_service_id = 'Keystone_K8s'
- return password_bytes.decode('ascii')
+ # params for identity_service creation
+ orchestrator = kwargs.get('orchestrator', 'multicloud')
+ identity_url = kwargs.get('identity_url', "http://1.2.3.4:5000/v2.0")
+ mso_id = kwargs.get('mso_id', 'onapsdk_user')
+ mso_pass = kwargs.get('mso_pass', 'mso_pass_onapsdk')
+ project_domain_name = kwargs.get("project_domain_name", None)
+ user_domain_name = kwargs.get("user_domain_name", None)
+ member_role = kwargs.get('member_role', 'admin')
+ admin_tenant = kwargs.get('admin_tenant', 'service')
+ identity_server_type = kwargs.get('identity_server_type', 'KEYSTONE')
+ identity_authentication_type = kwargs.get('identity_authentication_type', 'USERNAME_PASSWORD')
- def run_exec_request(self, exec_command):
- response = stream(self.api_instance.connect_get_namespaced_pod_exec,
- name=self.pod_name,
- # container="container-name",
- namespace="onap",
- command=exec_command,
- stdin=False,
- tty=False,
- stderr=True,
- stdout=True)
- return response
-
- def check_region_in_db(self):
- exec_command = [
- "/bin/sh",
- "-c",
- f"mysql -uroot -p{self.password} catalogdb -e 'SELECT * FROM cloud_sites;'"]
- response = self.run_exec_request(exec_command)
-
- is_region_found = False
- for line in response.split("\n"):
- if line.split("\t")[0] == self.CLOUD_REGION_ID:
- print(line)
- is_region_found = True
- return is_region_found
- return is_region_found
-
- def add_region_to_so_db(self):
- exec_command = [
- "/bin/sh",
- "-c",
- f"mysql -uroot -p{self.password} catalogdb -e "
- f"'insert into cloud_sites(ID, REGION_ID, IDENTITY_SERVICE_ID, CLOUD_VERSION, CLLI, ORCHESTRATOR ) "
- f"values (\"{self.CLOUD_REGION_ID}\", \"{self.CLOUD_REGION_ID}\", \"DEFAULT_KEYSTONE\", \"2.5\", "
- f"\"{self.COMPLEX_ID}\", \"multicloud\");'"]
+ data = {
+ "id": cloud_region_id,
+ "region_id": cloud_region_id,
+ "aic_version": "2.5",
+ "clli": complex_id,
+ "orchestrator": orchestrator,
+ "identityService": {
+ "id": identity_service_id,
+ "identityServerTypeAsString": "KEYSTONE",
+ "hibernateLazyInitializer": {},
+ "identity_url": identity_url,
+ "mso_id": mso_id,
+ "mso_pass": mso_pass,
+ "project_domain_name": project_domain_name,
+ "user_domain_name": user_domain_name,
+ "admin_tenant": admin_tenant,
+ "member_role": member_role,
+ "tenant_metadata": True,
+ "identity_server_type": identity_server_type,
+ "identity_authentication_type": identity_authentication_type
+ }
+ }
- response = self.run_exec_request(exec_command)
+ response = cls.send_message(
+ "POST",
+ "Create a region in SO db",
+ f"{cls.base_url}/cloudSite",
+ json=data,
+ headers=headers_so_creator(OnapService.headers),
+ exception=ValueError)
return response
Code Review / demo.git / blobdiff