introduce kcp into testsuite

config/settings.local.yaml.tpl

@@ -1,11 +1,14 @@
 #default:
 #  skip_cleanup: false
-#  openshift:
+#  openshift:                                  # Required for kuadrant tests: if no further configuration needed pass an empty dict
 #    project: "kuadrant"                       # Optional: namespace for tests to run, if None uses current project
 #    api_url: "https://api.openshift.com"      # Optional: OpenShift API URL, if None it will OpenShift that you are logged in
 #    token: "KUADRANT_RULEZ"                   # Optional: OpenShift Token, if None it will OpenShift that you are logged in
 #  openshift2:
 #    project: "kuadrant2"                      # Required: Secondary OpenShift project, for running tests across projects
+#  kcp:                                        # Required for glbc tests: if no further configuration needed pass en empty dict
+#    namespace: "glbc-test"                    # Optional: namespace for tests to run, if None uses current namespace
+#    kubeconfig_path: "~/.kube/config"         # Optional: kubeconfig to use for kcp if None default is used
 #  tools:
 #    project: "tools"                          # Optional: OpenShift project, where external tools are located
 #  rhsso:

config/settings.yaml

@@ -1,6 +1,7 @@
 default:
   skip_cleanup: false
   dynaconf_merge: true
+  openshift: {}
   tools:
     project: "tools"
   cfssl: "cfssl"

testsuite/config/__init__.py

@@ -20,6 +20,8 @@ def __init__(self, name, default, **kwargs) -> None:
                          **kwargs)
 
 
+_openshift_validator = Validator("openshift", eq=None)
+
 settings = Dynaconf(
     environments=True,
     lowercase_read=True,
@@ -28,10 +30,10 @@ def __init__(self, name, default, **kwargs) -> None:
     envvar_prefix="KUADRANT",
     merge_enabled=True,
     validators=[
-        Validator("authorino.deploy", eq=True) | Validator("authorino.url", must_exist=True),
-        DefaultValueValidator("rhsso.url", must_exist=True, default=fetch_route("no-ssl-sso")),
-        DefaultValueValidator("rhsso.password",
-                              must_exist=True, default=fetch_secret("credential-sso", "ADMIN_PASSWORD")),
+        _openshift_validator | Validator("authorino.url", must_exist=True),
+        _openshift_validator | DefaultValueValidator("rhsso.url", must_exist=True, default=fetch_route("no-ssl-sso")),
+        _openshift_validator | DefaultValueValidator("rhsso.password", must_exist=True,
+                                                     default=fetch_secret("credential-sso", "ADMIN_PASSWORD")),
     ],
     loaders=["testsuite.config.openshift_loader", "dynaconf.loaders.env_loader"]
 )

testsuite/config/openshift_loader.py

@@ -8,19 +8,32 @@
 def load(obj, env=None, silent=True, key=None, filename=None):
     """Creates all OpenShift clients"""
     config = weakget(obj)
-    section = config["openshift"]
-    client = OpenShiftClient(
-        section["project"] % None,
-        section["api_url"] % None,
-        section["token"] % None)
+    if "openshift" in obj:
+        section = config["openshift"]
+        client = OpenShiftClient(
+            section["project"] % None,
+            section["api_url"] % None,
+            section["token"] % None)
+    else:
+        client = None
     obj["openshift"] = client
 
     tools = None
-    if "tools" in obj and "project" in obj["tools"]:
+    if "tools" in obj and "project" in obj["tools"] and client:
         tools = client.change_project(obj["tools"]["project"])
     obj["tools"] = tools
 
     openshift2 = None
-    if "openshift2" in obj and "project" in obj["openshift2"]:
+    if "openshift2" in obj and "project" in obj["openshift2"] and client:
         openshift2 = client.change_project(obj["openshift2"]["project"])
     obj["openshift2"] = openshift2
+
+    kcp = None
+    if "kcp" in obj and "namespace" in obj["kcp"]:
+        kcp_section = config["kcp"]
+        kcp = OpenShiftClient(
+            project=kcp_section["namespace"] % None,
+            kubeconfig_path=kcp_section["kubeconfig_path"] % None
+        )
+        kcp.is_ready = lambda _: True
+    obj["kcp"] = kcp

testsuite/openshift/client.py

@@ -29,22 +29,26 @@ class OpenShiftClient:
 
     # pylint: disable=too-many-public-methods
 
-    def __init__(self, project: str, api_url: str = None, token: str = None):
+    def __init__(self, project: str, api_url: str = None, token: str = None, kubeconfig_path: str = None):
         self._project = project
         self._api_url = api_url
         self.token = token
+        self._kubeconfig_path = kubeconfig_path
 
     def change_project(self, project):
         """Return new OpenShiftClient with a different project"""
-        return OpenShiftClient(project, self._api_url, self.token)
+        return OpenShiftClient(project, self._api_url, self.token, self._kubeconfig_path)
 
     @cached_property
     def context(self):
         """Prepare context for command execution"""
         context = Context()
+
         context.project_name = self._project
         context.api_url = self._api_url
         context.token = self.token
+        context.kubeconfig_path = self._kubeconfig_path
+
         return context
 
     @property

testsuite/tests/conftest.py

@@ -36,6 +36,8 @@ def testconfig():
 def openshift(testconfig):
     """OpenShift client for the primary namespace"""
     client = testconfig["openshift"]
+    if client is None:
+        pytest.skip("Openshift required but was not configured")
     if not client.connected:
         pytest.fail("You are not logged into Openshift or the namespace doesn't exist")
     return client
@@ -52,6 +54,20 @@ def openshift2(testconfig):
     return client
 
 
+@pytest.fixture(scope="session")
+def kcp(testconfig):
+    """Modified OpenShift client acting as Kcp client"""
+    client = testconfig["kcp"]
+    if client is None:
+        pytest.skip("Kcp required but was not configured")
+
+    # does not work for kcp yet
+    # if not client.connected:
+    #     pytest.fail("You are not logged into Openshift or the namespace for Kcp doesn't exist")
+
+    return client
+
+
 @pytest.fixture(scope="session")
 def rhsso(request, testconfig, blame):
     """RHSSO OIDC Provider fixture"""