Merge pull request #240 from ficap/tlspolicy-test

Make mgc test_smoke use TLSPolicy

testsuite/openshift/objects/gateway_api/__init__.py

@@ -1,5 +1,9 @@
 """Module containing all Gateway API related classes"""
 from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from typing import Any, Optional
+
+from testsuite.objects import asdict
 
 
 class Referencable(ABC):
@@ -12,3 +16,22 @@ def reference(self) -> dict[str, str]:
         Returns dict, which can be used as reference in Gateway API Objects.
         https://gateway-api.sigs.k8s.io/references/spec/#gateway.networking.k8s.io/v1beta1.ParentReference
         """
+
+
+@dataclass
+class CustomReference(Referencable):
+    """
+    Manually creates Reference object.
+    https://gateway-api.sigs.k8s.io/references/spec/#gateway.networking.k8s.io%2fv1beta1.ParentReference
+    """
+
+    @property
+    def reference(self) -> dict[str, Any]:
+        return asdict(self)
+
+    group: str
+    kind: str
+    name: str
+    namespace: Optional[str] = None
+    sectionName: Optional[str] = None  # pylint: disable=invalid-name
+    port: Optional[int] = None

testsuite/openshift/objects/gateway_api/gateway.py

@@ -4,6 +4,7 @@
 
 from openshift import Selector, timeout, selector
 
+from testsuite.certificates import Certificate
 from testsuite.openshift.client import OpenShiftClient
 from testsuite.openshift.objects import OpenShiftObject
 from testsuite.openshift.objects.proxy import Proxy
@@ -93,7 +94,33 @@ def create_instance(
         if placement is not None:
             labels["cluster.open-cluster-management.io/placement"] = placement
 
-        return super(MGCGateway, cls).create_instance(openshift, name, gateway_class, hostname, labels)
+        instance = super(MGCGateway, cls).create_instance(openshift, name, gateway_class, hostname, labels)
+        instance.model["spec"]["listeners"] = [
+            {
+                "name": "api",
+                "port": 443,
+                "protocol": "HTTPS",
+                "hostname": hostname,
+                "allowedRoutes": {"namespaces": {"from": "All"}},
+                "tls": {
+                    "mode": "Terminate",
+                    "certificateRefs": [{"name": f"{name}-tls", "kind": "Secret"}],
+                },
+            }
+        ]
+
+        return instance
+
+    def get_tls_cert(self) -> Certificate:
+        """Returns TLS certificate used by the gateway"""
+        tls_cert_secret_name = self.model.spec.listeners[0].tls.certificateRefs[0].name
+        tls_cert_secret = self.openshift.get_secret(tls_cert_secret_name)
+        tls_cert = Certificate(
+            key=tls_cert_secret["tls.key"],
+            certificate=tls_cert_secret["tls.crt"],
+            chain=tls_cert_secret["ca.crt"],
+        )
+        return tls_cert
 
     def get_spoke_gateway(self, spokes: dict[str, OpenShiftClient]) -> "MGCGateway":
         """

testsuite/openshift/objects/tlspolicy.py

@@ -0,0 +1,31 @@
+"""Module for TLSPolicy related classes"""
+from testsuite.openshift.client import OpenShiftClient
+from testsuite.openshift.objects import OpenShiftObject
+from testsuite.openshift.objects.gateway_api import Referencable
+
+
+class TLSPolicy(OpenShiftObject):
+    """TLSPolicy object"""
+
+    @classmethod
+    def create_instance(
+        cls,
+        openshift: OpenShiftClient,
+        name: str,
+        parent: Referencable,
+        issuer: Referencable,
+        labels: dict[str, str] = None,
+    ):
+        """Creates new instance of TLSPolicy"""
+
+        model = {
+            "apiVersion": "kuadrant.io/v1alpha1",
+            "kind": "TLSPolicy",
+            "metadata": {"name": name, "labels": labels},
+            "spec": {
+                "targetRef": parent.reference,
+                "issuerRef": issuer.reference,
+            },
+        }
+
+        return cls(model, context=openshift.context)

testsuite/tests/mgc/conftest.py

@@ -5,10 +5,12 @@
 
 from testsuite.openshift.httpbin import Httpbin
 from testsuite.openshift.objects.dnspolicy import DNSPolicy
+from testsuite.openshift.objects.gateway_api import CustomReference
 from testsuite.openshift.objects.gateway_api.gateway import MGCGateway, GatewayProxy
 from testsuite.openshift.objects.gateway_api.route import HTTPRoute
 from testsuite.openshift.objects.proxy import Proxy
 from testsuite.openshift.objects.route import Route
+from testsuite.openshift.objects.tlspolicy import TLSPolicy
 
 
 @pytest.fixture(scope="module")
@@ -41,7 +43,8 @@ def upstream_gateway(request, openshift, blame, hostname, module_label):
     )
     request.addfinalizer(upstream_gateway.delete)
     upstream_gateway.commit()
-    upstream_gateway.wait_for_ready()
+    # we cannot wait here because of referencing not yet existent tls secret which would be provided later by tlspolicy
+    # upstream_gateway.wait_for_ready()
 
     return upstream_gateway
 
@@ -61,6 +64,16 @@ def initial_host(hostname):
     return f"route.{hostname}"
 
 
+@pytest.fixture(scope="session")
+def self_signed_cluster_issuer():
+    """Reference to cluster self-signed certificate issuer"""
+    return CustomReference(
+        group="cert-manager.io",
+        kind="ClusterIssuer",
+        name="selfsigned-cluster-issuer",
+    )
+
+
 @pytest.fixture(scope="module")
 def route(request, proxy, blame, gateway, initial_host, backend) -> Route:
     """Exposed Route object"""
@@ -77,9 +90,12 @@ def route(request, proxy, blame, gateway, initial_host, backend) -> Route:
     return route
 
 
+# pylint: disable=unused-argument
 @pytest.fixture(scope="module")
-def gateway(upstream_gateway, spokes):
+def gateway(upstream_gateway, spokes, hub_policies_commit):
     """Downstream gateway, e.g. gateway on a spoke cluster"""
+    # wait for upstream gateway here to be able to get spoke gateways
+    upstream_gateway.wait_for_ready()
     gw = upstream_gateway.get_spoke_gateway(spokes)
     gw.wait_for_ready()
     return gw
@@ -108,10 +124,23 @@ def dns_policy(blame, upstream_gateway, module_label):
     return policy
 
 
-@pytest.fixture(scope="module", autouse=True)
-def commit(request, dns_policy):
+@pytest.fixture(scope="module")
+def tls_policy(blame, upstream_gateway, module_label, self_signed_cluster_issuer):
+    """TLSPolicy fixture"""
+    policy = TLSPolicy.create_instance(
+        upstream_gateway.openshift,
+        blame("tls"),
+        parent=upstream_gateway,
+        issuer=self_signed_cluster_issuer,
+        labels={"app": module_label},
+    )
+    return policy
+
+
+@pytest.fixture(scope="module")
+def hub_policies_commit(request, upstream_gateway, dns_policy, tls_policy):
     """Commits all important stuff before tests"""
-    for component in [dns_policy]:
+    for component in [dns_policy, tls_policy]:
         if component is not None:
             request.addfinalizer(component.delete)
             component.commit()

testsuite/tests/mgc/test_basic.py

@@ -17,6 +17,8 @@
 
 import pytest
 
+from testsuite.httpx import HttpxBackoffClient
+
 pytestmark = [pytest.mark.mgc]
 
 
@@ -25,12 +27,16 @@ def test_gateway_readiness(gateway):
     assert gateway.is_ready()
 
 
-def test_smoke(route):
+def test_smoke(route, upstream_gateway):
     """
     Tests whether the backend, exposed using the HTTPRoute and Gateway, was exposed correctly,
     having a tls secured endpoint with a hostname managed by MGC
     """
-    backend_client = route.client(verify=False)  # self-signed certificate; TBD
+
+    tls_cert = upstream_gateway.get_tls_cert()
+
+    # assert that tls_cert is used by the server
+    backend_client = HttpxBackoffClient(base_url=f"https://{route.hostnames[0]}", verify=tls_cert)
 
     sleep(30)  # wait for DNS record to propagate correctly; TBD