content-explorer/tests/test_ingest.py

"""Ingestion gates for JSON, NDJSON, and Markdown frontmatter."""

from __future__ import annotations

import json
import tempfile
import unittest
from pathlib import Path

import sys

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))

from content_explorer.contract import validate_documents  # noqa: E402
from content_explorer.ingest import (  # noqa: E402
    load_json_records,
    load_markdown_records,
    load_ndjson_records,
    project_records,
    publish_directory,
)

LABELS = {"represented-by": {"forward": "Represented by", "inverse": "Represents"}}

KINDS = [
    {"id": "concept", "label": "Concept", "plural": "Concepts"},
    {"id": "method", "label": "Method", "plural": "Methods"},
]
SOURCE = {
    "id": "src:demo:notes",
    "uri": "https://example.invalid/content-explorer/ingest",
    "title": "Demo notes",
    "license": "CC0-1.0",
    "attribution": "Content Explorer demo",
}
RECORDS = [
    {
        "kind": "concept",
        "key": "feedback",
        "title": "Feedback",
        "tldr": "Results change the next action.",
        "summary": "A loop carries an outcome back to the actor.",
        "tags": ["regulation"],
        "relations": [{"predicate": "represented-by", "to": "method:loop-diagram"}],
    },
    {
        "kind": "method",
        "key": "loop-diagram",
        "title": "Loop diagram",
        "tldr": "Arrows that show influence.",
        "summary": "A drawing of influences among parts.",
        "tags": ["diagram"],
        "relations": [],
    },
]
def project(records, source=SOURCE, relation_labels=None):
    return project_records(
        records,
        dataspace_id="demo-notes",
        title="Demo notes",
        description="A connector test collection.",
        source=source,
        kinds=KINDS,
        observed_at="2026-10-12",
        relation_labels=LABELS if relation_labels is None else relation_labels,
    )


class IngestGates(unittest.TestCase):
    def test_json_ndjson_and_markdown_match_and_repeat(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            (root / "records.json").write_text(json.dumps(RECORDS), encoding="utf-8")
            (root / "records.ndjson").write_text(
                "\n".join(json.dumps(record) for record in RECORDS) + "\n",
                encoding="utf-8",
            )
            docs = root / "notes"
            docs.mkdir()
            for record in RECORDS:
                body = "---\n" + _yaml(record) + "---\n\n" + record["summary"] + "\n"
                (docs / f"{record['key']}.md").write_text(body, encoding="utf-8")
            json_records, json_digest = load_json_records(root / "records.json")
            ndjson_records, _ = load_ndjson_records(root / "records.ndjson")
            markdown_records, _ = load_markdown_records(docs)
            first = project_records(
                json_records,
                dataspace_id="demo-notes",
                title="Demo notes",
                description="A connector test collection.",
                source=SOURCE,
                kinds=KINDS,
                observed_at="2026-10-12",
                source_digest=json_digest,
                relation_labels=LABELS,
            )
            self.assertEqual(first["report"]["sourceDigest"], json_digest)
            second = project(ndjson_records)
            third = project(markdown_records)
            again = project(json_records)
            self.assertTrue(first["report"]["published"])
            self.assertEqual(first["report"]["contentHash"], second["report"]["contentHash"])
            self.assertEqual(first["report"]["contentHash"], third["report"]["contentHash"])
            self.assertEqual(first["report"]["contentHash"], again["report"]["contentHash"])
            self.assertEqual(first["snapshot"], third["snapshot"])
            for claim in first["snapshot"]["claims"]:
                self.assertTrue(claim["evidenceRefs"])
            errors = validate_documents(first["snapshot"], first["manifest"], require_size=False)
            self.assertEqual(errors, [])
            self.assertEqual(first["manifest"]["id"], "demo-notes")
            self.assertIn("Represented by", first["manifest"]["relationLabels"]["represented-by"]["forward"])

    def test_rights_and_dangling_refs_do_not_publish(self) -> None:
        blocked = project(RECORDS, source={**SOURCE, "license": "CC-BY-NC-4.0"})
        self.assertFalse(blocked["report"]["published"])
        self.assertIsNone(blocked["snapshot"])
        self.assertEqual(blocked["report"]["quarantined"][0]["reason"], "rights")
        dangling = project([
            {
                **RECORDS[0],
                "relations": [{"predicate": "represented-by", "to": "method:missing"}],
            }
        ])
        self.assertFalse(dangling["report"]["published"])
        self.assertTrue(dangling["report"]["dangling"])
        missing_credit = project(RECORDS, source={**SOURCE, "license": "CC-BY-4.0", "attribution": ""})
        self.assertFalse(missing_credit["report"]["published"])
        self.assertEqual(missing_credit["report"]["quarantined"][0]["detail"], "missing-attribution")
        duplicate = project(RECORDS + [{**RECORDS[0], "title": "Other feedback"}])
        self.assertFalse(duplicate["report"]["published"])
        self.assertEqual(duplicate["report"]["conflicts"][0]["reason"], "duplicate-identity")
        unlabeled = project(RECORDS, relation_labels={})
        self.assertFalse(unlabeled["report"]["published"])
        self.assertEqual(unlabeled["report"]["quarantined"][0]["reason"], "missing-relation-label")

    def test_tombstone_is_omitted_and_publish_writes_both_files(self) -> None:
        retired = project(RECORDS + [{
            "kind": "concept",
            "key": "retired",
            "title": "Retired",
            "tldr": "Gone.",
            "deleted": True,
        }])
        self.assertEqual(retired["report"]["tombstoned"], ["retired"])
        self.assertEqual(len(retired["snapshot"]["entities"]), 2)
        with tempfile.TemporaryDirectory() as tmp:
            destination = Path(tmp) / "out"
            publish_directory(destination, retired)
            self.assertTrue((destination / "snapshot.json").is_file())
            self.assertTrue((destination / "manifest.json").is_file())
            self.assertTrue((destination / "import-report.json").is_file())
            human = (destination / "import-report.md").read_text(encoding="utf-8")
            self.assertIn("Connector version: 0.1.0", human)
            self.assertIn("License: CC0-1.0", human)
            self.assertIn("retired", human)
            published = json.loads((destination / "snapshot.json").read_text(encoding="utf-8"))
            manifest = json.loads((destination / "manifest.json").read_text(encoding="utf-8"))
            self.assertEqual(validate_documents(published, manifest, require_size=False), [])
            with self.assertRaises(RuntimeError):
                publish_directory(destination, project(RECORDS, source={**SOURCE, "license": ""}))


def _yaml(record: dict) -> str:
    lines = [
        f"kind: {record['kind']}",
        f"key: {record['key']}",
        f"title: {record['title']}",
        f"tldr: {record['tldr']}",
        f"summary: {record['summary']}",
        "tags:",
        *[f"  - {tag}" for tag in record["tags"]],
        "relations:",
    ]
    if record["relations"]:
        for link in record["relations"]:
            lines.append(f"  - predicate: {link['predicate']}")
            lines.append(f"    to: {link['to']}")
    else:
        lines.append("  []")
    return "\n".join(lines) + "\n"


if __name__ == "__main__":
    unittest.main()