chemenu/tools/chemenu/tests/test_upstream_cmd.py

"""Tests for `wikitool upstream merge`/`upstream verify` - the code procedure
that replaces private-instance.md's prose merge script (Gitea #30).

Two real git repos stand in for a private instance (`repo`, remote name
`upstream`) and the public repo it takes updates from (`upstream`, a plain
repo committed to directly - a fetch-only remote does not need to be bare for
`git fetch` to work against it). Each scenario diverges the two by committing
independently on each side, exactly like a real fetch-only upstream would.
"""
from __future__ import annotations

import subprocess

import pytest
import typer

from chemenu import config, ownership
from chemenu.commands import git_publish, upstream_cmd


def _git(root, *args):
    result = subprocess.run(["git", *args], cwd=root, capture_output=True, text=True)
    assert result.returncode == 0, result.stderr
    return result


def _write(root, relative, content):
    path = root / relative
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(content, encoding="utf-8")


def _commit(root, message):
    _git(root, "add", "-A")
    _git(root, "commit", "-m", message)


@pytest.fixture
def two_repos(tmp_path, monkeypatch):
    """`repo`, a private instance, with a fetch-only `upstream` remote pointing
    at a second, independent repo. Both start from the same seed commit -
    kb/CONTRACT.md, kb/CONVENTIONS.md(.template), kb/entities/COLLECTION.md,
    raw/CONTRACT.md, work/CONTRACT.md, reports/CONTRACT.md, and one tools/
    file - which is what a private instance looks like right after the
    private-instance.md setup: the tracked machinery, plus its own filled
    instance files layered on top.
    """
    seed = tmp_path / "seed"
    seed.mkdir()
    _git(seed, "init", "-b", "main")
    _git(seed, "config", "user.name", "Seed")
    _git(seed, "config", "user.email", "seed@example.com")
    # .wikitool-remotes.json is gitignored in the real repo (it is per-checkout,
    # see config.PUBLISH_REMOTES_FILENAME) - without this, dropping one into the
    # fixture during a test would show up as an untracked file and trip the
    # dirty-working-tree precondition for a reason that has nothing to do with
    # what that test is checking.
    # Mirrors the real .gitignore in the two ways that matter here:
    # `.wikitool-remotes.json` is per-checkout (dropping one in during a test
    # must not read as a dirty tree), and `reports/` is derived output that is
    # ignored except for its contract - which is what makes a content stage
    # able to hold local, non-recomputable data a merge must not touch.
    _write(
        seed,
        ".gitignore",
        f"/{config.PUBLISH_REMOTES_FILENAME}\n/reports/*\n!/reports/CONTRACT.md\n",
    )
    _write(seed, "kb/CONTRACT.md", "stack kb contract v1\n")
    _write(seed, "kb/CONVENTIONS.md.template", "template v1\n")
    _write(seed, "kb/CONVENTIONS.md", "instance conventions v1\n")
    _write(seed, "kb/entities/COLLECTION.md", "instance collection contract v1\n")
    _write(seed, "kb/Both.md", "page both sides delete\n")
    _write(seed, "kb/ToDelete.md", "page the instance will delete\n")
    _write(seed, "kb/RegularPage.md", "an ordinary page neither side has touched yet\n")
    _write(seed, "raw/CONTRACT.md", "raw contract v1\n")
    _write(seed, "work/CONTRACT.md", "work contract v1\n")
    _write(seed, "reports/CONTRACT.md", "reports contract v1\n")
    _write(seed, "tools/wikitool.py", "line one\nline two\nline three\n")
    _commit(seed, "seed")

    upstream = tmp_path / "upstream"
    subprocess.run(["git", "clone", str(seed), str(upstream)], check=True, capture_output=True)
    _git(upstream, "config", "user.name", "Upstream")
    _git(upstream, "config", "user.email", "upstream@example.com")

    # Cloned from `upstream`, not from `seed` directly: the remote (renamed
    # below) must resolve to the path this fixture actually commits new
    # upstream state into, or a later `git fetch upstream main` silently
    # fetches from `seed` instead and never sees anything new.
    repo = tmp_path / "repo"
    subprocess.run(["git", "clone", str(upstream), str(repo)], check=True, capture_output=True)
    _git(repo, "config", "user.name", "Test")
    _git(repo, "config", "user.email", "test@example.com")
    _git(repo, "remote", "rename", "origin", "upstream")

    monkeypatch.setattr(config, "ROOT", repo)
    monkeypatch.setenv("WIKITOOL_SESSION_ID", "test-session")
    return upstream, repo


def _merge(**overrides):
    kwargs = dict(remote="upstream", branch="main", no_fetch=False)
    kwargs.update(overrides)
    upstream_cmd.merge_command(**kwargs)


# --- the four restbefund regressions, plus the baseline table from the issue ---


def test_upstream_edit_of_a_page_the_instance_deleted_does_not_land(two_repos):
    upstream, repo = two_repos
    _git(repo, "rm", "-q", "kb/ToDelete.md")
    _commit(repo, "instance deletes ToDelete")

    _write(upstream, "kb/ToDelete.md", "upstream edited it after the instance deleted it\n")
    _commit(upstream, "upstream edits ToDelete")

    _merge()

    assert not (repo / "kb/ToDelete.md").exists()


def test_upstream_new_page_does_not_land(two_repos):
    upstream, repo = two_repos
    _write(upstream, "kb/NewPage.md", "a demo page the upstream added\n")
    _commit(upstream, "upstream adds NewPage")

    _merge()

    assert not (repo / "kb/NewPage.md").exists()


def test_page_deleted_on_both_sides_is_a_noop(two_repos):
    upstream, repo = two_repos
    _git(repo, "rm", "-q", "kb/Both.md")
    _commit(repo, "instance deletes Both")
    _git(upstream, "rm", "-q", "kb/Both.md")
    _commit(upstream, "upstream deletes Both")

    _merge()  # must not raise

    assert not (repo / "kb/Both.md").exists()


def test_kb_contract_change_lands(two_repos):
    upstream, repo = two_repos
    _write(upstream, "kb/CONTRACT.md", "stack kb contract v2\n")
    _commit(upstream, "upstream changes kb/CONTRACT.md")

    _merge()

    assert (repo / "kb/CONTRACT.md").read_text(encoding="utf-8") == "stack kb contract v2\n"


def test_conventions_template_change_lands_local_conventions_untouched(two_repos):
    upstream, repo = two_repos
    _write(upstream, "kb/CONVENTIONS.md.template", "template v2\n")
    _commit(upstream, "upstream changes the conventions template")

    _merge()

    assert (repo / "kb/CONVENTIONS.md.template").read_text(encoding="utf-8") == "template v2\n"
    assert (repo / "kb/CONVENTIONS.md").read_text(encoding="utf-8") == "instance conventions v1\n"


def test_collection_contract_change_does_not_land(two_repos):
    """A COLLECTION.md is instance-owned since #39 - one level deeper than
    `<stage>/CONTRACT.md`, so `is_stack_owned` must say no to it."""
    upstream, repo = two_repos
    _write(repo, "kb/entities/COLLECTION.md", "instance collection contract v2 (local)\n")
    _commit(repo, "instance rewrites its own collection contract")

    _write(upstream, "kb/entities/COLLECTION.md", "upstream collection contract v2\n")
    _commit(upstream, "upstream changes the default collection contract")

    _merge()

    assert (repo / "kb/entities/COLLECTION.md").read_text(encoding="utf-8") == (
        "instance collection contract v2 (local)\n"
    )


def test_upstream_deletion_of_a_contract_file_lands(two_repos):
    """Restbefund 2: a machinery file the upstream deleted must not silently
    survive because `git checkout MERGE_HEAD -- <path>` has nothing to check
    out."""
    upstream, repo = two_repos
    _git(upstream, "rm", "-q", "raw/CONTRACT.md")
    _commit(upstream, "upstream drops raw/CONTRACT.md")

    _merge()

    assert not (repo / "raw/CONTRACT.md").exists()


def test_new_stack_template_under_a_content_stage_lands(two_repos):
    """Restbefund 4: a brand-new stack-owned path the local tree has never
    seen must still be recognised by the predicate, not by a literal list."""
    upstream, repo = two_repos
    _write(upstream, "kb/GLOSSARY.md.template", "a stack-owned template that never existed before\n")
    _commit(upstream, "upstream adds a new template")

    _merge()

    assert (repo / "kb/GLOSSARY.md.template").read_text(encoding="utf-8") == (
        "a stack-owned template that never existed before\n"
    )


def test_open_workshop_run_files_do_not_land(two_repos):
    upstream, repo = two_repos
    _write(upstream, "work/some-run/README.md", "an in-progress workshop run\n")
    _commit(upstream, "upstream ships an open work/ run")

    _merge()

    assert not (repo / "work/some-run").exists()


def test_merge_keeps_ignored_local_data_under_a_content_stage(two_repos):
    """`reports/` is gitignored except its contract, so a content stage's
    working tree holds local data that is in no git tree and is not
    recomputable - the telemetry traces `eval score` reads, saved eval
    reports, past lint reports. Forcing the stage back to the local side must
    not take those out as collateral: this instance had 497 trace directories
    under reports/telemetry/ when the first version of this command wiped the
    stage wholesale."""
    upstream, repo = two_repos
    _write(repo, "reports/telemetry/session-a/trace.jsonl", '{"event": "local"}\n')
    _write(repo, "reports/Lint Report 2026-09-04.md", "a local lint report\n")
    assert _git(repo, "status", "--porcelain").stdout == ""  # ignored, so the tree is clean

    _write(upstream, "kb/CONTRACT.md", "stack kb contract v2\n")
    _commit(upstream, "upstream changes kb/CONTRACT.md")

    _merge()

    assert (repo / "reports/telemetry/session-a/trace.jsonl").read_text(encoding="utf-8") == (
        '{"event": "local"}\n'
    )
    assert (repo / "reports/Lint Report 2026-09-04.md").exists()
    assert (repo / "reports/CONTRACT.md").read_text(encoding="utf-8") == "reports contract v1\n"


def test_upstream_content_under_a_stage_absent_from_head_does_not_land(two_repos):
    """The stage guard must not rest on the local side happening to track
    something under that stage: an instance whose `work/` holds no tracked
    file at all must still not receive the upstream's open run."""
    upstream, repo = two_repos
    _git(repo, "rm", "-q", "work/CONTRACT.md")
    _commit(repo, "instance has nothing tracked under work/")

    _write(upstream, "work/some-run/README.md", "an in-progress workshop run\n")
    _commit(upstream, "upstream ships an open work/ run")

    _merge()

    assert not (repo / "work/some-run").exists()


def test_one_upstream_commit_mixing_every_case_at_once(two_repos):
    """The acceptance test from the issue: a single upstream commit that edits
    a page the instance deleted, adds a new page, deletes an untouched page,
    changes a stack contract, changes a template, and deletes a different
    stack contract - all at once, all restored or discarded correctly by one
    `upstream merge` call."""
    upstream, repo = two_repos
    _git(repo, "rm", "-q", "kb/ToDelete.md")
    _commit(repo, "instance deletes ToDelete")

    _write(upstream, "kb/ToDelete.md", "upstream edited it after the instance deleted it\n")
    _write(upstream, "kb/BrandNewPage.md", "a demo page the upstream added\n")
    _git(upstream, "rm", "-q", "kb/RegularPage.md")
    _write(upstream, "kb/CONTRACT.md", "stack kb contract v2\n")
    _write(upstream, "kb/CONVENTIONS.md.template", "template v2\n")
    _git(upstream, "rm", "-q", "raw/CONTRACT.md")
    _commit(upstream, "one upstream commit: edit + add + delete + contract + template + contract-delete")

    _merge()

    assert not (repo / "kb/ToDelete.md").exists()
    assert not (repo / "kb/BrandNewPage.md").exists()
    assert (repo / "kb/RegularPage.md").exists()
    assert (repo / "kb/CONTRACT.md").read_text(encoding="utf-8") == "stack kb contract v2\n"
    assert (repo / "kb/CONVENTIONS.md.template").read_text(encoding="utf-8") == "template v2\n"
    assert (repo / "kb/CONVENTIONS.md").read_text(encoding="utf-8") == "instance conventions v1\n"
    assert not (repo / "raw/CONTRACT.md").exists()


def test_real_conflict_in_tools_leaves_the_merge_open(two_repos):
    upstream, repo = two_repos

    _write(repo, "tools/wikitool.py", "line one\nLOCAL CHANGE\nline three\n")
    _commit(repo, "local edits tools/wikitool.py")
    before = _git(repo, "rev-parse", "HEAD").stdout.strip()

    _write(upstream, "tools/wikitool.py", "line one\nUPSTREAM CHANGE\nline three\n")
    _commit(upstream, "upstream edits the same line")

    with pytest.raises(typer.Exit) as excinfo:
        _merge()
    assert excinfo.value.exit_code == 1

    assert _git(repo, "rev-parse", "HEAD").stdout.strip() == before
    assert (repo / ".git" / "MERGE_HEAD").exists()


def test_a_merge_git_refuses_to_open_deletes_nothing(two_repos, tmp_path):
    """The failure mode with the worst blast radius if it is not guarded:
    without MERGE_HEAD, every stack-owned path in HEAD reads as "the upstream
    deleted it", and the restore loop would remove kb/CONTRACT.md,
    raw/CONTRACT.md and every template. A merge git refuses to start must stop
    before that, with the tree untouched."""
    upstream, repo = two_repos
    unrelated = tmp_path / "unrelated"
    unrelated.mkdir()
    _git(unrelated, "init", "-b", "main")
    _git(unrelated, "config", "user.name", "Unrelated")
    _git(unrelated, "config", "user.email", "unrelated@example.com")
    _write(unrelated, "somefile.md", "no shared history with the instance\n")
    _commit(unrelated, "unrelated root commit")

    _git(repo, "remote", "set-url", "upstream", str(unrelated))
    before = _git(repo, "rev-parse", "HEAD").stdout.strip()

    with pytest.raises(typer.Exit) as excinfo:
        _merge()
    assert excinfo.value.exit_code == 1

    assert _git(repo, "rev-parse", "HEAD").stdout.strip() == before
    assert (repo / "kb/CONTRACT.md").read_text(encoding="utf-8") == "stack kb contract v1\n"
    assert (repo / "raw/CONTRACT.md").exists()
    assert (repo / "kb/CONVENTIONS.md.template").exists()
    assert _git(repo, "status", "--porcelain").stdout == ""


def test_success_message_reports_what_changed_not_what_was_restored(two_repos, capsys):
    """Restoring every stack-owned path from MERGE_HEAD touches all of them
    whether or not the upstream moved any, so the report has to ask git what
    changed - otherwise a one-file update is announced as five."""
    upstream, repo = two_repos
    _write(upstream, "kb/CONTRACT.md", "stack kb contract v2\n")
    _commit(upstream, "upstream changes exactly one stack path")

    _merge()

    out = capsys.readouterr().out
    assert "Stack paths changed (1)" in out
    assert "kb/CONTRACT.md" in out
    assert "kb/CONVENTIONS.md.template" not in out


def test_dirty_working_tree_is_refused_untouched(two_repos):
    upstream, repo = two_repos
    before = _git(repo, "rev-parse", "HEAD").stdout.strip()
    (repo / "kb/CONTRACT.md").write_text("uncommitted local edit\n", encoding="utf-8")

    _write(upstream, "kb/CONTRACT.md", "stack kb contract v2\n")
    _commit(upstream, "upstream changes kb/CONTRACT.md")

    with pytest.raises(typer.Exit) as excinfo:
        _merge()
    assert excinfo.value.exit_code == 1

    assert _git(repo, "rev-parse", "HEAD").stdout.strip() == before
    assert (repo / "kb/CONTRACT.md").read_text(encoding="utf-8") == "uncommitted local edit\n"


def test_already_up_to_date_is_a_noop(two_repos):
    upstream, repo = two_repos
    before = _git(repo, "rev-parse", "HEAD").stdout.strip()

    _merge()  # nothing new upstream at all

    assert _git(repo, "rev-parse", "HEAD").stdout.strip() == before


def test_merge_warns_when_the_publish_remote_gate_is_unarmed(two_repos, capsys):
    upstream, repo = two_repos
    assert git_publish.read_allowed_push_urls() is None  # no .wikitool-remotes.json in this repo

    _write(upstream, "kb/CONTRACT.md", "stack kb contract v2\n")
    _commit(upstream, "upstream changes kb/CONTRACT.md")

    _merge()

    captured = capsys.readouterr()
    assert "WARN" in captured.out
    assert ".wikitool-remotes.json" in captured.out


def test_merge_stays_silent_when_the_publish_remote_gate_is_armed(two_repos, capsys):
    upstream, repo = two_repos
    (repo / config.PUBLISH_REMOTES_FILENAME).write_text(
        '{"schema": 1, "allowed_push_urls": ["ssh://example/test.git"]}\n', encoding="utf-8"
    )

    _write(upstream, "kb/CONTRACT.md", "stack kb contract v2\n")
    _commit(upstream, "upstream changes kb/CONTRACT.md")

    _merge()

    captured = capsys.readouterr()
    assert "WARN" not in captured.out


def test_dist_cmd_contract_only_stages_agree_with_ownership(two_repos):
    """Consistency guard for the ownership refactor: `dist_cmd`'s own list of
    stage-contract paths and `ownership.is_stack_owned` must not be able to
    name a different set of stages - both are sourced from
    `ownership.CONTENT_STAGES` now, so a stage added to one and not the other
    fails this rather than only surfacing in a real merge."""
    from chemenu.commands import dist_cmd

    assert dist_cmd.CONTRACT_ONLY_STAGES  # sanity: the derivation still yields entries
    for relative in dist_cmd.CONTRACT_ONLY_STAGES:
        assert ownership.is_stack_owned(relative)


# --- upstream verify --------------------------------------------------------


def test_verify_is_clean_on_a_stack_owned_only_change(two_repos):
    upstream, repo = two_repos
    since = _git(repo, "rev-parse", "HEAD").stdout.strip()

    _write(repo, "kb/CONTRACT.md", "stack kb contract v2\n")
    _commit(repo, "advance kb/CONTRACT.md")

    upstream_cmd.verify_command(since=since, until="HEAD")  # must not raise


def test_verify_fails_on_a_hand_botched_merge(two_repos, capsys):
    upstream, repo = two_repos
    since = _git(repo, "rev-parse", "HEAD").stdout.strip()

    _write(repo, "kb/SneakedIn.md", "content that arrived outside a stack-owned path\n")
    _commit(repo, "a hand-resolved merge that let content through")

    with pytest.raises(typer.Exit) as excinfo:
        upstream_cmd.verify_command(since=since, until="HEAD")
    assert excinfo.value.exit_code == 1

    captured = capsys.readouterr()
    assert "kb/SneakedIn.md" in captured.out


# --- ownership predicate, exercised directly ---------------------------------


@pytest.mark.parametrize(
    "relative,expected",
    [
        ("kb/CONTRACT.md", True),
        ("raw/CONTRACT.md", True),
        ("work/CONTRACT.md", True),
        ("reports/CONTRACT.md", True),
        ("kb/CONVENTIONS.md.template", True),
        ("kb/entities/COLLECTION.md.template", True),
        ("kb/GLOSSARY.md.template", True),
        ("kb/CONVENTIONS.md", False),
        ("kb/entities/COLLECTION.md", False),
        ("kb/concepts/Some Page.md", False),
        ("raw/notes/x.md", False),
        ("tools/CONTRACT.md", False),  # not a content stage
        ("kb/log.md", False),  # export stub, not stack-owned
    ],
)
def test_is_stack_owned(relative, expected):
    assert ownership.is_stack_owned(relative) == expected