Merge pull request #99 from ficap/glbc

Preparations for glbc tests

config/settings.local.yaml.tpl

@@ -4,8 +4,11 @@
 #    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
+#    kubeconfig_path: "~/.kube/config"         # Optional: Kubeconfig to use, if None the default one is used
 #  openshift2:
 #    project: "kuadrant2"                      # Required: Secondary OpenShift project, for running tests across projects
+#  kcp:                                        # Required for glbc tests
+#    project: "glbc-test"                      # Optional: namespace for glbc tests to run, if None uses current namespace
 #  tools:
 #    project: "tools"                          # Optional: OpenShift project, where external tools are located
 #  rhsso:

testsuite/config/openshift_loader.py

@@ -12,7 +12,9 @@ def load(obj, env=None, silent=True, key=None, filename=None):
     client = OpenShiftClient(
         section["project"] % None,
         section["api_url"] % None,
-        section["token"] % None)
+        section["token"] % None,
+        section["kubeconfig_path"] % None
+    )
     obj["openshift"] = client
 
     tools = None
@@ -24,3 +26,12 @@ def load(obj, env=None, silent=True, key=None, filename=None):
     if "openshift2" in obj and "project" in obj["openshift2"]:
         openshift2 = client.change_project(obj["openshift2"]["project"])
     obj["openshift2"] = openshift2
+
+    kcp = None
+    if "kcp" in obj and "project" in obj["kcp"]:
+        kcp_section = config["kcp"]
+        kcp = client.change_project(kcp_section["project"] % None)
+        # when advanced scheduling is enabled on kcp/syncer, status field is not synced back from workload cluster
+        # deployment, is_ready method depends on status field that is not available yet hence we have to mock it
+        kcp.is_ready = lambda _: True
+    obj["kcp"] = kcp

testsuite/openshift/client.py

@@ -28,22 +28,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
@@ -108,6 +112,7 @@ def new_app(self, source, params: Dict[str, str] = None):
 
         if os.path.isfile(source):
             source = f"--filename={source}"
+            opt_args.append("--local=true")
         objects = self.do_action("process", source, opt_args).out()
         with self.context:
             created = oc.create(objects)

testsuite/openshift/objects/ingress.py

@@ -0,0 +1,74 @@
+"""Kubernetes Ingress object"""
+from typing import Any, Dict, List, Optional, TYPE_CHECKING
+
+from testsuite.objects import LifecycleObject
+from testsuite.openshift.objects import OpenShiftObject
+
+if TYPE_CHECKING:
+    # pylint: disable=cyclic-import
+    from testsuite.openshift.client import OpenShiftClient
+
+
+class Ingress(OpenShiftObject, LifecycleObject):
+    """Represents Kubernetes Ingress object"""
+
+    @classmethod
+    def create_instance(cls, openshift: 'OpenShiftClient', name, rules: Optional[List[Dict[str, Any]]] = None,
+                        tls: Optional[List[Dict[str, Any]]] = None):
+        """Creates base instance"""
+        if rules is None:
+            rules = []
+
+        if tls is None:
+            tls = []
+
+        model: Dict[str, Any] = {
+            "apiVersion": "networking.k8s.io/v1",
+            "kind": "Ingress",
+            "metadata": {
+                "name": name,
+                "namespace": openshift.project
+            },
+            "spec": {
+                "rules": rules,
+                "tls": tls
+            }
+        }
+
+        with openshift.context:
+            return cls(model)
+
+    @classmethod
+    def create_service_ingress(cls, openshift: 'OpenShiftClient', name, service_name, port_number=80, path="/",
+                               path_type="Prefix", host=None):
+        """
+        Creates Ingress instance for service without tls configured
+        """
+        rule = {
+            "http": {
+                "paths": [
+                    {
+                        "backend": {
+                            "service": {
+                                "name": service_name,
+                                "port": {
+                                    "number": port_number
+                                }
+                            }
+                        },
+                        "path": path,
+                        "pathType": path_type
+                    },
+                ]
+            }
+        }
+
+        if host is not None:
+            rule["host"] = host
+
+        return cls.create_instance(openshift, name, rules=[rule])
+
+    @property
+    def rules(self):
+        """Returns rules defined in the ingress"""
+        return self.model.spec.rules

testsuite/tests/conftest.py

@@ -72,6 +72,22 @@ 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
+    # internally implemented using `oc status` command that seems internally touching project kind
+    # that is not available on kcp
+    # 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"""