RHOAIENG-18459: chore(tests/containers): improve podman machine handling: detect the right one to use if user has multiple machines around

Author: jiridanek

tests/containers/conftest.py

@@ -10,6 +10,8 @@
 
 import pytest
 
+from tests.containers import docker_utils
+
 if TYPE_CHECKING:
     from pytest import ExitCode, Session, Parser, Metafunc
 
@@ -58,7 +60,7 @@ def pytest_sessionstart(session: Session) -> None:
 
     # determine the local socket path
     # NOTE: this will not work for remote docker, but we will cross the bridge when we come to it
-    socket_path = the_one(adapter.socket_path for adapter in client.client.api.adapters.values())
+    socket_path = docker_utils.get_socket_path(client.client)
 
     # set that socket path for ryuk's use, unless user overrode that
     if TESTCONTAINERS_DOCKER_SOCKET_OVERRIDE not in os.environ:
@@ -84,21 +86,6 @@ def pytest_sessionfinish(session: Session, exitstatus: int | ExitCode) -> None:
         testcontainers.core.container.Reaper.delete_instance()
 
 
-# https://docs.python.org/3/library/functions.html#iter
-def the_one[T](iterable: Iterable[T]) -> T:
-    """Checks that there is exactly one element in the iterable, and returns it."""
-    it = iter(iterable)
-    try:
-        v = next(it)
-    except StopIteration:
-        raise ValueError("No elements in iterable")
-    try:
-        next(it)
-    except StopIteration:
-        return v
-    raise ValueError("More than one element in iterable")
-
-
 @pytest.fixture(scope="function")
 def test_frame():
     class TestFrame:

tests/containers/docker_utils.py

@@ -6,7 +6,9 @@
 import sys
 import tarfile
 import time
-from typing import TYPE_CHECKING
+from typing import Iterable, TYPE_CHECKING
+
+import docker.client
 
 import testcontainers.core.container
 
@@ -133,10 +135,10 @@ def __init__(self, client, id, output: list[int] | list[str]):
     def inspect(self):
         return self.client.api.exec_inspect(self.id)
 
-    def poll(self):
+    def poll(self) -> int:
         return self.inspect()["ExitCode"]
 
-    def communicate(self, line_prefix=b""):
+    def communicate(self, line_prefix=b"") -> int:
         for data in self.output:
             if not data:
                 continue
@@ -155,3 +157,31 @@ def communicate(self, line_prefix=b""):
         while self.poll() is None:
             raise RuntimeError("Hm could that really happen?")
         return self.poll()
+
+def get_socket_path(client: docker.client.DockerClient) -> str:
+    """Determine the local socket path.
+    This works even when `podman machine` with its own host-mounts is involved
+    NOTE: this will not work for remote docker, but we will cross the bridge when we come to it"""
+    socket_path = _the_one(adapter.socket_path for adapter in client.api.adapters.values())
+    return socket_path
+
+def get_container_pid(container: Container) -> int | None:
+    """Get the network namespace of a Docker container."""
+    container.reload()
+    container_pid = container.attrs['State']['Pid']
+    return container_pid
+
+
+# https://docs.python.org/3/library/functions.html#iter
+def _the_one[T](iterable: Iterable[T]) -> T:
+    """Checks that there is exactly one element in the iterable, and returns it."""
+    it = iter(iterable)
+    try:
+        v = next(it)
+    except StopIteration:
+        raise ValueError("No elements in iterable")
+    try:
+        next(it)
+    except StopIteration:
+        return v
+    raise ValueError("More than one element in iterable")

tests/containers/podman_machine_utils.py

@@ -2,19 +2,22 @@
 import logging
 import socket
 import subprocess
-from tests.containers.schemas import PodmanMachineInspect
+from typing import Callable
+
+import tests.containers.schemas
 
 logging.basicConfig(level=logging.DEBUG)
 
-def open_ssh_tunnel(machine_name: str, local_port: int, remote_port: int, remote_interface: str = "localhost") -> subprocess.Popen:
+def open_ssh_tunnel(machine_predicate: Callable[[tests.containers.schemas.PodmanMachine], bool],
+                    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))
+    json_data = subprocess.check_output(["podman", "machine", "inspect"], text=True)
+    inspect = tests.containers.schemas.PodmanMachineInspect(machines=json.loads(json_data))
     machines = inspect.machines
 
-    machine = next((m for m in machines if m.Name == machine_name), None)
+    machine = next((m for m in machines if machine_predicate(m)), None)
     if not machine:
-        raise ValueError(f"Machine '{machine_name}' not found")
+        raise ValueError(f"Machine matching given predicate not found")
 
     ssh_command = [
         "ssh",
@@ -30,7 +33,7 @@ def open_ssh_tunnel(machine_name: str, local_port: int, remote_port: int, remote
     # 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}")
+    logging.info(f"SSH tunnel opened for {machine.Name}: {remote_interface}:{local_port} -> localhost:{remote_port}")
     return process
 
 def find_free_port() -> int:
@@ -46,8 +49,8 @@ def find_free_port() -> int:
 # Usage example
 if __name__ == "__main__":
     tunnel_process = open_ssh_tunnel(
-        "podman-machine-default",
-        8080, 8080,
+        machine_predicate=lambda m: m.Name == "podman-machine-default",
+        local_port=8080, remote_port=8080,
         remote_interface="[fc00::2]"
     )
 

tests/containers/workbenches/workbench_image_test.py

@@ -3,6 +3,7 @@
 import functools
 import http.cookiejar
 import logging
+import os
 import platform
 import urllib.error
 import urllib.request
@@ -87,9 +88,11 @@ def test_ipv6_only(self, image: str, test_frame):
                         # 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}]")
+                        socket_path = os.path.realpath(docker_utils.get_socket_path(client.client))
+                        process = podman_machine_utils.open_ssh_tunnel(
+                            machine_predicate=lambda m: os.path.realpath(m.ConnectionInfo.PodmanSocket.Path) == socket_path,
+                            local_port=port, remote_port=container.port,
+                            remote_interface=f"[{ipv6_address}]")
                         test_frame.append(process, lambda p: p.kill())
                     else:
                         host = ipv6_address