fix: test_kraft.test_listeners breaks if juju reports IPv6 address

tests/integration/helpers.py

@@ -3,6 +3,7 @@
 # See LICENSE file for licensing details.
 import json
 import logging
+import re
 import socket
 import subprocess
 from contextlib import closing
@@ -497,6 +498,19 @@ async def get_address(ops_test: OpsTest, app_name=APP_NAME, unit_num=0) -> str:
     return address
 
 
+async def get_unit_ipv4_address(ops_test: OpsTest, app_name=APP_NAME, unit_num=0) -> str | None:
+    """A safer alternative for `juju.unit.get_public_address()` which is robust to network changes."""
+    return_code, stdout, stderr = await ops_test.juju(
+        "ssh", f"{app_name}/{unit_num}", "hostname -i"
+    )
+    ipv4_matches = re.findall(r"[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}\.[0-9]{1,3}", stdout)
+
+    if ipv4_matches:
+        return ipv4_matches[0]
+
+    return None
+
+
 def balancer_exporter_is_up(model_full_name: str | None, app_name: str) -> bool:
     check_output(
         f"JUJU_MODEL={model_full_name} juju ssh {app_name}/leader sudo -i 'curl http://localhost:{JMX_CC_PORT}/metrics'",

tests/integration/test_kraft.py

@@ -19,7 +19,7 @@
 from .helpers import (
     APP_NAME,
     check_socket,
-    get_address,
+    get_unit_ipv4_address,
 )
 
 logger = logging.getLogger(__name__)
@@ -115,7 +115,7 @@ async def test_integrate(self, ops_test: OpsTest):
 
     @pytest.mark.abort_on_fail
     async def test_listeners(self, ops_test: OpsTest):
-        address = await get_address(ops_test=ops_test)
+        address = await get_unit_ipv4_address(ops_test=ops_test)
         assert check_socket(
             address, SECURITY_PROTOCOL_PORTS["SASL_PLAINTEXT", "SCRAM-SHA-512"].internal
         )  # Internal listener
@@ -127,6 +127,5 @@ async def test_listeners(self, ops_test: OpsTest):
 
         # Check controller socket
         if self.controller_app != APP_NAME:
-            address = await get_address(ops_test=ops_test, app_name=self.controller_app)
-
+            address = await get_unit_ipv4_address(ops_test=ops_test, app_name=self.controller_app)
         assert check_socket(address, CONTROLLER_PORT)