repo-manager/tests/test_record_command.py

"""Intake, decision, dependency, and human-flag receiving-surface coverage."""

from __future__ import annotations

import subprocess
from pathlib import Path

from repo_manager.commands.record import mutate_record
from repo_manager.observe import observe_repository


def _git(repo: Path, *args: str) -> None:
    subprocess.run(["git", *args], cwd=repo, check=True, capture_output=True)


def _fixture(tmp_path: Path) -> Path:
    repo = tmp_path / "pilot"
    repo.mkdir()
    _git(repo, "init")
    _git(repo, "config", "user.email", "test@example.com")
    _git(repo, "config", "user.name", "Test")
    (repo / ".repo-classification.yaml").write_text(
        "repo_classification:\n  category: tooling\n  domain: infotech\n",
        encoding="utf-8",
    )
    (repo / "workplans").mkdir()
    (repo / "workplans" / "P-WP-0001.md").write_text(
        """---
id: P-WP-0001
type: workplan
title: Pilot
status: active
depends_on: [P-WP-0000]
---

## Blocked task

```task
id: P-WP-0001-T01
status: wait
needs_human: true
intervention_note: Choose the API boundary.
blocking_reason: Awaiting decision.
depends_on: [P-WP-0000-T01]
```
""",
        encoding="utf-8",
    )
    _git(repo, "add", ".")
    _git(repo, "commit", "-m", "seed")
    return repo


def test_intake_lifecycle_is_file_backed_and_indexed(tmp_path: Path, monkeypatch) -> None:
    repo = _fixture(tmp_path)
    monkeypatch.setenv("RM_IDEMPOTENCY_PATH", str(tmp_path / "idempotency.json"))

    created = mutate_record(
        repo,
        "intake",
        "P-IN-0001",
        operation="create",
        title="Residual follow-up",
        data={"origin": "residual", "origin_ref": "P-WP-0000"},
        idempotency_key="intake-create",
    )
    replay = mutate_record(
        repo,
        "intake",
        "P-IN-0001",
        operation="create",
        title="Residual follow-up",
        data={"origin": "residual", "origin_ref": "P-WP-0000"},
        idempotency_key="intake-create",
    )
    assert created.status == replay.status == "applied"
    assert replay.evidence["git_sha"] == created.evidence["git_sha"]
    assert mutate_record(repo, "intake", "P-IN-0001", operation="route", route_to="P-WP-0002").status == "applied"
    assert mutate_record(repo, "intake", "P-IN-0001", operation="note", note="Owner accepted.").status == "applied"
    assert mutate_record(repo, "intake", "P-IN-0001", operation="close", outcome="promoted").status == "applied"

    snapshot, index = observe_repository(repo)
    record = next(item for item in index.work_records if item.kind == "intake")
    assert snapshot["index"]["intake_count"] == 1
    assert record.status == "closed"
    assert record.extra["record"]["routed_to"] == "P-WP-0002"
    assert record.extra["record"]["notes"][0]["content"] == "Owner accepted."


def test_decision_resolve_keeps_safe_guarantee(tmp_path: Path) -> None:
    repo = _fixture(tmp_path)
    created = mutate_record(
        repo,
        "decision",
        "P-DEC-0001",
        operation="create",
        title="Place the projection",
        data={"description": "Choose an aggregation owner."},
    )
    assert created.status == "applied"
    incomplete = mutate_record(repo, "decision", "P-DEC-0001", operation="resolve", rationale="Boundary")
    assert incomplete.status == "rejected"
    resolved = mutate_record(
        repo,
        "decision",
        "P-DEC-0001",
        operation="resolve",
        rationale="Repository authority stays local.",
        decided_by="operator",
    )
    assert resolved.status == "applied"
    snapshot, index = observe_repository(repo)
    record = next(item for item in index.work_records if item.kind == "decision")
    assert snapshot["index"]["decision_count"] == 1
    assert record.status == "resolved"
    assert record.extra["record"]["decided_by"] == "operator"


def test_dependency_and_human_flags_survive_indexing(tmp_path: Path) -> None:
    repo = _fixture(tmp_path)
    _snapshot, index = observe_repository(repo)
    workplan = next(item for item in index.work_records if item.kind == "workplan")
    task = next(item for item in index.work_records if item.kind == "task")
    assert workplan.extra["depends_on"] == ["P-WP-0000"]
    assert task.extra == {
        "depends_on": ["P-WP-0000-T01"],
        "needs_human": True,
        "intervention_note": "Choose the API boundary.",
        "blocking_reason": "Awaiting decision.",
    }