teleo-infrastructure/tests/test_kb_governance_concept_packet.py

"""Tests for scripts/kb_governance_concept_packet.py."""

from __future__ import annotations

import json
import sys
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(REPO_ROOT / "scripts"))

import kb_governance_concept_packet as packet  # noqa: E402


def test_packet_declares_minimal_schema_and_generated_rows():
    result = packet.build_packet()
    assert result["base_packet_required_first"] == "production-apply-packet-mapped-20260709"
    assert result["production_apply_status"] == "not_executed"
    assert result["schema_writes"] == [
        "public.concept_maps",
        "public.claim_concept_map_links",
        "public.claim_governance_gate_links",
    ]
    assert result["generated_counts"] == {
        "claim_concept_map_links": 1,
        "claim_governance_gate_links": 2,
        "concept_maps": 1,
    }
    assert result["required_existing_rows"] == {
        "public.claims": [
            packet.CLAIM_DISTRIBUTED_MARKET_INTELLIGENCE_ID,
            packet.CLAIM_NODES_NEED_BODIES_ID,
        ],
        "public.governance_gates": [packet.QUALITY_GATE_ID, packet.EVIDENCE_BAR_ID],
    }


def test_concept_map_row_and_links_represent_remaining_fragments():
    concept = packet.concept_map_row()
    assert concept["id"] == packet.CONCEPT_MAP_ID
    assert concept["key"] == "distributed_market_intelligence"
    assert concept["metadata"]["proposal_id"] == packet.PROPOSAL_CONCEPT_ID

    concept_link = packet.concept_links()[0]
    assert concept_link["claim_id"] == packet.CLAIM_DISTRIBUTED_MARKET_INTELLIGENCE_ID
    assert concept_link["concept_map_id"] == packet.CONCEPT_MAP_ID
    assert concept_link["relation"] == "derives_from"

    gate_links = {row["governance_gate_id"]: row for row in packet.governance_links()}
    assert gate_links[packet.QUALITY_GATE_ID]["claim_id"] == packet.CLAIM_NODES_NEED_BODIES_ID
    assert gate_links[packet.EVIDENCE_BAR_ID]["claim_id"] == packet.CLAIM_NODES_NEED_BODIES_ID
    assert {row["relation"] for row in gate_links.values()} == {"supports"}


def test_schema_sql_creates_first_class_tables_without_runtime_changes():
    sql = packet.build_schema_sql()
    assert "create table if not exists public.concept_maps" in sql
    assert "create table if not exists public.claim_concept_map_links" in sql
    assert "create table if not exists public.claim_governance_gate_links" in sql
    assert "references public.claims(id) on delete cascade" in sql
    assert "references public.governance_gates(id) on delete cascade" in sql
    assert "references public.concept_maps(id) on delete cascade" in sql


def test_apply_sql_is_guarded_and_commit_or_rollback_specific():
    commit_sql = packet.build_packet()["sql"]["apply_commit"]
    rollback_sql = packet.build_packet()["sql"]["apply_rollback_rehearsal"]
    assert "requires mapped base claims" in commit_sql
    assert "requires quality/evidence governance gates" in commit_sql
    assert "generated concept map already exists" in commit_sql
    assert "expected 1" in commit_sql
    assert "expected 2" in commit_sql
    assert commit_sql.strip().endswith("commit;")
    assert rollback_sql.strip().endswith("rollback;")


def test_delete_rollback_drops_schema_only_after_generated_rows_deleted():
    sql = packet.build_packet()["sql"]["delete_rollback"]
    concept_link_delete = sql.index("delete from public.claim_concept_map_links")
    concept_delete = sql.index("delete from public.concept_maps")
    drop_link = sql.index("drop table public.claim_concept_map_links")
    drop_concept = sql.index("drop table public.concept_maps")
    assert concept_link_delete < concept_delete < drop_concept
    assert concept_link_delete < drop_link
    assert "and not exists (select 1 from public.claim_governance_gate_links)" in sql
    assert "and not exists (select 1 from public.claim_concept_map_links)" in sql
    assert "and not exists (select 1 from public.concept_maps)" in sql


def test_write_packet_files(tmp_path: Path):
    result = packet.build_packet()
    packet.write_packet_files(result, tmp_path)
    expected = {
        "governance-concept-apply-authorized-commit.sql",
        "governance-concept-apply-rollback-rehearsal.sql",
        "governance-concept-delete-rollback.sql",
        "governance-concept-packet.json",
        "governance-concept-postflight.sql",
        "governance-concept-preflight.sql",
    }
    assert {path.name for path in tmp_path.iterdir()} == expected
    summary = json.loads((tmp_path / "governance-concept-packet.json").read_text())
    assert "sql" not in summary
    assert summary["generated_counts"]["claim_governance_gate_links"] == 2