info-tech-canon/tests/test_maintenance.py

from copy import deepcopy
import hashlib
import json
from pathlib import Path
import shutil
import tarfile

import pytest
import yaml

from info_tech_canon.cli import main
from info_tech_canon.generation import concept_ownership
from info_tech_canon.maintenance import (
    check_generated,
    concept_candidates,
    export_emission_bundle,
    source_evidence,
)
from info_tech_canon.contracts import coverage
from info_tech_canon.service import (
    DEFAULT_INFOSPACE_ROOT,
    concept_declaration_checks,
    load_context,
    validate_canon,
)


@pytest.fixture
def corpus(tmp_path):
    root = tmp_path / "infospace"
    shutil.copytree(DEFAULT_INFOSPACE_ROOT, root)
    return root


def test_emission_cli_checks_versions_duplicates_and_bad_types(tmp_path, capsys):
    source = DEFAULT_INFOSPACE_ROOT / "standards/emission-cadence/examples/qonto-assistant.yaml"
    valid = yaml.safe_load(source.read_text())
    path = tmp_path / "declaration.yaml"
    path.write_text(yaml.safe_dump(valid))
    assert main(["emission-review", str(path)]) == 0
    assert json.loads(capsys.readouterr().out)["operational_truth_assessed"] is False
    for change in ("version", "duplicate", "bad_type"):
        data = deepcopy(valid)
        if change == "version":
            data["schema_version"] = "99.0"
        elif change == "duplicate":
            data["sources"].append(deepcopy(data["sources"][0]))
        else:
            data["sources"][0]["source_id"] = ["unhashable"]
        path.write_text(yaml.safe_dump(data))
        assert main(["emission-review", str(path)]) == 1
        payload = json.loads(capsys.readouterr().out)
        assert payload["errors"]
        if change == "duplicate":
            assert payload["errors"][0]["code"] == "duplicate_emission_cadence_source_id"


def test_freshness_detects_missing_and_stale_without_repair(corpus):
    from info_tech_canon import generation
    context = load_context(corpus)
    for render in (generation.generate_indexes, generation.generate_tree, generation.generate_agent_briefs):
        render(context)
    assert check_generated(context)["ok"]
    stale = corpus / "views/by-concept.md"
    stale.write_text("stale\n")
    missing = corpus / "agent/retrieval-index.json"
    missing.unlink()
    result = check_generated(context)
    assert not result["ok"]
    assert "views/by-concept.md" in result["stale"]
    assert "agent/retrieval-index.json" in result["stale"]
    assert stale.read_text() == "stale\n"
    assert not missing.exists()


def test_bundle_is_reproducible_and_refuses_corruption(tmp_path):
    first = export_emission_bundle(DEFAULT_INFOSPACE_ROOT, tmp_path)
    second = export_emission_bundle(DEFAULT_INFOSPACE_ROOT, tmp_path)
    assert first == second
    path = Path(first["path"])
    assert hashlib.sha256(path.read_bytes()).hexdigest() == first["sha256"]
    with tarfile.open(path) as archive:
        for name, digest in first["manifest"]["files"].items():
            assert hashlib.sha256(archive.extractfile(name).read()).hexdigest() == digest
    path.write_bytes(b"corrupted")
    with pytest.raises(ValueError, match="Refusing"):
        export_emission_bundle(DEFAULT_INFOSPACE_ROOT, tmp_path)


def test_mapping_schema_and_explicit_ownership_are_enforced(corpus):
    from info_tech_canon.contracts import bound_artifact_errors
    mapping = corpus / "mappings/capability-anchors.yaml"
    data = yaml.safe_load(mapping.read_text())
    del data["target"]
    mapping.write_text(yaml.safe_dump(data))
    assert bound_artifact_errors(load_context(corpus))[0]["code"] == "schema_violation"
    pattern = corpus / "patterns/AgenticDrivesFunctional.md"
    text = pattern.read_text()
    # Reuse another artifact's title as an owned concept to create a real conflict.
    owner = next(a.title for a in load_context(corpus).infospace.artifacts if a.kind == "kernel")
    text = text.replace("owned_concepts:", f"owned_concepts:\n  - {owner}")
    pattern.write_text(text)
    assert any(e["code"] == "concept_ownership_conflict" for e in validate_canon(corpus)["errors"])


def test_evidence_digest_tracks_content_but_excludes_reports(corpus):
    before = source_evidence(corpus)
    (corpus / "validation/latest.json").write_text("{}")
    assert source_evidence(corpus)["corpus_sha256"] == before["corpus_sha256"]
    (corpus / "mappings/README.md").write_text("changed")
    assert source_evidence(corpus)["corpus_sha256"] != before["corpus_sha256"]
    assert before["generated_at"]


def test_explicit_root_controls_capability_catalog(corpus, tmp_path):
    from info_tech_canon.service import review_capability_record, CanonServiceError
    (corpus / "models/capability/capabilities.yaml").unlink()
    record = tmp_path / "record.json"
    record.write_text('{"record_id":"example", "requires":[], "provisions":[]}')
    with pytest.raises(CanonServiceError, match="catalog"):
        review_capability_record(record, corpus)


def test_missing_mapping_schema_is_a_validation_finding(corpus):
    (corpus / "schemas/mapping.schema.yaml").unlink()
    payload = validate_canon(corpus)
    assert not payload["ok"]
    assert any(e["code"] == "mapping_schema_unreadable" for e in payload["errors"])


def test_authority_is_declared_where_the_kernel_map_assigns_it():
    """The gap that let itc-org:Authority pass the SecurityCanon conflict check."""
    ownership = concept_ownership(load_context())
    owners = {item["concept"]: item["owner"] for item in ownership["concepts"]}

    assert owners["Authority"] == "model/organization"
    assert owners["Actor"] == "model/organization"
    assert not ownership["ownership_conflicts"]


def test_only_the_kernel_map_declares_no_concepts():
    """It assigns concepts to owners rather than defining them (T02)."""
    report = concept_candidates(load_context())

    assert report["silent_artifacts"] == ["kernel/itc-kernel-map"]
    assert report["undeclared_total"] < report["candidate_total"] / 2


def test_concept_candidates_ignore_preserved_source():
    report = concept_candidates(load_context())
    paths = [item["path"] for item in report["artifacts"]]
    assert paths
    assert not [path for path in paths if path.startswith(("assimilation/", "seeds/"))]


def test_concept_coverage_cli_reports_one_artifact(capsys):
    assert main(["concept-coverage", "--artifact", "model/identity"]) == 0
    payload = json.loads(capsys.readouterr().out)
    assert payload["ok"] is True
    assert [item["artifact"] for item in payload["artifacts"]] == ["model/identity"]


def test_every_defined_concept_has_an_owner():
    context = load_context()
    result = concept_declaration_checks(context, concept_ownership(context))

    assert not result["errors"]
    assert not result["warnings"]


def test_declaration_warning_fires_for_a_concept_nobody_owns(corpus):
    target = corpus / "standards/caring/InfoTechCanonCaringAccessGovernanceStandard.md"
    text = target.read_text(encoding="utf-8")
    target.write_text(text.replace("  - Effective Access\n", "", 1), encoding="utf-8")

    context = load_context(corpus)
    result = concept_declaration_checks(context, concept_ownership(context))

    warned = {item["artifact_id"]: item for item in result["warnings"]}
    assert "standard/caring" in warned
    assert "Effective Access" in warned["standard/caring"]["concepts"]
    assert warned["standard/caring"]["code"] == "concept_defined_without_owner"


def test_declaration_error_fires_when_an_artifact_declares_nothing(corpus):
    target = corpus / "models/security/InfoTechCanonSecurityModel.md"
    text = target.read_text(encoding="utf-8")
    target.write_text(text.split("---\n", 2)[2], encoding="utf-8")

    context = load_context(corpus)
    result = concept_declaration_checks(context, concept_ownership(context))

    assert [item["artifact_id"] for item in result["errors"]] == ["model/security"]


def test_validation_coverage_reports_the_declaration_ratio():
    report = coverage(load_context())["concept_declaration"]

    assert report["silent_artifacts"] == ["kernel/itc-kernel-map"]
    assert report["undeclared"] < report["declared"]
    assert report["limit"]