test: kubectl-dns secret-generation command

config/settings.local.yaml.tpl

@@ -1,6 +1,7 @@
 #default:
 #  tester: "someuser"                          # Optional: name of the user, who is running the tests, defaults to whoami/uid
 #  kuadrantctl: kuadrantctl
+#  kubectl-dns: "kubectl-dns"
 #  console:
 #    username: "CONSOLE_USERNAME"              # OpenShift console username for UI test login
 #    password: "CONSOLE_PASSWORD"              # OpenShift console password for UI tests login

config/settings.yaml

@@ -1,6 +1,7 @@
 default:
   dynaconf_merge: true
   kuadrantctl: "kuadrantctl"
+  kubectl-dns: "kubectl-dns"
   tools:
     project: "tools"
   cfssl: "cfssl"

pyproject.toml

@@ -56,6 +56,7 @@ markers = [
     "defaults_overrides: Defaults and Overrides feature tests",
     "observability: Tracing, metrics and logging tests",
     "ui: Test uses browser automation via Playwright to test the console plugin UI",
+    "cli: Test is using CLI tools (kubectl-dns, kuadrantctl)",
     "min_ocp_version: Minimum OpenShift version required for test (e.g., @pytest.mark.min_ocp_version((4, 20)))",
 ]
 filterwarnings = [

testsuite/cli/kubectl_dns.py

@@ -0,0 +1,25 @@
+"""Wrapper around kubectl-dns binary"""
+
+import os
+import subprocess
+
+
+class KubectlDNS:
+    """Wrapper on top of kubectl-dns binary"""
+
+    def __init__(self, binary) -> None:
+        super().__init__()
+        self.binary = binary
+
+    def run(self, *args, **kwargs):
+        """Passes arguments to subprocess.run()"""
+        args = (self.binary, *args)
+        kwargs.setdefault("capture_output", True)
+        kwargs.setdefault("text", True)
+
+        if "env" in kwargs:
+            env = os.environ.copy()
+            env.update(kwargs["env"])
+            kwargs["env"] = env
+
+        return subprocess.run(args, **kwargs)  # pylint: disable= subprocess-run-check

testsuite/kubernetes/client.py

@@ -2,6 +2,8 @@
 
 from functools import cached_property
 from urllib.parse import urlparse
+import tempfile
+import yaml
 
 import openshift_client as oc
 from openshift_client import Context, OpenShiftPythonException
@@ -45,6 +47,13 @@ def context(self):
 
         return context
 
+    @property
+    def current_context_name(self) -> str:
+        """Returns the current context name from the kubeconfig"""
+        if self._kubeconfig_path is None:
+            raise ValueError("Kubeconfig path is not set")
+        return self.do_action("config", "current-context").out().strip()
+
     @property
     def api_url(self):
         """Returns real API url"""
@@ -147,3 +156,25 @@ def apply_from_string(self, string, cls, cmd_args=None):
             obj = selector.object(cls=cls)
             obj.context = self.context
         return obj
+
+    def create_merged_kubeconfig(self, cluster2: "KubernetesClient") -> str:
+        """
+        Creates a merged kubeconfig from this instance and another KubernetesClient instance.
+        Returns the path to the temporary kubeconfig file.
+        """
+        with self.context:
+            config1_yaml = oc.invoke("config", ["view", "--minify=true", "--flatten=true"]).out()
+        with cluster2.context:
+            config2_yaml = oc.invoke("config", ["view", "--minify=true", "--flatten=true"]).out()
+
+        config1 = yaml.safe_load(config1_yaml)
+        config2 = yaml.safe_load(config2_yaml)
+
+        merged_config = config1.copy()
+        merged_config["clusters"].extend(config2.get("clusters", []))
+        merged_config["contexts"].extend(config2.get("contexts", []))
+        merged_config["users"].extend(config2.get("users", []))
+
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".kubeconfig", delete=False) as temp_file:
+            yaml.safe_dump(merged_config, temp_file)
+            return temp_file.name

testsuite/tests/kuadrantctl/conftest.py

@@ -10,7 +10,7 @@
 from testsuite.gateway.gateway_api.gateway import KuadrantGateway
 from testsuite.gateway.gateway_api.route import HTTPRoute
 from testsuite.gateway import Gateway, GatewayListener, Hostname
-from testsuite.kuadrantctl import KuadrantCTL
+from testsuite.cli.kuadrantctl import KuadrantCTL
 from testsuite.oas import OASWrapper
 
 

testsuite/tests/multicluster/coredns/two_clusters/kubectl_dns/test_secret_generation.py

@@ -0,0 +1,52 @@
+"""Test kubectl-dns secret-generate command with basic coredns setup with 1 primary and 1 secondary clusters"""
+
+import shutil
+
+import dns.resolver
+import pytest
+
+from testsuite.cli.kubectl_dns import KubectlDNS
+from testsuite.tests.multicluster.coredns.conftest import IP1, IP2
+
+pytestmark = [pytest.mark.cli]
+
+
+@pytest.fixture(scope="session")
+def kubectl_dns(testconfig, skip_or_fail):
+    """Return Kuadrantctl wrapper with merged kubeconfig"""
+    binary_path = testconfig["kubectl-dns"]
+    if not shutil.which(binary_path):
+        skip_or_fail("kubectl-dns binary not found")
+    return KubectlDNS(binary_path)
+
+
+@pytest.fixture(scope="module")
+def kubeconfig_secrets(request, testconfig, cluster, cluster2, kubectl_dns, blame):
+    """Run generate-secret command on merged kubeconfig to generate kubeconfig secret for the secondary cluster"""
+    system_project = testconfig["service_protection"]["system_project"]
+    secret_name = blame("kubecfg")
+    request.addfinalizer(
+        lambda: cluster.do_action("delete", "secret", secret_name, "-n", system_project, "--ignore-not-found")
+    )
+
+    merged_kubeconfig = cluster.create_merged_kubeconfig(cluster2)
+    result = kubectl_dns.run(
+        "secret-generation",
+        "--name",
+        secret_name,
+        "--context",
+        cluster2.current_context_name,
+        "--namespace",
+        system_project,
+        "--service-account",
+        "coredns",
+        env={"KUBECONFIG": merged_kubeconfig},
+    )
+    assert result.returncode == 0, f"kubectl-dns couldn't generate kubeconfig secret: {result.stderr}"
+    return []
+
+
+def test_kubectl_dns_secret_generation(hostname):
+    """IPs from both, primary and secondary, clusters should return in DNS A record set"""
+    dns_ips = {ip.address for ip in dns.resolver.resolve(hostname.hostname)}
+    assert {IP1, IP2} == dns_ips, "CoreDNS should have returned both IP addresses in A record set"