hub-core/tests/test_repository_navigation_contract.py

import json
from collections.abc import Mapping, Sequence
from typing import Any

from jsonschema import Draft202012Validator, FormatChecker

from hub_core.contracts import (
    REPOSITORY_NAVIGATION_CONTRACT_ID,
    REPOSITORY_NAVIGATION_CONTRACT_VERSION,
    repository_navigation_contract_root,
)


ROOT = repository_navigation_contract_root()
SCHEMAS = ROOT.joinpath("schemas")
INPUT_FIXTURE = ROOT.joinpath("fixtures", "repository-classification-page.json")
OUTPUT_FIXTURE = ROOT.joinpath("fixtures", "repository-navigation-projection.json")
OPENAPI = ROOT.joinpath("openapi", "repository-navigation.openapi.json")
COMPATIBILITY = ROOT.joinpath("compatibility-matrix.json")

FACET_KINDS = {
    "primary_domain",
    "secondary_domain",
    "category",
    "capability_tag",
    "business_stake",
    "business_mechanic",
}
SECRET_OR_LOCAL_KEYS = {
    "api_key",
    "credential",
    "locations",
    "password",
    "passwd",
    "private_key",
    "repo_root",
    "secret",
    "token",
    "work_records",
}


def load_json(resource: Any) -> Any:
    return json.loads(resource.read_text(encoding="utf-8"))


def validate(instance: Any, schema_name: str) -> None:
    schema = load_json(SCHEMAS.joinpath(schema_name))
    Draft202012Validator.check_schema(schema)
    Draft202012Validator(schema, format_checker=FormatChecker()).validate(instance)


def iter_keys(value: Any) -> set[str]:
    if isinstance(value, Mapping):
        keys = {str(key).lower() for key in value}
        for child in value.values():
            keys.update(iter_keys(child))
        return keys
    if isinstance(value, Sequence) and not isinstance(value, (str, bytes)):
        keys: set[str] = set()
        for child in value:
            keys.update(iter_keys(child))
        return keys
    return set()


def test_packaged_repository_navigation_contract_identity_and_artifacts() -> None:
    assert REPOSITORY_NAVIGATION_CONTRACT_ID == "helixforge.repository-navigation"
    assert REPOSITORY_NAVIGATION_CONTRACT_VERSION == "1.0.0"
    for resource in (
        ROOT.joinpath("README.md"),
        INPUT_FIXTURE,
        OUTPUT_FIXTURE,
        OPENAPI,
        COMPATIBILITY,
    ):
        assert resource.is_file()


def test_input_and_output_fixtures_validate() -> None:
    validate(load_json(INPUT_FIXTURE), "repository-classification-page.schema.json")
    validate(load_json(OUTPUT_FIXTURE), "repository-navigation-projection.schema.json")


def test_input_semantics_use_stable_unique_identity_and_consistent_final_page() -> None:
    page = load_json(INPUT_FIXTURE)
    repository_ids = [entry["repository_id"] for entry in page["repositories"]]

    assert repository_ids == sorted(repository_ids)
    assert len(repository_ids) == len(set(repository_ids))
    assert page["snapshot"]["final_page"] is (page["snapshot"]["next_cursor"] is None)
    assert page["snapshot"]["total_repository_count"] == len(repository_ids)
    assert all(
        entry["classification"].get("domain")
        not in entry["classification"].get("secondary_domains", [])
        for entry in page["repositories"]
        if entry["operation"] == "upsert"
    )


def test_output_has_every_derived_facet_and_per_repository_provenance() -> None:
    projection = load_json(OUTPUT_FIXTURE)
    repositories = projection["repositories"]
    repository_ids = [entry["repository_id"] for entry in repositories]

    assert repository_ids == sorted(repository_ids)
    assert len(repository_ids) == len(set(repository_ids))
    assert {facet["kind"] for facet in projection["facets"]} == FACET_KINDS
    assert projection["total_repository_count"] == len(repositories)
    assert all(
        entry["provenance"]["source_ref"].endswith(entry["repository_id"])
        for entry in repositories
    )
    assert all(
        facet["repository_count"] == len(facet["repository_ids"])
        for facet in projection["facets"]
    )


def test_contract_examples_exclude_secrets_and_host_local_authority_data() -> None:
    assert not (iter_keys(load_json(INPUT_FIXTURE)) & SECRET_OR_LOCAL_KEYS)
    assert not (iter_keys(load_json(OUTPUT_FIXTURE)) & SECRET_OR_LOCAL_KEYS)


def test_compatibility_accepts_only_repo_manager_v1_without_coercion() -> None:
    matrix = load_json(COMPATIBILITY)
    accepted = matrix["accepted_inputs"]

    assert matrix["contract_id"] == REPOSITORY_NAVIGATION_CONTRACT_ID
    assert matrix["current_version"] == REPOSITORY_NAVIGATION_CONTRACT_VERSION
    assert accepted == [
        {
            "contract_id": "helixforge.repository-classification-projection",
            "version": "1.0.0",
            "classification_contract_id": "RMGR-CONTRACT-CLASSIFICATION-0001",
            "classification_contract_version": "1.0",
            "status": "current",
            "coercion": False,
        }
    ]


def test_navigation_openapi_is_read_only_and_exposes_all_filter_families() -> None:
    document = load_json(OPENAPI)
    operations = [
        (method, operation)
        for path_item in document["paths"].values()
        for method, operation in path_item.items()
        if method in {"get", "post", "put", "patch", "delete"}
    ]
    query = document["paths"][
        "/ports/projections/repository-navigation/repositories"
    ]["get"]
    parameter_names = {
        reference["$ref"].rsplit("/", 1)[-1]
        for reference in query["parameters"]
        if "$ref" in reference
    }

    assert document["openapi"] == "3.1.0"
    assert all(method == "get" for method, _ in operations)
    assert all(operation["x-port-id"] == "port.projection.query" for _, operation in operations)
    assert {
        "PrimaryDomain",
        "SecondaryDomain",
        "Category",
        "CapabilityTag",
        "BusinessStake",
        "BusinessMechanic",
    } <= parameter_names
    assert set(document["components"]["schemas"]["FacetKind"]["enum"]) == FACET_KINDS


def test_legacy_topic_alias_is_expiring_read_only_metadata() -> None:
    projection = load_json(OUTPUT_FIXTURE)
    aliases = projection["legacy_topic_aliases"]
    matrix = load_json(COMPATIBILITY)

    assert aliases
    assert all(alias["expires_at"] and alias["residual_owner"] for alias in aliases)
    adapter = matrix["legacy_adapters"][0]
    assert adapter["authority"] is False
    assert adapter["status"] == "temporary-read-only"