tegwick
d8e0eddb22
STATE-WP-0080-T02: statehub register routes project-flavor scaffolding through rmgr scaffold and keeps only repo + host-path registration. T01 refuse remains when GOAL.md is missing and --wp-prefix is not set. RMGR-WP-0005-T01: C-06/C-11/C-32 skip mint+writeback unless this instance is the identifier registrar (STATEHUB_REGISTRAR or railiance hostname).
477 lines
14 KiB
Python
477 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
import custodian_cli
|
|
from custodian_cli import cmd_fix_consistency
|
|
from statehub_register import (
|
|
RegisterInference,
|
|
detect_project_flavor_signal,
|
|
project_registration_plan,
|
|
refuse_project_flavor_scaffold,
|
|
run_register,
|
|
_invoke_llm,
|
|
_normalise_inference,
|
|
_parse_json_object,
|
|
write_registration_files,
|
|
)
|
|
|
|
|
|
def test_parse_json_object_accepts_fenced_json():
|
|
parsed = _parse_json_object(
|
|
"""Here is the answer:
|
|
|
|
```json
|
|
{"repo_slug": "demo-repo", "in_scope": ["tests"]}
|
|
```
|
|
"""
|
|
)
|
|
|
|
assert parsed == {"repo_slug": "demo-repo", "in_scope": ["tests"]}
|
|
|
|
|
|
def test_normalise_inference_slugifies_and_normalises_prefix():
|
|
inference = _normalise_inference(
|
|
{
|
|
"project_description": "Provides demo automation.",
|
|
"domain_slug": "Custodian",
|
|
"repo_slug": "Demo Repo!",
|
|
"workplan_prefix": "demo",
|
|
"in_scope": ["one", "two"],
|
|
"out_of_scope": ["three"],
|
|
}
|
|
)
|
|
|
|
assert inference.domain_slug == "custodian"
|
|
assert inference.repo_slug == "demo-repo"
|
|
assert inference.workplan_prefix == "DEMO-WP"
|
|
assert inference.in_scope == ["one", "two"]
|
|
assert inference.out_of_scope == ["three"]
|
|
|
|
|
|
def test_invoke_llm_uses_mock_adapter(monkeypatch):
|
|
payload = {"project_description": "Mocked repo.", "repo_slug": "mocked-repo"}
|
|
monkeypatch.setenv("STATEHUB_REGISTER_MOCK_LLM_RESPONSE", json.dumps(payload))
|
|
args = argparse.Namespace(
|
|
llm_provider="mock",
|
|
llm_model=None,
|
|
llm_api_key=None,
|
|
llm_timeout=5,
|
|
)
|
|
|
|
assert json.loads(_invoke_llm("infer this repo", args)) == payload
|
|
|
|
|
|
def test_write_registration_files_primes_codex_repo(tmp_path: Path):
|
|
inference = RegisterInference(
|
|
in_scope=["Run the demo service."],
|
|
out_of_scope=["Operate unrelated services."],
|
|
current_state="Status: active; implementation is small.",
|
|
)
|
|
|
|
written = write_registration_files(
|
|
project_path=tmp_path,
|
|
project_name="demo-service",
|
|
project_description="Provides a demo service.",
|
|
domain="custodian",
|
|
topic_id="cee7bedf-2b48-46ef-8601-006474f2ad7a",
|
|
topic_slug="custodian",
|
|
repo_slug="demo-service",
|
|
wp_prefix="DEMO-WP",
|
|
intent_markdown="# INTENT\n\nDemo service intent.\n",
|
|
inference=inference,
|
|
)
|
|
|
|
assert {path.name for path in written} == {
|
|
"INTENT.md",
|
|
"SCOPE.md",
|
|
"AGENTS.md",
|
|
".custodian-brief.md",
|
|
"DEMO-WP-0001-statehub-bootstrap.md",
|
|
".gitignore",
|
|
}
|
|
assert "!.claude/rules/" in (tmp_path / ".gitignore").read_text()
|
|
assert (tmp_path / "INTENT.md").read_text() == "# INTENT\n\nDemo service intent.\n"
|
|
assert "**Repo slug:** demo-service" in (tmp_path / "AGENTS.md").read_text()
|
|
assert "Run the demo service." in (tmp_path / "SCOPE.md").read_text()
|
|
workplan = (tmp_path / "workplans" / "DEMO-WP-0001-statehub-bootstrap.md").read_text()
|
|
assert "id: DEMO-WP-0001" in workplan
|
|
assert "id: DEMO-WP-0001-T01" in workplan
|
|
assert "statehub fix-consistency" in workplan
|
|
|
|
|
|
def test_detect_project_flavor_from_classification(tmp_path: Path):
|
|
(tmp_path / ".repo-classification.yaml").write_text(
|
|
"repo_classification:\n category: project\n domain: infotech\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
assert detect_project_flavor_signal(tmp_path) == (
|
|
".repo-classification.yaml category=project"
|
|
)
|
|
|
|
|
|
def test_detect_project_flavor_from_goal_md(tmp_path: Path):
|
|
(tmp_path / "GOAL.md").write_text(
|
|
"---\nrepo_flavor: project\n---\n\n# Project goal\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
assert detect_project_flavor_signal(tmp_path) == "GOAL.md repo_flavor=project"
|
|
|
|
|
|
def test_detect_project_flavor_from_prj_directory_name(tmp_path: Path):
|
|
repo = tmp_path / "prj-example"
|
|
repo.mkdir()
|
|
|
|
assert detect_project_flavor_signal(repo) == "slug prefix prj- (prj-example)"
|
|
|
|
|
|
def test_detect_project_flavor_from_repo_slug_argument(tmp_path: Path):
|
|
repo = tmp_path / "example"
|
|
repo.mkdir()
|
|
|
|
assert detect_project_flavor_signal(repo) is None
|
|
assert detect_project_flavor_signal(repo, "prj-example") == (
|
|
"slug prefix prj- (prj-example)"
|
|
)
|
|
|
|
|
|
def test_classification_project_takes_precedence_over_slug(tmp_path: Path):
|
|
repo = tmp_path / "prj-example"
|
|
repo.mkdir()
|
|
(repo / ".repo-classification.yaml").write_text(
|
|
"repo_classification:\n category: project\n domain: infotech\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
assert detect_project_flavor_signal(repo) == (
|
|
".repo-classification.yaml category=project"
|
|
)
|
|
|
|
|
|
def test_durable_repo_is_not_project_flavor(tmp_path: Path):
|
|
(tmp_path / ".repo-classification.yaml").write_text(
|
|
"repo_classification:\n category: tooling\n domain: infotech\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
assert detect_project_flavor_signal(tmp_path, "demo-service") is None
|
|
|
|
|
|
def test_refuse_project_flavor_scaffold_points_at_rmgr(tmp_path: Path):
|
|
repo = tmp_path / "prj-example"
|
|
repo.mkdir()
|
|
|
|
with pytest.raises(SystemExit, match="rmgr scaffold") as exc:
|
|
refuse_project_flavor_scaffold(repo)
|
|
|
|
message = str(exc.value)
|
|
assert "refuses to scaffold a project-flavor repository" in message
|
|
assert "never PRJ-WP-" in message
|
|
assert str(repo) in message
|
|
|
|
|
|
def test_run_register_refuses_prj_repo_without_writing_or_calling_api(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
):
|
|
repo = tmp_path / "prj-example"
|
|
repo.mkdir()
|
|
(repo / ".repo-classification.yaml").write_text(
|
|
"repo_classification:\n category: project\n domain: infotech\n",
|
|
encoding="utf-8",
|
|
)
|
|
called: list[str] = []
|
|
monkeypatch.setattr(
|
|
"statehub_register._check_api",
|
|
lambda *_args, **_kwargs: called.append("api"),
|
|
)
|
|
monkeypatch.setattr(
|
|
"statehub_register.write_registration_files",
|
|
lambda **_kwargs: called.append("write") or [],
|
|
)
|
|
|
|
args = argparse.Namespace(path=str(repo), repo_slug=None)
|
|
|
|
with pytest.raises(SystemExit, match="rmgr scaffold"):
|
|
run_register(args)
|
|
|
|
assert called == []
|
|
assert not (repo / "INTENT.md").exists()
|
|
assert not (repo / "workplans").exists()
|
|
|
|
|
|
def test_project_registration_plan_durable_vs_refuse_vs_delegate(tmp_path: Path):
|
|
durable = tmp_path / "demo-service"
|
|
durable.mkdir()
|
|
assert project_registration_plan(durable) == ("durable", None)
|
|
|
|
fresh = tmp_path / "prj-example"
|
|
fresh.mkdir()
|
|
assert project_registration_plan(fresh)[0] == "refuse"
|
|
assert project_registration_plan(fresh, wp_prefix="EXCO-WP") == (
|
|
"delegate",
|
|
"EXCO-WP",
|
|
)
|
|
|
|
(fresh / "GOAL.md").write_text(
|
|
"---\nrepo_flavor: project\n---\n# Goal\n", encoding="utf-8"
|
|
)
|
|
assert project_registration_plan(fresh) == ("register-only", None)
|
|
assert project_registration_plan(fresh, wp_prefix="EXCO-WP") == (
|
|
"register-only",
|
|
None,
|
|
)
|
|
|
|
|
|
def _stub_register_io(monkeypatch: pytest.MonkeyPatch):
|
|
written: list[dict] = []
|
|
registered: list[tuple] = []
|
|
|
|
monkeypatch.setattr("statehub_register._check_api", lambda *_a, **_k: None)
|
|
monkeypatch.setattr(
|
|
"statehub_register._api_get",
|
|
lambda path, *_a, **_k: (
|
|
[{"slug": "infotech", "id": "dom-1"}]
|
|
if str(path).startswith("/domains")
|
|
else [{"id": "topic-1", "slug": "infotech", "domain_slug": "infotech"}]
|
|
),
|
|
)
|
|
monkeypatch.setattr(
|
|
"statehub_register.write_registration_files",
|
|
lambda **kwargs: written.append(kwargs) or [],
|
|
)
|
|
monkeypatch.setattr(
|
|
"statehub_register._register_or_update_repo",
|
|
lambda **kwargs: registered.append(("repo", kwargs)) or {"id": "repo-1"},
|
|
)
|
|
monkeypatch.setattr(
|
|
"statehub_register._register_host_path",
|
|
lambda *a, **k: registered.append(("path", a, k)),
|
|
)
|
|
monkeypatch.setattr("statehub_register._record_progress", lambda *_a, **_k: None)
|
|
return written, registered
|
|
|
|
|
|
def _register_args(path: Path, **overrides):
|
|
values = {
|
|
"path": str(path),
|
|
"repo_slug": None,
|
|
"wp_prefix": None,
|
|
"domain": "infotech",
|
|
"description": "Test project.",
|
|
"intent": None,
|
|
"api_base": "http://unused",
|
|
"no_llm": True,
|
|
"force": False,
|
|
"llm_provider": "mock",
|
|
"llm_model": None,
|
|
"llm_api_key": None,
|
|
"llm_timeout": 5,
|
|
}
|
|
values.update(overrides)
|
|
return argparse.Namespace(**values)
|
|
|
|
|
|
def test_run_register_delegates_project_scaffold_and_skips_templating(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
):
|
|
repo = tmp_path / "prj-example"
|
|
repo.mkdir()
|
|
written, registered = _stub_register_io(monkeypatch)
|
|
delegated: list[dict] = []
|
|
|
|
def fake_delegate(project_path, **kwargs):
|
|
delegated.append(kwargs)
|
|
(project_path / "GOAL.md").write_text(
|
|
"---\nrepo_flavor: project\n---\n# Goal\n", encoding="utf-8"
|
|
)
|
|
return {"status": "applied"}
|
|
|
|
monkeypatch.setattr("statehub_register.delegate_project_scaffold", fake_delegate)
|
|
|
|
run_register(_register_args(repo, wp_prefix="EXCO-WP"))
|
|
|
|
assert delegated == [
|
|
{
|
|
"slug": "prj-example",
|
|
"wp_prefix": "EXCO-WP",
|
|
"domain": "infotech",
|
|
"force": False,
|
|
}
|
|
]
|
|
assert written == []
|
|
assert [kind for kind, *_ in registered] == ["repo", "path"]
|
|
assert not (repo / "INTENT.md").exists()
|
|
assert (repo / "GOAL.md").is_file()
|
|
|
|
|
|
def test_run_register_existing_project_skips_scaffold_and_templating(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
):
|
|
repo = tmp_path / "prj-example"
|
|
repo.mkdir()
|
|
(repo / "GOAL.md").write_text(
|
|
"---\nrepo_flavor: project\n---\n# Goal\n", encoding="utf-8"
|
|
)
|
|
written, registered = _stub_register_io(monkeypatch)
|
|
monkeypatch.setattr(
|
|
"statehub_register.delegate_project_scaffold",
|
|
lambda *_a, **_k: (_ for _ in ()).throw(AssertionError("should not scaffold")),
|
|
)
|
|
|
|
run_register(_register_args(repo))
|
|
|
|
assert written == []
|
|
assert [kind for kind, *_ in registered] == ["repo", "path"]
|
|
assert not (repo / "INTENT.md").exists()
|
|
|
|
|
|
def test_write_registration_files_is_idempotent_without_force(tmp_path: Path):
|
|
inference = RegisterInference()
|
|
kwargs = {
|
|
"project_path": tmp_path,
|
|
"project_name": "demo",
|
|
"project_description": "Provides a demo.",
|
|
"domain": "custodian",
|
|
"topic_id": "topic",
|
|
"topic_slug": "custodian",
|
|
"repo_slug": "demo",
|
|
"wp_prefix": "DEMO-WP",
|
|
"intent_markdown": "# INTENT\n\nDemo.\n",
|
|
"inference": inference,
|
|
}
|
|
|
|
assert write_registration_files(**kwargs)
|
|
assert write_registration_files(**kwargs) == []
|
|
|
|
|
|
def _fix_args(**overrides):
|
|
values = {
|
|
"repo": None,
|
|
"all": False,
|
|
"path": None,
|
|
"repo_path": None,
|
|
"remote": False,
|
|
"max_seconds": None,
|
|
"no_writeback": False,
|
|
"archive_closed": False,
|
|
"archive_workplan": None,
|
|
"archive_date": None,
|
|
"api_base": "http://statehub.test",
|
|
"as_json": False,
|
|
"strict_warnings": False,
|
|
}
|
|
values.update(overrides)
|
|
return argparse.Namespace(**values)
|
|
|
|
|
|
def _install_fake_checker(monkeypatch, tmp_path: Path) -> Path:
|
|
checker = tmp_path / "scripts" / "consistency_check.py"
|
|
checker.parent.mkdir()
|
|
checker.write_text("#!/usr/bin/env python3\n", encoding="utf-8")
|
|
monkeypatch.setattr(custodian_cli, "STATE_HUB_DIR", tmp_path)
|
|
return checker
|
|
|
|
|
|
def test_fix_consistency_defaults_to_here_and_normalises_warning_exit(monkeypatch, tmp_path: Path):
|
|
checker = _install_fake_checker(monkeypatch, tmp_path)
|
|
repo = tmp_path / "repo"
|
|
repo.mkdir()
|
|
calls = []
|
|
|
|
def fake_run(cmd):
|
|
calls.append(cmd)
|
|
return argparse.Namespace(returncode=2)
|
|
|
|
monkeypatch.setattr(custodian_cli.subprocess, "run", fake_run)
|
|
|
|
with pytest.raises(SystemExit) as exc:
|
|
cmd_fix_consistency(_fix_args(path=str(repo)))
|
|
|
|
assert exc.value.code == 0
|
|
assert calls == [[
|
|
sys.executable,
|
|
str(checker),
|
|
"--here",
|
|
str(repo.resolve()),
|
|
"--fix",
|
|
"--api-base",
|
|
"http://statehub.test",
|
|
]]
|
|
|
|
|
|
def test_fix_consistency_strict_warnings_preserves_exit_two(monkeypatch, tmp_path: Path):
|
|
_install_fake_checker(monkeypatch, tmp_path)
|
|
repo = tmp_path / "repo"
|
|
repo.mkdir()
|
|
monkeypatch.setattr(
|
|
custodian_cli.subprocess,
|
|
"run",
|
|
lambda _cmd: argparse.Namespace(returncode=2),
|
|
)
|
|
|
|
with pytest.raises(SystemExit) as exc:
|
|
cmd_fix_consistency(_fix_args(path=str(repo), strict_warnings=True))
|
|
|
|
assert exc.value.code == 2
|
|
|
|
|
|
def test_fix_consistency_repo_remote_passes_pull_before_fix_options(monkeypatch, tmp_path: Path):
|
|
checker = _install_fake_checker(monkeypatch, tmp_path)
|
|
repo = tmp_path / "repo"
|
|
repo.mkdir()
|
|
calls = []
|
|
|
|
def fake_run(cmd):
|
|
calls.append(cmd)
|
|
return argparse.Namespace(returncode=0)
|
|
|
|
monkeypatch.setattr(custodian_cli.subprocess, "run", fake_run)
|
|
|
|
with pytest.raises(SystemExit) as exc:
|
|
cmd_fix_consistency(
|
|
_fix_args(
|
|
repo="demo-service",
|
|
repo_path=str(repo),
|
|
remote=True,
|
|
no_writeback=True,
|
|
as_json=True,
|
|
max_seconds=12,
|
|
)
|
|
)
|
|
|
|
assert exc.value.code == 0
|
|
assert calls == [[
|
|
sys.executable,
|
|
str(checker),
|
|
"--repo",
|
|
"demo-service",
|
|
"--repo-path",
|
|
str(repo.resolve()),
|
|
"--fix",
|
|
"--remote",
|
|
"--no-writeback",
|
|
"--api-base",
|
|
"http://statehub.test",
|
|
"--json",
|
|
"--max-seconds",
|
|
"12",
|
|
]]
|
|
|
|
|
|
def test_fix_consistency_remote_requires_explicit_repo_or_all(monkeypatch, tmp_path: Path):
|
|
_install_fake_checker(monkeypatch, tmp_path)
|
|
calls = []
|
|
monkeypatch.setattr(custodian_cli.subprocess, "run", lambda cmd: calls.append(cmd))
|
|
|
|
with pytest.raises(SystemExit) as exc:
|
|
cmd_fix_consistency(_fix_args(remote=True, path=str(tmp_path)))
|
|
|
|
assert exc.value.code == 1
|
|
assert calls == []
|