Create product routes

Author: mastastny

testsuite/gateways/apicast/system.py

@@ -1,17 +1,20 @@
 """System Apicast that comes deployed with 3scale"""
 
 from typing import TYPE_CHECKING
+from urllib.parse import urlparse
 
 import backoff
-from openshift_client import OpenShiftPythonException
+from openshift_client import Missing, OpenShiftPythonException
+from threescale_api.resources import Service
 
 from testsuite.capabilities import Capability
 from testsuite.gateways.apicast import AbstractApicast
 from testsuite.openshift.env import Properties
+from testsuite.openshift.objects import Routes
 from testsuite.utils import randomize
 
 if TYPE_CHECKING:
-    from typing import Optional
+    from typing import List, Optional
 
     # pylint: disable=cyclic-import
     from testsuite.openshift.client import OpenShiftClient
@@ -34,6 +37,54 @@ class SystemApicast(AbstractApicast):
     def __init__(self, staging: bool, openshift: "Optional[OpenShiftClient]" = None):
         self.staging = staging
         self.openshift: "Optional[OpenShiftClient]" = openshift
+        self._routes: "List[str]" = []
+
+    @property
+    def _zync_disabled(self) -> bool:
+        """Returns True if zync is disabled in the APIManager spec"""
+        if self.openshift is None:
+            return False
+        enabled = self.openshift.api_manager.get_path("spec/zync/enabled")
+        return enabled is not Missing and not enabled
+
+    @staticmethod
+    def _route_name(service: Service, production: bool = False) -> str:
+        env = "prod" if production else "stage"
+        return f"testsuite-{env}-{service.entity_id}"
+
+    def _create_route(self, name: str, endpoint_url: str, ocp_service: str):
+        """Creates an OCP route for the given endpoint URL pointing to the given OCP service"""
+        assert self.openshift is not None
+        hostname = urlparse(endpoint_url).hostname
+        if name in self.openshift.routes:
+            del self.openshift.routes[name]
+        self.openshift.routes.create(
+            name, Routes.Types.EDGE, service=ocp_service, hostname=hostname, **{"insecure-policy": "Allow"}
+        )
+        self._routes.append(name)
+
+    def on_service_create(self, service: Service):
+        """Creates staging OCP route when zync is disabled"""
+        if not self._zync_disabled:
+            return
+        proxy = service.proxy.fetch()
+        self._create_route(self._route_name(service), proxy["sandbox_endpoint"], "apicast-staging")
+
+    def on_proxy_promote(self, service: Service):
+        """Creates production OCP route when zync is disabled"""
+        if not self._zync_disabled:
+            return
+        proxy = service.proxy.fetch()
+        self._create_route(self._route_name(service, production=True), proxy["endpoint"], "apicast-production")
+
+    def on_service_delete(self, service: Service):
+        """Cleans up OCP routes created for this service"""
+        if self.openshift is None:
+            return
+        for name in [self._route_name(service), self._route_name(service, production=True)]:
+            if name in self._routes and name in self.openshift.routes:
+                del self.openshift.routes[name]
+                self._routes.remove(name)
 
     @property
     def deployment(self):

testsuite/lifecycle_hook.py

@@ -112,3 +112,9 @@ def on_application_delete(self, application: Application):
 
         Args:
             :param application: The application to be deleted"""
+
+    def on_proxy_promote(self, service: Service):
+        """Called after proxy configuration is promoted to production.
+
+        Args:
+            :param service: Service whose proxy was promoted to production"""

testsuite/tests/conftest.py

@@ -403,7 +403,7 @@ def tools(testconfig):
 
 
 @pytest.fixture(scope="module")
-def prod_client(production_gateway, application, request):
+def prod_client(production_gateway, application, request, lifecycle_hooks):
     """Prepares application and service for production use and creates new production client
 
     Parameters:
@@ -422,6 +422,8 @@ def _prod_client(app=application, promote: bool = True, version: int = -1, redep
             if version == -1:
                 version = app.service.proxy.list().configs.latest()["version"]
             app.service.proxy.list().promote(version=version)
+            for hook in _select_hooks("on_proxy_promote", lifecycle_hooks):
+                hook(app.service)
         if redeploy:
             production_gateway.reload()