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.maintenance import (
    check_generated,
    concept_candidates,
    export_emission_bundle,
    source_evidence,
)
from info_tech_canon.service import DEFAULT_INFOSPACE_ROOT, 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_concept_candidates_find_prose_definitions_the_declarations_miss():
    """The gap that let itc-org:Authority pass the SecurityCanon conflict check."""
    report = concept_candidates(load_context())
    organization = next(item for item in report["artifacts"]
                        if item["artifact"] == "model/organization")

    assert "Authority" in organization["undeclared"]
    assert organization["undeclared_count"] > organization["declared_count"]
    assert report["undeclared_total"] > 0
    assert "model/security" in report["silent_artifacts"]


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"]