Add ipv6 dualstack lb test

canonical · Jan 8, 2025 · 3ecb70e · 3ecb70e
1 parent 8234750
commit 3ecb70e
Show file tree

Hide file tree

Showing 2 changed files with 47 additions and 16 deletions.
diff --git a/tests/integration/tests/test_loadbalancer.py b/tests/integration/tests/test_loadbalancer.py
@@ -22,36 +22,67 @@ def test_loadbalancer_ipv4(instances: List[harness.Instance]):
 @pytest.mark.node_count(2)
 @pytest.mark.tags(tags.PULL_REQUEST)
 @pytest.mark.disable_k8s_bootstrapping()
-@pytest.mark.network_type("dualstack")
-def test_loadbalancer_ipv6(instances: List[harness.Instance]):
+def test_loadbalancer_ipv6_only(instances: List[harness.Instance]):
+    pytest.xfail(
+        "Cilium ipv6 only unsupported: https://github.com/cilium/cilium/issues/15082"
+    )
     _test_loadbalancer(instances, ipv6=True)
 
 
-def _test_loadbalancer(instances: List[harness.Instance], ipv6=False):
+@pytest.mark.node_count(2)
+@pytest.mark.tags(tags.PULL_REQUEST)
+@pytest.mark.disable_k8s_bootstrapping()
+@pytest.mark.dualstack()
+@pytest.mark.network_type("dualstack")
+def test_loadbalancer_ipv6_dualstack(instances: List[harness.Instance]):
+    _test_loadbalancer(instances, ipv6=True, dualstack=True)
+
+
+def _test_loadbalancer(instances: List[harness.Instance], ipv6=False, dualstack=False):
     instance = instances[0]
     tester_instance = instances[1]
 
     if ipv6:
-        bootstrap_config = (MANIFESTS_DIR / "bootstrap-ipv6-only.yaml").read_text()
+        bootstrap_args = []
+        if dualstack:
+            bootstrap_config = (MANIFESTS_DIR / "bootstrap-dualstack.yaml").read_text()
+        else:
+            bootstrap_config = (MANIFESTS_DIR / "bootstrap-ipv6-only.yaml").read_text()
+            bootstrap_args += ["--address", "::/0"]
         instance.exec(
-            ["k8s", "bootstrap", "--file", "-", "--address", "::/0"],
+            ["k8s", "bootstrap", "--file", "-", *bootstrap_args],
             input=str.encode(bootstrap_config),
         )
     else:
         instance.exec(["k8s", "bootstrap"])
 
-    instance_default_ip = util.get_default_ip(instance, ipv6=ipv6)
-    tester_instance_default_ip = util.get_default_ip(tester_instance, ipv6=ipv6)
+    lb_cidrs = []
 
-    instance_default_cidr = util.get_default_cidr(instance, instance_default_ip)
+    def get_lb_cidr(ipv6_cidr):
+        instance_default_ip = util.get_default_ip(instance, ipv6=ipv6_cidr)
+        tester_instance_default_ip = util.get_default_ip(
+            tester_instance, ipv6=ipv6_cidr
+        )
+        instance_default_cidr = util.get_default_cidr(instance, instance_default_ip)
+        lb_cidr = util.find_suitable_cidr(
+            parent_cidr=instance_default_cidr,
+            excluded_ips=[instance_default_ip, tester_instance_default_ip],
+        )
+        return lb_cidr
 
-    lb_cidr = util.find_suitable_cidr(
-        parent_cidr=instance_default_cidr,
-        excluded_ips=[instance_default_ip, tester_instance_default_ip],
-    )
+    if dualstack or not ipv6:
+        lb_cidrs.append(get_lb_cidr(ipv6_cidr=False))
+    if ipv6:
+        lb_cidrs.append(get_lb_cidr(ipv6_cidr=True))
+    lb_cidr_str = ",".join(lb_cidrs)
 
     instance.exec(
-        ["k8s", "set", f"load-balancer.cidrs={lb_cidr}", "load-balancer.l2-mode=true"]
+        [
+            "k8s",
+            "set",
+            f"load-balancer.cidrs={lb_cidr_str}",
+            "load-balancer.l2-mode=true",
+        ]
     )
     instance.exec(["k8s", "enable", "load-balancer"])
 

diff --git a/tests/integration/tests/test_util/util.py b/tests/integration/tests/test_util/util.py
@@ -378,13 +378,13 @@ def get_default_ip(instance: harness.Instance, ipv6=False):
     # default via 10.42.254.1 dev eth0 proto dhcp src 10.42.254.197 metric 100
     # ---
     # Fetching the default IP address from the output, e.g. 10.42.254.197
-    addr_family = "-6" if ipv6 else "-4"
     if ipv6:
         p = instance.exec(
-            ["ip", "-json", "-6", "addr", "show", "scope", "global"], capture_output=True
+            ["ip", "-json", "-6", "addr", "show", "scope", "global"],
+            capture_output=True,
         )
         addr_json = json.loads(p.stdout.decode())
-        return addr_json[0]['addr_info'][0]['local']
+        return addr_json[0]["addr_info"][0]["local"]
     else:
         p = instance.exec(
             ["ip", "-o", "-4", "route", "show", "to", "default"], capture_output=True