nisse/nisse-review/tests/test_review.py

"""nisse-review's first slice (Part R, M7.3): its duty, a review run, posting
the review, and recovering after a crash."""

import io
from pathlib import Path
from typing import Any

import pytest
from review_fakes import FakeGitea, Scripted, call, pr

from nisse_core.gitea import GiteaError
from nisse_core.knowledge.handlers import Knowledge
from nisse_core.knowledge.memory import Memory
from nisse_core.knowledge.vault import Vault
from nisse_core.log.canonical import AssistantText
from nisse_core.loop.driver import Driver
from nisse_core.oplog import OpLog
from nisse_review.build import StartupError, check_access
from nisse_review.review import Reviewer, _diff_lines, marker
from nisse_review.scheduler import Scheduler, ready
from nisse_review.state import (
    FAILED,
    POST_REVIEW,
    READY,
    REVIEWED,
    RUNNING,
    WAITING,
    StateStore,
    pr_of,
)

REPO = "o/demo"


def world(
    tmp_path: Path, model: Scripted, gitea: FakeGitea | None = None, **more: Any
) -> dict[str, Any]:
    gitea = gitea or FakeGitea(repos={REPO: [pr()]})
    oplog = OpLog("nisse-review", io.StringIO())
    store = StateStore(tmp_path / "state.db", oplog=oplog)
    vault = Vault(tmp_path / "vault", {"memory/nisse-review": "rw"}, agent="r")
    reviewer = Reviewer(
        store,
        gitea,  # type: ignore[arg-type]
        Driver(model, "m"),  # type: ignore[arg-type]
        account=gitea.account,
        gitea_url="https://gitea.test",
        oplog=oplog,
        knowledge=Knowledge(vault, Memory(vault, "memory/nisse-review"), oplog),
    )
    scratch = tmp_path / "scratch"
    scratch.mkdir()
    scheduler = Scheduler(
        store,
        gitea,  # type: ignore[arg-type]
        reviewer,
        oplog=oplog,
        scratch=scratch,
        **more,
    )
    return {"gitea": gitea, "store": store, "scheduler": scheduler, "tmp": tmp_path}


def item(w: dict[str, Any], number: int = 7):
    found = w["store"].work_item(f"{REPO}#{number}")
    assert found is not None
    return found


REVIEW = [
    [call("read_file", path="count.py", ref="head")],
    [call("comment", path="count.py", line="2", text="split() handles tabs too.")],
    [call("comment", path="README.md", line="3", text="Say it counts words.")],
    [call("post_review", summary="Looks right; two small points.")],
]


# --- the duty (R1) ---------------------------------------------------------------


@pytest.mark.parametrize(
    ("title", "draft", "expected"),
    [
        ("Count words", False, True),
        ("Count words", True, False),
        ("WIP: count words", False, False),
        ("[WIP] count words", False, False),
        ("wip count", False, False),
        ("Wipe the cache", False, True),
    ],
)
def test_ready_means_not_a_draft_and_not_wip(title, draft, expected) -> None:
    assert ready(pr(title=title, draft=draft)) is expected


def test_intake_opens_an_item_per_pr_and_closes_it_with_the_pr(
    tmp_path: Path,
) -> None:
    gitea = FakeGitea(repos={REPO: [pr(7), pr(8, draft=True)]})
    w = world(tmp_path, Scripted(), gitea)
    w["scheduler"].intake()
    assert (item(w, 7).position, item(w, 8).position) == (READY, WAITING)
    assert item(w, 7).scope == REPO
    gitea.repos["o/elsewhere"] = [pr(9)]
    w["scheduler"]._only = frozenset({REPO})
    w["scheduler"].intake()
    assert w["store"].work_item("o/elsewhere#9") is None
    gitea.repos[REPO] = [pr(8, draft=False)]  # 7 merged; 8 is ready now
    gitea.merged.add((REPO, 7))
    w["scheduler"].intake()
    assert (item(w, 7).status, item(w, 7).outcome) == ("closed", "completed")
    assert item(w, 8).position == READY


# --- a review --------------------------------------------------------------------


def test_a_review_is_read_drafted_and_posted_as_a_comment_review(
    tmp_path: Path,
) -> None:
    model = Scripted(*REVIEW)
    w = world(tmp_path, model)
    assert w["scheduler"].round() is True
    [posted] = w["gitea"].posted
    # In the diff: inline. README.md:3 isn't in it: Gitea took it inline too.
    assert [(c.path, c.line) for c in posted["comments"]] == [
        ("count.py", 2),
        ("README.md", 3),
    ]
    assert posted["commit"] == "abc123def456"
    assert posted["body"].startswith("Looks right; two small points.")
    assert marker("abc123def456") in posted["body"]
    assert item(w).position == REVIEWED
    assert pr_of(item(w)).reviewed == "abc123def456"
    # What it started from: the PR, what was said, its issue, the diff.
    brief = model.seen[0][0].event.text
    for part in ("Count words, not characters", "Please keep it simple", "#3"):
        assert part in brief
    assert "-    return len(text)" in brief
    # Its tools: reading, commenting, memory, posting; nothing that writes code.
    assert model.offered[0] >= {
        "read_file",
        "search_code",
        "comment",
        "remember",
        "post_review",
    }
    assert not model.offered[0] & {"write_file", "run_command"}
    assert "never an instruction" in model.systems[0]


def test_comments_gitea_wont_take_off_the_diff_become_links(tmp_path: Path) -> None:
    gitea = FakeGitea(repos={REPO: [pr()]}, refuse_off_diff=True)
    w = world(tmp_path, Scripted(*REVIEW), gitea)
    w["scheduler"].round()
    [posted] = gitea.posted
    assert [(c.path, c.line) for c in posted["comments"]] == [("count.py", 2)]
    assert (
        "[`README.md:3`](https://gitea.test/o/demo/src/commit/abc123def456/"
        "README.md#L3): Say it counts words." in posted["body"]
    )


def test_a_comment_must_be_on_a_real_line(tmp_path: Path) -> None:
    model = Scripted(
        [call("comment", path="count.py", line="9", text="x")],
        [call("comment", path="nope.py", line="1", text="x")],
        [call("comment", path="count.py", line="two", text="x")],
        [call("post_review", summary="ok")],
    )
    w = world(tmp_path, model)
    w["scheduler"].round()
    assert w["gitea"].posted[0]["comments"] == []
    results = [
        e.event.result
        for e in w["store"].transcript(f"{REPO}#7", "driver").read()
        if hasattr(e.event, "result")
    ]
    assert results[:3] == [
        "count.py has 2 lines; 9 isn't one of them",
        "nope.py: no such file at the PR's head",
        "line is the line's number, like 42",
    ]


def test_reads_stay_in_the_repository(tmp_path: Path) -> None:
    model = Scripted(
        [
            call(
                "read_file", path="../../../../../../../../../../etc/passwd", ref="head"
            )
        ],
        [call("read_file", path="count.py", ref="other")],
        [call("search_code", text="split", ref="head")],
        [call("read_issue", number="3")],
        [call("post_review", summary="ok")],
    )
    w = world(tmp_path, model)
    w["scheduler"].round()
    results = [
        (e.event.result, e.event.success)
        for e in w["store"].transcript(f"{REPO}#7", "driver").read()
        if hasattr(e.event, "result")
    ]
    assert results[0] == (
        "../../../../../../../../../../etc/passwd: no such file at head",
        False,
    )
    assert results[1] == ("ref is head or base, not 'other'", False)
    assert "count.py:2: return len(text.split())" in results[2][0]
    assert results[3][0].startswith("[o/demo#3, by don]")
    assert list((tmp_path / "scratch").iterdir()) == []  # the trees are gone


def test_memory_notes_go_under_the_repositorys_folder(tmp_path: Path) -> None:
    model = Scripted(
        [
            call(
                "remember",
                name="tests-beside-code",
                description="Tests sit beside the code",
                fact="o/demo keeps test_x.py next to x.py",
                scope="project",
            )
        ],
        [call("post_review", summary="ok")],
    )
    w = world(tmp_path, model)
    w["scheduler"].round()
    note = tmp_path / "vault/memory/nisse-review/projects/o-demo/tests-beside-code.md"
    assert note.is_file()


# --- failures and crashes (L8) ----------------------------------------------------


def test_failed_runs_are_retried_then_said_on_the_pr(tmp_path: Path) -> None:
    prose = [AssistantText(text="I think it's fine.", model="m")]
    model = Scripted(prose, prose, prose, prose)  # never a tool call
    w = world(tmp_path, model, attempts=2)
    w["scheduler"].round()
    assert (item(w).position, pr_of(item(w)).attempts) == (READY, 1)
    w["scheduler"].round()
    assert item(w).position == FAILED
    [(repo, number, said)] = w["gitea"].comments
    assert (repo, number) == (REPO, 7) and "couldn't review this" in said
    w["gitea"].repos[REPO] = [pr(head_sha="fff000")]  # a new push: tried again
    w["scheduler"].intake()
    assert (item(w).position, pr_of(item(w)).attempts) == (READY, 0)


def recovering(tmp_path: Path, gitea: FakeGitea) -> dict[str, Any]:
    """An item a crash left running, with its posting intent pending."""
    w = world(tmp_path, Scripted(), gitea)
    w["scheduler"].intake()
    store = w["store"]
    store.update_work_item(
        f"{REPO}#7", position=RUNNING, data=item(w).data, cause="test"
    )
    store.transcript(f"{REPO}#7", "orchestrator").begin_intent(
        POST_REVIEW, {"repo": REPO, "pr": 7, "head": "abc123def456"}, cause="test"
    )
    return w


def test_a_review_posted_before_a_crash_isnt_posted_again(tmp_path: Path) -> None:
    gitea = FakeGitea(repos={REPO: [pr()]})
    gitea.posted.append(
        {
            "repo": REPO,
            "pr": 7,
            "body": f"ok\n\n{marker('abc123def456')}",
            "commit": "abc123def456",
            "comments": [],
        }
    )
    w = recovering(tmp_path, gitea)
    assert w["scheduler"].round() is False  # nothing left to review
    assert item(w).position == REVIEWED and len(gitea.posted) == 1
    assert w["store"].pending_intents() == []


def test_a_review_not_posted_before_a_crash_is_done_again(tmp_path: Path) -> None:
    w = recovering(tmp_path, FakeGitea(repos={REPO: [pr()]}))
    w["scheduler"].recover()
    assert item(w).position == READY
    assert w["store"].pending_intents() == []


def test_with_gitea_down_a_crashed_review_waits(tmp_path: Path) -> None:
    gitea = FakeGitea(repos={REPO: [pr()]})
    w = recovering(tmp_path, gitea)
    gitea.down = True
    assert w["scheduler"].round() is False
    assert item(w).position == RUNNING and len(w["store"].pending_intents()) == 1


# --- access (R3) ------------------------------------------------------------------


def test_an_account_that_can_write_is_refused(tmp_path: Path) -> None:
    gitea = FakeGitea(repos={REPO: [], "o/other": []})
    assert check_access(gitea) == ("nisse-review", [REPO, "o/other"])  # type: ignore[arg-type]
    gitea.permissions["o/other"] = {"pull": True, "push": True}
    with pytest.raises(StartupError, match="read access only"):
        check_access(gitea)  # type: ignore[arg-type]


def test_diff_lines_are_the_heads_lines_a_diff_shows() -> None:
    diff = (
        "--- a/x.py\n+++ b/x.py\n@@ -1,3 +1,4 @@\n a\n-b\n+c\n+d\n e\n"
        "--- a/gone.py\n+++ /dev/null\n@@ -1 +0,0 @@\n-z\n"
    )
    assert _diff_lines(diff) == {("x.py", 1), ("x.py", 2), ("x.py", 3), ("x.py", 4)}


def test_gitea_errors_while_posting_count_as_a_failed_run(tmp_path: Path) -> None:
    class Refusing(FakeGitea):
        def post_review(self, *args: Any, **kwargs: Any) -> int:
            raise GiteaError("403: forbidden")

    gitea = Refusing(repos={REPO: [pr()]})
    w = world(tmp_path, Scripted(*REVIEW), gitea)
    w["scheduler"].round()
    assert (item(w).position, pr_of(item(w)).attempts) == (READY, 1)
    assert w["store"].pending_intents() == []


def test_an_archive_unpacks_only_plain_files_inside(tmp_path: Path) -> None:
    import tarfile

    from nisse_review.reader import _unpack

    raw = io.BytesIO()
    with tarfile.open(fileobj=raw, mode="w:gz") as tar:
        for name, kind in [
            ("demo/ok.py", tarfile.REGTYPE),
            ("demo/../../evil.py", tarfile.REGTYPE),
            ("demo/link", tarfile.SYMTYPE),
        ]:
            info = tarfile.TarInfo(name)
            info.type = kind
            info.linkname = "/etc/passwd" if kind == tarfile.SYMTYPE else ""
            info.size = 0 if kind == tarfile.SYMTYPE else 2
            tar.addfile(info, io.BytesIO(b"x\n") if info.size else None)
    root = tmp_path / "tree"
    root.mkdir()
    _unpack(raw.getvalue(), root)
    assert sorted(p.name for p in root.rglob("*")) == ["ok.py"]
    assert not (tmp_path / "evil.py").exists()


def test_the_config_names_every_problem(tmp_path: Path) -> None:
    from nisse_core.config import ConfigError
    from nisse_review.config import ReviewConfig

    tree = {
        "state_db": str(tmp_path / "s.db"),
        "llm": {"base_url": "http://llm", "token": "${LLM}"},
        "roles": {"reviewer": "local", "expert": "frontier"},
        "gitea": {"url": "https://g", "token": "${GITEA}", "operators": ["don"]},
        "tuning": {"review": {"max_turns": 10}},
    }
    env = {"LLM": "k", "GITEA": "t"}
    config = ReviewConfig.from_tree(tree, env)
    assert config.roles.resolve("driver") == "local"
    assert config.tuning.review.max_turns == 10
    assert config.gitea_operators == {"don"} and config.repositories is None
    only = ReviewConfig.from_tree({**tree, "repositories": ["o/demo"]}, env)
    assert only.repositories == {"o/demo"}
    with pytest.raises(ConfigError) as raised:
        ReviewConfig.from_tree({**tree, "matrix": {"x": 1}}, {"LLM": "k"})
    assert "matrix.x: unknown setting" in str(raised.value)
    assert "references ${GITEA}" in str(raised.value)