Add reconciliation tests for DNSPolicy and TLSPolicy

testsuite/openshift/__init__.py

@@ -4,7 +4,7 @@
 from dataclasses import dataclass, field
 from typing import Optional, Literal
 
-from openshift_client import APIObject, timeout
+from openshift_client import APIObject, timeout, OpenShiftPythonException
 
 from testsuite.lifecycle import LifecycleObject
 
@@ -32,6 +32,17 @@ def delete(self, ignore_not_found=True, cmd_args=None):
             self.committed = False
             return deleted
 
+    def wait_until(self, test_function, timelimit=90):
+        """Waits until the test function succeeds for this object"""
+        try:
+            with timeout(timelimit):
+                success, _, _ = self.self_selector().until_all(
+                    success_func=lambda obj: test_function(self.__class__(obj.model))
+                )
+                return success
+        except OpenShiftPythonException:
+            return False
+
 
 def modify(func):
     """Wraps method of a subclass of OpenShiftObject to use modify_and_apply when the object

testsuite/tests/mgc/reconciliation/__init__.py

@@ -0,0 +1,8 @@
+"""Module containing tests for Reconciliation of MGC policies"""
+
+from testsuite.policy.tls_policy import TLSPolicy
+
+
+def tls_policy(openshift, name, parent, labels: dict[str, str] = None):
+    """Creates TLS Policy with issuer same as the parent to match the DNSPolicy constructor"""
+    return TLSPolicy.create_instance(openshift, name, parent, labels=labels, issuer=parent)

testsuite/tests/mgc/reconciliation/test_gw_doesnt_exist.py

@@ -0,0 +1,40 @@
+"""Tests that DNSPolicy/TLSPolicy is rejected if the Gateway does not exist at all"""
+
+import pytest
+
+from testsuite.gateway import CustomReference
+from testsuite.policy.dns_policy import DNSPolicy
+from testsuite.tests.mgc.reconciliation import tls_policy
+
+pytestmark = [pytest.mark.mgc]
+
+
+@pytest.fixture(scope="module", autouse=True)
+def commit():
+    """We want to manage Policies on our own here"""
+    return
+
+
+@pytest.mark.parametrize(
+    "create_cr", [pytest.param(DNSPolicy.create_instance, id="DNSPolicy"), pytest.param(tls_policy, id="TLSPolicy")]
+)
+@pytest.mark.issue("https://github.com/Kuadrant/multicluster-gateway-controller/issues/361")
+def test_dns_no_gw(request, create_cr, hub_openshift, blame, module_label):
+    """Tests that DNSPolicy is rejected if the Gateway does not exist at all"""
+
+    def target_not_found(policy):
+        for condition in policy.model.status.conditions:
+            if condition.type == "Ready" and condition.status == "False" and condition.reason == "TargetNotFound":
+                return True
+        return False
+
+    policy = create_cr(
+        hub_openshift,
+        blame("resource"),
+        CustomReference(group="gateway.networking.k8s.io", kind="Gateway", name="does-not-exist"),
+        labels={"app": module_label},
+    )
+    request.addfinalizer(policy.delete)
+    policy.commit()
+
+    assert policy.wait_until(target_not_found), "Policy did not reach expected status"

testsuite/tests/mgc/reconciliation/test_same_target.py

@@ -0,0 +1,49 @@
+"""Tests that DNSPolicy/TLSPolicy is rejected when the Gateway already has a policy of the same kind"""
+
+import pytest
+from openshift_client import selector
+
+from testsuite.policy.dns_policy import DNSPolicy
+from testsuite.tests.mgc.reconciliation import tls_policy
+
+pytestmark = [pytest.mark.mgc]
+
+
+@pytest.fixture(scope="module")
+def base_domain(hub_openshift):
+    """Returns preconfigured base domain"""
+    zone = selector("managedzone/aws-mz", static_context=hub_openshift.context).object()
+    return zone.model["spec"]["domainName"]
+
+
+@pytest.mark.parametrize(
+    "create_cr", [pytest.param(DNSPolicy.create_instance, id="DNSPolicy"), pytest.param(tls_policy, id="TLSPolicy")]
+)
+def test_two_policies_one_gw(request, create_cr, hub_gateway, client, blame, module_label):
+    """Tests that DNSPolicy is rejected when the Gateway already has a DNSPolicy"""
+
+    def two_dns_policies_error(policy: DNSPolicy):
+        for condition in policy.model.status.conditions:
+            if condition.type == "Ready" and condition.status == "False" and condition.reason == "ReconciliationError":
+                return True
+        return False
+
+    # test that it works before the policy
+    response = client.get("get")
+    assert response.status_code == 200, "Original DNSPolicy does not work"
+
+    policy = create_cr(
+        hub_gateway.openshift,
+        blame("dns2"),
+        hub_gateway,
+        labels={"app": module_label},
+    )
+    request.addfinalizer(policy.delete)
+    policy.commit()
+
+    # Wait for expected status
+    assert policy.wait_until(two_dns_policies_error), "Policy did not reach expected status"
+
+    # Test that the original policy still works
+    response = client.get("get")
+    assert response.status_code == 200