gateway_schema_v2_capability_parity_test.py

# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

"""Validate the schema-v2 live capability-parity manifest without a runtime."""

from __future__ import annotations

import re
import tomllib
from copy import deepcopy
from pathlib import Path
from typing import Any

import pytest

REPO_ROOT = Path(__file__).resolve().parents[2]
MANIFEST_PATH = REPO_ROOT / "e2e/configs/gateway/schema-v2-capability-parity.toml"

REQUIRED_MANIFEST_FIELDS = {
    "manifest_version",
    "baseline_ref",
    "baseline_commit",
    "baseline_schema_version",
    "candidate_ref",
    "candidate_commit",
    "candidate_schema_version",
    "capabilities",
}
REQUIRED_CAPABILITY_FIELDS = {
    "id",
    "topics",
    "origin_main_access_paths",
    "schema_v2_access_paths",
    "behavioral_oracle",
    "required_environment",
    "test_lane",
    "status",
}
REQUIRED_TOPICS = {
    "auth_tls_jwt",
    "configuration_producers",
    "credentials",
    "database",
    "docker",
    "external_drivers",
    "inference",
    "interceptors",
    "kubernetes",
    "listeners",
    "middleware",
    "mxc",
    "observability",
    "packaging_upgrades",
    "podman",
    "vm",
}
REQUIRED_CAPABILITY_IDS = {
    "configuration-source-precedence",
    "schema-version-and-strict-layout",
    "gateway-identity-and-logging",
    "primary-health-and-metrics-listeners",
    "database-url-and-persistence-backends",
    "ssh-rate-limit-and-policy-posture",
    "sandbox-service-routing",
    "gateway-listener-tls-and-sni",
    "plaintext-listener-mode",
    "guest-callback-tls-ownership",
    "oidc-bearer-authentication",
    "mtls-user-authentication",
    "unsafe-unauthenticated-user-mode",
    "gateway-minted-sandbox-jwt",
    "otlp-observability",
    "gateway-interceptor-registration",
    "supervisor-middleware-registration",
    "provider-profile-sources",
    "inference-control-plane-configuration",
    "credential-driver-selection-and-kek",
    "credential-driver-backend-tables",
    "docker-image-and-callback-configuration",
    "docker-security-and-provider-configuration",
    "podman-image-and-callback-configuration",
    "podman-runtime-security-and-health",
    "kubernetes-core-placement-and-images",
    "kubernetes-workspace-isolation",
    "kubernetes-supervisor-topology",
    "kubernetes-egress-spiffe-and-security",
    "vm-launch-and-resource-configuration",
    "vm-guest-security-and-spiffe",
    "mxc-windows-driver-configuration",
    "external-compute-driver-socket",
    "helm-configuration-producer",
    "local-launch-script-producers",
    "e2e-fixture-producers",
    "rpm-schema-upgrade",
    "homebrew-debian-and-snap-upgrades",
}
ALLOWED_LANES = {
    "deterministic",
    "e2e-docker",
    "e2e-podman",
    "e2e-kubernetes",
    "e2e-vm",
    "windows-mxc",
    "extension-driver",
    "auth-oidc",
    "observability",
    "packaging",
}
# This inventory plans execution. Live results belong in the execution record,
# not in this baseline manifest, so a PASS cannot be accidentally implied.
ALLOWED_STATUSES = {"not_run", "blocked", "planned"}


def load_manifest() -> dict[str, Any]:
    with MANIFEST_PATH.open("rb") as manifest_file:
        return tomllib.load(manifest_file)


def require_nonempty_string(value: Any, field: str, entry_id: str) -> None:
    assert isinstance(value, str) and value.strip(), f"{entry_id}: {field} is required"


def require_string_list(value: Any, field: str, entry_id: str) -> None:
    assert isinstance(value, list) and value, f"{entry_id}: {field} must be non-empty"
    assert all(isinstance(item, str) and item.strip() for item in value), (
        f"{entry_id}: {field} must contain only non-empty strings"
    )
    assert len(value) == len(set(value)), f"{entry_id}: {field} contains duplicates"


def validate_manifest(manifest: dict[str, Any]) -> None:
    assert set(manifest) == REQUIRED_MANIFEST_FIELDS, (
        "unexpected or missing manifest metadata"
    )
    assert manifest["manifest_version"] == 1
    assert manifest["baseline_ref"] == "origin/main"
    assert manifest["baseline_commit"] == "74960ebfaeec4673885089ed995fad902459749f"
    assert manifest["baseline_schema_version"] == 1
    assert manifest["candidate_ref"] == "HEAD"
    assert manifest["candidate_commit"] == "8c868e430e9cd3284d7e274628419ab484ebcee0"
    assert manifest["candidate_schema_version"] == 2

    capabilities = manifest["capabilities"]
    assert isinstance(capabilities, list) and capabilities, (
        "capabilities must be non-empty"
    )
    ids: list[str] = []
    topics: set[str] = set()
    for capability in capabilities:
        assert isinstance(capability, dict), "each capability must be a TOML table"
        assert set(capability) == REQUIRED_CAPABILITY_FIELDS, (
            "capability has unexpected or missing metadata"
        )
        entry_id = capability["id"]
        require_nonempty_string(entry_id, "id", "capability")
        ids.append(entry_id)
        require_string_list(capability["topics"], "topics", entry_id)
        topics.update(capability["topics"])
        require_string_list(
            capability["origin_main_access_paths"],
            "origin_main_access_paths",
            entry_id,
        )
        require_string_list(
            capability["schema_v2_access_paths"],
            "schema_v2_access_paths",
            entry_id,
        )
        require_nonempty_string(
            capability["behavioral_oracle"], "behavioral_oracle", entry_id
        )
        require_nonempty_string(
            capability["required_environment"], "required_environment", entry_id
        )
        assert capability["test_lane"] in ALLOWED_LANES, (
            f"{entry_id}: unknown test lane {capability['test_lane']!r}"
        )
        assert capability["status"] in ALLOWED_STATUSES, (
            f"{entry_id}: live PASS results are not valid in this planning manifest"
        )

    assert len(ids) == len(set(ids)), "capability IDs must be unique"
    assert set(ids) == REQUIRED_CAPABILITY_IDS, (
        "capability inventory is incomplete or stale"
    )
    assert topics == REQUIRED_TOPICS, (
        "topic inventory is incomplete or contains an unknown topic"
    )


def capability_by_id(manifest: dict[str, Any], capability_id: str) -> dict[str, Any]:
    return next(
        capability
        for capability in manifest["capabilities"]
        if capability["id"] == capability_id
    )


def test_schema_v2_capability_parity_manifest_is_well_formed() -> None:
    validate_manifest(load_manifest())


def test_frozen_comparison_commits_are_full_git_object_ids() -> None:
    manifest = load_manifest()

    for field in ("baseline_commit", "candidate_commit"):
        commit = manifest[field]
        assert len(commit) == 40
        assert all(character in "0123456789abcdef" for character in commit)


def test_vm_gateway_inventory_excludes_standalone_driver_only_fields() -> None:
    capability = capability_by_id(
        load_manifest(), "vm-launch-and-resource-configuration"
    )
    gateway_paths = " ".join(
        capability["origin_main_access_paths"] + capability["schema_v2_access_paths"]
    )

    gateway_fields = set(re.findall(r"[a-z][a-z0-9_]*", gateway_paths))
    for standalone_field in ("launcher_bin", "log_level", "gpu_enabled", "gpu_mem_mib"):
        assert standalone_field not in gateway_fields
    assert "driver_dir" in gateway_fields


def test_new_docker_capabilities_are_not_attributed_to_origin_main() -> None:
    capability = capability_by_id(
        load_manifest(), "docker-security-and-provider-configuration"
    )
    origin_paths = " ".join(capability["origin_main_access_paths"])
    candidate_paths = " ".join(capability["schema_v2_access_paths"])

    assert "no origin/main Docker equivalent" in origin_paths
    for added_field in (
        "https_proxy",
        "provider_spiffe_workload_api_socket",
        "app_armor_profile",
    ):
        assert added_field not in origin_paths
        assert added_field in candidate_paths


@pytest.mark.parametrize("field", sorted(REQUIRED_MANIFEST_FIELDS - {"capabilities"}))
def test_manifest_rejects_missing_header_metadata(field: str) -> None:
    manifest = deepcopy(load_manifest())
    del manifest[field]

    with pytest.raises(AssertionError, match="missing manifest metadata"):
        validate_manifest(manifest)


@pytest.mark.parametrize("field", sorted(REQUIRED_CAPABILITY_FIELDS - {"id"}))
def test_manifest_rejects_missing_capability_metadata(field: str) -> None:
    manifest = deepcopy(load_manifest())
    del manifest["capabilities"][0][field]

    with pytest.raises(AssertionError, match=r"metadata|required|must be"):
        validate_manifest(manifest)


@pytest.mark.parametrize(
    ("field", "value", "match"),
    [
        ("topics", [], "must be non-empty"),
        ("behavioral_oracle", "", "is required"),
        ("required_environment", "", "is required"),
        ("test_lane", "not-a-lane", "unknown test lane"),
    ],
)
def test_manifest_rejects_malformed_capability_metadata(
    field: str, value: Any, match: str
) -> None:
    manifest = deepcopy(load_manifest())
    manifest["capabilities"][0][field] = value

    with pytest.raises(AssertionError, match=match):
        validate_manifest(manifest)


def test_manifest_rejects_duplicate_capability_id() -> None:
    manifest = deepcopy(load_manifest())
    manifest["capabilities"][1]["id"] = manifest["capabilities"][0]["id"]

    with pytest.raises(AssertionError, match=r"unique|incomplete"):
        validate_manifest(manifest)


def test_manifest_does_not_claim_live_pass_results() -> None:
    manifest = deepcopy(load_manifest())
    manifest["capabilities"][0]["status"] = "pass"

    with pytest.raises(AssertionError, match="live PASS"):
        validate_manifest(manifest)