Modify Python ONAP SDK tests to not require kubeconfig

Modify Python ONAP SDK tests to not require kubeconfig

Issue-ID: INT-2247
Signed-off-by: pawel.denst <pawel.denst@external.t-mobile.pl>
Signed-off-by: Lukasz Rajewski <lukasz.rajewski@t-mobile.pl>
Change-Id: I0d61250a2373d8549696cedaa2c0c5fb721d0e3b

6 files changed, 62 insertions, 14 deletions

diff --git a/src/onaptests/configuration/settings.py b/src/onaptests/configuration/settings.py
index 436e82c..83ae792 100644
--- a/src/onaptests/configuration/settings.py
+++ b/src/onaptests/configuration/settings.py
@@ -53,4 +53,5 @@ K8S_ADDITIONAL_RESOURCES_NAMESPACE = K8S_ONAP_NAMESPACE  # Resources created on
 ORCHESTRATION_REQUEST_TIMEOUT = 60.0 * 15  # 15 minutes in seconds
 SERVICE_DISTRIBUTION_NUMBER_OF_TRIES = 30
 SERVICE_DISTRIBUTION_SLEEP_TIME = 60
-EXPOSE_SERVICES_NODE_PORTS = True
\ No newline at end of file
+EXPOSE_SERVICES_NODE_PORTS = True
+IN_CLUSTER = False
\ No newline at end of file
diff --git a/src/onaptests/steps/cloud/check_status.py b/src/onaptests/steps/cloud/check_status.py
index a72c992..80205b5 100644
--- a/src/onaptests/steps/cloud/check_status.py
+++ b/src/onaptests/steps/cloud/check_status.py
@@ -120,10 +120,10 @@ class CheckNamespaceStatusStep(BaseStep):
         else:
             self.res_dir = f"{testcase.TestCase.dir_results}/kubernetes-status"
 
-        if settings.K8S_CONFIG:
-            config.load_kube_config(config_file=settings.K8S_CONFIG)
+        if settings.IN_CLUSTER:
+            config.load_incluster_config()
         else:
-            config.load_kube_config()
+            config.load_kube_config(config_file=settings.K8S_CONFIG)
 
         self.core = client.CoreV1Api()
         self.batch = client.BatchV1Api()
diff --git a/src/onaptests/steps/cloud/k8s_connectivity_info_create.py b/src/onaptests/steps/cloud/k8s_connectivity_info_create.py
index dfcda05..8c361e4 100644
--- a/src/onaptests/steps/cloud/k8s_connectivity_info_create.py
+++ b/src/onaptests/steps/cloud/k8s_connectivity_info_create.py
@@ -1,9 +1,12 @@
 """Connectivity info creation module."""
+from jinja2 import Environment, PackageLoader, select_autoescape
+
 from onapsdk.configuration import settings
 from onapsdk.exceptions import APIError
 from onapsdk.k8s import ConnectivityInfo
+from onapsdk.utils.jinja import jinja_env
+from onaptests.steps.base import BaseStep
 
-from ..base import BaseStep
 
 class K8SConnectivityInfoStep(BaseStep):
     """CreateConnnectivityInfoStep."""
@@ -31,17 +34,34 @@ class K8SConnectivityInfoStep(BaseStep):
         ######## Create Connectivity Info #########################################
         try:
             self._logger.info("Check if k8s connectivity information exists")
-            ConnectivityInfo.get_connectivity_info_by_region_id(settings.CLOUD_REGION_ID)
+            ConnectivityInfo.get_connectivity_info_by_region_id(
+                settings.CLOUD_REGION_ID)
         except APIError:
-            self._logger.info("Create the k8s connectivity information")
-            ConnectivityInfo.create(settings.CLOUD_REGION_ID,
-                                    settings.CLOUD_REGION_CLOUD_OWNER,
-                                    open(settings.K8S_CONFIG, 'rb').read())
+            if settings.IN_CLUSTER:
+                token_file = "/var/run/secrets/kubernetes.io/serviceaccount/token"
+                with open(token_file, "r") as file:
+                    user_token_value = file.read().strip()
+                jinja_env = Environment(autoescape=select_autoescape(['json.j2']),
+                                        loader=PackageLoader('onaptests.templates', 'kubeconfig'))
+                kubeconfig_data = jinja_env.get_template("kube_config.json.j2").render(
+                    user_token_value=user_token_value
+                )
+                # Create the k8s connectivity information with the kubeconfig data
+                self._logger.info("Create the k8s connectivity information")
+                ConnectivityInfo.create(settings.CLOUD_REGION_ID,
+                                        settings.CLOUD_REGION_CLOUD_OWNER,
+                                        kubeconfig_data.encode('utf-8'))
+            else:
+                self._logger.info("Create the k8s connectivity information")
+                ConnectivityInfo.create(settings.CLOUD_REGION_ID,
+                                        settings.CLOUD_REGION_CLOUD_OWNER,
+                                        open(settings.K8S_CONFIG, 'rb').read())
 
     @BaseStep.store_state(cleanup=True)
     def cleanup(self) -> None:
         """Cleanup K8S Connectivity information."""
         self._logger.info("Clean the k8s connectivity information")
-        connectinfo = ConnectivityInfo.get_connectivity_info_by_region_id(settings.CLOUD_REGION_ID)
+        connectinfo = ConnectivityInfo.get_connectivity_info_by_region_id(
+            settings.CLOUD_REGION_ID)
         connectinfo.delete()
         super().cleanup()
diff --git a/src/onaptests/steps/onboard/cds.py b/src/onaptests/steps/onboard/cds.py
index ad40fe5..5256eac 100644
--- a/src/onaptests/steps/onboard/cds.py
+++ b/src/onaptests/steps/onboard/cds.py
@@ -34,7 +34,10 @@ class ExposeCDSBlueprintprocessorNodePortStep(CDSBaseStep):
         """Initialize step."""
         super().__init__(cleanup=cleanup)
         self.service_name: str = "cds-blueprints-processor-http"
-        config.load_kube_config(settings.K8S_CONFIG)
+        if settings.IN_CLUSTER:
+            config.load_incluster_config()
+        else:
+            config.load_kube_config(config_file=settings.K8S_CONFIG)
         self.k8s_client: client.CoreV1Api = client.CoreV1Api()
 
     @property
diff --git a/src/onaptests/steps/simulator/pnf_simulator_cnf/pnf_register.py b/src/onaptests/steps/simulator/pnf_simulator_cnf/pnf_register.py
index 0e6e001..3a846e3 100644
--- a/src/onaptests/steps/simulator/pnf_simulator_cnf/pnf_register.py
+++ b/src/onaptests/steps/simulator/pnf_simulator_cnf/pnf_register.py
@@ -46,7 +46,10 @@ class PnfSimulatorCnfRegisterStep(BaseStep):
             bool: True if PNF simulator pod is running, False otherwise
 
         """
-        config.load_kube_config(settings.K8S_CONFIG)
+        if settings.IN_CLUSTER:
+            config.load_incluster_config()
+        else:
+            config.load_kube_config(config_file=settings.K8S_CONFIG)
         k8s_client: "CoreV1API" = client.CoreV1Api()
         k8s_watch: "Watch" =  watch.Watch()
         status = False
@@ -75,7 +78,10 @@ class PnfSimulatorCnfRegisterStep(BaseStep):
             Tuple[str, str, str]: VES protocol, IP and port
 
         """
-        config.load_kube_config(settings.K8S_CONFIG)
+        if settings.IN_CLUSTER:
+            config.load_incluster_config()
+        else:
+            config.load_kube_config(config_file=settings.K8S_CONFIG)
         k8s_client: "CoreV1API" = client.CoreV1Api()
         try:
             for service in k8s_client.list_namespaced_service(namespace=settings.K8S_ONAP_NAMESPACE).items:
diff --git a/src/onaptests/templates/kubeconfig/kube_config.json.j2 b/src/onaptests/templates/kubeconfig/kube_config.json.j2
new file mode 100644
index 0000000..9614e2f
--- /dev/null
+++ b/src/onaptests/templates/kubeconfig/kube_config.json.j2
@@ -0,0 +1,18 @@
+apiVersion: v1
+kind: Config
+current-context: default
+contexts:
+- name: default
+context:
+    cluster: cluster
+    user: cluster-admin
+    namespace: default
+clusters:
+- name: cluster
+cluster:
+    insecure-skip-tls-verify: true
+    server: https://kubernetes.default.svc.cluster.local
+users:
+- name: cluster-admin
+user:
+token: {{ user_token_value }}
\ No newline at end of file