Add basic tests for DNSPolicy health checks

testsuite/openshift/objects/dnspolicy.py

@@ -1,9 +1,36 @@
 """Module for DNSPolicy related classes"""
+from dataclasses import dataclass
+from typing import Optional, Literal
+
+import openshift as oc
+
+from testsuite.objects import asdict
 from testsuite.openshift.client import OpenShiftClient
 from testsuite.openshift.objects import OpenShiftObject
 from testsuite.openshift.objects.gateway_api import Referencable
 
 
+@dataclass
+class HealthCheck:  # pylint: disable=invalid-name
+    """Object representing DNSPolicy health check specification"""
+
+    allowInsecureCertificates: Optional[bool] = None
+    endpoint: Optional[str] = None
+    expectedResponses: Optional[list[int]] = None
+    failureThreshold: Optional[int] = None
+    interval: Optional[str] = None
+    port: Optional[int] = None
+    protocol: Literal["http", "https"] = "https"
+
+
+class DNSHealthCheckProbe(OpenShiftObject):
+    """DNSHealthCheckProbe object"""
+
+    def is_healthy(self) -> bool:
+        """Returns True if DNSHealthCheckProbe endpoint is healthy"""
+        return self.model.status.healthy
+
+
 class DNSPolicy(OpenShiftObject):
     """DNSPolicy object"""
 
@@ -17,11 +44,21 @@ def create_instance(
     ):
         """Creates new instance of DNSPolicy"""
 
-        model = {
+        model: dict = {
             "apiVersion": "kuadrant.io/v1alpha1",
             "kind": "DNSPolicy",
             "metadata": {"name": name, "labels": labels},
             "spec": {"targetRef": parent.reference},
         }
 
         return cls(model, context=openshift.context)
+
+    def set_health_check(self, health_check: HealthCheck):
+        """Sets health check for DNSPolicy"""
+        self.model["spec"]["healthCheck"] = asdict(health_check)
+
+    def get_dns_health_probe(self) -> oc.APIObject:
+        """Returns DNSHealthCheckProbe object for the created DNSPolicy"""
+        with self.context:
+            dns_probe = oc.selector("DNSHealthCheckProbe", labels={"kuadrant.io/dnspolicy": self.name()}).object()
+        return DNSHealthCheckProbe(dns_probe.model, context=self.context)

testsuite/openshift/objects/gateway_api/gateway.py

@@ -85,6 +85,7 @@ def create_instance(
         gateway_class: str,
         hostname: str,
         labels: dict[str, str] = None,
+        tls: bool = False,
         placement: typing.Optional[str] = None,
     ):
         """Creates new instance of Gateway"""
@@ -95,19 +96,21 @@ def create_instance(
             labels["cluster.open-cluster-management.io/placement"] = placement
 
         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"}],
-                },
-            }
-        ]
+
+        if tls:
+            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
 

testsuite/tests/mgc/conftest.py

@@ -38,6 +38,7 @@ def upstream_gateway(request, openshift, blame, hostname, module_label):
         name=blame("mgc-gateway"),
         gateway_class="kuadrant-multi-cluster-gateway-instance-per-cluster",
         hostname=f"*.{hostname}",
+        tls=True,
         placement="http-gateway",
         labels={"app": module_label},
     )

testsuite/tests/mgc/dnspolicy/conftest.py

@@ -0,0 +1,28 @@
+"""Conftest for DNSPolicy tests"""
+import pytest
+
+from testsuite.openshift.objects.gateway_api.gateway import MGCGateway
+
+
+@pytest.fixture(scope="module")
+def upstream_gateway(request, openshift, blame, hostname, module_label):
+    """Creates and returns configured and ready upstream Gateway with disabled tls"""
+    upstream_gateway = MGCGateway.create_instance(
+        openshift=openshift,
+        name=blame("mgc-gateway"),
+        gateway_class="kuadrant-multi-cluster-gateway-instance-per-cluster",
+        hostname=f"*.{hostname}",
+        tls=False,
+        placement="http-gateway",
+        labels={"app": module_label},
+    )
+    request.addfinalizer(upstream_gateway.delete)
+    upstream_gateway.commit()
+
+    return upstream_gateway
+
+
+@pytest.fixture(scope="module")
+def tls_policy():
+    """Don't need TLSPolicy in the DNSPolicy only tests"""
+    return None

testsuite/tests/mgc/dnspolicy/health_check/conftest.py

@@ -0,0 +1,40 @@
+"""Conftest for DNSPolicy health checks"""
+import time
+import pytest
+
+from testsuite.openshift.objects.gateway_api.gateway import MGCGateway
+
+
+@pytest.fixture(scope="module")
+def upstream_gateway(request, openshift, blame, module_label, initial_host):
+    """
+    Creates and returns configured and ready upstream Gateway with FQDN hostname
+    Health checks available only with Fully Qualified Domain Names in gateway (no wildcards are allowed)
+    """
+    upstream_gateway = MGCGateway.create_instance(
+        openshift=openshift,
+        name=blame("mgc-gateway"),
+        gateway_class="kuadrant-multi-cluster-gateway-instance-per-cluster",
+        hostname=initial_host,
+        tls=False,
+        placement="http-gateway",
+        labels={"app": module_label},
+    )
+    request.addfinalizer(upstream_gateway.delete)
+    upstream_gateway.commit()
+
+    return upstream_gateway
+
+
+@pytest.fixture(scope="module")
+def dns_policy(dns_policy, health_check):
+    """Add health check to DNSPolicy"""
+    dns_policy.set_health_check(health_check)
+    return dns_policy
+
+
+@pytest.fixture(scope="module")
+def dns_health_probe(dns_policy, route):  # pylint: disable=unused-argument
+    """Wait for health check to start monitoring endpoint and return according DNSHealthCheckProbe object"""
+    time.sleep(10)
+    return dns_policy.get_dns_health_probe()

testsuite/tests/mgc/dnspolicy/health_check/test_healthy_endpoint.py

@@ -0,0 +1,23 @@
+"""Tests for DNSPolicy health checks - healthy endpoint"""
+import pytest
+
+from testsuite.openshift.objects.dnspolicy import HealthCheck
+
+pytestmark = [pytest.mark.mgc]
+
+
+@pytest.fixture(scope="module")
+def health_check():
+    """Returns healthy endpoint specification for DNSPolicy health check"""
+    return HealthCheck(
+        allowInsecureCertificates=True,
+        endpoint="/get",
+        interval="5s",
+        port=80,
+        protocol="http",
+    )
+
+
+def test_healthy_endpoint(dns_health_probe):
+    """Test healthy endpoint check"""
+    assert dns_health_probe.is_healthy()

testsuite/tests/mgc/dnspolicy/health_check/test_unhealthy_endpoint.py

@@ -0,0 +1,23 @@
+"""Tests for DNSPolicy health checks - unhealthy endpoint"""
+import pytest
+
+from testsuite.openshift.objects.dnspolicy import HealthCheck
+
+pytestmark = [pytest.mark.mgc]
+
+
+@pytest.fixture(scope="module")
+def health_check():
+    """Returns unhealthy endpoint specification for DNSPolicy health check"""
+    return HealthCheck(
+        allowInsecureCertificates=True,
+        endpoint="/unknown-endpoint",
+        interval="5s",
+        port=80,
+        protocol="http",
+    )
+
+
+def test_unhealthy_endpoint(dns_health_probe):
+    """Test unhealthy endpoint check"""
+    assert not dns_health_probe.is_healthy()