Container testing framework

.config/dictionary.txt

@@ -1,4 +1,6 @@
 accesslog
+addinivalue
+addoption
 adt
 ansibuddy
 antsibull
@@ -13,6 +15,7 @@ endgroup
 gunicorn
 libera
 microdnf
+modifyitems
 netcommon
 pkgmgr
 pylibssh

tests/conftest.py

@@ -18,10 +18,13 @@
 """
 
 import os
+import shutil
 import subprocess
 import sys
 import time
 
+from collections.abc import Callable
+from dataclasses import dataclass
 from pathlib import Path
 
 import pytest
@@ -30,7 +33,137 @@
 import ansible_dev_tools  # noqa: F401
 
 
-PROC: None | subprocess.Popen[bytes] = None
+@dataclass
+class Infrastructure:
+    """Structure for instance infrastructure.
+
+    Attributes:
+        session: The pytest session
+        container_engine: The container engine
+        container_name: The container name
+        container: Container required
+        image_name: The image name
+        include_container: Include container tests
+        only_container: Only container tests
+        proc: The server process
+        server: Server required
+    """
+
+    session: pytest.Session
+    container_engine: str = ""
+    container_name: str = ""
+    container: bool = False
+    image_name: str = ""
+    include_container: bool = False
+    only_container: bool = False
+    proc: None | subprocess.Popen[bytes] = None
+    server: bool = False
+
+    def __post_init__(self) -> None:
+        """Initialize the infrastructure.
+
+        Raises:
+            ValueError: If the container engine is not found.
+            ValueError: If the container name is not set.
+            ValueError: If both only_container and include_container are set.
+        """
+        self.container_engine = self.session.config.getoption("--container-engine")
+        self.container_name = self.session.config.getoption("--container-name", "")
+        self.image_name = self.session.config.getoption("--image-name", "")
+        self.include_container = self.session.config.getoption("--include-container")
+        self.only_container = self.session.config.getoption("--only-container")
+        if self.only_container or self.include_container:
+            if not self.container_name:
+                err = "ADT_CONTAINER_NAME must be set for container tests"
+                raise ValueError(err)
+            if not self.container_engine:
+                err = "No container engine found, required for container tests"
+                raise ValueError(err)
+        elif self.only_container and self.include_container:
+            err = "Cannot use both --only-container and --include-container"
+            raise ValueError(err)
+
+        if self.only_container:
+            self.container = True
+            self.server = False
+        elif self.include_container:
+            self.container = True
+            self.server = True
+        else:
+            self.container = False
+            self.server = True
+
+
+INFRASTRUCTURE: Infrastructure
+
+
+def pytest_addoption(parser: pytest.Parser) -> None:
+    """Add options to pytest.
+
+    Args:
+        parser: The pytest parser.
+    """
+    parser.addoption(
+        "--container-engine",
+        action="store",
+        default=os.environ.get(
+            "ADT_CONTAINER_ENGINE",
+            shutil.which("podman") or shutil.which("docker") or "",
+        ),
+        help="Container engine to use. (default=ADT_CONTAINER_ENGINE, podman, docker, '')",
+    )
+    parser.addoption(
+        "--container-name",
+        action="store",
+        default=os.environ.get("ADT_CONTAINER_NAME", "adt-test-container"),
+        help="Container name to use for the running container. (default=ADT_CONTAINER_NAME)",
+    )
+    parser.addoption(
+        "--image-name",
+        action="store",
+        default=os.environ.get("ADT_IMAGE_NAME", ""),
+        help="Container name to use. (default=ADT_IMAGE_NAME)",
+    )
+    parser.addoption(
+        "--only-container",
+        action="store_true",
+        default=False,
+        help="Only run container tests",
+    )
+    parser.addoption(
+        "--include-container",
+        action="store_true",
+        default=False,
+        help="Include container tests",
+    )
+
+
+def pytest_configure(config: pytest.Config) -> None:
+    """Configure pytest.
+
+    Args:
+        config: The pytest configuration.
+    """
+    config.addinivalue_line("markers", "container: container tests")
+
+
+def pytest_collection_modifyitems(config: pytest.Config, items: list[pytest.Item]) -> None:
+    """Modify the collection of items.
+
+    Args:
+        config: The pytest configuration.
+        items: The list of items.
+    """
+    if config.getoption("--only-container"):
+        skip_container = pytest.mark.skip(reason="--only-container specified")
+        for item in items:
+            if "container" not in item.keywords:
+                item.add_marker(skip_container)
+    elif not config.getoption("--include-container"):
+        skip_container = pytest.mark.skip(reason="need --include-container option to run")
+        for item in items:
+            if "container" in item.keywords:
+                item.add_marker(skip_container)
 
 
 @pytest.fixture(scope="session")
@@ -48,18 +181,140 @@ def pytest_sessionstart(session: pytest.Session) -> None:
 
     Args:
         session: The pytest session.
-
-    Raises:
-        RuntimeError: If the server could not be started.
     """
     assert session
-    bin_path = Path(sys.executable).parent / "adt"
 
     if os.environ.get("PYTEST_XDIST_WORKER"):
         return
 
-    global PROC  # noqa: PLW0603
-    PROC = subprocess.Popen(  # noqa: S603
+    global INFRASTRUCTURE  # noqa: PLW0603
+
+    INFRASTRUCTURE = Infrastructure(session)
+
+    if INFRASTRUCTURE.container:
+        _start_container()
+    if INFRASTRUCTURE.server:
+        _start_server()
+
+
+def pytest_sessionfinish(session: pytest.Session) -> None:
+    """Stop the server.
+
+    Args:
+        session: The pytest session.
+    """
+    assert session
+    if os.environ.get("PYTEST_XDIST_WORKER"):
+        return
+
+    if INFRASTRUCTURE.container:
+        _stop_container()
+    if INFRASTRUCTURE.server:
+        _stop_server()
+
+
+PODMAN_CMD = """{container_engine} run -d --rm
+ --cap-add=SYS_ADMIN
+ --cap-add=SYS_RESOURCE
+ --device "/dev/fuse"
+ --hostname=ansible-dev-container
+ --name={container_name}
+ --security-opt "apparmor=unconfined"
+ --security-opt "label=disable"
+ --security-opt "seccomp=unconfined"
+ --user=root
+ --userns=host
+ -v $PWD:/workdir
+ {image_name}
+ sleep infinity"""
+
+DOCKER_CMD = """{container_engine} run -d --rm
+ --cap-add=SYS_ADMIN
+ --cap-add=SYS_RESOURCE
+ --device "/dev/fuse"
+ --hostname=ansible-dev-container
+ --name={container_name}
+ --security-opt "apparmor=unconfined"
+ --security-opt "label=disable"
+ --security-opt "seccomp=unconfined"
+ --user=podman
+ -v $PWD:/workdir
+ {image_name}
+ sleep infinity"""
+
+
+def _start_container() -> None:
+    """Start the container.
+
+    Raises:
+        ValueError: If the container engine is not podman or docker.
+    """
+    if "podman" in INFRASTRUCTURE.container_engine:
+        cmd = PODMAN_CMD.format(
+            container_engine=INFRASTRUCTURE.container_engine,
+            container_name=INFRASTRUCTURE.container_name,
+            image_name=INFRASTRUCTURE.image_name,
+        )
+    elif "docker" in INFRASTRUCTURE.container_engine:
+        cmd = DOCKER_CMD.format(
+            container_engine=INFRASTRUCTURE.container_engine,
+            container_name=INFRASTRUCTURE.container_name,
+            image_name=INFRASTRUCTURE.image_name,
+        )
+    else:
+        err = f"Container engine {INFRASTRUCTURE.container_engine} not found."
+        raise ValueError(err)
+    cmd = cmd.replace("\n", " ")
+    subprocess.run(cmd, check=True, capture_output=True, shell=True)
+
+
+def _stop_container() -> None:
+    """Stop the container."""
+    cmd = [
+        INFRASTRUCTURE.container_engine,
+        "stop",
+        INFRASTRUCTURE.container_name,
+    ]
+    subprocess.run(cmd, check=True, capture_output=True)  # noqa: S603
+
+
+def _exec_container(command: str) -> subprocess.CompletedProcess[str]:
+    """Run the container.
+
+    Args:
+        command: The command to run
+
+    Returns:
+        subprocess.CompletedProcess: The completed process.
+    """
+    cmd = f"{INFRASTRUCTURE.container_engine} exec -t {INFRASTRUCTURE.container_name} {command}"
+    return subprocess.run(
+        cmd,
+        check=True,
+        capture_output=True,
+        text=True,
+        shell=True,
+    )
+
+
+@pytest.fixture()
+def exec_container() -> Callable[[str], subprocess.CompletedProcess[str]]:
+    """Run the container.
+
+    Returns:
+        callable: The container executor.
+    """
+    return _exec_container
+
+
+def _start_server() -> None:
+    """Start the server.
+
+    Raises:
+        RuntimeError: If the server could not be started.
+    """
+    bin_path = Path(sys.executable).parent / "adt"
+    INFRASTRUCTURE.proc = subprocess.Popen(  # noqa: S603
         [bin_path, "server", "-p", "8000"],
         env=os.environ,
     )
@@ -78,23 +333,15 @@ def pytest_sessionstart(session: pytest.Session) -> None:
     raise RuntimeError(msg)
 
 
-def pytest_sessionfinish(session: pytest.Session) -> None:
+def _stop_server() -> None:
     """Stop the server.
 
-    Args:
-        session: The pytest session.
-
     Raises:
-        RuntimeError: If the server could not be stopped.
+        RuntimeError: If the server is not running.
     """
-    assert session
-    if os.environ.get("PYTEST_XDIST_WORKER"):
-        return
-
-    global PROC  # noqa: PLW0603
-    if PROC is None:
+    if INFRASTRUCTURE.proc is None:
         msg = "The server is not running."
         raise RuntimeError(msg)
-    PROC.terminate()
-    PROC.wait()
-    PROC = None
+    INFRASTRUCTURE.proc.terminate()
+    INFRASTRUCTURE.proc.wait()
+    INFRASTRUCTURE.proc = None

tests/integration/test_container.py

@@ -0,0 +1,21 @@
+"""Run tests against the container."""
+
+import subprocess
+
+from collections.abc import Callable
+
+import pytest
+
+from ansible_dev_tools.version_builder import PKGS
+
+
+@pytest.mark.container()
+def test_versions(exec_container: Callable[[str], subprocess.CompletedProcess[str]]) -> None:
+    """Test the versions.
+
+    Args:
+        exec_container: The container executor.
+    """
+    versions = exec_container("adt --version")
+    for pkg in PKGS:
+        assert pkg in versions.stdout, f"{pkg} not found in version output"