infospace-bench/tests/test_content_explorer_projection.py

"""Projection of the Lefevre infospace into a content-explorer presentation."""

from __future__ import annotations

import sys
from pathlib import Path

import json

import yaml

from infospace_bench.markdown_adapter import extract_section_text, parse_markdown_artifact
from infospace_bench.presentation import display_title, project_infospace
from infospace_bench.semantics import slugify

ROOT = Path(__file__).resolve().parents[1]
LEFEVRE = ROOT / "infospaces" / "lefevre-reminiscences-of-a-stock-operator"
CONTENT_EXPLORER = Path("/home/worsch/content-explorer")
sys.path.insert(0, str(CONTENT_EXPLORER))

from content_explorer.contract import validate_documents  # noqa: E402


def test_lefevre_projection_matches_stored_artifacts_and_contract() -> None:
    first = project_infospace(LEFEVRE)
    second = project_infospace(LEFEVRE)
    snapshot = first.snapshot
    manifest = first.manifest

    assert [item["id"] for item in snapshot["entities"]] == [
        item["id"] for item in second.snapshot["entities"]
    ]
    assert [item["id"] for item in snapshot["relations"]] == [
        item["id"] for item in second.snapshot["relations"]
    ]

    index = yaml.safe_load((LEFEVRE / "artifacts" / "index.yaml").read_text(encoding="utf-8"))
    indexed = index["artifacts"]
    source_rows = [item for item in indexed if item["kind"] == "source"]
    entity_rows = [item for item in indexed if item["kind"] == "entity"]
    generated_rows = [item for item in indexed if item["kind"] == "generated"]
    relation_rows = [item for item in indexed if item["kind"] == "relation"]

    assert len(entity_rows) == 14
    assert len(snapshot["entities"]) == 14
    assert {item["kindId"] for item in snapshot["entities"]} >= {
        "instrument",
        "strategy",
        "market",
        "event",
        "evidence-bearing-claim",
        "institution",
        "error",
        "trader",
    }
    assert {item["id"] for item in snapshot["kinds"]} == {
        item["kindId"] for item in snapshot["entities"]
    }
    assert {item["id"] for item in snapshot["kinds"]}.isdisjoint(
        {"source", "entity", "relation", "generated"}
    )

    book = source_rows[0]["provenance"]["book_metadata"]
    assert len(snapshot["sources"]) == 1
    work = snapshot["sources"][0]
    assert work["license"] == book["rights"] == "Public domain in the USA."
    assert work["uri"] == book["source_url"]
    assert work["title"] == book["title"]
    attribution = work["attribution"]
    assert "provenance:" in attribution
    assert len(source_rows) == 6
    for source in source_rows:
        provenance = source["provenance"]
        assert provenance["digest"] in attribution
        assert provenance["original_path"] in attribution
        assert provenance["chunk_id"] in attribution

    generated_titles = {item["title"] for item in generated_rows}
    generated_ids = {item["id"] for item in generated_rows}
    by_path = {item["properties"]["artifact_path"]: item for item in snapshot["entities"]}
    assert len(by_path) == 14
    for row in entity_rows:
        projected = by_path[row["path"]]
        parsed = parse_markdown_artifact(LEFEVRE / row["path"])
        heading = next(item.text for item in parsed.headings if item.level == 1)
        definition = extract_section_text(parsed, "Definition").strip()
        category = extract_section_text(parsed, "Category").strip()
        assert projected["title"] == display_title(heading)
        assert projected["summary"] == definition
        assert projected["tldr"]
        assert definition.startswith(projected["tldr"])
        assert projected["kindId"] == slugify(category)
        assert projected["id"] == (
            f"urn:ce:lefevre-reminiscences-of-a-stock-operator:"
            f"{projected['kindId']}:{slugify(projected['title'])}"
        )
        assert projected["sourceRefs"] == [work["id"]]
        assert projected["properties"]["workflow_id"] == row["provenance"]["workflow_id"]
        assert projected["properties"]["input_artifact_id"] == row["provenance"]["input_artifact_id"]
        assert projected["title"] not in generated_titles
        assert projected["id"] not in generated_ids

    block_count = 0
    for row in relation_rows:
        text = (LEFEVRE / row["path"]).read_text(encoding="utf-8")
        block_count += text.count("\n## Subject") + int(text.startswith("## Subject"))
    assert block_count == 27
    assert len(first.resolved_relations) + len(first.quarantined_relations) == block_count
    assert first.quarantined_relations
    assert len(snapshot["relations"]) == len(first.resolved_relations)
    entities_by_id = {item["id"]: item for item in snapshot["entities"]}
    relations_by_id = {item["id"]: item for item in snapshot["relations"]}
    for record in first.resolved_relations:
        projected = relations_by_id[record.relation_id]
        assert projected["predicate"] == record.relation_type.replace("_", "-")
        assert " " not in projected["predicate"]
        if record.predicate_text != projected["predicate"]:
            assert projected["predicate"] != record.predicate_text
        assert slugify(record.subject) == slugify(entities_by_id[record.from_id]["title"])
        assert slugify(record.object) == slugify(entities_by_id[record.to_id]["title"])
        assert projected["from"] == record.from_id
        assert projected["to"] == record.to_id
        assert projected["sourceRefs"] == [work["id"]]
    resolved_keys = {(item.artifact_id, item.block_index) for item in first.resolved_relations}
    for record in first.quarantined_relations:
        assert (record.artifact_id, record.block_index) not in resolved_keys
        subject_hits = [
            item
            for item in snapshot["entities"]
            if slugify(item["title"]) == slugify(record.subject)
        ]
        object_hits = [
            item
            for item in snapshot["entities"]
            if slugify(item["title"]) == slugify(record.object)
        ]
        assert not (len(subject_hits) == 1 and len(object_hits) == 1)

    assert manifest["schemaVersion"] == "0.1"
    assert manifest["id"] == "lefevre-reminiscences-of-a-stock-operator"
    assert manifest["panels"] == ["summary", "relations", "evidence", "claims"]
    assert manifest["factPolicy"] == "show-all"
    assert set(manifest["kindViews"]) == {item["id"] for item in snapshot["kinds"]}
    for view in manifest["kindViews"].values():
        assert view["columns"]
        assert "facets" in view
    assert set(manifest["home"]["featuredKinds"]) == {item["id"] for item in snapshot["kinds"]}
    assert set(manifest["relationLabels"]) == {item["predicate"] for item in snapshot["relations"]}
    for wording in manifest["relationLabels"].values():
        assert wording["forward"]
        assert wording["inverse"]

    export_dir = LEFEVRE / "exports" / "content-explorer"
    assert json.loads((export_dir / "snapshot.json").read_text(encoding="utf-8")) == snapshot
    assert json.loads((export_dir / "manifest.json").read_text(encoding="utf-8")) == manifest

    errors = validate_documents(snapshot, manifest, require_size=True)
    assert errors == []
    raw_errors = validate_documents(index, index, require_size=True)
    assert raw_errors

    note = (LEFEVRE / "exports" / "content-explorer" / "PROJECTION.md").read_text(encoding="utf-8")
    assert "lefevre-reminiscences-of-a-stock-operator" in note
    assert "content-explorer only presents" in note.lower()
    assert "infospace-bench still owns" in note.lower()

    print("ENTITY_IDS " + ",".join(item["id"] for item in snapshot["entities"]))
    print("RELATION_IDS " + ",".join(item["id"] for item in snapshot["relations"]))
    print("VALIDATOR_ERRORS 0")
    print("RAW_ARTIFACTS_ERRORS " + str(len(raw_errors)))