Merge pull request #55 from pehala/types

Add Routes and Secrets objects

testsuite/openshift/client.py

@@ -9,6 +9,8 @@
 from openshift import Context, Selector, OpenShiftPythonException
 
 from testsuite.certificates import Certificate
+from testsuite.openshift.types.routes import Routes
+from testsuite.openshift.types.secrets import Secrets
 
 
 class ServiceTypes(enum.Enum):
@@ -61,6 +63,16 @@ def connected(self):
             return False
         return True
 
+    @cached_property
+    def routes(self):
+        """Return dict-like interface for Routes"""
+        return Routes(self)
+
+    @cached_property
+    def secrets(self):
+        """Return dict-like interface for Secrets"""
+        return Secrets(self)
+
     def do_action(self, verb: str, *args,
                   auto_raise: bool = True, parse_output: bool = False):
         """Run an oc command."""

testsuite/openshift/httpbin.py

@@ -9,7 +9,7 @@
 
 class Envoy(LifecycleObject):
     """Envoy deployed from template"""
-    def __init__(self, openshift, authorino, name, label, httpbin_hostname) -> None:
+    def __init__(self, openshift: "OpenShiftClient", authorino, name, label, httpbin_hostname) -> None:
         self.openshift = openshift
         self.authorino = authorino
         self.name = name
@@ -27,8 +27,7 @@ def route(self):
     def create_route(self, name):
         """Creates another route pointing to this Envoy"""
         service_name = f"envoy-{self.name}"
-        route = self.openshift.do_action("expose", "service", f"--name={name}", "-o", "json",
-                                         service_name, parse_output=True)
+        route = self.openshift.routes.expose(name, service_name)
         with self.openshift.context:
             self.envoy_objects = self.envoy_objects.union(route.self_selector())
         return route

testsuite/openshift/types/__init__.py

@@ -0,0 +1,42 @@
+"""Module containing helper classes for OpenShift objects (C)RUD"""
+from typing import TYPE_CHECKING, List
+
+if TYPE_CHECKING:
+    # pylint: disable=cyclic-import
+    from testsuite.openshift.client import OpenShiftClient
+
+
+class RemoteMapping:
+    """Dict-like interface to different types of OpenShift Objects"""
+
+    def __init__(self, client: 'OpenShiftClient', resource_name: str):
+        self._client = client
+        self._resource_name = resource_name
+
+    def fetch_resource(self, name, *cmd_args: str, auto_raise: bool = True):
+        """Executes command and returns output in yaml format"""
+        args: List[str] = []
+        args.extend([self._resource_name, name, "-o", "json", "--ignore-not-found=true"])
+        args.extend(cmd_args)
+        return self._client.do_action("get", *args, auto_raise=auto_raise, parse_output=True)
+
+    def __iter__(self):
+        """Return iterator for requested resource"""
+        data = self._client.do_action("get", self._resource_name, parse_output=True)
+        return iter(data["items"])
+
+    def __getitem__(self, name):
+        """Return requested resource as APIObject"""
+        res = self.fetch_resource(name)
+        if res is None:
+            raise KeyError()
+        return res
+
+    def __contains__(self, name):
+        res = self.fetch_resource(name)
+        return res is not None
+
+    def __delitem__(self, name):
+        if name not in self:
+            raise KeyError()
+        self._client.do_action("delete", self._resource_name, name)

testsuite/openshift/types/routes.py

@@ -0,0 +1,19 @@
+"""Module containing classes for manipulation with OpenShift routes"""
+from testsuite.openshift.types import RemoteMapping
+
+
+class Routes(RemoteMapping):
+    """Dict-like interface to OpenShift routes"""
+
+    def __init__(self, client) -> None:
+        super().__init__(client, "route")
+
+    def expose(self, name, service, hostname=None):
+        """Expose containers internally as services or externally via routes.
+        Returns requested route in yaml format.
+        """
+        extra_args = []
+        if hostname is not None:
+            extra_args.append(f"--hostname={hostname}")
+        return self._client.do_action("expose", "service", f"--name={name}", "-o", "json",
+                                      service, *extra_args, parse_output=True)

testsuite/openshift/types/secrets.py

@@ -0,0 +1,27 @@
+"""Module containing classes for manipulation with OpenShift secrets"""
+import base64
+
+from testsuite.openshift.types import RemoteMapping
+
+
+class Secrets(RemoteMapping):
+    """Dict-like interface to openshift secrets"""
+
+    def __init__(self, client):
+        super().__init__(client, "secret")
+
+    def __getitem__(self, name: str):
+        """Return requested secret in yaml format"""
+
+        # pylint: disable=too-few-public-methods
+        class _DecodedSecrets:
+            def __init__(self, data):
+                self._data = data
+
+            def __getitem__(self, name):
+                return base64.b64decode(self._data[name]).decode("utf-8")
+
+            def __contains__(self, name):
+                return name in self._data
+
+        return _DecodedSecrets(super().__getitem__(name).model.data)