tests: add container registry pytest tests

Add pytest tests for registry functionality:
- test_vdkr_registry.py: vconfig registry, image commands, CLI override
- test_container_registry_script.py: start/stop/push/import/list/tags
- conftest.py: --registry-url, --registry-script options

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

5 files changed, 1106 insertions, 5 deletions

diff --git a/recipes-containers/container-registry/container-oci-registry-config.bb b/recipes-containers/container-registry/container-oci-registry-config.bb
index ee6760f4..ba8cf4c3 100644
--- a/recipes-containers/container-registry/container-oci-registry-config.bb
+++ b/recipes-containers/container-registry/container-oci-registry-config.bb
@@ -22,16 +22,18 @@
 #
 # IMPORTANT: This recipe:
 #   - Does NOT modify docker-distribution or container-host-config
-#   - Does NOT install automatically - user must add to IMAGE_INSTALL
 #   - Does NOT clobber public registry access (docker.io, quay.io, etc.)
 #   - Uses drop-in files in /etc/containers/registries.conf.d/
 #   - Skips entirely if CONTAINER_REGISTRY_URL is not set
 #
 # Usage:
-#   # In local.conf or image recipe - BOTH required:
+#   # In local.conf or image recipe:
 #   CONTAINER_REGISTRY_URL = "localhost:5000"
 #   CONTAINER_REGISTRY_INSECURE = "1"
-#   IMAGE_INSTALL:append = " container-oci-registry-config"
+#   IMAGE_FEATURES += "container-registry"
+#
+#   The IMAGE_FEATURES mechanism auto-selects this recipe for Podman/CRI-O
+#   or docker-registry-config for Docker based on VIRTUAL-RUNTIME_container_engine.
 #
 # ===========================================================================
 
diff --git a/recipes-containers/container-registry/docker-registry-config.bb b/recipes-containers/container-registry/docker-registry-config.bb
index eee74c98..cbe2e13f 100644
--- a/recipes-containers/container-registry/docker-registry-config.bb
+++ b/recipes-containers/container-registry/docker-registry-config.bb
@@ -20,7 +20,6 @@
 # This config only handles insecure registry trust.
 #
 # IMPORTANT: This recipe:
-#   - Does NOT install automatically - user must add to IMAGE_INSTALL
 #   - Skips entirely if DOCKER_REGISTRY_INSECURE is not set
 #   - Creates /etc/docker/daemon.json (will be merged if docker recipe
 #     also creates one, or may need RCONFLICTS handling)
@@ -28,7 +27,11 @@
 # Usage:
 #   # In local.conf or image recipe:
 #   DOCKER_REGISTRY_INSECURE = "10.0.2.2:5000 myregistry.local:5000"
-#   IMAGE_INSTALL:append = " docker-registry-config"
+#   IMAGE_FEATURES += "container-registry"
+#
+#   The IMAGE_FEATURES mechanism auto-selects this recipe for Docker
+#   or container-oci-registry-config for Podman/CRI-O based on
+#   VIRTUAL-RUNTIME_container_engine.
 #
 # ===========================================================================
 
diff --git a/tests/conftest.py b/tests/conftest.py
index 712700ee..accb8f17 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -226,6 +226,25 @@ def pytest_addoption(parser):
         default=24.0,
         help="Max rootfs age in hours before warning (default: 24)",
     )
+    # Container registry options
+    parser.addoption(
+        "--registry-url",
+        action="store",
+        default=os.environ.get("TEST_REGISTRY_URL"),
+        help="Registry URL for vdkr registry tests (e.g., 10.0.2.2:5000/yocto)",
+    )
+    parser.addoption(
+        "--registry-script",
+        action="store",
+        default=os.environ.get("CONTAINER_REGISTRY_SCRIPT"),
+        help="Path to container-registry.sh script",
+    )
+    parser.addoption(
+        "--skip-registry-network",
+        action="store_true",
+        default=False,
+        help="Skip registry tests that require network access to docker.io",
+    )
 
 
 def _cleanup_stale_test_state():
diff --git a/tests/test_container_registry_script.py b/tests/test_container_registry_script.py
new file mode 100644
index 00000000..444c5d3b
--- /dev/null
+++ b/tests/test_container_registry_script.py
@@ -0,0 +1,663 @@
+# SPDX-FileCopyrightText: Copyright (C) 2025 Bruce Ashfield
+#
+# SPDX-License-Identifier: MIT
+"""
+Tests for container-registry.sh helper script.
+
+These tests verify the container registry helper script commands:
+- start/stop/status - Registry server lifecycle
+- push - Push OCI images to registry (with tag/strategy options)
+- import - Import 3rd party images
+- delete - Delete tagged images
+- gc - Garbage collection
+- list/tags/catalog - Query registry contents
+
+Prerequisites:
+    # Generate the script first:
+    bitbake container-registry-index -c generate_registry_script
+
+    # The script location is:
+    $TOPDIR/container-registry/container-registry.sh
+
+Run with:
+    pytest tests/test_container_registry_script.py -v
+
+Run with specific registry script:
+    pytest tests/test_container_registry_script.py -v \\
+        --registry-script /path/to/container-registry.sh
+
+Environment variables:
+    CONTAINER_REGISTRY_SCRIPT: Path to the registry script
+    TOPDIR: Yocto build directory (script at $TOPDIR/container-registry/)
+"""
+
+import pytest
+import subprocess
+import os
+import time
+from pathlib import Path
+
+
+# Note: Registry options (--registry-script, --skip-registry-network)
+# are defined in conftest.py
+
+
+@pytest.fixture(scope="module")
+def registry_script(request):
+    """Get path to the registry script.
+
+    Looks in order:
+    1. --registry-script command line option
+    2. CONTAINER_REGISTRY_SCRIPT environment variable
+    3. $TOPDIR/container-registry/container-registry.sh
+    4. Common locations based on cwd
+    """
+    # Check command line option
+    script_path = request.config.getoption("--registry-script", default=None)
+
+    if script_path is None:
+        # Check environment variable
+        script_path = os.environ.get("CONTAINER_REGISTRY_SCRIPT")
+
+    if script_path is None:
+        # Try TOPDIR-based path
+        topdir = os.environ.get("TOPDIR")
+        if topdir:
+            script_path = os.path.join(topdir, "container-registry", "container-registry.sh")
+
+    if script_path is None:
+        # Try common locations relative to cwd
+        candidates = [
+            "container-registry/container-registry.sh",
+            "../container-registry/container-registry.sh",
+            "build/container-registry/container-registry.sh",
+        ]
+        for candidate in candidates:
+            if os.path.exists(candidate):
+                script_path = candidate
+                break
+
+    if script_path is None or not os.path.exists(script_path):
+        pytest.skip(
+            "Registry script not found. Generate it with: "
+            "bitbake container-registry-index -c generate_registry_script\n"
+            "Or specify path with --registry-script or CONTAINER_REGISTRY_SCRIPT env var"
+        )
+
+    script_path = Path(script_path).resolve()
+    if not script_path.exists():
+        pytest.skip(f"Registry script not found at: {script_path}")
+
+    return script_path
+
+
+@pytest.fixture(scope="module")
+def skip_network(request):
+    """Check if network tests should be skipped."""
+    return request.config.getoption("--skip-registry-network", default=False)
+
+
+class RegistryScriptRunner:
+    """Helper class for running registry script commands."""
+
+    def __init__(self, script_path: Path):
+        self.script_path = script_path
+        self._was_running = None
+
+    def run(self, *args, timeout=30, check=True, capture_output=True):
+        """Run a registry script command."""
+        cmd = [str(self.script_path)] + list(args)
+        result = subprocess.run(
+            cmd,
+            timeout=timeout,
+            check=False,
+            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"
+            raise AssertionError(error_msg)
+        return result
+
+    def start(self, timeout=30):
+        """Start the registry."""
+        return self.run("start", timeout=timeout)
+
+    def stop(self, timeout=10):
+        """Stop the registry."""
+        return self.run("stop", timeout=timeout, check=False)
+
+    def status(self, timeout=10):
+        """Check registry status."""
+        return self.run("status", timeout=timeout, check=False)
+
+    def is_running(self):
+        """Check if registry is running."""
+        result = self.status()
+        return result.returncode == 0 and "running" in result.stdout.lower()
+
+    def ensure_running(self, timeout=30):
+        """Ensure registry is running, starting if needed."""
+        if not self.is_running():
+            result = self.start(timeout=timeout)
+            if result.returncode != 0:
+                raise RuntimeError(f"Failed to start registry: {result.stderr}")
+            time.sleep(2)
+
+    def push(self, timeout=120):
+        """Push OCI images to registry."""
+        return self.run("push", timeout=timeout)
+
+    def import_image(self, source, dest_name=None, timeout=300):
+        """Import a 3rd party image."""
+        args = ["import", source]
+        if dest_name:
+            args.append(dest_name)
+        return self.run(*args, timeout=timeout)
+
+    def list_images(self, timeout=30):
+        """List images in registry."""
+        return self.run("list", timeout=timeout)
+
+    def tags(self, image, timeout=30):
+        """Get tags for an image."""
+        return self.run("tags", image, timeout=timeout, check=False)
+
+    def catalog(self, timeout=30):
+        """Get raw catalog."""
+        return self.run("catalog", timeout=timeout)
+
+    def help(self):
+        """Show help."""
+        return self.run("help", check=False)
+
+    def delete(self, image_tag, timeout=30):
+        """Delete a tagged image."""
+        return self.run("delete", image_tag, timeout=timeout, check=False)
+
+    def gc(self, timeout=60):
+        """Run garbage collection (non-interactive)."""
+        # gc prompts for confirmation, so we can't easily test interactive mode
+        # Just test that the command exists and shows dry-run
+        return self.run("gc", timeout=timeout, check=False)
+
+    def push_with_args(self, *args, timeout=120):
+        """Push with custom arguments."""
+        return self.run("push", *args, timeout=timeout, check=False)
+
+
+@pytest.fixture(scope="module")
+def registry(registry_script):
+    """Create a RegistryScriptRunner instance."""
+    return RegistryScriptRunner(registry_script)
+
+
+@pytest.fixture(scope="module")
+def registry_session(registry):
+    """Module-scoped fixture that ensures registry is running.
+
+    Starts the registry if not running and stops it at the end
+    if we started it.
+    """
+    was_running = registry.is_running()
+
+    if not was_running:
+        result = registry.start(timeout=30)
+        if result.returncode != 0:
+            pytest.skip(f"Failed to start registry: {result.stderr}")
+        # Wait a moment for registry to be ready
+        time.sleep(2)
+
+    yield registry
+
+    # Only stop if we started it
+    if not was_running:
+        registry.stop()
+
+
+class TestRegistryScriptBasic:
+    """Test basic registry script functionality."""
+
+    def test_script_exists_and_executable(self, registry_script):
+        """Test that the script exists and is executable."""
+        assert registry_script.exists()
+        assert os.access(registry_script, os.X_OK)
+
+    def test_help_command(self, registry):
+        """Test help command shows usage info."""
+        result = registry.help()
+        assert result.returncode == 0
+        assert "start" in result.stdout
+        assert "stop" in result.stdout
+        assert "push" in result.stdout
+        assert "import" in result.stdout
+        assert "list" in result.stdout
+
+    def test_unknown_command_shows_error(self, registry):
+        """Test that unknown command shows error and help."""
+        result = registry.run("invalid-command", check=False)
+        assert result.returncode != 0
+        assert "unknown" in result.stdout.lower() or "usage" in result.stdout.lower()
+
+
+class TestRegistryLifecycle:
+    """Test registry start/stop/status commands."""
+
+    def test_start_registry(self, registry):
+        """Test starting the registry."""
+        # Stop first if running
+        registry.stop()
+        time.sleep(1)
+
+        result = registry.start()
+        assert result.returncode == 0
+        assert "started" in result.stdout.lower() or "running" in result.stdout.lower()
+
+        # Verify it's running
+        assert registry.is_running()
+
+    def test_status_when_running(self, registry):
+        """Test status command when registry is running."""
+        # Ensure running
+        if not registry.is_running():
+            registry.start()
+            time.sleep(2)
+
+        result = registry.status()
+        assert result.returncode == 0
+        assert "running" in result.stdout.lower()
+        assert "healthy" in result.stdout.lower() or "url" in result.stdout.lower()
+
+    def test_stop_registry(self, registry):
+        """Test stopping the registry."""
+        # Ensure running first
+        if not registry.is_running():
+            registry.start()
+            time.sleep(2)
+
+        result = registry.stop()
+        assert result.returncode == 0
+        assert "stop" in result.stdout.lower()
+
+        # Verify it's stopped
+        assert not registry.is_running()
+
+    def test_status_when_stopped(self, registry):
+        """Test status command when registry is stopped."""
+        # Ensure stopped
+        registry.stop()
+        time.sleep(1)
+
+        result = registry.status()
+        assert result.returncode != 0
+        assert "not running" in result.stdout.lower()
+
+    def test_start_when_already_running(self, registry):
+        """Test that starting when already running is idempotent."""
+        # Start once
+        if not registry.is_running():
+            registry.start()
+            time.sleep(2)
+
+        # Start again
+        result = registry.start()
+        assert result.returncode == 0
+        assert "already running" in result.stdout.lower() or "running" in result.stdout.lower()
+
+    def test_stop_when_not_running(self, registry):
+        """Test that stopping when not running is idempotent."""
+        # Ensure stopped
+        registry.stop()
+        time.sleep(1)
+
+        # Stop again
+        result = registry.stop()
+        assert result.returncode == 0
+        assert "not running" in result.stdout.lower()
+
+
+class TestRegistryPush:
+    """Test pushing OCI images to the registry.
+
+    Note: This requires OCI images in the deploy directory.
+    Tests will skip if no images are available.
+    """
+
+    def test_push_requires_running_registry(self, registry):
+        """Test that push fails when registry is not running."""
+        registry.stop()
+        time.sleep(1)
+
+        result = registry.run("push", check=False, timeout=10)
+        assert result.returncode != 0
+        assert "not responding" in result.stdout.lower() or "start" in result.stdout.lower()
+
+    def test_push_with_no_images(self, registry_session):
+        """Test push when no OCI images are in deploy directory.
+
+        This may succeed (with "no images" message) or actually push
+        images if they exist. Either is acceptable.
+        """
+        registry_session.ensure_running()
+        result = registry_session.push(timeout=120)
+        # Either succeeds (with images) or shows message (without)
+        # Both are valid outcomes
+        assert result.returncode == 0
+
+
+class TestRegistryImport:
+    """Test importing 3rd party images.
+
+    Note: Import tests require network access to docker.io.
+    Use --skip-registry-network to skip these tests.
+    """
+
+    def test_import_requires_running_registry(self, registry):
+        """Test that import fails when registry is not running."""
+        registry.stop()
+        time.sleep(1)
+
+        result = registry.run("import", "docker.io/library/alpine:latest",
+                              check=False, timeout=10)
+        assert result.returncode != 0
+        assert "not responding" in result.stdout.lower() or "start" in result.stdout.lower()
+
+    def test_import_no_args_shows_usage(self, registry_session):
+        """Test that import without args shows usage."""
+        registry_session.ensure_running()
+        result = registry_session.run("import", check=False)
+        assert result.returncode != 0
+        assert "usage" in result.stdout.lower()
+        assert "docker.io" in result.stdout.lower() or "example" in result.stdout.lower()
+
+    @pytest.mark.network
+    @pytest.mark.slow
+    def test_import_alpine(self, registry_session, skip_network):
+        """Test importing alpine from docker.io."""
+        if skip_network:
+            pytest.skip("Skipping network test (--skip-registry-network)")
+
+        registry_session.ensure_running()
+        result = registry_session.import_image(
+            "docker.io/library/alpine:latest",
+            timeout=300
+        )
+        assert result.returncode == 0
+        assert "import complete" in result.stdout.lower() or "importing" in result.stdout.lower()
+
+        # Verify it appears in list
+        list_result = registry_session.list_images()
+        assert "alpine" in list_result.stdout
+
+    @pytest.mark.network
+    @pytest.mark.slow
+    def test_import_with_custom_name(self, registry_session, skip_network):
+        """Test importing with a custom local name."""
+        if skip_network:
+            pytest.skip("Skipping network test (--skip-registry-network)")
+
+        registry_session.ensure_running()
+        result = registry_session.import_image(
+            "docker.io/library/busybox:latest",
+            "my-busybox",
+            timeout=300
+        )
+        assert result.returncode == 0
+
+        # Verify it appears with custom name
+        list_result = registry_session.list_images()
+        assert "my-busybox" in list_result.stdout
+
+
+class TestRegistryQuery:
+    """Test registry query commands (list, tags, catalog)."""
+
+    def test_catalog_requires_running_registry(self, registry):
+        """Test that catalog fails when registry is not running."""
+        registry.stop()
+        time.sleep(1)
+
+        result = registry.run("catalog", check=False, timeout=10)
+        # May fail or return empty/error JSON
+        # Just verify it doesn't hang
+
+    def test_list_requires_running_registry(self, registry):
+        """Test that list fails when registry is not running."""
+        registry.stop()
+        time.sleep(1)
+
+        result = registry.run("list", check=False, timeout=10)
+        assert result.returncode != 0
+        assert "not responding" in result.stdout.lower()
+
+    def test_catalog_returns_json(self, registry_session):
+        """Test that catalog returns JSON format."""
+        registry_session.ensure_running()
+        result = registry_session.catalog()
+        assert result.returncode == 0
+
+        # Should be valid JSON with repositories key
+        import json
+        try:
+            data = json.loads(result.stdout)
+            assert "repositories" in data
+        except json.JSONDecodeError:
+            # May be pretty-printed, try parsing lines
+            assert "repositories" in result.stdout
+
+    def test_list_shows_images(self, registry_session):
+        """Test that list shows images with their tags."""
+        registry_session.ensure_running()
+        result = registry_session.list_images()
+        assert result.returncode == 0
+        # Should show header or images
+        assert "images" in result.stdout.lower() or ":" in result.stdout or "(none)" in result.stdout
+
+    def test_tags_for_nonexistent_image(self, registry_session):
+        """Test tags command for nonexistent image."""
+        registry_session.ensure_running()
+        result = registry_session.tags("nonexistent-image-xyz")
+        # Either returns non-zero with "not found", or returns empty/error JSON
+        # The important thing is it doesn't crash and indicates the image doesn't exist
+        if result.returncode == 0:
+            # If it returns 0, stdout should be empty or contain error info
+            assert "nonexistent" not in result.stdout.lower() or "error" in result.stdout.lower() or result.stdout.strip() == ""
+        else:
+            assert "not found" in result.stdout.lower() or "error" in result.stdout.lower()
+
+    def test_tags_usage_without_image(self, registry_session):
+        """Test tags command without image argument shows usage."""
+        registry_session.ensure_running()
+        result = registry_session.run("tags", check=False)
+        assert result.returncode != 0
+        assert "usage" in result.stdout.lower()
+
+
+class TestRegistryDelete:
+    """Test delete command for removing tagged images."""
+
+    def test_delete_requires_running_registry(self, registry):
+        """Test that delete fails when registry is not running."""
+        registry.stop()
+        time.sleep(1)
+
+        result = registry.delete("container-base:latest")
+        assert result.returncode != 0
+        assert "not responding" in result.stdout.lower()
+
+    def test_delete_no_args_shows_usage(self, registry_session):
+        """Test that delete without args shows usage."""
+        registry_session.ensure_running()
+        result = registry_session.run("delete", check=False)
+        assert result.returncode != 0
+        assert "usage" in result.stdout.lower()
+
+    def test_delete_requires_tag(self, registry_session):
+        """Test that delete requires image:tag format."""
+        registry_session.ensure_running()
+        result = registry_session.delete("container-base")  # No tag
+        assert result.returncode != 0
+        assert "tag required" in result.stdout.lower()
+
+    def test_delete_nonexistent_tag(self, registry_session):
+        """Test deleting a nonexistent tag."""
+        registry_session.ensure_running()
+        result = registry_session.delete("container-base:nonexistent-tag-xyz")
+        assert result.returncode != 0
+        assert "not found" in result.stdout.lower()
+
+    @pytest.mark.network
+    @pytest.mark.slow
+    def test_delete_workflow(self, registry_session, skip_network):
+        """Test importing an image, then deleting it."""
+        if skip_network:
+            pytest.skip("Skipping network test (--skip-registry-network)")
+
+        registry_session.ensure_running()
+
+        # Import an image with unique name
+        result = registry_session.import_image(
+            "docker.io/library/alpine:latest",
+            "delete-test",
+            timeout=300
+        )
+        assert result.returncode == 0
+
+        # Verify it exists
+        result = registry_session.tags("delete-test")
+        assert result.returncode == 0
+        assert "latest" in result.stdout
+
+        # Delete it
+        result = registry_session.delete("delete-test:latest")
+        assert result.returncode == 0
+        assert "deleted successfully" in result.stdout.lower()
+
+        # Verify it's gone
+        result = registry_session.tags("delete-test")
+        assert result.returncode != 0 or "not found" in result.stdout.lower()
+
+
+class TestRegistryGC:
+    """Test garbage collection command."""
+
+    def test_gc_help_in_help_output(self, registry):
+        """Test that gc command is listed in help."""
+        result = registry.help()
+        assert "gc" in result.stdout.lower()
+
+    def test_gc_requires_registry_binary(self, registry_session):
+        """Test that gc checks for registry binary.
+
+        This test just verifies gc command runs and either:
+        - Works (shows dry-run output)
+        - Fails with useful error message
+        """
+        # gc stops registry first, so just run it and check output
+        result = registry_session.gc(timeout=30)
+        # Should either work or show error about binary/not running
+        output = result.stdout.lower()
+        assert any([
+            "garbage" in output,
+            "collecting" in output,
+            "registry" in output,
+            "error" in output,
+            "not found" in output,
+        ])
+
+
+class TestRegistryPushOptions:
+    """Test push command with various options."""
+
+    def test_push_tag_requires_image_name(self, registry_session):
+        """Test that --tag without image name fails."""
+        registry_session.ensure_running()
+        result = registry_session.push_with_args("--tag", "v1.0.0")
+        assert result.returncode != 0
+        assert "--tag requires an image name" in result.stdout.lower()
+
+    def test_push_with_image_filter(self, registry_session):
+        """Test pushing a specific image by name."""
+        registry_session.ensure_running()
+        result = registry_session.push_with_args("container-base")
+        # Should either succeed or report image not found
+        # (depending on whether container-base exists)
+        output = result.stdout.lower()
+        assert any([
+            "pushing" in output,
+            "not found" in output,
+            "done" in output,
+        ])
+
+    def test_push_with_strategy(self, registry_session):
+        """Test pushing with explicit strategy."""
+        registry_session.ensure_running()
+        result = registry_session.push_with_args("--strategy", "latest")
+        assert result.returncode == 0 or "pushing" in result.stdout.lower()
+
+    def test_push_help_shows_options(self, registry):
+        """Test that help shows push options."""
+        result = registry.help()
+        assert "--tag" in result.stdout
+        assert "--strategy" in result.stdout
+        assert "image" in result.stdout.lower()
+
+
+class TestRegistryIntegration:
+    """Integration tests for full registry workflow.
+
+    These tests require:
+    - Registry script generated
+    - docker-distribution-native built
+    - skopeo-native built
+    - Network access (for import tests)
+    """
+
+    @pytest.mark.network
+    @pytest.mark.slow
+    def test_full_workflow(self, registry, skip_network):
+        """Test complete workflow: start -> import -> list -> stop."""
+        if skip_network:
+            pytest.skip("Skipping network test (--skip-registry-network)")
+
+        # Start fresh
+        registry.stop()
+        time.sleep(1)
+
+        try:
+            # Start
+            result = registry.start()
+            assert result.returncode == 0
+            time.sleep(2)
+
+            # Import an image
+            result = registry.import_image(
+                "docker.io/library/alpine:latest",
+                "workflow-test",
+                timeout=300
+            )
+            assert result.returncode == 0
+
+            # List should show it
+            result = registry.list_images()
+            assert result.returncode == 0
+            assert "workflow-test" in result.stdout
+
+            # Tags should work
+            result = registry.tags("workflow-test")
+            assert result.returncode == 0
+            assert "latest" in result.stdout
+
+            # Catalog should include it
+            result = registry.catalog()
+            assert result.returncode == 0
+            assert "workflow-test" in result.stdout
+
+        finally:
+            # Always stop
+            registry.stop()
diff --git a/tests/test_vdkr_registry.py b/tests/test_vdkr_registry.py
new file mode 100644
index 00000000..9073444d
--- /dev/null
+++ b/tests/test_vdkr_registry.py
@@ -0,0 +1,414 @@
+# SPDX-FileCopyrightText: Copyright (C) 2025 Bruce Ashfield
+#
+# SPDX-License-Identifier: MIT
+"""
+Tests for vdkr registry functionality.
+
+These tests verify vdkr registry configuration and pulling from registries:
+- CLI override: --registry <url> pull <image>
+- Persistent config: vconfig registry <url> + pull <image>
+- Config reset: vconfig registry --reset
+- Image compound commands: image ls/rm/pull/inspect/tag
+
+Note: Tests that require a running registry use mock endpoints or skip
+if no registry is available. For full integration testing, start the
+registry first:
+    $TOPDIR/container-registry/container-registry.sh start
+
+Run with:
+    pytest tests/test_vdkr_registry.py -v --vdkr-dir /tmp/vcontainer
+
+Run specific test class:
+    pytest tests/test_vdkr_registry.py::TestVconfigRegistry -v
+"""
+
+import pytest
+import json
+import subprocess
+
+
+class TestVconfigRegistry:
+    """Test vconfig registry configuration commands."""
+
+    def test_vconfig_registry_show_empty(self, vdkr):
+        """Test showing registry config when not set."""
+        # Reset first to ensure clean state
+        vdkr.run("vconfig", "registry", "--reset", check=False)
+
+        result = vdkr.run("vconfig", "registry")
+        assert result.returncode == 0
+        # Should show empty or "not set" message
+        output = result.stdout.strip()
+        assert output == "" or "not set" in output.lower() or "registry:" in output.lower()
+
+    def test_vconfig_registry_set(self, vdkr):
+        """Test setting registry configuration."""
+        test_registry = "10.0.2.2:5000/test"
+
+        result = vdkr.run("vconfig", "registry", test_registry)
+        assert result.returncode == 0
+
+        # Verify it was set
+        result = vdkr.run("vconfig", "registry")
+        assert result.returncode == 0
+        assert test_registry in result.stdout
+
+    def test_vconfig_registry_reset(self, vdkr):
+        """Test resetting registry configuration."""
+        # Set a value first
+        vdkr.run("vconfig", "registry", "10.0.2.2:5000/test")
+
+        # Reset it
+        result = vdkr.run("vconfig", "registry", "--reset")
+        assert result.returncode == 0
+
+        # Verify it was reset
+        result = vdkr.run("vconfig", "registry")
+        assert result.returncode == 0
+        # Should be empty after reset
+        output = result.stdout.strip()
+        assert "10.0.2.2:5000/test" not in output
+
+    def test_vconfig_show_all_includes_registry(self, vdkr):
+        """Test that vconfig (no args) shows registry in output."""
+        result = vdkr.run("vconfig")
+        assert result.returncode == 0
+        # Should list registry as one of the config keys
+        assert "registry" in result.stdout.lower()
+
+
+class TestImageCompoundCommands:
+    """Test vdkr image compound commands.
+
+    These translate docker image <subcommand> to the appropriate docker command:
+    - image ls → images
+    - image rm → rmi
+    - image pull → pull (with registry transform)
+    - image inspect → inspect
+    - image tag → tag
+    - image prune → image prune
+    - image history → history
+
+    Note: These tests reset any registry config to ensure commands work with
+    unqualified image names. Registry-specific tests are in TestRegistryTransform.
+    """
+
+    @pytest.fixture(autouse=True)
+    def setup_memres(self, memres_session):
+        """Ensure memres is running and registry config is reset."""
+        self.vdkr = memres_session
+        self.vdkr.ensure_memres()
+        # Reset any baked-in or configured registry to test basic commands
+        self.vdkr.run("vconfig", "registry", "--reset", check=False)
+
+    def get_alpine_info(self):
+        """Get alpine image info (ref and ID).
+
+        Returns tuple of (full_ref, image_id) or (None, None) if not found.
+        """
+        # Use raw 'images' to see what's actually stored
+        images_result = self.vdkr.run("images", check=False)
+        if images_result.returncode != 0:
+            return None, None
+
+        # Parse the images output to find alpine
+        for line in images_result.stdout.splitlines():
+            if "alpine" in line.lower() and "REPOSITORY" not in line:
+                # Parse: REPOSITORY TAG IMAGE_ID CREATED SIZE
+                parts = line.split()
+                if len(parts) >= 3:
+                    repo = parts[0]
+                    tag = parts[1]
+                    image_id = parts[2]
+                    return f"{repo}:{tag}", image_id
+        return None, None
+
+    def test_image_ls(self):
+        """Test 'image ls' command."""
+        result = self.vdkr.run("image", "ls")
+        assert result.returncode == 0
+        # Should have header line
+        assert "REPOSITORY" in result.stdout or "IMAGE" in result.stdout
+
+    def ensure_alpine_and_get_info(self):
+        """Ensure alpine is present and return (ref, image_id)."""
+        # First check if alpine is already present
+        alpine_ref, alpine_id = self.get_alpine_info()
+        if alpine_ref:
+            return alpine_ref, alpine_id
+
+        # Try to pull alpine
+        result = self.vdkr.run("pull", "alpine:latest", timeout=300, check=False)
+        if result.returncode != 0:
+            return None, None
+
+        # Get the info again
+        return self.get_alpine_info()
+
+    @pytest.mark.network
+    def test_image_ls_with_filter(self):
+        """Test 'image ls' with filter.
+
+        After the transform fix, 'images alpine' should show alpine directly
+        without registry prefix transformation.
+        """
+        alpine_ref, alpine_id = self.ensure_alpine_and_get_info()
+        if not alpine_ref:
+            pytest.skip("Could not get alpine image (network issue?)")
+
+        # Get the simple name for filtering (e.g., "alpine" from "alpine:latest")
+        simple_name = alpine_ref.split("/")[-1].split(":")[0]  # "alpine"
+
+        # Filter should work with simple name after transform fix
+        result = self.vdkr.run("images", simple_name)
+        assert result.returncode == 0
+
+        # Should show the alpine image (check for image ID to be robust)
+        if alpine_id not in result.stdout and "alpine" not in result.stdout:
+            # If filter doesn't work (old vdkr), at least verify full list works
+            full_result = self.vdkr.run("images")
+            assert alpine_id in full_result.stdout, f"Alpine ({alpine_id}) not in images"
+            pytest.skip("Filter transform not fixed yet - vdkr rebuild needed")
+
+    @pytest.mark.network
+    def test_image_pull(self):
+        """Test 'image pull' command."""
+        # Remove if exists
+        self.vdkr.run("image", "rm", "-f", "busybox:latest", check=False)
+
+        # Pull via image command
+        result = self.vdkr.run("image", "pull", "busybox:latest", timeout=300, check=False)
+        if result.returncode != 0:
+            pytest.skip(f"Could not pull busybox (network issue?): {result.stderr}")
+
+        # Verify it appears in images
+        images = self.vdkr.run("image", "ls")
+        assert "busybox" in images.stdout
+
+    @pytest.mark.network
+    def test_image_inspect(self):
+        """Test 'image inspect' command."""
+        alpine_ref, alpine_id = self.ensure_alpine_and_get_info()
+        if not alpine_id:
+            pytest.skip("Could not get alpine image (network issue?)")
+
+        # Use image ID to avoid registry transform
+        result = self.vdkr.run("image", "inspect", alpine_id)
+        assert result.returncode == 0
+
+        # Should be valid JSON
+        data = json.loads(result.stdout)
+        assert isinstance(data, list)
+        assert len(data) > 0
+
+    @pytest.mark.network
+    def test_image_history(self):
+        """Test 'image history' command."""
+        alpine_ref, alpine_id = self.ensure_alpine_and_get_info()
+        if not alpine_id:
+            pytest.skip("Could not get alpine image (network issue?)")
+
+        # Use image ID to avoid registry transform
+        result = self.vdkr.run("image", "history", alpine_id)
+        assert result.returncode == 0
+        # Should show history with IMAGE or CREATED columns
+        assert "IMAGE" in result.stdout or "CREATED" in result.stdout
+
+    @pytest.mark.network
+    def test_image_tag(self):
+        """Test 'image tag' command."""
+        alpine_ref, alpine_id = self.ensure_alpine_and_get_info()
+        if not alpine_id:
+            pytest.skip("Could not get alpine image (network issue?)")
+
+        # Use image ID as source to avoid registry transform
+        result = self.vdkr.run("image", "tag", alpine_id, "my-test-alpine:v1")
+        assert result.returncode == 0
+
+        # Verify the new tag exists
+        images = self.vdkr.run("image", "ls")
+        assert "my-test-alpine" in images.stdout
+
+        # Clean up using image ID of the new tag
+        self.vdkr.run("rmi", "my-test-alpine:v1", check=False)
+
+    @pytest.mark.network
+    def test_image_rm(self):
+        """Test 'image rm' command."""
+        alpine_ref, alpine_id = self.ensure_alpine_and_get_info()
+        if not alpine_id:
+            pytest.skip("Could not get alpine image (network issue?)")
+
+        # Tag using image ID to create a removable image
+        result = self.vdkr.run("tag", alpine_id, "test-rm:latest", check=False)
+        if result.returncode != 0:
+            pytest.skip(f"Could not tag image: {result.stderr}")
+
+        # Remove it using rmi (not image rm) to avoid transform
+        result = self.vdkr.run("rmi", "test-rm:latest", check=False)
+        # Verify it succeeded or at least didn't crash
+        assert result.returncode == 0 or "No such image" in result.stdout
+
+    def test_image_prune(self):
+        """Test 'image prune' command."""
+        # Prune dangling images (-f to skip confirmation)
+        result = self.vdkr.run("image", "prune", "-f")
+        assert result.returncode == 0
+
+    def test_image_requires_subcommand(self):
+        """Test that 'image' without subcommand shows error."""
+        result = self.vdkr.run("image", check=False)
+        assert result.returncode != 0
+        assert "subcommand" in result.stderr.lower() or "requires" in result.stderr.lower()
+
+
+class TestRegistryCLIOverride:
+    """Test --registry CLI flag for one-off registry usage.
+
+    Note: These tests require a running registry. They skip if no registry
+    is available.
+    """
+
+    @pytest.fixture
+    def registry_url(self, request):
+        """Get registry URL from command line or environment, or skip."""
+        url = request.config.getoption("--registry-url", default=None)
+        if url is None:
+            import os
+            url = os.environ.get("TEST_REGISTRY_URL")
+        if url is None:
+            pytest.skip("No registry URL provided (use --registry-url or TEST_REGISTRY_URL)")
+        return url
+
+    @pytest.mark.network
+    def test_registry_flag_pull(self, memres_session, registry_url):
+        """Test pulling with --registry flag."""
+        vdkr = memres_session
+        vdkr.ensure_memres()
+
+        # Pull with explicit registry
+        result = vdkr.run("--registry", registry_url, "pull", "alpine", timeout=300, check=False)
+        # May succeed or fail depending on whether image exists in registry
+        # Just verify the command is accepted
+        assert "unknown flag" not in result.stderr.lower()
+
+    @pytest.mark.network
+    def test_registry_flag_run(self, memres_session, registry_url):
+        """Test run with --registry flag."""
+        vdkr = memres_session
+        vdkr.ensure_memres()
+
+        # Try to run with explicit registry
+        result = vdkr.run("--registry", registry_url, "run", "--rm", "alpine",
+                          "echo", "hello", timeout=300, check=False)
+        # Just verify the flag is accepted
+        assert "unknown flag" not in result.stderr.lower()
+
+
+class TestRegistryPersistentConfig:
+    """Test persistent registry configuration with vconfig."""
+
+    @pytest.fixture(autouse=True)
+    def reset_registry_config(self, vdkr):
+        """Reset registry config before and after each test."""
+        vdkr.run("vconfig", "registry", "--reset", check=False)
+        yield
+        vdkr.run("vconfig", "registry", "--reset", check=False)
+
+    def test_config_persists_across_commands(self, vdkr):
+        """Test that registry config persists across multiple vdkr invocations."""
+        test_registry = "10.0.2.2:5000/persistent-test"
+
+        # Set registry
+        vdkr.run("vconfig", "registry", test_registry)
+
+        # Verify in new command
+        result = vdkr.run("vconfig", "registry")
+        assert test_registry in result.stdout
+
+        # Verify in vconfig all
+        result = vdkr.run("vconfig")
+        assert test_registry in result.stdout
+
+    def test_pull_uses_config(self, memres_session):
+        """Test that pull command uses configured registry.
+
+        Note: This test verifies the configuration is passed to the VM.
+        It may fail if the registry doesn't have the image, but we can
+        check the error message to verify the registry was used.
+        """
+        vdkr = memres_session
+        vdkr.ensure_memres()
+
+        # Set a fake registry
+        fake_registry = "10.0.2.2:9999/fake"
+        vdkr.run("vconfig", "registry", fake_registry)
+
+        # Try to pull - should fail because registry doesn't exist
+        # but error should reference the fake registry
+        result = vdkr.run("pull", "nonexistent-image", timeout=60, check=False)
+
+        # The important thing is that it tried to use our registry
+        # (connection refused or similar error indicates it tried)
+        assert result.returncode != 0
+        # Error should indicate connection issue to our fake registry
+
+
+class TestInsecureRegistry:
+    """Test --insecure-registry flag."""
+
+    def test_insecure_registry_flag_accepted(self, vdkr):
+        """Test that --insecure-registry flag is accepted."""
+        vdkr.ensure_memres()
+
+        # Just verify the flag is recognized
+        result = vdkr.run("--insecure-registry", "10.0.2.2:5000", "images", check=False)
+        assert "unknown flag" not in result.stderr.lower()
+        assert "unrecognized" not in result.stderr.lower()
+
+    def test_multiple_insecure_registries(self, vdkr):
+        """Test multiple --insecure-registry flags."""
+        vdkr.ensure_memres()
+
+        result = vdkr.run(
+            "--insecure-registry", "10.0.2.2:5000",
+            "--insecure-registry", "10.0.2.2:5001",
+            "images", check=False
+        )
+        assert "unknown flag" not in result.stderr.lower()
+
+
+class TestRegistryTransform:
+    """Test image name transformation with registry prefix.
+
+    When a default registry is configured, unqualified image names
+    should be transformed to include the registry prefix.
+    """
+
+    @pytest.fixture(autouse=True)
+    def reset_registry_config(self, vdkr):
+        """Reset registry config before and after each test."""
+        vdkr.run("vconfig", "registry", "--reset", check=False)
+        yield
+        vdkr.run("vconfig", "registry", "--reset", check=False)
+
+    def test_qualified_names_not_transformed(self, memres_session):
+        """Test that fully qualified image names are not transformed."""
+        vdkr = memres_session
+        vdkr.ensure_memres()
+
+        # Set a registry
+        vdkr.run("vconfig", "registry", "10.0.2.2:5000/test")
+
+        # Pull fully qualified image - should use docker.io, not our registry
+        result = vdkr.run("pull", "docker.io/library/alpine:latest",
+                          timeout=300, check=False)
+
+        # If successful, alpine from docker.io should be present
+        # If failed, should NOT be a connection error to 10.0.2.2
+        if result.returncode != 0:
+            # Should not have tried our fake registry
+            assert "10.0.2.2:5000/test" not in result.stderr
+
+
+# Note: Registry options (--registry-url) are defined in conftest.py