chemenu/tools/chemenu/tests/test_migrate_cmd.py

"""Tests for `wikitool migrate`: the KB version, the migration chain, its
ordering rule, and `verify` against real git history."""
from __future__ import annotations

import json
import subprocess
from pathlib import Path

import pytest
import typer

from chemenu import config, kb_state
from chemenu.commands import migrate_cmd
from chemenu.version import Version

CHANGES = "# Changelog\n\n---\n\n## 1.0.0 - 2026-08-30 - First\n\nBody.\n"


def write_migration(
    directory: Path,
    target: str,
    slug: str,
    kind: str = "assisted",
    obligation: str = "required",
) -> Path:
    directory.mkdir(parents=True, exist_ok=True)
    path = directory / f"{target}-{slug}.md"
    path.write_text(
        "---\n"
        "type: types/instruction.md\n"
        f"name: {target}-{slug}\n"
        f"description: Migration to {target}.\n"
        "manual: true\n"
        f"migrates_to: {target}\n"
        f"migration_kind: {kind}\n"
        f"obligation: {obligation}\n"
        "---\n\n# Migration\n\nSteps.\n",
        encoding="utf-8",
    )
    return path


@pytest.fixture
def instance(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
    """A tree with a stack version, a changelog and three migrations, so the
    chain has something to order."""
    (tmp_path / "VERSION").write_text("2.0.0\n", encoding="utf-8")
    (tmp_path / "CHANGES.md").write_text(CHANGES, encoding="utf-8")
    instructions = tmp_path / "instructions"
    migrations = instructions / "migrations"
    for target, slug in (("1.4.0", "rename-field"), ("1.7.0", "split-sources"), ("2.0.0", "retype")):
        write_migration(migrations, target, slug)
    # Below the range and above the machinery: neither belongs in a chain.
    write_migration(migrations, "1.2.0", "ancient")
    write_migration(migrations, "2.1.0", "not-installed-yet")

    monkeypatch.setattr(config, "ROOT", tmp_path)
    monkeypatch.setattr(config, "INSTRUCTIONS_DIR", instructions)
    monkeypatch.setenv("WIKI_AUTHOR", "Test Author")
    return tmp_path


def set_kb_version(root: Path, version: str) -> None:
    (root / kb_state.KB_STATE_FILENAME).write_text(
        json.dumps({"schema": 1, "kb_version": version, "applied": []}), encoding="utf-8"
    )


# --- the chain -------------------------------------------------------------


def test_chain_is_the_open_interval_up_to_the_installed_machinery(instance):
    migrations = kb_state.load_migrations()
    pending = kb_state.chain(migrations, Version(1, 3, 1), Version(2, 0, 0))
    assert [str(m.target) for m in pending] == ["1.4.0", "1.7.0", "2.0.0"]


def test_a_version_with_no_migration_of_its_own_is_not_a_special_case(instance):
    """1.3.1 has no migration targeting it - it simply is not in the interval,
    and the chain starts at the next one that is."""
    pending = kb_state.chain(kb_state.load_migrations(), Version(1, 3, 1), Version(2, 0, 0))
    assert str(pending[0].target) == "1.4.0"


def test_migrations_above_the_installed_machinery_are_excluded(instance):
    """2.1.0 exists as a document but this instance has no code for it."""
    pending = kb_state.chain(kb_state.load_migrations(), Version(1, 0, 0), Version(2, 0, 0))
    assert "2.1.0" not in [str(m.target) for m in pending]


def test_nothing_outstanding_when_content_matches_machinery(instance):
    assert kb_state.chain(kb_state.load_migrations(), Version(2, 0, 0), Version(2, 0, 0)) == []


# --- status ----------------------------------------------------------------


def test_status_refuses_to_guess_an_undeclared_kb_version(instance):
    with pytest.raises(typer.Exit):
        migrate_cmd.status_command(json_out=False)


def test_status_lists_the_chain_in_order(instance, capsys):
    set_kb_version(instance, "1.3.1")
    migrate_cmd.status_command(json_out=True)
    result = json.loads(capsys.readouterr().out)
    assert result["kb_version"] == "1.3.1"
    assert [m["migrates_to"] for m in result["pending"]] == ["1.4.0", "1.7.0", "2.0.0"]


def test_list_reports_every_document_sorted_by_target(instance, capsys):
    migrate_cmd.list_command(json_out=True)
    targets = [m["migrates_to"] for m in json.loads(capsys.readouterr().out)]
    assert targets == ["1.2.0", "1.4.0", "1.7.0", "2.0.0", "2.1.0"]


# --- done: the ordering rule ----------------------------------------------


def test_done_advances_the_kb_version_and_records_the_entry(instance):
    set_kb_version(instance, "1.3.1")
    migrate_cmd.done_command(version="1.4.0", pages=38, dry_run=False)
    state = kb_state.read_kb_state()
    assert state["kb_version"] == "1.4.0"
    assert state["applied"][-1]["migration"] == "1.4.0-rename-field"
    assert state["applied"][-1]["pages"] == 38


def test_done_refuses_a_migration_that_is_not_next(instance):
    """Skipping a link leaves the corpus in a shape no version describes."""
    set_kb_version(instance, "1.3.1")
    with pytest.raises(typer.Exit):
        migrate_cmd.done_command(version="2.0.0", dry_run=False)
    assert kb_state.read_kb_version() == Version(1, 3, 1)


def test_an_interrupted_upgrade_resumes_where_it_stopped(instance):
    set_kb_version(instance, "1.3.1")
    migrate_cmd.done_command(version="1.4.0", pages=None, dry_run=False)
    pending = kb_state.chain(kb_state.load_migrations(), Version(1, 4, 0), Version(2, 0, 0))
    assert [str(m.target) for m in pending] == ["1.7.0", "2.0.0"]


def test_done_dry_run_writes_nothing(instance):
    set_kb_version(instance, "1.3.1")
    migrate_cmd.done_command(version="1.4.0", pages=None, dry_run=True)
    assert kb_state.read_kb_version() == Version(1, 3, 1)


def test_done_without_a_declared_kb_version_is_refused(instance):
    with pytest.raises(typer.Exit):
        migrate_cmd.done_command(version="1.4.0", pages=None, dry_run=False)


# --- baseline --------------------------------------------------------------


def test_baseline_declares_the_version_once(instance):
    migrate_cmd.baseline_command(version="1.3.1", force=False)
    assert kb_state.read_kb_version() == Version(1, 3, 1)


def test_baseline_refuses_to_overwrite_without_force(instance):
    """Advancing after a migration is `done`, which checks the chain; baseline
    does not, so it must not become the quiet way around it."""
    migrate_cmd.baseline_command(version="1.3.1", force=False)
    with pytest.raises(typer.Exit):
        migrate_cmd.baseline_command(version="2.0.0", force=False)
    assert kb_state.read_kb_version() == Version(1, 3, 1)
    migrate_cmd.baseline_command(version="2.0.0", force=True)
    assert kb_state.read_kb_version() == Version(2, 0, 0)


# --- verify against real git history --------------------------------------


PAGE = """---
type: types/entity.md
entity_type: system
created: 2026-07-31
provenance: sourced
---

# Aurora

Links to [[Borealis]] and again to [[Borealis]].
"""


@pytest.fixture
def git_instance(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
    kb = tmp_path / "kb" / "entities"
    kb.mkdir(parents=True)
    (tmp_path / "kb" / "entities" / "COLLECTION.md").write_text("# c\n", encoding="utf-8")
    page = kb / "Aurora.md"
    page.write_text(PAGE, encoding="utf-8")

    subprocess.run(["git", "init", "-q", "-b", "main"], cwd=tmp_path, check=True)
    subprocess.run(["git", "config", "user.name", "T"], cwd=tmp_path, check=True)
    subprocess.run(["git", "config", "user.email", "t@e.invalid"], cwd=tmp_path, check=True)
    subprocess.run(["git", "add", "-A"], cwd=tmp_path, check=True)
    subprocess.run(
        ["git", "commit", "-q", "-m", "seed"], cwd=tmp_path, check=True, capture_output=True
    )

    monkeypatch.setattr(config, "ROOT", tmp_path)
    monkeypatch.setattr(config, "KB_DIR", tmp_path / "kb")
    return tmp_path


def test_verify_is_clean_when_only_prose_moved(git_instance, capsys):
    page = git_instance / "kb" / "entities" / "Aurora.md"
    page.write_text(PAGE.replace("Links to", "Verweist auf"), encoding="utf-8")
    migrate_cmd.verify_command(
        from_rev="HEAD", path=None, expect_body_change=False, json_out=True, fail_on_error=True
    )
    result = json.loads(capsys.readouterr().out)
    assert result["compared"] == 1
    assert result["findings"] == []


def test_verify_catches_a_dropped_link_against_history(git_instance, capsys):
    page = git_instance / "kb" / "entities" / "Aurora.md"
    page.write_text(PAGE.replace(" and again to [[Borealis]]", ""), encoding="utf-8")
    with pytest.raises(typer.Exit):
        migrate_cmd.verify_command(
            from_rev="HEAD", path=None, expect_body_change=False, json_out=True, fail_on_error=True
        )
    result = json.loads(capsys.readouterr().out)
    assert result["findings"][0]["kind"] == "wikilinks"
    assert "'Borealis' 2->1" in result["findings"][0]["detail"]


def test_verify_scopes_to_a_subtree(git_instance, capsys):
    page = git_instance / "kb" / "entities" / "Aurora.md"
    page.write_text(PAGE.replace(" and again to [[Borealis]]", ""), encoding="utf-8")
    migrate_cmd.verify_command(
        from_rev="HEAD",
        path=["kb/concepts"],
        expect_body_change=False,
        json_out=True,
        fail_on_error=True,
    )
    assert json.loads(capsys.readouterr().out)["compared"] == 0


def test_verify_does_not_mistake_routing_files_for_removed_pages(git_instance, capsys):
    """Regression: the historical side listed every `.md` under kb/ while the
    working-tree side skipped COLLECTION.md/INDEX.md, so a clean run reported
    13 phantom removals. Both sides now answer with `kb_scan.is_page_path`."""
    migrate_cmd.verify_command(
        from_rev="HEAD", path=None, expect_body_change=False, json_out=True, fail_on_error=True
    )
    result = json.loads(capsys.readouterr().out)
    assert result["removed"] == []
    assert result["added"] == []
    assert result["compared"] == 1


def test_verify_reports_an_unknown_revision(git_instance):
    with pytest.raises(typer.Exit):
        migrate_cmd.verify_command(
            from_rev="no-such-rev",
            path=None,
            expect_body_change=False,
            json_out=False,
            fail_on_error=False,
        )


# --- obligation: required vs offered ---------------------------------------


def test_an_offered_migration_is_not_in_the_outstanding_chain(instance):
    """An offer is the stack proposing a better default for a file the instance
    owns. Declining it leaves the content in a shape the machinery accepts, so
    counting it as owed would make `kb_version` unreachable for an instance that
    simply kept its own file."""
    write_migration(
        instance / "instructions" / "migrations", "1.9.0", "nicer-template",
        kind="mechanical", obligation="offered",
    )
    migrations = kb_state.load_migrations()
    pending = kb_state.chain(migrations, Version.parse("1.3.0"), Version.parse("2.0.0"))
    assert "1.9.0-nicer-template" not in [m.name for m in pending]
    assert [str(m.target) for m in pending] == ["1.4.0", "1.7.0", "2.0.0"]


def test_an_offered_migration_is_listed_separately(instance):
    write_migration(
        instance / "instructions" / "migrations", "1.9.0", "nicer-template",
        kind="mechanical", obligation="offered",
    )
    offered = kb_state.offers(kb_state.load_migrations(), applied=set())
    assert [m.name for m in offered] == ["1.9.0-nicer-template"]


def test_an_offer_stays_on_the_table_regardless_of_the_version(instance):
    """Offers are bounded by the applied ledger, not by kb_version - taking one
    deliberately does not move the version, so the version can say nothing about
    whether it was taken. Nor are they bounded above by the stack: an offer is
    about a file the instance owns, not about the content shape."""
    directory = instance / "instructions" / "migrations"
    write_migration(directory, "1.1.0", "old-default", obligation="offered")
    write_migration(directory, "2.1.0", "later-default", obligation="offered")
    offered = kb_state.offers(kb_state.load_migrations(), applied=set())
    assert [m.name for m in offered] == ["1.1.0-old-default", "2.1.0-later-default"]


def test_taking_an_offer_records_it_without_moving_the_version(instance):
    write_migration(
        instance / "instructions" / "migrations", "1.9.0", "nicer-template",
        obligation="offered",
    )
    set_kb_version(instance, "1.3.1")
    migrate_cmd.done_command(version="1.9.0", pages=None, dry_run=False)

    state = kb_state.read_kb_state()
    assert state["kb_version"] == "1.3.1"
    assert state["applied"][-1]["migration"] == "1.9.0-nicer-template"
    assert kb_state.offers(kb_state.load_migrations(), kb_state.applied_names(state)) == []


def test_an_offer_out_of_order_is_not_refused(instance):
    """The chain's ordering rule exists because skipping a link leaves the
    corpus in an undescribed shape. An offer is not a link, so there is nothing
    to skip - and refusing it would make the required chain a prerequisite for
    an unrelated file upgrade."""
    write_migration(
        instance / "instructions" / "migrations", "1.9.0", "nicer-template",
        obligation="offered",
    )
    set_kb_version(instance, "1.3.1")  # 1.4.0 is the next *required* link
    migrate_cmd.done_command(version="1.9.0", pages=None, dry_run=False)
    assert kb_state.read_kb_version() == Version(1, 3, 1)


def test_obligation_defaults_to_required_when_undeclared(instance):
    """Every migration written before this axis existed is mandatory, and an
    unreadable value must not silently downgrade one."""
    directory = instance / "instructions" / "migrations"
    path = write_migration(directory, "1.5.0", "legacy")
    path.write_text(
        path.read_text(encoding="utf-8").replace("obligation: required\n", ""), encoding="utf-8"
    )
    bogus = write_migration(directory, "1.6.0", "bogus", obligation="whatever")
    assert bogus.is_file()

    by_name = {m.name: m for m in kb_state.load_migrations()}
    assert by_name["1.5.0-legacy"].obligation == kb_state.REQUIRED
    assert by_name["1.6.0-bogus"].obligation == kb_state.REQUIRED


def test_status_never_blocks_on_an_offer(instance, capsys):
    write_migration(
        instance / "instructions" / "migrations", "1.9.0", "nicer-template",
        obligation="offered",
    )
    set_kb_version(instance, "2.0.0")
    migrate_cmd.status_command(json_out=True)
    result = json.loads(capsys.readouterr().out)
    assert result["pending"] == []
    assert [m["name"] for m in result["offered"]] == ["1.9.0-nicer-template"]
    # The chain is empty and the offer is listed: `status` reports both without
    # the offer ever counting as owed.


# --- divergence against the release stamp ----------------------------------


def _write_stamp(root: Path, files: dict[str, str]) -> None:
    import hashlib

    digests = {}
    for relative, content in files.items():
        path = root / relative
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(content, encoding="utf-8")
        digests[relative] = "sha256:" + hashlib.sha256(content.encode("utf-8")).hexdigest()
    (root / ".wikitool-release.json").write_text(
        json.dumps({"schema": 1, "version": "2.0.0", "files": digests}), encoding="utf-8"
    )


def test_divergent_files_tells_an_edited_file_from_a_received_one(instance):
    """The half of the release stamp that has existed since it was written and
    that nothing read until offers needed it: may this file be overwritten, or
    does a person have to reconcile it?"""
    _write_stamp(instance, {"types/entity.md": "shipped\n", "types/concept.md": "shipped\n"})
    (instance / "types" / "entity.md").write_text("locally changed\n", encoding="utf-8")

    assert kb_state.divergent_files() == ["types/entity.md"]


def test_a_deleted_file_counts_as_divergent(instance):
    _write_stamp(instance, {"types/entity.md": "shipped\n"})
    (instance / "types" / "entity.md").unlink()
    assert kb_state.divergent_files() == ["types/entity.md"]


def test_divergence_is_unanswerable_without_a_stamp(instance):
    """None, not []. A development tree carries no stamp, and reporting
    "nothing diverged" there would be a fabricated answer."""
    assert kb_state.divergent_files() is None