trycua--cua/libs/python/cua-sandbox/tests/test_oci.py

"""OCI registry tests — manifest inspection, kind detection, image formats.

    pytest tests/test_oci.py -v -s

Unit tests use synthetic manifests. Integration tests (marked `oci_live`)
fetch real manifests from ghcr.io and require network access.
"""

from __future__ import annotations

import pytest
from cua_sandbox.registry.manifest import (
    ImageFormat,
    detect_format,
    detect_kind,
    detect_os,
    get_layer_info,
)
from cua_sandbox.registry.media_types import (
    DOCKER_IMAGE_CONFIG,
    DOCKER_IMAGE_LAYER,
    LEGACY_DISK_CHUNK,
    OCI_IMAGE_CONFIG,
    OCI_IMAGE_LAYER,
    OCI_VM_AUX,
    OCI_VM_CONFIG,
    OCI_VM_DISK,
    QEMU_CONFIG,
    QEMU_DISK_GZIP,
)
from cua_sandbox.registry.ref import parse_ref

# ═════════════════════════════════════════════════════════════════════════════
# parse_ref
# ═════════════════════════════════════════════════════════════════════════════


class TestParseRef:
    def test_full_ref(self):
        assert parse_ref("ghcr.io/trycua/macos-sequoia-cua:latest") == (
            "ghcr.io",
            "trycua",
            "macos-sequoia-cua",
            "latest",
        )

    def test_org_name(self):
        assert parse_ref("trycua/cua-xfce:v2") == ("ghcr.io", "trycua", "cua-xfce", "v2")

    def test_short_name(self):
        assert parse_ref("cua-xfce") == ("ghcr.io", "trycua", "cua-xfce", "latest")

    def test_short_name_with_tag(self):
        assert parse_ref("cua-xfce:nightly") == ("ghcr.io", "trycua", "cua-xfce", "nightly")


# ═════════════════════════════════════════════════════════════════════════════
# detect_format / detect_kind — synthetic manifests
# ═════════════════════════════════════════════════════════════════════════════


def _make_manifest(config_mt: str, layer_mts: list[str], **kwargs) -> dict:
    m = {
        "config": {"mediaType": config_mt, "digest": "sha256:aaa", "size": 100},
        "layers": [
            {"mediaType": mt, "digest": f"sha256:{i:03x}", "size": 1000}
            for i, mt in enumerate(layer_mts)
        ],
    }
    m.update(kwargs)
    return m


class TestDetectFormat:
    def test_oci_layered_by_config(self):
        m = _make_manifest(OCI_VM_CONFIG, [OCI_VM_DISK, OCI_VM_AUX])
        assert detect_format(m) == ImageFormat.OCI_LAYERED

    def test_oci_layered_by_layer(self):
        m = _make_manifest("application/json", [OCI_VM_DISK])
        assert detect_format(m) == ImageFormat.OCI_LAYERED

    def test_legacy_lz4_by_layer(self):
        m = _make_manifest("application/json", [LEGACY_DISK_CHUNK, LEGACY_DISK_CHUNK])
        assert detect_format(m) == ImageFormat.LEGACY_LZ4

    def test_chunked_parts(self):
        m = _make_manifest(
            "application/json",
            [
                f"{OCI_IMAGE_LAYER};part.number=1;part.total=3",
                f"{OCI_IMAGE_LAYER};part.number=2;part.total=3",
                f"{OCI_IMAGE_LAYER};part.number=3;part.total=3",
            ],
        )
        assert detect_format(m) == ImageFormat.CHUNKED_PARTS

    def test_container_oci(self):
        m = _make_manifest(OCI_IMAGE_CONFIG, [OCI_IMAGE_LAYER])
        assert detect_format(m) == ImageFormat.CONTAINER

    def test_container_docker(self):
        m = _make_manifest(DOCKER_IMAGE_CONFIG, [DOCKER_IMAGE_LAYER])
        assert detect_format(m) == ImageFormat.CONTAINER

    def test_qemu_by_config(self):
        m = _make_manifest(QEMU_CONFIG, [QEMU_DISK_GZIP])
        assert detect_format(m) == ImageFormat.QEMU

    def test_qemu_by_layer(self):
        m = _make_manifest("application/json", [QEMU_DISK_GZIP])
        assert detect_format(m) == ImageFormat.QEMU

    def test_unknown(self):
        m = _make_manifest("application/json", ["application/octet-stream"])
        assert detect_format(m) == ImageFormat.UNKNOWN


class TestDetectKind:
    def test_vm_oci_layered(self):
        m = _make_manifest(OCI_VM_CONFIG, [OCI_VM_DISK])
        assert detect_kind(m) == "vm"

    def test_vm_legacy(self):
        m = _make_manifest("application/json", [LEGACY_DISK_CHUNK])
        assert detect_kind(m) == "vm"

    def test_vm_chunked(self):
        m = _make_manifest(
            "application/json",
            [
                f"{OCI_IMAGE_LAYER};part.number=1;part.total=2",
            ],
        )
        assert detect_kind(m) == "vm"

    def test_vm_qemu(self):
        m = _make_manifest(QEMU_CONFIG, [QEMU_DISK_GZIP])
        assert detect_kind(m) == "vm"

    def test_container(self):
        m = _make_manifest(OCI_IMAGE_CONFIG, [OCI_IMAGE_LAYER])
        assert detect_kind(m) == "container"


class TestDetectOs:
    def test_agoda_is_macos(self):
        m = _make_manifest(OCI_VM_CONFIG, [OCI_VM_DISK])
        assert detect_os(m) == "macos"

    def test_annotation_linux(self):
        m = _make_manifest(
            "application/json",
            [],
            annotations={
                "org.trycua.lume.os": "Linux",
            },
        )
        assert detect_os(m) == "linux"

    def test_no_os(self):
        m = _make_manifest(OCI_IMAGE_CONFIG, [OCI_IMAGE_LAYER])
        assert detect_os(m) is None


# ═════════════════════════════════════════════════════════════════════════════
# get_layer_info
# ═════════════════════════════════════════════════════════════════════════════


class TestGetLayerInfo:
    def test_part_from_annotations(self):
        m = {
            "layers": [
                {
                    "mediaType": OCI_VM_DISK,
                    "digest": "sha256:abc",
                    "size": 500_000_000,
                    "annotations": {
                        "org.trycua.lume.part.number": "3",
                        "org.trycua.lume.part.total": "10",
                        "org.opencontainers.image.title": "disk.img.003",
                    },
                }
            ],
        }
        info = get_layer_info(m)
        assert len(info) == 1
        assert info[0]["part_number"] == 3
        assert info[0]["part_total"] == 10
        assert info[0]["title"] == "disk.img.003"

    def test_part_from_mediatype(self):
        mt = f"{OCI_IMAGE_LAYER};part.number=7;part.total=164"
        m = {"layers": [{"mediaType": mt, "digest": "sha256:def", "size": 100}]}
        info = get_layer_info(m)
        assert info[0]["part_number"] == 7
        assert info[0]["part_total"] == 164


# ═════════════════════════════════════════════════════════════════════════════
# Live registry tests (require network)
# ═════════════════════════════════════════════════════════════════════════════

oci_live = pytest.mark.skipif(
    not pytest.importorskip("oras", reason="oras-py not installed"),
    reason="oras-py not installed",
)


@oci_live
class TestLiveRegistry:
    """Fetch real manifests from ghcr.io. Requires network + oras."""

    def test_macos_sparse_oci_layered(self):
        from cua_sandbox.registry.manifest import get_manifest

        manifest = get_manifest("ghcr.io/trycua/macos-sequoia-cua-sparse:latest-oci-layered")
        assert detect_format(manifest) == ImageFormat.OCI_LAYERED
        assert detect_kind(manifest) == "vm"
        assert detect_os(manifest) == "macos"

        info = get_layer_info(manifest)
        assert len(info) > 0
        # Should have disk layers with part numbers
        disk_parts = [layer for layer in info if layer["part_number"] is not None]
        assert len(disk_parts) > 0

    def test_macos_chunked_parts(self):
        from cua_sandbox.registry.manifest import get_manifest

        manifest = get_manifest("ghcr.io/trycua/macos-sequoia-cua:latest")
        fmt = detect_format(manifest)
        assert fmt in (ImageFormat.CHUNKED_PARTS, ImageFormat.OCI_LAYERED, ImageFormat.LEGACY_LZ4)
        assert detect_kind(manifest) == "vm"

    def test_ubuntu_container(self):
        from cua_sandbox.registry.manifest import get_manifest

        manifest = get_manifest("docker.io/trycua/cua-xfce:latest")
        assert detect_format(manifest) == ImageFormat.CONTAINER
        assert detect_kind(manifest) == "container"