repo-manager/tests/test_workplan_command.py

"""Governed workplan mutation coverage for RMGR-WP-0008-T01."""

from __future__ import annotations

import subprocess
from pathlib import Path

from repo_manager.commands.workplan import archive_workplan, create_workplan, update_workplan
from repo_manager.gitops import head_sha
from repo_manager.index_store import default_index_path, load_index
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 / "workplans").mkdir()
    (repo / ".repo-classification.yaml").write_text(
        "repo_classification:\n  category: tooling\n  domain: infotech\n",
        encoding="utf-8",
    )
    (repo / "workplans" / "P-WP-0001-first.md").write_text(
        """---
id: P-WP-0001
type: workplan
title: "First"
domain: infotech
repo: pilot
status: proposed
owner: codex
topic_slug: infotech
created: "2026-08-21"
updated: "2026-08-21"
state_hub_workstream_id: "11111111-1111-4111-8111-111111111111"
---

# First

## Task

```task
id: P-WP-0001-T01
status: todo
priority: high
```

Do it.
""",
        encoding="utf-8",
    )
    _git(repo, "add", ".")
    _git(repo, "commit", "-m", "seed")
    return repo


def test_create_workplan_commits_indexes_and_meters(tmp_path: Path, monkeypatch) -> None:
    repo = _fixture(tmp_path)
    meter = tmp_path / "meter.jsonl"
    monkeypatch.setenv("RM_METER_PATH", str(meter))

    result = create_workplan(
        repo,
        "P-WP-0002",
        "Second plan",
        "Deliver the second governed slice.",
        status="ready",
        correlation_id="00000000-0000-4000-8000-000000000002",
        repo_slug="pilot",
    )

    assert result.status == "applied"
    assert result.evidence["git_sha"] == head_sha(repo)
    path = repo / "workplans" / "P-WP-0002-second-plan.md"
    assert path.is_file()
    text = path.read_text(encoding="utf-8")
    assert "status: ready" in text
    assert "Deliver the second governed slice." in text
    _snapshot, index = observe_repository(repo, slug="pilot")
    assert next(r for r in index.work_records if r.id == "P-WP-0002").status == "ready"
    stored = load_index(default_index_path(repo))
    event = next(e for e in stored.events if e.get("type") == "repo.command.applied")
    assert event["command"] == "repo.work.create_workplan"
    assert '"kind": "workplan_create"' in meter.read_text(encoding="utf-8")


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

    rejected = update_workplan(
        repo,
        "11111111-1111-4111-8111-111111111111",
        status="active",
        expected_head_sha="0" * 40,
    )
    assert rejected.status == "rejected"
    assert rejected.error and rejected.error["code"] == "precondition_failed"

    applied = update_workplan(
        repo,
        "11111111-1111-4111-8111-111111111111",
        title="Renamed",
        status="active",
        expected_head_sha=before,
        idempotency_key="workplan-update-1",
    )
    assert applied.status == "applied"
    assert applied.evidence["git_sha"] != before
    text = (repo / "workplans" / "P-WP-0001-first.md").read_text(encoding="utf-8")
    assert 'title: "Renamed"' in text
    assert "status: active" in text

    replay = update_workplan(
        repo,
        "11111111-1111-4111-8111-111111111111",
        title="Renamed",
        status="active",
        expected_head_sha=before,
        idempotency_key="workplan-update-1",
    )
    assert replay.status == "applied"
    assert replay.evidence["git_sha"] == applied.evidence["git_sha"]


def test_delete_is_guarded_recoverable_archive(tmp_path: Path) -> None:
    repo = _fixture(tmp_path)

    rejected = archive_workplan(repo, "P-WP-0001")
    assert rejected.status == "rejected"
    assert rejected.error and rejected.error["code"] == "confirmation_required"
    assert (repo / "workplans" / "P-WP-0001-first.md").is_file()

    applied = archive_workplan(repo, "P-WP-0001", confirm_archive=True)
    assert applied.status == "applied"
    assert not (repo / "workplans" / "P-WP-0001-first.md").exists()
    archived = list((repo / "workplans" / "archived").glob("*-P-WP-0001-first.md"))
    assert len(archived) == 1
    assert "status: archived" in archived[0].read_text(encoding="utf-8")
    _snapshot, index = observe_repository(repo)
    record = next(r for r in index.work_records if r.id == "P-WP-0001")
    assert record.status == "archived"
    assert record.source_path.startswith("workplans/archived/")


def test_create_rejects_unsafe_filename_and_duplicate_id(tmp_path: Path) -> None:
    repo = _fixture(tmp_path)

    unsafe = create_workplan(
        repo,
        "P-WP-0002",
        "Second",
        "Goal",
        filename="../outside.md",
    )
    assert unsafe.status == "rejected"
    assert unsafe.error and unsafe.error["code"] == "validation_error"

    duplicate = create_workplan(repo, "P-WP-0001", "Duplicate", "Goal")
    assert duplicate.status == "rejected"
    assert duplicate.error and duplicate.error["code"] == "conflict"