coulomb/state-hub
2
0
Fork 0
generated from coulomb/repo-seed
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: add local consistency sync cli

Browse Source
This commit is contained in:
Bernd Worsch
2026-07-02 00:15:16 +02:00
parent 1f61008837
commit a361ce8731
15 changed files with 422 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

118
tests/test_consistency_check.py
View File

@@ -1016,6 +1016,124 @@ class TestLifecycleRenormalization:
class TestC12OrphanDbTasks:
def _make_repo(self, tmp_path: Path, status: str = "finished") -> Path:
repo = tmp_path / "repo"
workplans = repo / "workplans"
workplans.mkdir(parents=True)
(workplans / "STATE-WP-0001-demo.md").write_text(
"---\n"
"id: STATE-WP-0001\n"
"type: workplan\n"
"title: Demo\n"
"domain: infotech\n"
"repo: state-hub\n"
f"status: {status}\n"
"owner: codex\n"
"state_hub_workstream_id: \"ws-1\"\n"
"---\n\n"
"## Keep Task\n\n"
"```task\n"
"id: STATE-WP-0001-T01\n"
"status: done\n"
"priority: high\n"
"state_hub_task_id: \"task-linked\"\n"
"```\n",
encoding="utf-8",
)
return repo
def _api_get_for_repo(self, repo: Path, orphan_status: str):
ws = {
"id": "ws-1",
"repo_id": "repo-1",
"topic_id": "topic-1",
"slug": "state-wp-0001",
"title": "Demo",
"status": "finished",
"planning_priority": None,
"planning_order": None,
}
linked = {
"id": "task-linked",
"title": "Keep Task",
"status": "done",
"description": None,
}
orphan = {
"id": "task-orphan",
"title": "Legacy Duplicate",
"status": orphan_status,
"description": None,
}
def fake_get(_api_base, path, params=None, **_kwargs):
if path == "/repos/state-hub":
import socket
return {
"id": "repo-1",
"slug": "state-hub",
"domain_slug": "infotech",
"local_path": str(repo),
"host_paths": {socket.gethostname(): str(repo)},
}
if path == "/workstreams/ws-1":
return ws
if path == "/tasks/task-linked":
return linked
if path == "/tasks" and params == {"workstream_id": "ws-1"}:
return [linked, orphan]
if path == "/workstreams/ws-1/dependencies":
return []
if path == "/workstreams" and params == {"repo_id": "repo-1"}:
return [ws]
if path == "/workstreams" and params and params.get("topic_id") == "topic-1":
return []
return []
return fake_get
def _quiet_classification(self, monkeypatch):
monkeypatch.setattr("consistency_check.load_classification_file", lambda _repo_dir: ({}, [], []))
def test_closed_workstream_suppresses_terminal_orphan_task(self, tmp_path, monkeypatch):
repo = self._make_repo(tmp_path)
self._quiet_classification(monkeypatch)
monkeypatch.setattr("consistency_check._api_get", self._api_get_for_repo(repo, "cancel"))
report = check_repo("http://unused", "state-hub")
assert [issue for issue in report.issues if issue.check_id == "C-12"] == []
def test_closed_workstream_reports_open_orphan_task_as_fixable(self, tmp_path, monkeypatch):
repo = self._make_repo(tmp_path)
self._quiet_classification(monkeypatch)
monkeypatch.setattr("consistency_check._api_get", self._api_get_for_repo(repo, "todo"))
report = check_repo("http://unused", "state-hub")
issue = next(issue for issue in report.issues if issue.check_id == "C-12")
assert issue.db_id == "task-orphan"
assert issue.fixable is True
def test_fix_repo_cancels_open_orphan_task_in_closed_workstream(self, tmp_path, monkeypatch):
repo = self._make_repo(tmp_path)
patches = []
self._quiet_classification(monkeypatch)
monkeypatch.setattr("consistency_check._api_get", self._api_get_for_repo(repo, "todo"))
monkeypatch.setattr("consistency_check._api_patch", lambda _api_base, path, body: patches.append((path, body)) or {"ok": True})
monkeypatch.setattr("consistency_check._detect_behind_remote", lambda _repo_path: False)
monkeypatch.setattr("consistency_check._detect_ahead_of_remote", lambda _repo_path: 0)
monkeypatch.setattr("consistency_check._write_custodian_brief", lambda *args, **kwargs: False)
monkeypatch.setattr("consistency_check._git_push", lambda _repo_path: (True, "pushed"))
report = fix_repo("http://unused", "state-hub")
assert ("/tasks/task-orphan", {"status": "cancel"}) in patches
assert any("C-12 fixed: orphan task task-or" in fix for fix in report.fixes_applied)
class TestC20DependencyDetection:
def test_canonical_dependency_fields_satisfy_workplan_dependency(self, tmp_path, monkeypatch):
repo = tmp_path / "repo"

133
tests/test_statehub_register_cli.py
View File

@@ -2,8 +2,13 @@ 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,
_invoke_llm,
@@ -91,7 +96,7 @@ def test_write_registration_files_primes_codex_repo(tmp_path: Path):
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 "make fix-consistency REPO=demo-service" in workplan
assert "statehub fix-consistency" in workplan
def test_write_registration_files_is_idempotent_without_force(tmp_path: Path):
@@ -111,3 +116,129 @@ def test_write_registration_files_is_idempotent_without_force(tmp_path: Path):
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 == []