chemenu/tools/chemenu/tests/test_xref.py

from chemenu import blocks, links
from chemenu.frontmatter_io import read_page
from chemenu.commands.xref import (
    apply_links_block,
    remove_link_bullets,
    remove_related,
    render_links_block,
)
from chemenu.kb_scan import load_kb_pages
from chemenu.page import Page
from pathlib import Path


def _page(related, body="\n# X\n\nProse.\n"):
    return Page(Path("kb/entities/X.md"), {"related": related}, body)


def test_an_edge_carries_its_label_in_the_data():
    fm = {}
    assert links.upsert(fm, "related", links.Edge("B", "depends-on")) is True
    assert links.upsert(fm, "related", links.Edge("B", "depends-on")) is False
    assert fm["related"] == [{"depends-on": "B"}]


def test_relabelling_replaces_rather_than_appends():
    """One page asserts one thing about another. Two edges to the same target
    would render two bullets with no way to say which is meant."""
    fm = {"related": [{"uses": "B"}]}
    assert links.upsert(fm, "related", links.Edge("B", "depends-on")) is True
    assert fm["related"] == [{"depends-on": "B"}]


def test_a_bare_title_reads_as_an_unlabelled_edge():
    """The shape every page is in between this machinery landing and the
    migration reaching it. Readers must not crash on it, and it must stay
    visibly unlabelled so `lint` can report it."""
    fm = {"related": ["B", {"uses": "C"}]}
    edges = links.edges(fm, "related")
    assert [(e.target, e.label) for e in edges] == [("B", None), ("C", "uses")]
    assert links.targets(fm, "related") == ["B", "C"]
    assert [e.is_labelled for e in edges] == [False, True]


def test_a_malformed_entry_is_reported_rather_than_guessed_at():
    fm = {"related": [{"a": "X", "b": "Y"}, 42, "Fine"]}
    assert links.targets(fm, "related") == ["Fine"]
    assert len(links.malformed(fm, "related")) == 2


def test_remove_related_is_the_inverse_of_upsert():
    fm = {"related": [{"uses": "A"}, {"uses": "B"}]}
    assert remove_related(fm, "B") is True
    assert fm["related"] == [{"uses": "A"}]
    assert remove_related(fm, "B") is False


def test_remove_related_tolerates_a_missing_field():
    assert remove_related({}, "B") is False


def test_retarget_keeps_the_label():
    fm = {"related": [{"depends-on": "Old"}]}
    assert links.retarget(fm, "related", "Old", "New") is True
    assert fm["related"] == [{"depends-on": "New"}]


def test_the_body_block_is_rendered_from_the_frontmatter():
    """The body is a rendering of the graph, not a second place it is stored.
    That is what removed the need to parse a German bullet back into a
    relationship."""
    page = _page([{"depends-on": "Hermes"}, "Unlabelled"])
    body = apply_links_block(page)
    region = blocks.find(body, blocks.LINKS)
    assert "- **depends-on:** [[Hermes]]" in region
    assert "- [[Unlabelled]]" in region


def test_rendering_is_idempotent_and_replaces_rather_than_appends():
    page = _page([{"uses": "A"}])
    once = apply_links_block(page)
    twice = apply_links_block(page, once)
    assert once == twice
    page.frontmatter["related"] = [{"uses": "B"}]
    thrice = apply_links_block(page, once)
    assert thrice.count("<!-- wikitool:links -->") == 1
    assert "[[A]]" not in thrice and "[[B]]" in thrice


def test_a_page_with_no_edges_carries_no_region():
    page = _page([])
    body = apply_links_block(page)
    assert "wikitool:links" not in body
    assert render_links_block(page) == ""


def test_prose_after_the_region_survives_a_rewrite():
    """The failure the marker pair exists to make impossible. The old block ran
    to the next heading - and before that to the end of the file - so a section
    sitting after it was deleted on the next write. Eight pages were carrying
    content in that position when it was found."""
    page = _page([{"uses": "A"}])
    body = apply_links_block(page) + "\n## Afterwards\n\nKeep me.\n"
    page.frontmatter["related"] = [{"uses": "B"}]
    rewritten = apply_links_block(page, body)
    assert "Keep me." in rewritten
    assert rewritten.count("## Afterwards") == 1


def test_remove_link_bullets_removes_what_the_renderer_wrote():
    body = "\n# X\n\n## Beziehungen\n\n- **uses:** [[B]]\n- [[B]]\n"
    result = remove_link_bullets(body, "B")
    assert "[[B]]" not in result


def test_xref_add_updates_both_pages_on_disk(kb_dir):
    from typer.testing import CliRunner
    from chemenu.cli import app
    import chemenu.config as config

    config.KB_DIR = kb_dir
    config.INDEX_FILE = kb_dir / "index.md"

    runner = CliRunner()
    modbus_before = (kb_dir / "concepts/protocols/Modbus.md").read_text(encoding="utf-8")
    result = runner.invoke(app, ["xref", "add", "--a", "gdeploy", "--b", "Modbus", "--rel", "uses"])
    assert result.exit_code == 0, result.output

    pages = load_kb_pages(kb_dir)
    assert links.targets(pages["gdeploy"].frontmatter, "related") == ["Modbus"]
    assert "- **uses:** [[Modbus]]" in pages["gdeploy"].body

    # B is not touched at all. Its inbound view is rendered from the graph, so
    # nothing has to be written there for a reader to find its way back.
    assert (kb_dir / "concepts/protocols/Modbus.md").read_text(encoding="utf-8") == modbus_before

    _fm_before, body_before = read_page(kb_dir / "entities/tools/gdeploy.md")
    link_count_before = body_before.count("[[Modbus]]")

    result2 = runner.invoke(app, ["xref", "add", "--a", "gdeploy", "--b", "Modbus", "--rel", "uses"])
    assert result2.exit_code == 0
    fm_after, body_after = read_page(kb_dir / "entities/tools/gdeploy.md")
    assert links.targets(fm_after, "related").count("Modbus") == 1
    assert body_after.count("[[Modbus]]") == link_count_before


def test_xref_remove_undoes_xref_add(kb_dir):
    from typer.testing import CliRunner
    from chemenu.cli import app
    import chemenu.config as config

    config.KB_DIR = kb_dir
    config.INDEX_FILE = kb_dir / "index.md"

    runner = CliRunner()
    before = (kb_dir / "entities/tools/gdeploy.md").read_text(encoding="utf-8")

    added = runner.invoke(app, ["xref", "add", "--a", "gdeploy", "--b", "Modbus", "--rel", "uses"])
    assert added.exit_code == 0, added.output
    removed = runner.invoke(app, ["xref", "remove", "--a", "gdeploy", "--b", "Modbus"])
    assert removed.exit_code == 0, removed.output

    pages = load_kb_pages(kb_dir)
    assert "Modbus" not in links.targets(pages["gdeploy"].frontmatter, "related")
    assert "gdeploy" not in links.targets(pages["Modbus"].frontmatter, "related")
    assert "[[Modbus]]" not in pages["gdeploy"].body
    assert before  # sanity: fixture page was non-empty


def test_xref_remove_clears_a_ref_to_a_page_that_no_longer_exists(kb_dir):
    """The cleanup case: a hand-deleted or hand-renamed page leaves `related:`
    pointing at nothing, and only `--a` can still be loaded."""
    from typer.testing import CliRunner
    from chemenu.cli import app
    import chemenu.config as config
    from chemenu.frontmatter_io import write_page

    config.KB_DIR = kb_dir
    config.INDEX_FILE = kb_dir / "index.md"

    write_page(
        kb_dir / "entities/tools/gdeploy.md",
        {"type": "types/entity.md", "entity_type": "tool", "tags": [],
         "created": "2026-07-25", "modified": "2026-07-25",
         "related": ["Ghost Page"], "sources": []},
        "\n# gdeploy\n\n## See Also\n\n- [[Ghost Page]]\n",
    )

    runner = CliRunner()
    result = runner.invoke(app, ["xref", "remove", "--a", "gdeploy", "--b", "Ghost Page"])
    assert result.exit_code == 0, result.output
    assert "not a page" in result.output

    pages = load_kb_pages(kb_dir)
    assert pages["gdeploy"].frontmatter["related"] == []
    assert "[[Ghost Page]]" not in pages["gdeploy"].body


def test_xref_remove_is_idempotent(kb_dir):
    from typer.testing import CliRunner
    from chemenu.cli import app
    import chemenu.config as config

    config.KB_DIR = kb_dir
    config.INDEX_FILE = kb_dir / "index.md"

    runner = CliRunner()
    result = runner.invoke(app, ["xref", "remove", "--a", "gdeploy", "--b", "Modbus"])
    assert result.exit_code == 0, result.output
    assert "nothing changed" in result.output


def test_xref_remove_dry_run_writes_nothing(kb_dir):
    from typer.testing import CliRunner
    from chemenu.cli import app
    import chemenu.config as config

    config.KB_DIR = kb_dir
    config.INDEX_FILE = kb_dir / "index.md"

    runner = CliRunner()
    runner.invoke(app, ["xref", "add", "--a", "gdeploy", "--b", "Modbus"])
    before = (kb_dir / "entities/tools/gdeploy.md").read_text(encoding="utf-8")

    result = runner.invoke(
        app, ["xref", "remove", "--a", "gdeploy", "--b", "Modbus", "--dry-run"]
    )
    assert result.exit_code == 0, result.output
    assert "No files written" in result.output
    assert (kb_dir / "entities/tools/gdeploy.md").read_text(encoding="utf-8") == before


def test_xref_add_dry_run_writes_nothing(kb_dir):
    from typer.testing import CliRunner
    from chemenu.cli import app
    import chemenu.config as config

    config.KB_DIR = kb_dir
    config.INDEX_FILE = kb_dir / "index.md"

    gdeploy_path = kb_dir / "entities/tools/gdeploy.md"
    modbus_path = kb_dir / "concepts/protocols/Modbus.md"
    gdeploy_before = gdeploy_path.read_text(encoding="utf-8")
    modbus_before = modbus_path.read_text(encoding="utf-8")

    runner = CliRunner()
    result = runner.invoke(
        app,
        ["xref", "add", "--a", "gdeploy", "--b", "Modbus", "--rel", "uses", "--dry-run"],
    )
    assert result.exit_code == 0, result.output
    assert "would declare" in result.output
    assert "No files written" in result.output

    assert gdeploy_path.read_text(encoding="utf-8") == gdeploy_before
    assert modbus_path.read_text(encoding="utf-8") == modbus_before


def test_xref_link_source_dry_run_writes_nothing(kb_dir):
    from typer.testing import CliRunner
    from chemenu.cli import app
    import chemenu.config as config

    config.KB_DIR = kb_dir
    config.INDEX_FILE = kb_dir / "index.md"

    gdeploy_path = kb_dir / "entities/tools/gdeploy.md"
    gdeploy_before = gdeploy_path.read_text(encoding="utf-8")

    runner = CliRunner()
    result = runner.invoke(
        app,
        ["xref", "link-source", "--source", "Source - Aurora", "--entities", "gdeploy", "--dry-run"],
    )
    assert result.exit_code == 0, result.output
    assert "Would link" in result.output
    assert "No files written" in result.output
    assert gdeploy_path.read_text(encoding="utf-8") == gdeploy_before


def test_xref_add_reports_a_write_failure(kb_dir, monkeypatch):
    """One edge, one write - so there is no half-written pair to report any
    more. The old two-sided `xref add` could update A and fail on B, leaving a
    link the user had to be told was one-directional; a directional edge has
    nothing to be half of."""
    from typer.testing import CliRunner
    from chemenu.cli import app
    from chemenu.commands import xref as xref_module
    import chemenu.config as config

    config.KB_DIR = kb_dir
    config.INDEX_FILE = kb_dir / "index.md"

    def flaky_write_page(path, frontmatter, body):
        raise OSError("disk full")

    monkeypatch.setattr(xref_module, "write_page", flaky_write_page)

    runner = CliRunner()
    result = runner.invoke(
        app, ["xref", "add", "--a", "gdeploy", "--b", "Modbus", "--rel", "uses"]
    )
    assert result.exit_code == 1
    assert "disk full" in result.output

    pages = load_kb_pages(kb_dir)
    assert "Modbus" not in links.targets(pages["gdeploy"].frontmatter, "related")


def test_xref_add_refuses_a_label_the_collection_does_not_authorise(kb_dir):
    """The source collection decides which labels may be used from it. Refused
    here rather than only in `lint`, because this is the moment the author is
    present and can pick a better one."""
    from typer.testing import CliRunner
    from chemenu.cli import app
    import chemenu.config as config

    config.KB_DIR = kb_dir
    config.INDEX_FILE = kb_dir / "index.md"

    runner = CliRunner()
    result = runner.invoke(
        app, ["xref", "add", "--a", "gdeploy", "--b", "Modbus", "--rel", "hängt ab von"]
    )
    assert result.exit_code == 1
    assert "not authorised" in result.output
    assert "link-taxonomy" in result.output


def test_xref_link_source_distinguishes_write_failures_from_missing_pages(kb_dir, monkeypatch):
    from typer.testing import CliRunner
    from chemenu.cli import app
    from chemenu.commands import xref as xref_module
    import chemenu.config as config

    config.KB_DIR = kb_dir
    config.INDEX_FILE = kb_dir / "index.md"

    real_write_page = xref_module.write_page

    def flaky_write_page(path, frontmatter, body):
        if path.name == "gdeploy.md":
            raise OSError("disk full")
        return real_write_page(path, frontmatter, body)

    monkeypatch.setattr(xref_module, "write_page", flaky_write_page)

    runner = CliRunner()
    result = runner.invoke(
        app,
        ["xref", "link-source", "--source", "Source - Aurora", "--entities", "gdeploy,Modbus"],
    )
    assert result.exit_code == 1
    assert "Failed to write" in result.output
    assert "gdeploy" in result.output

    pages = load_kb_pages(kb_dir)
    assert "Source - Aurora" in pages["Modbus"].frontmatter["sources"]  # unaffected by gdeploy's failure


def test_xref_link_source_reports_missing_entity_but_links_the_rest(kb_dir):
    from typer.testing import CliRunner
    from chemenu.cli import app
    import chemenu.config as config

    config.KB_DIR = kb_dir
    config.INDEX_FILE = kb_dir / "index.md"

    runner = CliRunner()
    result = runner.invoke(
        app,
        ["xref", "link-source", "--source", "Source - Aurora", "--entities", "gdeploy,Nonexistent Page"],
    )
    assert result.exit_code == 1
    assert "Linked source" in result.output
    assert "gdeploy" in result.output
    assert "Skipped (page not found): Nonexistent Page" in result.output

    # The valid entity must still have been linked despite the other one being missing.
    pages = load_kb_pages(kb_dir)
    assert "Source - Aurora" in pages["gdeploy"].frontmatter["sources"]


# --- reference fields a type does not declare (issue #18) --------------------


def _runner_env(kb_dir):
    from typer.testing import CliRunner
    from chemenu.cli import app
    import chemenu.config as config

    config.KB_DIR = kb_dir
    config.INDEX_FILE = kb_dir / "index.md"
    return CliRunner(), app


def test_xref_add_refuses_a_type_without_a_related_field(kb_dir):
    """`types/source.md` declares `page_ref_fields: [entities, concepts]`. Writing
    `related:` there produced frontmatter the schema rejects, and `xref remove`
    could not clear it - one command creating a state another could not undo."""
    runner, app = _runner_env(kb_dir)
    result = runner.invoke(
        app, ["xref", "add", "--a", "Source - Aurora", "--b", "aurora", "--rel", "uses"]
    )
    assert result.exit_code == 1
    # Rich wraps the message to the terminal width, so compare on collapsed
    # whitespace rather than pinning the line breaks.
    output = " ".join(result.output.split())
    assert "does not declare a `related:` field" in output
    assert "entities, concepts" in output
    assert "link-source" in output

    pages = load_kb_pages(kb_dir)
    assert "related" not in pages["Source - Aurora"].frontmatter


def test_xref_add_refuses_before_writing_either_side(kb_dir):
    """A refusal must not leave a half-link behind."""
    runner, app = _runner_env(kb_dir)
    before = (kb_dir / "entities/systems/aurora.md").read_text(encoding="utf-8")
    runner.invoke(app, ["xref", "add", "--a", "aurora", "--b", "Source - Aurora"])
    assert (kb_dir / "entities/systems/aurora.md").read_text(encoding="utf-8") == before


def test_xref_remove_clears_an_undeclared_leftover_field(kb_dir):
    """The state 1.6.0 stopped producing still has to be repairable, or every
    page that already carries one is a dead end."""
    from chemenu.frontmatter_io import write_page

    path = kb_dir / "sources/notes/Source - Aurora.md"
    frontmatter, body = read_page(path)
    frontmatter["related"] = ["aurora"]
    write_page(path, frontmatter, body)

    runner, app = _runner_env(kb_dir)
    result = runner.invoke(app, ["xref", "remove", "--a", "Source - Aurora", "--b", "aurora"])
    assert result.exit_code == 0, result.output

    frontmatter, _ = read_page(path)
    # Removed outright, not left as `related: []` - the key was never valid for
    # this type, and an empty list keeps the page failing schema validation.
    assert "related" not in frontmatter


def test_xref_link_source_records_the_targets_on_the_source_page(kb_dir):
    """The way back. Without it an ingest that creates its concept pages after
    the source page - which it must, since their titles come out of the
    extraction - left `concepts:` empty with no command able to fill it."""
    runner, app = _runner_env(kb_dir)
    result = runner.invoke(
        app, ["xref", "link-source", "--source", "Source - Aurora", "--entities", "aurora"]
    )
    assert result.exit_code == 0, result.output

    pages = load_kb_pages(kb_dir)
    source = pages["Source - Aurora"].frontmatter
    # Routed by the target's collection: kb/entities/ -> entities:
    assert "aurora" in source["entities"]
    assert "aurora" not in source.get("concepts", [])
    # And the direction it always wrote.
    assert "Source - Aurora" in pages["aurora"].frontmatter["sources"]


def test_xref_link_source_is_idempotent_on_the_source_page(kb_dir):
    runner, app = _runner_env(kb_dir)
    for _ in range(2):
        runner.invoke(
            app, ["xref", "link-source", "--source", "Source - Aurora", "--entities", "aurora"]
        )
    source = load_kb_pages(kb_dir)["Source - Aurora"].frontmatter
    assert source["entities"].count("aurora") == 1


def test_xref_link_source_dry_run_leaves_the_source_page_alone(kb_dir):
    runner, app = _runner_env(kb_dir)
    path = kb_dir / "sources/notes/Source - Aurora.md"
    before = path.read_text(encoding="utf-8")
    runner.invoke(
        app,
        ["xref", "link-source", "--source", "Source - Aurora", "--entities", "aurora",
         "--dry-run"],
    )
    assert path.read_text(encoding="utf-8") == before


def test_xref_link_source_refuses_a_target_that_is_itself_a_source(kb_dir):
    """`link-source` means "A mentions X" and writes `X.sources += A`; for X =
    another source that claims "B cites A", and the back-reference would add
    `A.sources += B` too. A citation between sources has a direction, and only
    `cite add` says which one - so the target is refused, the rest still linked."""
    from chemenu.frontmatter_io import write_page

    runner, app = _runner_env(kb_dir)
    path_b = kb_dir / "sources/notes/Source - B.md"
    write_page(
        path_b,
        {"type": "types/source.md", "source_type": "notes", "author": "Fixture Author",
         "raw_files": ["raw/notes/Uningested.md"], "date": "2026-08-02", "summary": "B"},
        "\n# Source: B\n",
    )
    result = runner.invoke(
        app,
        ["xref", "link-source", "--source", "Source - Aurora", "--entities", "gdeploy,Source - B"],
    )
    assert result.exit_code == 1
    assert "Source - B" in result.output
    assert 'cite add --page "Source - Aurora" --source "Source - B"' in result.output

    pages = load_kb_pages(kb_dir)
    assert "Source - Aurora" in pages["gdeploy"].frontmatter["sources"]
    source_a = pages["Source - Aurora"].frontmatter
    source_b = pages["Source - B"].frontmatter
    for field in ("sources", "entities", "concepts"):
        assert "Source - B" not in (source_a.get(field) or [])
        assert "Source - Aurora" not in (source_b.get(field) or [])


# --- the inbound view ------------------------------------------------------


def test_the_inbound_view_is_derived_not_stored(kb_dir):
    """The load-bearing half of dropping mirrored edges. Nothing writes an edge
    onto the target, so the only way "what points at this page" can be answered
    completely is by computing it - which is also why it cannot go stale or be
    half-written the way a mirror could."""
    from typer.testing import CliRunner
    from chemenu.cli import app
    import chemenu.config as config

    config.KB_DIR = kb_dir
    config.INDEX_FILE = kb_dir / "index.md"

    runner = CliRunner()
    assert runner.invoke(
        app, ["xref", "add", "--a", "gdeploy", "--b", "Modbus", "--rel", "uses"]
    ).exit_code == 0

    import json

    result = runner.invoke(app, ["links", "show", "--page", "Modbus", "--json"])
    assert result.exit_code == 0, result.output
    data = json.loads(result.output)
    assert data["inbound"] == [{"source": "gdeploy", "label": "uses", "collection": "entities"}]
    # Nothing was written onto Modbus itself to make that answer possible.
    assert "gdeploy" not in links.targets(
        load_kb_pages(kb_dir)["Modbus"].frontmatter, "related"
    )


def test_an_unresolvable_outbound_edge_is_marked(kb_dir):
    """`lint` reports dangling references corpus-wide; this reports it for the
    one page someone is looking at, which is where it gets fixed."""
    from typer.testing import CliRunner
    from chemenu.cli import app
    import chemenu.config as config
    import json

    config.KB_DIR = kb_dir
    config.INDEX_FILE = kb_dir / "index.md"

    page = load_kb_pages(kb_dir)["gdeploy"]
    links.upsert(page.frontmatter, "related", links.Edge("Gone", "uses"))
    from chemenu.frontmatter_io import write_page

    write_page(page.path, page.frontmatter, page.body)

    result = CliRunner().invoke(app, ["links", "show", "--page", "gdeploy", "--json"])
    assert json.loads(result.output)["outbound"] == [
        {"target": "Gone", "label": "uses", "resolves": False}
    ]