RHOAIENG-18459: chore(tests/containers/workbenches): make the ipv6 listening test work on macOS

Author: jiridanek

tests/containers/podman_machine_utils.py

@@ -0,0 +1,59 @@
+import json
+import logging
+import socket
+import subprocess
+from tests.containers.schemas import PodmanMachineInspect
+
+logging.basicConfig(level=logging.DEBUG)
+
+def open_ssh_tunnel(machine_name: str, local_port: int, remote_port: int, remote_interface: str = "localhost") -> subprocess.Popen:
+    # Load and parse the Podman machine data
+    json_data = subprocess.check_output(["podman", "machine", "inspect", machine_name], text=True)
+    inspect = PodmanMachineInspect(machines=json.loads(json_data))
+    machines = inspect.machines
+
+    machine = next((m for m in machines if m.Name == machine_name), None)
+    if not machine:
+        raise ValueError(f"Machine '{machine_name}' not found")
+
+    ssh_command = [
+        "ssh",
+        "-i", machine.SSHConfig.IdentityPath,
+        "-p", str(machine.SSHConfig.Port),
+        "-L", f"{local_port}:{remote_interface}:{remote_port}",
+        "-N",  # Do not execute a remote command
+        "-o", "UserKnownHostsFile=/dev/null",
+        "-o", "StrictHostKeyChecking=no",
+        f"{machine.SSHConfig.RemoteUsername}@localhost"
+    ]
+
+    # Open the SSH tunnel
+    process = subprocess.Popen(ssh_command)
+
+    logging.info(f"SSH tunnel opened for {machine_name}: {remote_interface}:{local_port} -> localhost:{remote_port}")
+    return process
+
+def find_free_port() -> int:
+    """Find a free port on the local machine.
+    :return: A port number that is currently free and available for use.
+    """
+    with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as s:
+        s.bind(('', 0))  # Bind to a free port provided by the system
+        s.listen(1)
+        port = s.getsockname()[1]
+    return port
+
+# Usage example
+if __name__ == "__main__":
+    tunnel_process = open_ssh_tunnel(
+        "podman-machine-default",
+        8080, 8080,
+        remote_interface="[fc00::2]"
+    )
+
+    # Keep the tunnel open until user interrupts
+    try:
+        tunnel_process.wait()
+    except KeyboardInterrupt:
+        tunnel_process.terminate()
+        print("SSH tunnel closed")

tests/containers/schemas.py

@@ -0,0 +1,41 @@
+from pydantic import BaseModel
+from typing import Optional
+from datetime import datetime
+
+class ConfigDir(BaseModel):
+    Path: str
+
+class PodmanSocket(BaseModel):
+    Path: str
+
+class ConnectionInfo(BaseModel):
+    PodmanSocket: PodmanSocket
+    PodmanPipe: Optional[str] = None
+
+class Resources(BaseModel):
+    CPUs: int
+    DiskSize: int
+    Memory: int
+    USBs: list[str] = []
+
+class SSHConfig(BaseModel):
+    IdentityPath: str
+    Port: int
+    RemoteUsername: str
+
+class PodmanMachine(BaseModel):
+    ConfigDir: ConfigDir
+    ConnectionInfo: ConnectionInfo
+    Created: datetime
+    LastUp: datetime
+    Name: str
+    Resources: Resources
+    SSHConfig: SSHConfig
+    State: str
+    UserModeNetworking: bool
+    Rootful: bool
+    Rosetta: bool
+
+# generated from `podman machine inspect` output by smart tooling
+class PodmanMachineInspect(BaseModel):
+    machines: list[PodmanMachine]

tests/containers/workbenches/workbench_image_test.py

@@ -18,7 +18,7 @@
 
 import pytest
 
-from tests.containers import docker_utils
+from tests.containers import docker_utils, podman_machine_utils
 
 
 class TestWorkbenchImage:
@@ -53,10 +53,6 @@ def test_ipv6_only(self, image: str, test_frame):
         Workarounds for macOS will be needed, so that's why it's a separate test."""
         skip_if_not_workbench_image(image)
 
-        if platform.system().lower() == 'darwin':
-            pytest.skip("Podman on macOS does not support exposing IPv6 ports,"
-                        " see https://github.com/containers/podman/issues/15140")
-
         # network is made ipv6 by only defining the ipv6 subnet for it
         # do _not_ set the ipv6=true option, that would actually make it dual-stack
         # https://github.com/containers/podman/issues/22359#issuecomment-2196817604
@@ -87,8 +83,19 @@ def test_ipv6_only(self, image: str, test_frame):
                     container.get_wrapped_container().reload()
                     ipv6_address = (container.get_wrapped_container().attrs
                         ["NetworkSettings"]["Networks"][network.name]["GlobalIPv6Address"])
-
-                    container._connect(container_host=ipv6_address, container_port=container.port)
+                    if platform.system().lower() == 'darwin':
+                        # the container host is a podman machine, we need to expose port on podman machine first
+                        host = "localhost"
+                        port = podman_machine_utils.find_free_port()
+                        process = podman_machine_utils.open_ssh_tunnel("podman-machine-default",
+                                                                       local_port=port, remote_port=container.port,
+                                                                       remote_interface=f"[{ipv6_address}]")
+                        test_frame.append(process, lambda p: p.kill())
+                    else:
+                        host = ipv6_address
+                        port = container.port
+
+                    container._connect(container_host=host, container_port=port)
             finally:
                 # try to grab logs regardless of whether container started or not
                 stdout, stderr = container.get_logs()