awoooi/apps/api/tests/test_docker_build_surface_inventory.py

from __future__ import annotations

import json

import pytest

from src.services.docker_build_surface_inventory import load_latest_docker_build_surface_inventory


def test_load_latest_docker_build_surface_inventory_reads_newest_file(tmp_path):
    older = _snapshot(generated_at="2026-06-03T00:00:00+08:00", completion=95)
    newer = _snapshot(generated_at="2026-06-04T00:00:00+08:00", completion=97)
    (tmp_path / "docker_build_surface_inventory_2026-06-03.json").write_text(
        json.dumps(older),
        encoding="utf-8",
    )
    (tmp_path / "docker_build_surface_inventory_2026-06-04.json").write_text(
        json.dumps(newer),
        encoding="utf-8",
    )

    loaded = load_latest_docker_build_surface_inventory(tmp_path)

    assert loaded["generated_at"] == "2026-06-04T00:00:00+08:00"
    assert loaded["program_status"]["overall_completion_percent"] == 97
    assert loaded["rollups"]["total_surfaces"] == 2
    assert loaded["operation_boundaries"]["docker_build_allowed"] is False


def test_docker_build_surface_inventory_requires_read_only_mode(tmp_path):
    snapshot = _snapshot()
    snapshot["program_status"]["read_only_mode"] = False
    (tmp_path / "docker_build_surface_inventory_2026-06-04.json").write_text(
        json.dumps(snapshot),
        encoding="utf-8",
    )

    with pytest.raises(ValueError, match="read_only_mode"):
        load_latest_docker_build_surface_inventory(tmp_path)


def test_docker_build_surface_inventory_requires_blocked_operations(tmp_path):
    snapshot = _snapshot()
    snapshot["operation_boundaries"]["image_pull_allowed"] = True
    (tmp_path / "docker_build_surface_inventory_2026-06-04.json").write_text(
        json.dumps(snapshot),
        encoding="utf-8",
    )

    with pytest.raises(ValueError, match="operation boundaries"):
        load_latest_docker_build_surface_inventory(tmp_path)


def test_docker_build_surface_inventory_requires_action_required_consistency(tmp_path):
    snapshot = _snapshot()
    snapshot["rollups"]["action_required_surface_ids"] = []
    (tmp_path / "docker_build_surface_inventory_2026-06-04.json").write_text(
        json.dumps(snapshot),
        encoding="utf-8",
    )

    with pytest.raises(ValueError, match="action_required_surface_ids"):
        load_latest_docker_build_surface_inventory(tmp_path)


def test_docker_build_surface_inventory_requires_network_fetch_consistency(tmp_path):
    snapshot = _snapshot()
    snapshot["rollups"]["build_time_network_fetch_count"] = 999
    (tmp_path / "docker_build_surface_inventory_2026-06-04.json").write_text(
        json.dumps(snapshot),
        encoding="utf-8",
    )

    with pytest.raises(ValueError, match="build_time_network_fetch_count"):
        load_latest_docker_build_surface_inventory(tmp_path)


def test_docker_build_surface_inventory_requires_healthcheck_consistency(tmp_path):
    snapshot = _snapshot()
    snapshot["rollups"]["healthcheck_count"] = 999
    (tmp_path / "docker_build_surface_inventory_2026-06-04.json").write_text(
        json.dumps(snapshot),
        encoding="utf-8",
    )

    with pytest.raises(ValueError, match="healthcheck_count"):
        load_latest_docker_build_surface_inventory(tmp_path)


def test_docker_build_surface_inventory_fails_when_missing(tmp_path):
    with pytest.raises(FileNotFoundError):
        load_latest_docker_build_surface_inventory(tmp_path)


def _snapshot(
    *,
    generated_at: str = "2026-06-04T00:00:00+08:00",
    completion: int = 97,
) -> dict:
    return {
        "schema_version": "docker_build_surface_inventory_v1",
        "generated_at": generated_at,
        "program_status": {
            "overall_completion_percent": completion,
            "current_priority": "P1",
            "current_task_id": "P1-203",
            "next_task_id": "P1-204",
            "read_only_mode": True,
        },
        "source_refs": ["apps/api/Dockerfile", "apps/web/Dockerfile"],
        "rollups": {
            "total_surfaces": 2,
            "dockerfile_count": 2,
            "external_image_ref_count": 2,
            "from_instruction_count": 2,
            "copy_from_external_image_count": 0,
            "digest_pinned_image_count": 0,
            "tag_pinned_image_count": 2,
            "build_time_network_fetch_count": 2,
            "non_root_runtime_count": 2,
            "healthcheck_count": 1,
            "by_status": {"action_required": 2},
            "action_required_surface_ids": ["api_dockerfile", "web_dockerfile"],
            "planned_next_surface_ids": [],
        },
        "surfaces": [
            _surface("api_dockerfile", healthcheck=True),
            _surface("web_dockerfile", healthcheck=False),
        ],
        "risk_findings": [
            {
                "finding_id": "base_images_not_digest_pinned",
                "severity": "high",
                "status": "action_required",
                "summary": "not pinned",
                "evidence_refs": ["apps/api/Dockerfile"],
                "next_action": "policy",
            }
        ],
        "operation_boundaries": {
            "read_only_api_allowed": True,
            "docker_build_allowed": False,
            "image_pull_allowed": False,
            "image_rebuild_allowed": False,
            "registry_push_allowed": False,
            "external_cve_lookup_allowed": False,
            "package_installation_allowed": False,
            "production_routing_allowed": False,
        },
        "approval_boundaries": {
            "sdk_installation_allowed": False,
            "paid_api_call_allowed": False,
            "shadow_or_canary_allowed": False,
            "production_routing_allowed": False,
            "destructive_operation_allowed": False,
        },
    }


def _surface(surface_id: str, *, healthcheck: bool) -> dict:
    return {
        "surface_id": surface_id,
        "display_name": surface_id,
        "dockerfile_ref": "Dockerfile",
        "status": "action_required",
        "risk_level": "high",
        "stage_count": 1,
        "external_image_refs": ["python:3.11-slim"],
        "digest_pinned_image_refs": [],
        "tag_pinned_image_refs": ["python:3.11-slim"],
        "build_time_network_fetches": ["curl"],
        "binary_sources": ["python:3.11-slim"],
        "non_root_runtime": True,
        "healthcheck_present": healthcheck,
        "cache_controls": ["CACHE_BUST"],
        "gate_status": "image_rebuild_blocked",
        "evidence_refs": ["Dockerfile"],
        "next_action": "next",
    }