tests: add pytest framework for vdkr and vpdmn

Add pytest-based test suite for testing vdkr and vpdmn CLI tools.
Tests use a separate state directory (~/.vdkr-test/) to avoid
interfering with production images.

Test files:
- conftest.py: Pytest fixtures for VdkrRunner and VpdmnRunner
- test_vdkr.py: Docker CLI tests (images, vimport, vrun, volumes, etc.)
- test_vpdmn.py: Podman CLI tests (mirrors vdkr test coverage)
- memres-test.sh: Helper script for running tests with memres
- pytest.ini: Pytest configuration and markers

Test categories:
- Basic operations: images, info, version
- Import/export: vimport, load, save
- Container execution: vrun, run, exec
- Storage management: system df, vstorage
- Memory resident mode: memres/vmemres start/stop/status

Running tests:
    pytest tests/test_vdkr.py -v --vdkr-dir /tmp/vcontainer-standalone
    pytest tests/test_vpdmn.py -v --vdkr-dir /tmp/vcontainer-standalone

Signed-off-by: Bruce Ashfield <bruce.ashfield@gmail.com>

1 files changed, 609 insertions, 0 deletions

diff --git a/tests/conftest.py b/tests/conftest.py
new file mode 100644
index 00000000..39d92322
--- /dev/null
+++ b/tests/conftest.py
@@ -0,0 +1,609 @@
+# SPDX-FileCopyrightText: Copyright (C) 2025 Bruce Ashfield
+#
+# SPDX-License-Identifier: MIT
+"""
+Pytest configuration and fixtures for vdkr, vpdmn and container-cross-install testing.
+
+Usage:
+    # Run all tests (default path: /tmp/vcontainer)
+    pytest tests/ --vdkr-dir /tmp/vcontainer
+
+    # Run vdkr tests only
+    pytest tests/test_vdkr.py -v --vdkr-dir /tmp/vcontainer
+
+    # Run vpdmn tests only
+    pytest tests/test_vpdmn.py -v --vdkr-dir /tmp/vcontainer
+
+    # Run with memres pre-started (faster)
+    ./tests/memres-test.sh start --vdkr-dir /tmp/vcontainer
+    pytest tests/test_vdkr.py --vdkr-dir /tmp/vcontainer --skip-destructive
+
+    # Run specific test
+    pytest tests/test_vdkr.py::TestMemresBasic -v --vdkr-dir /tmp/vcontainer
+
+Requirements:
+    pip install pytest
+
+Environment:
+    VDKR_STANDALONE_DIR: Path to extracted vdkr/vpdmn standalone tarball
+    VDKR_ARCH: Architecture to test (x86_64 or aarch64), default: x86_64
+
+Notes:
+    - Tests use separate state directories (~/.vdkr-test/, ~/.vpdmn-test/) to avoid
+      interfering with user's images in ~/.vdkr/ and ~/.vpdmn/.
+    - If memres is already running, tests reuse it and don't stop it at the end.
+    - Tests pull required images (alpine) automatically if not present.
+"""
+
+import os
+import subprocess
+import shutil
+import tempfile
+import signal
+import atexit
+import pytest
+from pathlib import Path
+
+
+# Test state directories - separate from user's ~/.vdkr/ and ~/.vpdmn/
+TEST_STATE_BASE = os.path.expanduser("~/.vdkr-test")
+VPDMN_TEST_STATE_BASE = os.path.expanduser("~/.vpdmn-test")
+
+# Track test memres PIDs for cleanup
+_test_memres_pids = set()
+
+
+def _cleanup_test_memres():
+    """
+    Clean up any test memres processes that may have been left running.
+    Called on exit (atexit) and signal handlers.
+    """
+    for state_base in [TEST_STATE_BASE, VPDMN_TEST_STATE_BASE]:
+        for arch_dir in Path(state_base).glob("*"):
+            pid_file = arch_dir / "daemon.pid"
+            if pid_file.exists():
+                try:
+                    pid = int(pid_file.read_text().strip())
+                    # Check if process is still running
+                    if Path(f"/proc/{pid}").exists():
+                        os.kill(pid, signal.SIGTERM)
+                        # Give it a moment to clean up
+                        import time
+                        time.sleep(0.5)
+                        # Force kill if still running
+                        if Path(f"/proc/{pid}").exists():
+                            os.kill(pid, signal.SIGKILL)
+                except (ValueError, ProcessLookupError, PermissionError):
+                    pass
+                # Remove stale PID file
+                try:
+                    pid_file.unlink()
+                except OSError:
+                    pass
+
+
+def _signal_handler(signum, frame):
+    """Handle SIGINT/SIGTERM by cleaning up test memres before exit."""
+    _cleanup_test_memres()
+    # Re-raise the signal to trigger default behavior
+    signal.signal(signum, signal.SIG_DFL)
+    os.kill(os.getpid(), signum)
+
+
+# Register cleanup handlers
+atexit.register(_cleanup_test_memres)
+signal.signal(signal.SIGINT, _signal_handler)
+signal.signal(signal.SIGTERM, _signal_handler)
+
+
+def pytest_addoption(parser):
+    """Add custom command line options."""
+    # vdkr/vpdmn options
+    parser.addoption(
+        "--vdkr-dir",
+        action="store",
+        default=os.environ.get("VDKR_STANDALONE_DIR", "/tmp/vcontainer"),
+        help="Path to vcontainer standalone directory",
+    )
+    parser.addoption(
+        "--arch",
+        action="store",
+        default=os.environ.get("VDKR_ARCH", "x86_64"),
+        choices=["x86_64", "aarch64"],
+        help="Target architecture to test",
+    )
+    parser.addoption(
+        "--oci-image",
+        action="store",
+        default=os.environ.get("TEST_OCI_IMAGE"),
+        help="Path to OCI image for import tests",
+    )
+    parser.addoption(
+        "--skip-destructive",
+        action="store_true",
+        default=False,
+        help="Skip tests that stop memres or clean state (useful when reusing test memres)",
+    )
+    # container-cross-install options
+    parser.addoption(
+        "--poky-dir",
+        action="store",
+        default=os.environ.get("POKY_DIR", "/opt/bruce/poky"),
+        help="Path to poky directory",
+    )
+    parser.addoption(
+        "--build-dir",
+        action="store",
+        default=os.environ.get("BUILD_DIR"),
+        help="Path to build directory",
+    )
+    parser.addoption(
+        "--machine",
+        action="store",
+        default=os.environ.get("MACHINE", "qemux86-64"),
+        help="Target machine",
+    )
+    parser.addoption(
+        "--image",
+        action="store",
+        default=os.environ.get("TEST_IMAGE", "container-image-host"),
+        help="Image to boot for container verification tests",
+    )
+    parser.addoption(
+        "--image-fstype",
+        action="store",
+        default=os.environ.get("TEST_IMAGE_FSTYPE", "ext4"),
+        help="Image filesystem type (default: ext4)",
+    )
+    parser.addoption(
+        "--boot-timeout",
+        action="store",
+        type=int,
+        default=120,
+        help="Timeout in seconds for image boot (default: 120)",
+    )
+    parser.addoption(
+        "--no-kvm",
+        action="store_true",
+        default=False,
+        help="Disable KVM acceleration",
+    )
+    parser.addoption(
+        "--fail-stale",
+        action="store_true",
+        default=False,
+        help="Fail if rootfs is stale (OCI containers or bbclass newer than rootfs)",
+    )
+    parser.addoption(
+        "--max-age",
+        action="store",
+        type=float,
+        default=24.0,
+        help="Max rootfs age in hours before warning (default: 24)",
+    )
+
+
+@pytest.fixture(scope="session")
+def vdkr_dir(request):
+    """Path to vdkr standalone directory."""
+    path = Path(request.config.getoption("--vdkr-dir"))
+    if not path.exists():
+        pytest.skip(f"vdkr standalone directory not found: {path}")
+    return path
+
+
+@pytest.fixture(scope="session")
+def arch(request):
+    """Target architecture."""
+    return request.config.getoption("--arch")
+
+
+@pytest.fixture(scope="session")
+def vdkr_bin(vdkr_dir, arch):
+    """Path to vdkr binary for the target architecture.
+
+    Tries arch-specific symlink first (vdkr-x86_64), then main vdkr binary.
+    """
+    # Try arch-specific symlink first
+    binary = vdkr_dir / f"vdkr-{arch}"
+    if binary.exists():
+        return binary
+
+    # Fall back to main vdkr binary
+    binary = vdkr_dir / "vdkr"
+    if binary.exists():
+        return binary
+
+    pytest.skip(f"vdkr binary not found: {vdkr_dir}/vdkr or {vdkr_dir}/vdkr-{arch}")
+
+
+@pytest.fixture(scope="session")
+def test_state_dir(arch):
+    """Test-specific state directory to avoid interfering with user's state."""
+    state_dir = Path(TEST_STATE_BASE) / arch
+    state_dir.mkdir(parents=True, exist_ok=True)
+    return state_dir
+
+
+@pytest.fixture(scope="session")
+def vdkr_env(vdkr_dir):
+    """Environment variables for running vdkr."""
+    env = os.environ.copy()
+
+    # Source init-env.sh equivalent
+    # Ensure vdkr_dir is a string for PATH concatenation
+    vdkr_path = str(vdkr_dir)
+
+    # Support both old layout (qemu/, lib/) and new SDK layout (sysroots/)
+    sysroot_dir = vdkr_dir / "sysroots" / "x86_64-pokysdk-linux"
+    if sysroot_dir.exists():
+        # New SDK layout: sysroots/x86_64-pokysdk-linux/usr/bin/
+        env["PATH"] = f"{vdkr_path}:{sysroot_dir}/usr/bin:/usr/bin:/bin:{env.get('PATH', '')}"
+        # No LD_LIBRARY_PATH needed - SDK uses proper RPATH
+    else:
+        # Old layout: qemu/, lib/
+        env["PATH"] = f"{vdkr_path}:{vdkr_path}/qemu:/usr/bin:/bin:{env.get('PATH', '')}"
+        env["LD_LIBRARY_PATH"] = f"{vdkr_path}/lib:{env.get('LD_LIBRARY_PATH', '')}"
+
+    return env
+
+
+@pytest.fixture(scope="session")
+def oci_image(request):
+    """Path to test OCI image, if available."""
+    path = request.config.getoption("--oci-image")
+    if path:
+        path = Path(path)
+        if not path.exists():
+            pytest.skip(f"OCI image not found: {path}")
+        return path
+    return None
+
+
+class VdkrRunner:
+    """Helper class for running vdkr commands."""
+
+    def __init__(self, binary: Path, env: dict, arch: str, state_dir: Path):
+        self.binary = binary
+        self.env = env
+        self.arch = arch
+        self.state_dir = state_dir
+        self._user_memres_was_running = None
+        # Check if we're using main vdkr (needs --arch) vs arch-specific symlink
+        self._needs_arch_flag = binary.name == "vdkr"
+
+    def run(self, *args, timeout=120, check=True, capture_output=True):
+        """Run a vdkr command with test state directory."""
+        cmd = [str(self.binary)]
+        if self._needs_arch_flag:
+            cmd.extend(["--arch", self.arch])
+        cmd.extend(["--state-dir", str(self.state_dir)])
+        cmd.extend(list(args))
+        result = subprocess.run(
+            cmd,
+            env=self.env,
+            timeout=timeout,
+            check=False,  # Don't raise immediately, check manually for better error messages
+            capture_output=capture_output,
+            text=True,
+        )
+        if check and result.returncode != 0:
+            error_msg = f"Command failed: {' '.join(cmd)}\n"
+            error_msg += f"Exit code: {result.returncode}\n"
+            if result.stdout:
+                error_msg += f"stdout: {result.stdout}\n"
+            if result.stderr:
+                error_msg += f"stderr: {result.stderr}\n"
+            # Print error so it's visible in test output
+            print(error_msg)
+            raise AssertionError(error_msg)
+        return result
+
+    def memres_start(self, timeout=120):
+        """Start memory resident mode."""
+        return self.run("memres", "start", timeout=timeout)
+
+    def memres_stop(self, timeout=30):
+        """Stop memory resident mode."""
+        return self.run("memres", "stop", timeout=timeout, check=False)
+
+    def memres_status(self):
+        """Check memory resident status."""
+        return self.run("memres", "status", check=False)
+
+    def is_memres_running(self):
+        """Check if memres is running (in test state dir)."""
+        result = self.memres_status()
+        return result.returncode == 0 and "running" in result.stdout.lower()
+
+    def ensure_memres(self, timeout=180):
+        """Ensure memres is running, starting it if needed."""
+        if not self.is_memres_running():
+            result = self.memres_start(timeout=timeout)
+            if result.returncode != 0:
+                raise RuntimeError(f"Failed to start memres: {result.stderr}")
+
+    def is_user_memres_running(self):
+        """Check if user's memres is running (in default ~/.vdkr/)."""
+        # Check without --state-dir to see user's memres
+        cmd = [str(self.binary)]
+        if self._needs_arch_flag:
+            cmd.extend(["--arch", self.arch])
+        cmd.extend(["memres", "status"])
+        result = subprocess.run(
+            cmd, env=self.env, capture_output=True, text=True, timeout=10
+        )
+        return result.returncode == 0 and "running" in result.stdout.lower()
+
+    def images(self, timeout=120):
+        """List images."""
+        return self.run("images", timeout=timeout)
+
+    def clean(self):
+        """Clean state."""
+        return self.run("clean", check=False)
+
+    def vimport(self, path, name, timeout=120):
+        """Import an OCI image."""
+        return self.run("vimport", str(path), name, timeout=timeout)
+
+    def pull(self, image, timeout=180):
+        """Pull an image from registry."""
+        return self.run("pull", image, timeout=timeout)
+
+    def rmi(self, image, timeout=60):
+        """Remove an image."""
+        return self.run("rmi", image, timeout=timeout, check=False)
+
+    def vrun(self, image, *cmd, timeout=120):
+        """Run a command in a container."""
+        return self.run("vrun", image, *cmd, timeout=timeout)
+
+    def inspect(self, target, timeout=60):
+        """Inspect an image or container."""
+        return self.run("inspect", target, timeout=timeout)
+
+    def save(self, output_file, image, timeout=120):
+        """Save an image to a tar file."""
+        return self.run("save", "-o", str(output_file), image, timeout=timeout)
+
+    def load(self, input_file, timeout=120):
+        """Load an image from a tar file."""
+        return self.run("load", "-i", str(input_file), timeout=timeout)
+
+    def has_image(self, image_name):
+        """Check if an image exists."""
+        self.ensure_memres()
+        result = self.images()
+        return image_name.split(":")[0] in result.stdout
+
+    def ensure_alpine(self, timeout=300):
+        """Ensure alpine:latest is available, pulling if necessary."""
+        # Ensure memres is running first (in case a previous test stopped it)
+        self.ensure_memres()
+        if not self.has_image("alpine"):
+            self.pull("alpine:latest", timeout=timeout)
+
+
+@pytest.fixture(scope="session")
+def vdkr(vdkr_bin, vdkr_env, arch, test_state_dir):
+    """VdkrRunner instance for running vdkr commands."""
+    return VdkrRunner(vdkr_bin, vdkr_env, arch, test_state_dir)
+
+
+@pytest.fixture(scope="session")
+def memres_session(vdkr):
+    """
+    Session-scoped fixture that ensures memres is running for tests.
+    Uses separate test state directory (~/.vdkr-test/).
+
+    Note: TestMemresBasic tests may stop/restart memres during the session.
+    Tests using this fixture should call ensure_memres() or ensure_alpine()
+    to guarantee memres is running before executing commands.
+    """
+    # Check if memres was already running at session start
+    was_running_at_start = vdkr.is_memres_running()
+
+    # Ensure memres is running
+    vdkr.ensure_memres()
+
+    yield vdkr
+
+    # Only stop memres if it wasn't running when we started
+    if not was_running_at_start:
+        vdkr.memres_stop()
+
+
+@pytest.fixture
+def temp_dir():
+    """Create a temporary directory for test files."""
+    tmpdir = tempfile.mkdtemp(prefix="vdkr-test-")
+    yield Path(tmpdir)
+    shutil.rmtree(tmpdir, ignore_errors=True)
+
+
+# Markers
+def pytest_configure(config):
+    """Register custom markers."""
+    config.addinivalue_line(
+        "markers", "slow: marks tests as slow (deselect with '-m \"not slow\"')"
+    )
+    config.addinivalue_line(
+        "markers", "memres: marks tests that require memory resident mode"
+    )
+    config.addinivalue_line(
+        "markers", "network: marks tests that require network access"
+    )
+
+
+# ============================================================================
+# vpdmn (Podman) fixtures
+# ============================================================================
+
+@pytest.fixture(scope="session")
+def vpdmn_bin(vdkr_dir, arch):
+    """Path to vpdmn binary for the target architecture.
+
+    Tries arch-specific symlink first (vpdmn-x86_64), then main vpdmn binary.
+    """
+    # Try arch-specific symlink first
+    binary = vdkr_dir / f"vpdmn-{arch}"
+    if binary.exists():
+        return binary
+
+    # Fall back to main vpdmn binary
+    binary = vdkr_dir / "vpdmn"
+    if binary.exists():
+        return binary
+
+    pytest.skip(f"vpdmn binary not found: {vdkr_dir}/vpdmn or {vdkr_dir}/vpdmn-{arch}")
+
+
+@pytest.fixture(scope="session")
+def vpdmn_test_state_dir(arch):
+    """Test-specific state directory for vpdmn to avoid interfering with user's state."""
+    state_dir = Path(VPDMN_TEST_STATE_BASE) / arch
+    state_dir.mkdir(parents=True, exist_ok=True)
+    return state_dir
+
+
+class VpdmnRunner:
+    """Helper class for running vpdmn commands."""
+
+    def __init__(self, binary: Path, env: dict, arch: str, state_dir: Path):
+        self.binary = binary
+        self.env = env
+        self.arch = arch
+        self.state_dir = state_dir
+        self._user_memres_was_running = None
+        # Check if we're using main vpdmn (needs --arch) vs arch-specific symlink
+        self._needs_arch_flag = binary.name == "vpdmn"
+
+    def run(self, *args, timeout=120, check=True, capture_output=True):
+        """Run a vpdmn command with test state directory."""
+        cmd = [str(self.binary)]
+        if self._needs_arch_flag:
+            cmd.extend(["--arch", self.arch])
+        cmd.extend(["--state-dir", str(self.state_dir)])
+        cmd.extend(list(args))
+        result = subprocess.run(
+            cmd,
+            env=self.env,
+            timeout=timeout,
+            check=False,  # Don't raise immediately, check manually for better error messages
+            capture_output=capture_output,
+            text=True,
+        )
+        if check and result.returncode != 0:
+            error_msg = f"Command failed: {' '.join(cmd)}\n"
+            error_msg += f"Exit code: {result.returncode}\n"
+            if result.stdout:
+                error_msg += f"stdout: {result.stdout}\n"
+            if result.stderr:
+                error_msg += f"stderr: {result.stderr}\n"
+            # Print error so it's visible in test output
+            print(error_msg)
+            raise AssertionError(error_msg)
+        return result
+
+    def memres_start(self, timeout=120):
+        """Start memory resident mode."""
+        return self.run("memres", "start", timeout=timeout)
+
+    def memres_stop(self, timeout=30):
+        """Stop memory resident mode."""
+        return self.run("memres", "stop", timeout=timeout, check=False)
+
+    def memres_status(self):
+        """Check memory resident status."""
+        return self.run("memres", "status", check=False)
+
+    def is_memres_running(self):
+        """Check if memres is running (in test state dir)."""
+        result = self.memres_status()
+        return result.returncode == 0 and "running" in result.stdout.lower()
+
+    def ensure_memres(self, timeout=180):
+        """Ensure memres is running, starting it if needed."""
+        if not self.is_memres_running():
+            result = self.memres_start(timeout=timeout)
+            if result.returncode != 0:
+                raise RuntimeError(f"Failed to start memres: {result.stderr}")
+
+    def images(self, timeout=120):
+        """List images."""
+        return self.run("images", timeout=timeout)
+
+    def clean(self):
+        """Clean state."""
+        return self.run("clean", check=False)
+
+    def vimport(self, path, name, timeout=120):
+        """Import an OCI image."""
+        return self.run("vimport", str(path), name, timeout=timeout)
+
+    def pull(self, image, timeout=180):
+        """Pull an image from registry."""
+        return self.run("pull", image, timeout=timeout)
+
+    def rmi(self, image, timeout=60):
+        """Remove an image."""
+        return self.run("rmi", image, timeout=timeout, check=False)
+
+    def vrun(self, image, *cmd, timeout=120):
+        """Run a command in a container."""
+        return self.run("vrun", image, *cmd, timeout=timeout)
+
+    def inspect(self, target, timeout=60):
+        """Inspect an image or container."""
+        return self.run("inspect", target, timeout=timeout)
+
+    def save(self, output_file, image, timeout=120):
+        """Save an image to a tar file."""
+        return self.run("save", "-o", str(output_file), image, timeout=timeout)
+
+    def load(self, input_file, timeout=120):
+        """Load an image from a tar file."""
+        return self.run("load", "-i", str(input_file), timeout=timeout)
+
+    def has_image(self, image_name):
+        """Check if an image exists."""
+        self.ensure_memres()
+        result = self.images()
+        return image_name.split(":")[0] in result.stdout
+
+    def ensure_alpine(self, timeout=300):
+        """Ensure alpine:latest is available, pulling if necessary."""
+        # Ensure memres is running first (in case a previous test stopped it)
+        self.ensure_memres()
+        if not self.has_image("alpine"):
+            self.pull("alpine:latest", timeout=timeout)
+
+
+@pytest.fixture(scope="session")
+def vpdmn(vpdmn_bin, vdkr_env, arch, vpdmn_test_state_dir):
+    """VpdmnRunner instance for running vpdmn commands."""
+    return VpdmnRunner(vpdmn_bin, vdkr_env, arch, vpdmn_test_state_dir)
+
+
+@pytest.fixture(scope="session")
+def vpdmn_memres_session(vpdmn):
+    """
+    Session-scoped fixture that ensures memres is running for vpdmn tests.
+    Uses separate test state directory (~/.vpdmn-test/).
+
+    Note: TestMemresBasic tests may stop/restart memres during the session.
+    Tests using this fixture should call ensure_memres() or ensure_alpine()
+    to guarantee memres is running before executing commands.
+    """
+    # Check if memres was already running at session start
+    was_running_at_start = vpdmn.is_memres_running()
+
+    # Ensure memres is running
+    vpdmn.ensure_memres()
+
+    yield vpdmn
+
+    # Only stop memres if it wasn't running when we started
+    if not was_running_at_start:
+        vpdmn.memres_stop()