chemenu/tools/chemenu/tests/test_raw_capture.py

"""`raw capture`, `raw status` and `raw accept --replaces-bundle` (Gitea #177),
against local test repositories - never the network.

The `file` transport those repositories need is unlocked by the `capture`
fixture widening `repo_capture.ALLOWED_SCHEMES`, the one switch the command
itself cannot reach: it sets `GIT_ALLOW_PROTOCOL` from that tuple on every git
call, so no option and no environment variable opens it.
"""
from __future__ import annotations

import datetime
import http.server
import json
import os
import subprocess
import threading
import time
from pathlib import Path

import pytest
import typer

from chemenu import config, repo_capture
from chemenu.commands.raw_cmd import (
    raw_accept_command,
    raw_capture_command,
    raw_pending_command,
    raw_status_command,
)
from chemenu.errors import BackendError, ValidationError
from chemenu.frontmatter_io import read_page, write_page
from chemenu.kb_scan import load_kb_pages
from chemenu.provenance import uncovered_raw_files

posix_only = pytest.mark.skipif(os.name != "posix", reason="needs a POSIX shell and symlinks")


def _git(cwd: Path, *args: str) -> str:
    return subprocess.run(
        ["git", "-c", "user.name=Fixture", "-c", "user.email=f@example.org", *args],
        cwd=cwd, check=True, capture_output=True, text=True,
    ).stdout.strip()


class Repo:
    """A work repository whose `file://` URL is what the commands are given."""

    def __init__(self, path: Path) -> None:
        self.path = path
        path.mkdir(parents=True)
        _git(path, "init", "-q", "-b", "main")

    @property
    def url(self) -> str:
        return self.path.as_uri()

    def write(self, rel: str, data) -> None:
        target = self.path / rel
        target.parent.mkdir(parents=True, exist_ok=True)
        if isinstance(data, str):
            data = data.encode("utf-8")
        target.write_bytes(data)

    def remove(self, rel: str) -> None:
        _git(self.path, "rm", "-q", rel)

    def commit(self, message: str = "change") -> str:
        _git(self.path, "add", "-A")
        _git(self.path, "commit", "-q", "--allow-empty", "-m", message)
        return _git(self.path, "rev-parse", "HEAD")

    def tag(self, name: str, annotated: bool = False) -> None:
        if annotated:
            _git(self.path, "tag", "-a", name, "-m", name)
        else:
            _git(self.path, "tag", name)


def _shard() -> str:
    today = datetime.date.today()
    return f"raw/{today.year:04d}/{today.month:02d}"


def _out(capsys) -> str:
    return " ".join(capsys.readouterr().out.split())


@pytest.fixture
def capture(kb_dir, monkeypatch):
    root = kb_dir.parent
    (root / "raw").mkdir(exist_ok=True)
    (root / "incoming").mkdir(exist_ok=True)
    monkeypatch.setattr(repo_capture, "ALLOWED_SCHEMES", ("ssh", "https", "file"))
    return root


@pytest.fixture
def repo(tmp_path):
    r = Repo(tmp_path / "upstream")
    r.write("README.md", "# Service\r\nCRLF line endings, kept as they are.\r\n")
    r.write("docs/a.md", "# A\n")
    r.write("docs/x/y/b.md", "# B\n")
    r.write("docs/config.yaml", "key: value\n")
    r.write("src/main.py", "print('hi')\n")
    r.commit("initial")
    return r


def _capture(url=None, ref="main", paths=("docs/**/*.md", "README.md"), name="svc",
             fidelity="verbatim", authority="normative", update=None):
    return raw_capture_command(
        repo_url=url, ref=None if update else ref, paths=None if update else list(paths),
        name=None if update else name, fidelity=fidelity, authority=authority,
        update=Path(update) if update else None,
    )


def _update(bundle, fidelity=None, authority=None):
    return raw_capture_command(
        repo_url=None, ref=None, paths=None, name=None, fidelity=fidelity, authority=authority,
        update=Path(bundle),
    )


def _accept(path, **kwargs):
    defaults = dict(fidelity=None, authority=None, page=None, replaces=None, dry_run=False)
    defaults.update(kwargs)
    return raw_accept_command(files=[Path(path)], **defaults)


def _tree(*dirs: Path) -> dict[str, bytes]:
    """Every file below `dirs`, with its bytes - what "nothing changed" is
    asserted against."""
    snapshot = {}
    for top in dirs:
        for path in sorted(top.rglob("*")):
            if path.is_file():
                snapshot[path.as_posix()] = path.read_bytes()
    return snapshot


def _write_source(kb_dir, title, raw_files, fidelity="verbatim", authority="normative"):
    write_page(
        kb_dir / "sources" / f"{title}.md",
        {
            "type": "types/source.md", "source_type": "document", "author": "Fixture",
            "raw_files": list(raw_files), "date": "2026-10-01", "tags": [], "entities": [],
            "concepts": [], "summary": "Test source.", "fidelity": fidelity, "authority": authority,
        },
        f"\n# {title}\n\n## Summary\n\nTest.\n",
    )


def _captured_and_accepted(capture, repo, **kwargs) -> Path:
    _capture(repo.url, **kwargs)
    _accept(capture / "incoming" / kwargs.get("name", "svc"))
    return capture / _shard() / kwargs.get("name", "svc")


# --- globs --------------------------------------------------------------------


GLOB_FILES = (
    "README.md", "docs/a.md", "docs/x/b.md", "docs/x/y/c.md", "docs/x/y/c.yaml", "docs/.h.md",
    ".github/w.md", "docx/n.md", "docs/x[1].md",
)
GLOB_PATTERNS = (
    "docs/*.md", "docs/**/*.md", "docs", "docs/", "docs/x", "**/*.md", "docs/**", "*.md", "do*",
    "docs/x*", "docs/**b.md", "*", "**", "docs/?.md", "docs/[ab].md", "docs/[!a].md", "**/c.*",
    "docs/x/**/c.md",
)


def test_glob_semantics_are_gits_own(tmp_path):
    """Every pattern selects exactly what `git ls-files ':(glob)<p>'` selects -
    the reference the issue names, asked of git itself rather than restated."""
    work = tmp_path / "globs"
    work.mkdir()
    _git(work, "init", "-q", "-b", "main")
    for rel in GLOB_FILES:
        (work / rel).parent.mkdir(parents=True, exist_ok=True)
        (work / rel).write_text("x\n", encoding="utf-8")
    _git(work, "add", "-A")
    for pattern in GLOB_PATTERNS:
        expected = set(_git(work, "ls-files", "--", f":(glob){pattern}").splitlines())
        actual = {rel for rel in GLOB_FILES if repo_capture.glob_matches(pattern, rel)}
        assert actual == expected, pattern


def test_named_glob_cases():
    assert repo_capture.glob_matches("docs/**/*.md", "docs/a.md")
    assert repo_capture.glob_matches("docs/**/*.md", "docs/x/y/b.md")
    assert not repo_capture.glob_matches("docs/*.md", "docs/x/b.md")
    assert not repo_capture.glob_matches("docs/**/*.md", "docs/config.yaml")


@pytest.mark.parametrize("pattern", ["", "/docs/**", "docs/../x", "docs//a.md", "./docs"])
def test_unusable_globs_are_refused(pattern):
    with pytest.raises(ValidationError):
        repo_capture.check_glob(pattern)


# --- capture + accept ---------------------------------------------------------


def test_capture_and_accept_hold_exactly_the_matching_files_byte_for_byte(capture, repo, monkeypatch, tmp_path):
    # The host's git would convert line endings on a checkout; a blob read must not.
    gitconfig = tmp_path / "gitconfig"
    gitconfig.write_text("[core]\n\tautocrlf = true\n", encoding="utf-8")
    monkeypatch.setenv("GIT_CONFIG_GLOBAL", str(gitconfig))

    _capture(repo.url)
    incoming = capture / "incoming" / "svc"
    manifest = json.loads((incoming / "_capture.json").read_text(encoding="utf-8"))
    assert manifest["schema"] == 1
    assert manifest["repo"] == repo.url
    assert manifest["ref"] == "main"
    assert manifest["commit"] == _git(repo.path, "rev-parse", "HEAD")
    assert manifest["paths"] == ["docs/**/*.md", "README.md"]
    assert manifest["fidelity"] == "verbatim" and manifest["authority"] == "normative"
    assert manifest["files"] == ["README.md", "docs/a.md", "docs/x/y/b.md"]
    assert manifest["captured"].endswith("Z")

    _accept(incoming)
    bundle = capture / _shard() / "svc"
    on_disk = sorted(p.relative_to(bundle).as_posix() for p in bundle.rglob("*") if p.is_file())
    assert on_disk == ["README.md", "_capture.json", "docs/a.md", "docs/x/y/b.md"]
    for rel in ("README.md", "docs/a.md", "docs/x/y/b.md"):
        assert (bundle / rel).read_bytes() == (repo.path / rel).read_bytes()
    assert (bundle / "README.md").read_bytes().count(b"\r\n") == 2
    assert not incoming.exists()


def test_accepting_a_captured_folder_takes_the_capture_fields_from_its_manifest(capture, repo, capsys):
    _capture(repo.url, fidelity="published", authority="reporting")
    _accept(capture / "incoming" / "svc")
    out = _out(capsys)
    assert "--set fidelity=published --set authority=reporting" in out
    assert "_capture.json" not in out.split("--set raw_files=")[1].split()[0]


def test_accepting_a_captured_folder_with_capture_flags_is_refused(capture, repo):
    _capture(repo.url)
    before = _tree(capture / "incoming", capture / "raw")
    with pytest.raises(typer.Exit):
        _accept(capture / "incoming" / "svc", fidelity="verbatim", authority="normative")
    assert _tree(capture / "incoming", capture / "raw") == before


def test_a_captured_folder_edited_after_capture_is_refused(capture, repo, capsys):
    _capture(repo.url)
    (capture / "incoming" / "svc" / "docs" / "extra.md").write_text("added by hand\n", encoding="utf-8")
    with pytest.raises(typer.Exit):
        _accept(capture / "incoming" / "svc")
    assert "no longer matches" in _out(capsys)


def test_pending_judges_a_captured_folder_by_its_manifest(capture, repo, capsys):
    _capture(repo.url)
    capsys.readouterr()
    raw_pending_command(json_out=True)
    [candidate] = json.loads(capsys.readouterr().out)
    assert candidate["kind"] == "folder" and candidate["acceptable"] is True


def test_capture_refuses_an_existing_incoming_folder(capture, repo):
    (capture / "incoming" / "svc").mkdir()
    with pytest.raises(typer.Exit):
        _capture(repo.url)
    assert list((capture / "incoming" / "svc").iterdir()) == []


def test_capture_refuses_a_name_a_captured_bundle_holds_and_names_update(capture, repo, capsys):
    _captured_and_accepted(capture, repo)
    capsys.readouterr()
    with pytest.raises(typer.Exit):
        _capture(repo.url)
    assert "raw capture --update" in _out(capsys)
    assert not (capture / "incoming" / "svc").exists()


def test_capture_refuses_when_nothing_matches(capture, repo):
    with pytest.raises(typer.Exit):
        _capture(repo.url, paths=("nothing/**",))
    assert list((capture / "incoming").iterdir()) == []


def test_capture_refuses_a_path_over_the_budget_before_writing(capture, tmp_path, capsys):
    r = Repo(tmp_path / "deep")
    deep = "/".join(["d" * 30] * 6) + "/page.md"
    r.write(deep, "# deep\n")
    r.commit()
    with pytest.raises(typer.Exit):
        _capture(r.url, paths=("**",))
    assert "path budget" in _out(capsys)
    assert list((capture / "incoming").iterdir()) == []


def test_capture_follows_the_newest_tag_by_version_order(capture, repo):
    repo.tag("v1.9")
    repo.write("docs/a.md", "# A at 1.10\n")
    v110 = repo.commit()
    repo.tag("v1.10", annotated=True)
    repo.write("docs/a.md", "# A on main, untagged\n")
    repo.commit()
    _capture(repo.url, ref="v*")
    incoming = capture / "incoming" / "svc"
    manifest = json.loads((incoming / "_capture.json").read_text(encoding="utf-8"))
    assert manifest["commit"] == v110  # the peeled commit, not the tag object
    assert (incoming / "docs/a.md").read_text(encoding="utf-8") == "# A at 1.10\n"


def test_capture_refuses_a_branch_that_does_not_exist(capture, repo):
    with pytest.raises(typer.Exit):
        _capture(repo.url, ref="nope")
    assert list((capture / "incoming").iterdir()) == []


# --- exclusions ---------------------------------------------------------------


@posix_only
def test_excluded_paths_are_never_captured_and_each_is_named(capture, repo, capsys):
    repo.write("docs/guideline.md", "<!-- wikitool:export from=kb -->\n# Exported\n")
    repo.write(".github/workflow.md", "# CI\n")
    repo.write("docs/big.bin", "version https://git-lfs.github.com/spec/v1\noid sha256:abc\nsize 9\n")
    repo.write("docs/sub/_capture.json", "{}\n")
    os.symlink("a.md", repo.path / "docs" / "link.md")
    sub = repo.commit()
    _git(repo.path, "update-index", "--add", "--cacheinfo", f"160000,{sub},vendor/lib")
    _git(repo.path, "commit", "-q", "-m", "submodule")

    _capture(repo.url, paths=("**",))
    out = _out(capsys)
    for path, reason in (
        ("docs/guideline.md", "guideline export"),
        (".github/workflow.md", "hidden path segment"),
        ("docs/big.bin", "Git LFS pointer"),
        ("docs/sub/_capture.json", "reserved name"),
        ("docs/link.md", "symlink"),
        ("vendor/lib", "submodule"),
    ):
        assert f"excluded {path} ({reason}" in out
    files = json.loads((capture / "incoming" / "svc" / "_capture.json").read_text(encoding="utf-8"))["files"]
    assert files == ["README.md", "docs/a.md", "docs/config.yaml", "docs/x/y/b.md", "src/main.py"]

    _accept(capture / "incoming" / "svc")
    repo.write("docs/a.md", "# A, changed\n")
    repo.commit()
    capsys.readouterr()
    raw_status_command(json_out=True)
    [row] = json.loads(capsys.readouterr().out)
    assert row["files"] == [{"path": f"{_shard()}/svc/docs/a.md", "status": "M"}]


def test_a_file_over_the_size_limit_is_excluded(capture, repo, monkeypatch, capsys):
    from chemenu import web_capture

    monkeypatch.setattr(web_capture, "MAX_BYTES", 5)
    _capture(repo.url, paths=("docs/x/**",))
    assert "excluded docs/x/y/b.md (over" not in _out(capsys)  # 4 bytes: under
    with pytest.raises(typer.Exit):
        _capture(repo.url, paths=("README.md",), name="svc2")
    assert "excluded README.md (over" in _out(capsys)


# --- URLs and credentials -----------------------------------------------------


@pytest.mark.parametrize("url", [
    "ext::sh -c touch% /tmp/x",
    "fd::17",
    "file:///srv/repo.git",
    "/srv/repo.git",
    "http://example.org/repo.git",
    "https://user:token@example.org/repo.git",
    "ssh://user:secret@example.org/repo.git",
    "-uhelp@example.org:x",
])
def test_refused_urls(url):
    with pytest.raises(ValidationError):
        repo_capture.check_repo_url(url)


@pytest.mark.parametrize("url", [
    "https://example.org/team/repo.git",
    "https://user@example.org/team/repo.git",
    "ssh://git@example.org:2222/team/repo.git",
    "git@example.org:team/repo.git",
])
def test_accepted_urls(url):
    repo_capture.check_repo_url(url)


def _ext_url(marker: Path) -> str:
    return f"ext::sh -c touch% {marker}"


@posix_only
def test_capture_refuses_an_ext_url_and_runs_nothing(capture, tmp_path):
    marker = tmp_path / "ext-ran"
    with pytest.raises(typer.Exit):
        _capture(_ext_url(marker))
    assert not marker.exists()


@posix_only
def test_git_itself_refuses_the_ext_transport(capture, tmp_path):
    """Below the URL check: the git layer alone, handed the URL directly."""
    marker = tmp_path / "ext-ran"
    with pytest.raises(BackendError):
        with repo_capture.cache_repo("ext-probe") as cache:
            repo_capture.resolve_ref(_ext_url(marker), "main", cache)
    assert not marker.exists()


@posix_only
def test_the_ext_probe_would_run_if_git_allowed_it(capture, tmp_path, monkeypatch):
    """The control for the two tests above: the same URL does run its command
    once `ext` is allowed, so their empty marker means something."""
    marker = tmp_path / "ext-ran"
    monkeypatch.setattr(repo_capture, "ALLOWED_SCHEMES", ("ext",))
    with pytest.raises(BackendError):
        with repo_capture.cache_repo("ext-probe") as cache:
            repo_capture.resolve_ref(_ext_url(marker), "main", cache)
    assert marker.exists()


def test_a_refused_url_leaves_no_credential_anywhere(capture, tmp_path):
    with pytest.raises(typer.Exit):
        _capture("https://user:s3cr3t-token@example.org/repo.git")
    for path in tmp_path.rglob("*"):
        if path.is_file():
            assert b"s3cr3t-token" not in path.read_bytes(), path


def _plant_bundle(capture, name, url, commit="0" * 40, files=("README.md",)):
    """A captured bundle under raw/ as an earlier accept would have left it."""
    bundle = capture / _shard() / name
    bundle.mkdir(parents=True)
    for rel in files:
        (bundle / rel).write_text("planted\n", encoding="utf-8")
    manifest = {
        "schema": 1, "repo": url, "ref": "main", "commit": commit, "paths": ["README.md"],
        "captured": "2026-10-01T00:00:00Z", "fidelity": "verbatim", "authority": "normative",
        "files": list(files),
    }
    (bundle / "_capture.json").write_text(json.dumps(manifest), encoding="utf-8")
    return bundle


@posix_only
def test_update_and_status_refuse_an_ext_url_read_from_a_manifest(capture, tmp_path, capsys):
    marker = tmp_path / "ext-ran"
    bundle = _plant_bundle(capture, "evil", _ext_url(marker))
    with pytest.raises(typer.Exit):
        _update(bundle)
    capsys.readouterr()
    raw_status_command(json_out=True)
    [row] = json.loads(capsys.readouterr().out)
    assert "refused" in row["error"]
    assert not marker.exists()


@pytest.fixture
def asks_for_credentials(monkeypatch):
    """An HTTP server that answers every request with a Basic-auth challenge -
    the repository that would make an unguarded git prompt for a password."""

    class Handler(http.server.BaseHTTPRequestHandler):
        def do_GET(self):  # noqa: N802 - http.server's name
            self.send_response(401)
            self.send_header("WWW-Authenticate", 'Basic realm="repo"')
            self.send_header("Content-Length", "0")
            self.end_headers()

        def log_message(self, *args):
            pass

    server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    monkeypatch.setattr(repo_capture, "ALLOWED_SCHEMES", ("ssh", "https", "file", "http"))
    yield f"http://127.0.0.1:{server.server_address[1]}/repo.git"
    server.shutdown()


def test_a_repository_asking_for_credentials_is_unreachable_not_a_hang(capture, asks_for_credentials, capsys):
    _plant_bundle(capture, "locked", asks_for_credentials)
    started = time.monotonic()
    raw_status_command(json_out=True)
    [row] = json.loads(capsys.readouterr().out)
    assert row["error"].startswith("not reachable")
    with pytest.raises(typer.Exit):
        _capture(asks_for_credentials, name="locked2")
    assert time.monotonic() - started < 30


# --- raw status ---------------------------------------------------------------


def test_status_reports_changed_docs_and_goes_quiet_after_the_replacement(capture, repo, capsys):
    bundle = _captured_and_accepted(capture, repo)
    repo.write("docs/a.md", "# A, second edition\n")
    repo.remove("docs/x/y/b.md")
    repo.write("docs/new.md", "# New\n")
    repo.commit()
    capsys.readouterr()

    raw_status_command(json_out=True)
    [row] = json.loads(capsys.readouterr().out)
    prefix = f"{_shard()}/svc"
    assert row["changed"] is True and row["error"] is None
    assert row["new"] == _git(repo.path, "rev-parse", "HEAD")
    assert row["files"] == [
        {"path": f"{prefix}/docs/a.md", "status": "M"},
        {"path": f"{prefix}/docs/new.md", "status": "A"},
        {"path": f"{prefix}/docs/x/y/b.md", "status": "D"},
    ]

    raw_status_command(json_out=False)
    out = _out(capsys)
    assert f"raw capture --update {prefix}" in out
    assert f"--replaces-bundle {prefix}" in out

    _update(bundle)
    _accept(capture / "incoming" / "svc", replaces_bundle=bundle)
    capsys.readouterr()
    raw_status_command(json_out=True)
    [row] = json.loads(capsys.readouterr().out)
    assert row["changed"] is False and row["files"] == []


def test_status_ignores_a_change_outside_the_globs(capture, repo, capsys):
    _captured_and_accepted(capture, repo)
    repo.write("src/main.py", "print('changed')\n")
    repo.commit()
    capsys.readouterr()
    raw_status_command(json_out=True)
    [row] = json.loads(capsys.readouterr().out)
    assert row["old"] != row["new"] and row["changed"] is False
    raw_status_command(json_out=False)
    out = _out(capsys)
    assert "1 unchanged" in out and "svc" not in out.split("raw/:")[1].replace("1 unchanged.", "")


def test_status_follows_new_tags_only_not_new_commits_on_main(capture, repo, capsys):
    repo.tag("v1.0")
    _captured_and_accepted(capture, repo, ref="v*")
    repo.write("docs/a.md", "# A on main\n")
    repo.commit()
    capsys.readouterr()
    raw_status_command(json_out=True)
    assert json.loads(capsys.readouterr().out)[0]["changed"] is False
    repo.tag("v1.1")
    raw_status_command(json_out=True)
    [row] = json.loads(capsys.readouterr().out)
    assert row["changed"] is True
    assert row["files"] == [{"path": f"{_shard()}/svc/docs/a.md", "status": "M"}]


def test_an_unreachable_repository_is_one_line_beside_the_others(capture, repo, tmp_path, capsys):
    _captured_and_accepted(capture, repo)
    _plant_bundle(capture, "gone", (tmp_path / "does-not-exist").as_uri())
    repo.write("docs/a.md", "# A, changed\n")
    repo.commit()
    capsys.readouterr()
    raw_status_command(json_out=True)
    rows = {Path(r["bundle"]).name: r for r in json.loads(capsys.readouterr().out)}
    assert rows["gone"]["error"].startswith("not reachable")
    assert rows["svc"]["changed"] is True and rows["svc"]["error"] is None
    raw_status_command(json_out=False)  # exits normally, both on their own line
    out = capsys.readouterr().out
    assert "not reachable" in out and "docs/a.md" in out


def test_status_with_no_captured_bundle(capture, capsys):
    raw_status_command(json_out=False)
    assert "No captured bundle" in capsys.readouterr().out


# --- raw accept --replaces-bundle ---------------------------------------------


def test_replaces_bundle_leaves_exactly_the_new_edition_in_place(capture, repo, kb_dir, capsys):
    bundle = _captured_and_accepted(capture, repo)
    prefix = f"{_shard()}/svc"
    owned = [f"{prefix}/README.md", f"{prefix}/docs/a.md", f"{prefix}/docs/x/y/b.md"]
    _write_source(kb_dir, "Source - Svc", owned)

    repo.write("docs/a.md", "# A, second edition\n")
    repo.remove("docs/x/y/b.md")
    repo.write("docs/new.md", "# New\n")
    new_commit = repo.commit()
    _update(bundle)
    capsys.readouterr()
    _accept(capture / "incoming" / "svc", replaces_bundle=bundle)
    out = _out(capsys)

    on_disk = sorted(p.relative_to(bundle).as_posix() for p in bundle.rglob("*") if p.is_file())
    assert on_disk == ["README.md", "_capture.json", "docs/a.md", "docs/new.md"]
    for rel in ("README.md", "docs/a.md", "docs/new.md"):
        assert (bundle / rel).read_bytes() == (repo.path / rel).read_bytes()
    assert not (bundle / "docs" / "x").exists()
    assert json.loads((bundle / "_capture.json").read_text(encoding="utf-8"))["commit"] == new_commit
    assert not (capture / "incoming" / "svc").exists()
    assert read_page(kb_dir / "sources" / "Source - Svc.md")[0]["raw_files"] == owned

    assert f"M {prefix}/docs/a.md" in out and f"D {prefix}/docs/x/y/b.md" in out
    assert f"A {prefix}/docs/new.md" in out
    assert f'touch --page "Source - Svc" --remove raw_files={prefix}/docs/x/y/b.md' in out
    assert f'touch --page "Source - Svc" --add raw_files={prefix}/docs/new.md' in out


def test_replaces_bundle_overwrites_changed_capture_fields_on_the_owning_page(capture, repo, kb_dir):
    bundle = _captured_and_accepted(capture, repo)
    _write_source(kb_dir, "Source - Svc", [f"{_shard()}/svc/README.md"])
    _update(bundle, authority="reporting")
    _accept(capture / "incoming" / "svc", replaces_bundle=bundle)
    assert read_page(kb_dir / "sources" / "Source - Svc.md")[0]["authority"] == "reporting"


def test_replaces_bundle_refuses_another_repository(capture, repo, tmp_path):
    bundle = _captured_and_accepted(capture, repo)
    other = Repo(tmp_path / "other")
    other.write("README.md", "# Other\n")
    other.commit()
    _capture(other.url, paths=("README.md",), name="other")
    (capture / "incoming" / "other").rename(capture / "incoming" / "svc")
    before = _tree(capture / "incoming", capture / "raw")
    with pytest.raises(typer.Exit):
        _accept(capture / "incoming" / "svc", replaces_bundle=bundle)
    assert _tree(capture / "incoming", capture / "raw") == before


def test_replaces_bundle_refuses_a_different_folder_name(capture, repo):
    bundle = _captured_and_accepted(capture, repo)
    _update(bundle)
    (capture / "incoming" / "svc").rename(capture / "incoming" / "svc-renamed")
    before = _tree(capture / "incoming", capture / "raw")
    with pytest.raises(typer.Exit):
        _accept(capture / "incoming" / "svc-renamed", replaces_bundle=bundle)
    assert _tree(capture / "incoming", capture / "raw") == before


def test_replaces_bundle_refuses_a_bundle_without_a_manifest(capture, repo):
    plain = capture / _shard() / "svc"
    plain.mkdir(parents=True)
    (plain / "README.md").write_text("hand-made folder bundle\n", encoding="utf-8")
    _capture(repo.url, name="svc-new")
    (capture / "incoming" / "svc-new").rename(capture / "incoming" / "svc")
    before = _tree(capture / "incoming", capture / "raw")
    with pytest.raises(typer.Exit):
        _accept(capture / "incoming" / "svc", replaces_bundle=plain)
    assert _tree(capture / "incoming", capture / "raw") == before


def test_replaces_bundle_refuses_page_replaces_and_capture_flags(capture, repo):
    bundle = _captured_and_accepted(capture, repo)
    _update(bundle)
    before = _tree(capture / "incoming", capture / "raw")
    for extra in ({"page": "Source - Svc"}, {"replaces": bundle / "README.md"}, {"fidelity": "verbatim"}):
        with pytest.raises(typer.Exit):
            _accept(capture / "incoming" / "svc", replaces_bundle=bundle, **extra)
    assert _tree(capture / "incoming", capture / "raw") == before


def test_replaces_bundle_refuses_a_file_with_two_owners(capture, repo, kb_dir):
    bundle = _captured_and_accepted(capture, repo)
    readme = f"{_shard()}/svc/README.md"
    _write_source(kb_dir, "Source - One", [readme])
    _write_source(kb_dir, "Source - Two", [readme])
    _update(bundle)
    before = _tree(capture / "incoming", capture / "raw")
    with pytest.raises(typer.Exit):
        _accept(capture / "incoming" / "svc", replaces_bundle=bundle)
    assert _tree(capture / "incoming", capture / "raw") == before


# --- --replaces / --page inside a captured bundle -----------------------------


def test_replaces_refuses_a_target_inside_a_captured_bundle(capture, repo, capsys):
    bundle = _captured_and_accepted(capture, repo)
    (capture / "incoming" / "b.md").write_text("# B by hand\n", encoding="utf-8")
    before = _tree(capture / "incoming", capture / "raw")
    with pytest.raises(typer.Exit):
        _accept(capture / "incoming" / "b.md", replaces=bundle / "docs" / "x" / "y" / "b.md")
    assert _tree(capture / "incoming", capture / "raw") == before
    assert "--replaces-bundle" in _out(capsys)


def test_page_refuses_to_grow_a_captured_bundle(capture, tmp_path, kb_dir):
    flat = Repo(tmp_path / "flat")
    flat.write("README.md", "# R\n")
    flat.write("AGENTS.md", "# A\n")
    flat.commit()
    _capture(flat.url, paths=("*.md",), name="flat")
    _accept(capture / "incoming" / "flat")
    prefix = f"{_shard()}/flat"
    _write_source(kb_dir, "Source - Flat", [f"{prefix}/AGENTS.md", f"{prefix}/README.md"])
    (capture / "incoming" / "notes.md").write_text("# notes\n", encoding="utf-8")
    before = _tree(capture / "incoming", capture / "raw")
    with pytest.raises(typer.Exit):
        _accept(capture / "incoming" / "notes.md", fidelity="verbatim", authority="normative",
                page="Source - Flat")
    assert _tree(capture / "incoming", capture / "raw") == before


def test_a_loose_file_named_like_the_manifest_is_refused(capture):
    (capture / "incoming" / "_capture.json").write_text("{}\n", encoding="utf-8")
    with pytest.raises(typer.Exit):
        _accept(capture / "incoming" / "_capture.json", fidelity="verbatim", authority="normative")


# --- coverage -----------------------------------------------------------------


def test_coverage_skips_the_manifest_and_sees_a_bundled_contract(capture, tmp_path):
    r = Repo(tmp_path / "with-contract")
    r.write("README.md", "# R\n")
    r.write("raw/CONTRACT.md", "# a repository's own stage contract\n")
    r.commit()
    _capture(r.url, paths=("README.md", "raw/CONTRACT.md"), name="wc")
    _accept(capture / "incoming" / "wc")
    (capture / "raw" / "CONTRACT.md").write_text("# the stage contract\n", encoding="utf-8")

    uncovered = uncovered_raw_files(config.RAW_DIR, load_kb_pages(config.KB_DIR))
    prefix = f"{_shard()}/wc"
    assert f"{prefix}/raw/CONTRACT.md" in uncovered
    assert f"{prefix}/README.md" in uncovered
    assert f"{prefix}/_capture.json" not in uncovered
    assert "raw/CONTRACT.md" not in uncovered