Add Gateway listener dns test

Signed-off-by: Alex Zgabur <[email protected]>

testsuite/tests/singlecluster/gateway/reconciliation/test_gateway_listeners_dns.py

@@ -0,0 +1,77 @@
+"""Testing specific bug that happens when listener hostname in Gateway gets changed while DNSPolicy is applied."""
+
+import pytest
+from testsuite.gateway.gateway_api.hostname import StaticHostname
+from testsuite.utils import is_nxdomain, sleep_ttl
+
+pytestmark = [pytest.mark.kuadrant_only, pytest.mark.dnspolicy, pytest.mark.tlspolicy]
+
+
+@pytest.fixture(scope="module")
+def wildcard_domain(base_domain, blame):
+    """
+    For this test we want specific domain, not wildcard.
+    This will be used in the first iteration of Gateway and HTTPRoute.
+    """
+    return f"{blame('dnsbug1')}.{base_domain}"
+
+
+@pytest.fixture(scope="module")
+def new_domain(base_domain, blame):
+    """In the test the Gateway and HTTPRoute will change their hostnames to this one."""
+    return f"{blame('dnsbug2')}.{base_domain}"
+
+
+@pytest.fixture(scope="module")
+def client(gateway, wildcard_domain):
+    """Make client point to correct domain route domain."""
+    return StaticHostname(wildcard_domain, gateway.get_tls_cert).client()
+
+
+@pytest.fixture(scope="module")
+def client_new(gateway, new_domain):
+    """Second client that will be created after Gateway gets updated pointing to new_domain."""
+
+    def _client_new():
+        return StaticHostname(new_domain, gateway.get_tls_cert).client()
+
+    return _client_new
+
+
+@pytest.fixture(scope="module")
+def route(route, wildcard_domain):
+    """So that route hostname matches the gateway hostname."""
+    route.remove_all_hostnames()
+    route.add_hostname(wildcard_domain)
+    return route
+
+
+@pytest.mark.xfail()
+@pytest.mark.issue("https://github.com/Kuadrant/kuadrant-operator/issues/794")
+def test_change_hostname(client, client_new, auth, gateway, route, new_domain, wildcard_domain):
+    """
+    This test checks if after change of listener hostname in a Gateway while having DNSPolicy applied, that
+    the old hostname gets deleted from DNS provider. After editing the hostname in HTTPRoute to the new value
+    this test checks the reconciliation of such procedure.
+
+    WARNING
+        Running this test in unpatched Kuadrant will leave orphaned DNS records in DNS provider.
+        If you want to delete them you need to do it manually. The DNS records will contain string 'dnsbug'
+    """
+    result = client.get("/get", auth=auth)
+    assert not result.has_dns_error()
+    assert not result.has_cert_verify_error()
+    assert result.status_code == 200
+
+    gateway.refresh().model.spec.listeners[0].hostname = new_domain
+    gateway.apply()
+    route.refresh().model.spec.hostnames[0] = new_domain
+    route.apply()
+
+    result = client_new().get("/get", auth=auth)
+    assert not result.has_dns_error()
+    assert not result.has_cert_verify_error()
+    assert result.status_code == 200
+
+    sleep_ttl(wildcard_domain)
+    assert is_nxdomain(wildcard_domain)

testsuite/utils.py

@@ -6,6 +6,7 @@
 import os
 import getpass
 import secrets
+from time import sleep
 from collections.abc import Collection
 from copy import deepcopy
 from dataclasses import is_dataclass, fields
@@ -14,6 +15,7 @@
 from typing import Dict, Union
 from urllib.parse import urlparse, ParseResult
 
+import dns.resolver
 from weakget import weakget
 
 from testsuite.certificates import Certificate, CFSSLClient, CertInfo
@@ -176,3 +178,26 @@ def check_condition(condition, condition_type, status, reason=None, message=None
     ):
         return True
     return False
+
+
+def is_nxdomain(hostname: str):
+    """
+    Returns True if hostname has no `A` record in DNS. False otherwise.
+    Will raise exception `dns.resolver.NoAnswer` if there exists different record type under the hostname.
+    """
+    try:
+        dns.resolver.resolve(hostname)
+    except dns.resolver.NXDOMAIN:
+        return True
+    return False
+
+
+def sleep_ttl(hostname: str):
+    """Sleeps for duration of TTL for given hostname."""
+    try:
+        response = dns.resolver.resolve(hostname, raise_on_no_answer=False)
+    except dns.resolver.NXDOMAIN:
+        return
+    if response.rrset is None:
+        return
+    sleep(response.rrset.ttl)