coulomb/state-hub
2
0
Fork 0
generated from coulomb/repo-seed
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
6cbf2d2c56da5623f1c30f2dca3d9d4759906b24
state-hub/tests/test_consistency_check.py
tegwick 6cbf2d2c56 feat(consistency): T04 push seal — closed-loop writeback for automated commits 
Root cause of the 501-commit pile-up in inter-hub: fix_repo() created
git commits (brief updates, T03 writebacks) but never pushed them, so
the 15-minute timer accumulated local commits indefinitely. Once real
development landed on remote the repos diverged with no self-healing path.

Changes
-------
repo_sync.py (new module)
  Extracts all git lifecycle primitives: pull_ff, push_ff,
  count_remote_ahead (C-16 input), count_local_ahead (C-17/T04 input).
  Module docstring documents the push-seal invariant and stable state.

consistency_check.py
  - Imports primitives from repo_sync; thin _detect_behind_remote wrapper
    preserves backward compat for existing callers and tests.
  - C-17 backlog guard: if local has unpushed commits from a prior failed
    push, retry before making more; skip all writes if push still fails.
  - T04 push seal: unconditional push_ff() at end of every fix_repo() run.
  - _report_needs_action: ahead_of_remote param so repos with unpushed
    backlogs are not silently skipped as "clean" by fix_all_remote().
  - Domain-slug fallback: brief no longer degrades to "(unknown)" when all
    workplans are completed — falls back to any workstream for domain context.
  - Service switched from --all --fix to --remote --all (pulls before
    fixing, skips already-clean repos).

push-seal.md (new)
  Capability documentation: the problem, the invariant, all three checks
  (C-16/C-17/T04), stable-state description, API reference, and test map.

test_repo_sync.py (new, 32 tests)
  Full coverage of all four primitives via real git repos (tmp_path).
  Includes C-17 scenario, push-seal invariant, and four end-to-end
  loop-stability tests.

test_consistency_check.py
  Four new _report_needs_action cases for the ahead_of_remote parameter.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-21 01:43:40 +02:00

702 lines
27 KiB
Python
Raw Blame History

"""Unit tests for the stateless / pure-function layer of consistency_check.py.
Covers:
- parse_frontmatter — YAML frontmatter splitting
- parse_task_blocks — ```task``` block extraction
- get_tasks_from_workplan — block vs. frontmatter fallback
- ConsistencyReport — issue accumulation and severity filtering
- render_text — text rendering smoke tests
- report_to_dict — serialisation shape
No network calls, no DB, no live API — these tests run fully offline.
"""
from __future__ import annotations
import sys
from pathlib import Path
import pytest
# Make scripts/ importable without installing
sys.path.insert(0, str(Path(__file__).parent.parent / "scripts"))
from consistency_check import (
ConsistencyReport,
Issue,
FILE_TO_DB_WORKSTREAM_STATUS,
STATUS_ORDER,
_BACKGROUND_CHECKS,
_detect_behind_remote,
_git_pull,
_patch_task_status_in_file,
_report_needs_action,
get_tasks_from_workplan,
normalise_workstream_status,
parse_frontmatter,
parse_task_blocks,
render_text,
report_to_dict,
)
# _detect_behind_remote and _git_pull are re-exported from consistency_check
# for backward compat; their canonical implementations live in repo_sync.py.
# ---------------------------------------------------------------------------
# parse_frontmatter
# ---------------------------------------------------------------------------
class TestParseFrontmatter:
def test_valid_frontmatter(self):
text = "---\nid: WP-001\ntitle: Test\nstatus: active\n---\n\n# Body"
meta, body = parse_frontmatter(text)
assert meta["id"] == "WP-001"
assert meta["title"] == "Test"
assert meta["status"] == "active"
assert "# Body" in body
def test_no_frontmatter_returns_empty_meta(self):
text = "# Just a heading\n\nSome body text."
meta, body = parse_frontmatter(text)
assert meta == {}
assert body == text
def test_empty_frontmatter(self):
text = "---\n---\n\nBody only."
meta, body = parse_frontmatter(text)
assert isinstance(meta, dict)
assert "Body only." in body
def test_frontmatter_with_quoted_strings(self):
text = '---\nid: "WP-007"\ntitle: "My Workplan"\n---\nBody'
meta, _ = parse_frontmatter(text)
assert meta["id"] == "WP-007"
assert meta["title"] == "My Workplan"
def test_incomplete_frontmatter_delimiter(self):
"""Only one --- means no valid frontmatter block."""
text = "---\nid: foo\nNo closing delimiter"
meta, body = parse_frontmatter(text)
assert meta == {}
def test_body_preserved_fully(self):
text = "---\nid: X\n---\nLine1\nLine2\nLine3"
_, body = parse_frontmatter(text)
assert "Line1" in body
assert "Line2" in body
assert "Line3" in body
# ---------------------------------------------------------------------------
# parse_task_blocks
# ---------------------------------------------------------------------------
class TestParseTaskBlocks:
def test_single_task_block(self):
body = "## T01\n\n```task\nid: WP-T01\nstatus: done\npriority: high\n```\n"
tasks = parse_task_blocks(body)
assert len(tasks) == 1
assert tasks[0]["id"] == "WP-T01"
assert tasks[0]["status"] == "done"
def test_multiple_task_blocks(self):
body = (
"## T01\n```task\nid: T01\nstatus: done\n```\n\n"
"## T02\n```task\nid: T02\nstatus: todo\n```\n"
)
tasks = parse_task_blocks(body)
assert len(tasks) == 2
assert tasks[0]["id"] == "T01"
assert tasks[1]["id"] == "T02"
def test_no_task_blocks(self):
body = "# Workplan\n\nNo task blocks here.\n"
tasks = parse_task_blocks(body)
assert tasks == []
def test_task_block_with_all_fields(self):
body = (
"```task\n"
"id: CUST-WP-0008-T01\n"
"status: done\n"
"priority: critical\n"
"assignee: custodian\n"
'state_hub_task_id: "abc-123"\n'
"```\n"
)
tasks = parse_task_blocks(body)
assert len(tasks) == 1
t = tasks[0]
assert t["id"] == "CUST-WP-0008-T01"
assert t["status"] == "done"
assert t["priority"] == "critical"
assert t["state_hub_task_id"] == "abc-123"
def test_ignores_non_task_fenced_blocks(self):
body = (
"```python\nprint('hello')\n```\n\n"
"```task\nid: T01\nstatus: todo\n```\n"
"```bash\necho hi\n```\n"
)
tasks = parse_task_blocks(body)
assert len(tasks) == 1
assert tasks[0]["id"] == "T01"
# ---------------------------------------------------------------------------
# get_tasks_from_workplan
# ---------------------------------------------------------------------------
class TestGetTasksFromWorkplan:
def test_prefers_task_blocks_over_frontmatter(self):
meta = {"tasks": [{"id": "from-frontmatter", "status": "todo"}]}
body = "```task\nid: from-block\nstatus: done\n```\n"
tasks = get_tasks_from_workplan(meta, body)
assert len(tasks) == 1
assert tasks[0]["id"] == "from-block"
def test_falls_back_to_frontmatter_tasks(self):
meta = {"tasks": [{"id": "FM-01", "status": "todo"}, {"id": "FM-02", "status": "done"}]}
body = "# No task blocks here\n"
tasks = get_tasks_from_workplan(meta, body)
assert len(tasks) == 2
assert tasks[0]["id"] == "FM-01"
assert tasks[1]["id"] == "FM-02"
def test_returns_empty_when_no_tasks(self):
meta = {}
body = "# Just prose\n"
tasks = get_tasks_from_workplan(meta, body)
assert tasks == []
def test_ignores_non_list_frontmatter_tasks(self):
meta = {"tasks": "not-a-list"}
body = "# No blocks\n"
tasks = get_tasks_from_workplan(meta, body)
assert tasks == []
# ---------------------------------------------------------------------------
# ConsistencyReport — issue accumulation and severity filtering
# ---------------------------------------------------------------------------
class TestConsistencyReport:
def _report(self):
return ConsistencyReport(repo_slug="test-repo", repo_path="/tmp/test-repo")
def test_add_fail_issue(self):
r = self._report()
r.add(severity="FAIL", check_id="C-01", message="No workplans/ dir")
assert len(r.issues) == 1
assert len(r.failures) == 1
assert len(r.warnings) == 0
assert len(r.infos) == 0
def test_add_warn_issue(self):
r = self._report()
r.add(severity="WARN", check_id="C-04", message="Status drift", fixable=True)
assert len(r.warnings) == 1
assert r.warnings[0].fixable is True
def test_add_info_issue(self):
r = self._report()
r.add(severity="INFO", check_id="C-08", message="Completed orphan")
assert len(r.infos) == 1
def test_mixed_severities(self):
r = self._report()
r.add(severity="FAIL", check_id="C-01", message="F1")
r.add(severity="FAIL", check_id="C-03", message="F2")
r.add(severity="WARN", check_id="C-04", message="W1")
r.add(severity="INFO", check_id="C-08", message="I1")
assert len(r.failures) == 2
assert len(r.warnings) == 1
assert len(r.infos) == 1
assert len(r.issues) == 4
def test_add_returns_issue(self):
r = self._report()
issue = r.add(severity="FAIL", check_id="C-01", message="msg")
assert isinstance(issue, Issue)
assert issue.severity == "FAIL"
def test_empty_report(self):
r = self._report()
assert r.failures == []
assert r.warnings == []
assert r.infos == []
assert r.issues == []
# ---------------------------------------------------------------------------
# render_text
# ---------------------------------------------------------------------------
class TestRenderText:
def _clean_report(self):
return ConsistencyReport(repo_slug="my-repo", repo_path="/path/to/repo")
def test_pass_result_shown(self):
r = self._clean_report()
text = render_text(r)
assert "PASS" in text
assert "my-repo" in text
def test_fail_result_shown(self):
r = self._clean_report()
r.add(severity="FAIL", check_id="C-01", message="No workplans/ directory")
text = render_text(r)
assert "FAIL" in text
assert "C-01" in text
assert "No workplans/ directory" in text
def test_warn_result_shown(self):
r = self._clean_report()
r.add(severity="WARN", check_id="C-04", message="Status drift", fixable=True)
text = render_text(r)
assert "WARN" in text
assert "[fixable]" in text
def test_fix_applied_shown(self):
r = self._clean_report()
r.fixes_applied.append("C-04: updated status active→done for WP-001")
text = render_text(r)
assert "FIXES APPLIED" in text
assert "C-04" in text
def test_summary_counts_correct(self):
r = self._clean_report()
r.add(severity="FAIL", check_id="C-01", message="f")
r.add(severity="WARN", check_id="C-04", message="w")
r.add(severity="INFO", check_id="C-08", message="i")
text = render_text(r)
assert "1 fail" in text
assert "1 warn" in text
assert "1 info" in text
def test_info_hidden_when_show_info_false(self):
r = self._clean_report()
r.add(severity="INFO", check_id="C-08", message="Completed orphan")
text = render_text(r, show_info=False)
assert "C-08" not in text
# ---------------------------------------------------------------------------
# report_to_dict
# ---------------------------------------------------------------------------
class TestReportToDict:
def test_clean_report_result_is_pass(self):
r = ConsistencyReport(repo_slug="r", repo_path="/p")
d = report_to_dict(r)
assert d["result"] == "pass"
assert d["summary"] == {"fail": 0, "warn": 0, "info": 0}
assert d["issues"] == []
def test_fail_result(self):
r = ConsistencyReport(repo_slug="r", repo_path="/p")
r.add(severity="FAIL", check_id="C-01", message="missing")
d = report_to_dict(r)
assert d["result"] == "fail"
assert d["summary"]["fail"] == 1
def test_warn_only_result(self):
r = ConsistencyReport(repo_slug="r", repo_path="/p")
r.add(severity="WARN", check_id="C-04", message="drift")
d = report_to_dict(r)
assert d["result"] == "warn"
assert d["summary"]["warn"] == 1
assert d["summary"]["fail"] == 0
def test_issue_fields_serialised(self):
r = ConsistencyReport(repo_slug="r", repo_path="/p")
r.add(
severity="FAIL",
check_id="C-03",
message="stale ref",
file_path="workplans/WP.md",
db_id="abc-123",
file_value="abc-123",
db_value="",
)
d = report_to_dict(r)
issue = d["issues"][0]
assert issue["severity"] == "FAIL"
assert issue["check_id"] == "C-03"
assert issue["file_path"] == "workplans/WP.md"
assert issue["db_id"] == "abc-123"
def test_fixes_applied_in_dict(self):
r = ConsistencyReport(repo_slug="r", repo_path="/p")
r.fixes_applied.append("C-04: status fixed")
d = report_to_dict(r)
assert "C-04: status fixed" in d["fixes_applied"]
def test_repo_slug_and_path_preserved(self):
r = ConsistencyReport(repo_slug="the-custodian", repo_path="/home/worsch/the-custodian")
d = report_to_dict(r)
assert d["repo_slug"] == "the-custodian"
assert d["repo_path"] == "/home/worsch/the-custodian"
# ---------------------------------------------------------------------------
# Status vocabulary normalisation
# ---------------------------------------------------------------------------
class TestNormaliseWorkstreamStatus:
"""FILE_TO_DB_WORKSTREAM_STATUS maps workplan file vocabulary to DB vocabulary.
Workplan files use task-style "done"; the DB workstream API uses "completed".
The C-04 check and fix code must normalise before comparing or PATCHing.
"""
def test_done_maps_to_completed(self):
assert normalise_workstream_status("done") == "completed"
def test_completed_is_identity(self):
assert normalise_workstream_status("completed") == "completed"
def test_active_is_identity(self):
assert normalise_workstream_status("active") == "active"
def test_blocked_is_identity(self):
assert normalise_workstream_status("blocked") == "blocked"
def test_archived_is_identity(self):
assert normalise_workstream_status("archived") == "archived"
def test_unknown_value_returned_as_is(self):
# Don't crash on unexpected values — return them unchanged
assert normalise_workstream_status("foobar") == "foobar"
def test_map_constant_covers_done(self):
assert "done" in FILE_TO_DB_WORKSTREAM_STATUS
assert FILE_TO_DB_WORKSTREAM_STATUS["done"] == "completed"
def test_c04_no_spurious_drift_when_done_vs_completed(self):
"""done (file) vs completed (DB) must NOT be reported as C-04 drift."""
assert normalise_workstream_status("done") == normalise_workstream_status("completed")
def test_c04_real_drift_still_detected(self):
"""done (file) vs active (DB) IS real drift and must be detected."""
assert normalise_workstream_status("done") != normalise_workstream_status("active")
# ---------------------------------------------------------------------------
# STATUS_ORDER / no-regress rule (T01 / C-15)
# ---------------------------------------------------------------------------
class TestStatusOrder:
"""STATUS_ORDER drives the no-regress rule: DB-ahead wins, file-ahead syncs."""
def test_todo_is_lowest(self):
assert STATUS_ORDER["todo"] == 0
def test_done_and_cancelled_are_highest(self):
assert STATUS_ORDER["done"] == STATUS_ORDER["cancelled"] == 2
def test_in_progress_and_blocked_are_mid(self):
assert STATUS_ORDER["in_progress"] == STATUS_ORDER["blocked"] == 1
def test_db_ahead_detected(self):
"""done (DB) vs todo (file) — DB is ahead."""
assert STATUS_ORDER["done"] > STATUS_ORDER["todo"]
def test_file_ahead_detected(self):
"""done (file) vs todo (DB) — file is ahead, should sync."""
assert STATUS_ORDER["todo"] < STATUS_ORDER["done"]
def test_same_rank_treated_as_db_ahead(self):
"""in_progress (DB) vs blocked (file) — same rank, no regression."""
assert STATUS_ORDER["in_progress"] >= STATUS_ORDER["blocked"]
def test_todo_to_done_is_regression(self):
"""Applying file=todo to DB=done would be a regression."""
db_rank = STATUS_ORDER["done"]
file_rank = STATUS_ORDER["todo"]
assert db_rank >= file_rank # → should emit C-15, not C-10
# ---------------------------------------------------------------------------
# _detect_behind_remote (T02 / C-16)
# ---------------------------------------------------------------------------
class TestDetectBehindRemote:
"""_detect_behind_remote must return False on any error (best-effort)."""
def test_returns_false_for_nonexistent_path(self):
assert _detect_behind_remote("/nonexistent/path/xyz") is False
def test_returns_false_for_non_git_dir(self, tmp_path):
assert _detect_behind_remote(str(tmp_path)) is False
def test_returns_false_for_repo_without_remote(self, tmp_path):
"""A local-only repo with no remote tracking branch → not behind."""
import subprocess
subprocess.run(["git", "init", str(tmp_path)], capture_output=True)
subprocess.run(
["git", "-C", str(tmp_path), "commit", "--allow-empty", "-m", "init"],
capture_output=True,
)
# No remote configured → rev-parse @{u} fails → best-effort returns False
assert _detect_behind_remote(str(tmp_path)) is False
def test_returns_false_when_up_to_date(self, tmp_path):
"""Clone from a local bare repo, no new commits → not behind."""
import subprocess
bare = tmp_path / "bare.git"
clone = tmp_path / "clone"
bare.mkdir()
subprocess.run(["git", "init", "--bare", str(bare)], capture_output=True)
subprocess.run(["git", "clone", str(bare), str(clone)], capture_output=True)
subprocess.run(
["git", "-C", str(clone), "commit", "--allow-empty", "-m", "init"],
capture_output=True,
)
subprocess.run(["git", "-C", str(clone), "push", "origin", "HEAD"], capture_output=True)
assert _detect_behind_remote(str(clone)) is False
def test_returns_false_when_local_ahead(self, tmp_path):
"""Local has unpushed commits — NOT considered behind."""
import subprocess
bare = tmp_path / "bare.git"
clone = tmp_path / "clone"
bare.mkdir()
subprocess.run(["git", "init", "--bare", str(bare)], capture_output=True)
subprocess.run(["git", "clone", str(bare), str(clone)], capture_output=True)
subprocess.run(
["git", "-C", str(clone), "commit", "--allow-empty", "-m", "init"],
capture_output=True,
)
subprocess.run(["git", "-C", str(clone), "push", "origin", "HEAD"], capture_output=True)
# Add a local-only commit (not pushed)
subprocess.run(
["git", "-C", str(clone), "commit", "--allow-empty", "-m", "local only"],
capture_output=True,
)
assert _detect_behind_remote(str(clone)) is False
def test_returns_true_when_behind(self, tmp_path):
"""Remote has a commit the clone doesn't → clone is behind."""
import subprocess
bare = tmp_path / "bare.git"
clone = tmp_path / "clone"
bare.mkdir()
subprocess.run(["git", "init", "--bare", str(bare)], capture_output=True)
subprocess.run(["git", "clone", str(bare), str(clone)], capture_output=True)
# Initial commit pushed so there's an upstream ref
subprocess.run(
["git", "-C", str(clone), "commit", "--allow-empty", "-m", "init"],
capture_output=True,
)
subprocess.run(["git", "-C", str(clone), "push", "origin", "HEAD"], capture_output=True)
# Add a commit directly in the bare repo (simulates remote-only progress)
work = tmp_path / "work"
subprocess.run(["git", "clone", str(bare), str(work)], capture_output=True)
subprocess.run(
["git", "-C", str(work), "commit", "--allow-empty", "-m", "remote commit"],
capture_output=True,
)
subprocess.run(["git", "-C", str(work), "push"], capture_output=True)
# After fetch the clone should appear behind
assert _detect_behind_remote(str(clone)) is True
# ---------------------------------------------------------------------------
# _patch_task_status_in_file (T03 writeback)
# ---------------------------------------------------------------------------
class TestPatchTaskStatusInFile:
"""_patch_task_status_in_file must only change the status field in the
matching task block and leave everything else untouched."""
def _make_workplan(self, tmp_path, content: str) -> Path:
f = tmp_path / "workplan.md"
f.write_text(content, encoding="utf-8")
return f
def test_patches_matching_task_block(self, tmp_path):
content = (
"---\nid: WP-001\n---\n"
"## My Task\n\n"
"```task\n"
"id: T01\n"
"status: todo\n"
"priority: high\n"
"```\n"
)
f = self._make_workplan(tmp_path, content)
result = _patch_task_status_in_file(f, "T01", "done")
assert result is True
patched = f.read_text()
assert "status: done" in patched
assert "status: todo" not in patched
assert "priority: high" in patched # other fields untouched
def test_does_not_patch_non_matching_block(self, tmp_path):
content = (
"---\nid: WP-001\n---\n"
"```task\n"
"id: T02\n"
"status: todo\n"
"```\n"
)
f = self._make_workplan(tmp_path, content)
result = _patch_task_status_in_file(f, "T01", "done")
assert result is False
assert "status: todo" in f.read_text()
def test_patches_correct_block_among_multiple(self, tmp_path):
content = (
"---\nid: WP-001\n---\n"
"```task\nid: T01\nstatus: todo\n```\n"
"```task\nid: T02\nstatus: in_progress\n```\n"
)
f = self._make_workplan(tmp_path, content)
_patch_task_status_in_file(f, "T02", "done")
patched = f.read_text()
assert "id: T01\nstatus: todo" in patched
assert "id: T02\nstatus: done" in patched
def test_does_not_touch_non_status_fields(self, tmp_path):
content = (
"---\nid: WP\n---\n"
"```task\nid: T01\nstatus: todo\npriority: high\nstate_hub_task_id: \"abc\"\n```\n"
)
f = self._make_workplan(tmp_path, content)
_patch_task_status_in_file(f, "T01", "done")
patched = f.read_text()
assert "priority: high" in patched
assert 'state_hub_task_id: "abc"' in patched
def test_idempotent_when_already_correct(self, tmp_path):
content = "---\nid: WP\n---\n```task\nid: T01\nstatus: done\n```\n"
f = self._make_workplan(tmp_path, content)
result = _patch_task_status_in_file(f, "T01", "done")
# status matches, re.sub produces same text → no write → False
assert result is False
# ---------------------------------------------------------------------------
# _git_pull (T02 remote fix helper)
# ---------------------------------------------------------------------------
class TestGitPull:
def test_returns_false_for_nonexistent_path(self):
ok, msg = _git_pull("/nonexistent/path")
assert ok is False
assert msg
def test_returns_true_for_up_to_date_repo(self, tmp_path):
import subprocess
bare = tmp_path / "bare.git"
clone = tmp_path / "clone"
bare.mkdir()
subprocess.run(["git", "init", "--bare", str(bare)], capture_output=True)
subprocess.run(["git", "clone", str(bare), str(clone)], capture_output=True)
subprocess.run(
["git", "-C", str(clone), "commit", "--allow-empty", "-m", "init"],
capture_output=True,
)
subprocess.run(["git", "-C", str(clone), "push", "origin", "HEAD"], capture_output=True)
ok, msg = _git_pull(str(clone))
assert ok is True
def test_pulls_new_remote_commit(self, tmp_path):
import subprocess
bare = tmp_path / "bare.git"
clone = tmp_path / "clone"
work = tmp_path / "work"
bare.mkdir()
subprocess.run(["git", "init", "--bare", str(bare)], capture_output=True)
subprocess.run(["git", "clone", str(bare), str(clone)], capture_output=True)
subprocess.run(
["git", "-C", str(clone), "commit", "--allow-empty", "-m", "init"],
capture_output=True,
)
subprocess.run(["git", "-C", str(clone), "push", "origin", "HEAD"], capture_output=True)
# Push a new commit from a second clone
subprocess.run(["git", "clone", str(bare), str(work)], capture_output=True)
subprocess.run(
["git", "-C", str(work), "commit", "--allow-empty", "-m", "remote work"],
capture_output=True,
)
subprocess.run(["git", "-C", str(work), "push"], capture_output=True)
# Pull should succeed and bring in the new commit
ok, msg = _git_pull(str(clone))
assert ok is True
head = subprocess.run(
["git", "-C", str(clone), "log", "--oneline", "-1"],
capture_output=True, text=True,
).stdout
assert "remote work" in head
# ---------------------------------------------------------------------------
# _report_needs_action (smart skip logic)
# ---------------------------------------------------------------------------
class TestReportNeedsAction:
def _make_report(self, issues: list[tuple[str, str, bool]]) -> ConsistencyReport:
"""Build a report from (severity, check_id, fixable) tuples."""
r = ConsistencyReport(repo_slug="test", repo_path="/tmp/test")
for severity, check_id, fixable in issues:
r.add(severity=severity, check_id=check_id, message="test", fixable=fixable)
return r
def test_clean_report_not_behind_is_skipped(self):
r = self._make_report([])
assert _report_needs_action(r, behind_remote=False) is False
def test_behind_remote_always_needs_action(self):
r = self._make_report([])
assert _report_needs_action(r, behind_remote=True) is True
def test_fail_always_needs_action(self):
r = self._make_report([("FAIL", "C-07", False)])
assert _report_needs_action(r, behind_remote=False) is True
def test_c08_only_is_background_noise(self):
"""C-08 (legacy archived workstream) alone should not trigger a fix cycle."""
r = self._make_report([("INFO", "C-08", False)])
assert _report_needs_action(r, behind_remote=False) is False
def test_c08_plus_actionable_warn_needs_action(self):
r = self._make_report([("INFO", "C-08", False), ("WARN", "C-10", True)])
assert _report_needs_action(r, behind_remote=False) is True
def test_fixable_warn_needs_action(self):
r = self._make_report([("WARN", "C-15", True)])
assert _report_needs_action(r, behind_remote=False) is True
def test_non_fixable_warn_needs_action(self):
"""Non-fixable WARNs (e.g. C-07 ghost) still warrant attention."""
r = self._make_report([("WARN", "C-14", False)])
assert _report_needs_action(r, behind_remote=False) is True
def test_background_checks_constant_contains_c08(self):
assert "C-08" in _BACKGROUND_CHECKS
# --- ahead_of_remote (T04 push-seal integration) ---
def test_ahead_of_remote_triggers_action(self):
"""Unpushed commits from a prior failed push must re-trigger a fix cycle."""
r = self._make_report([])
assert _report_needs_action(r, behind_remote=False, ahead_of_remote=3) is True
def test_zero_ahead_clean_is_skipped(self):
"""Fully in-sync repo with no issues is skipped."""
r = self._make_report([])
assert _report_needs_action(r, behind_remote=False, ahead_of_remote=0) is False
def test_ahead_default_is_zero(self):
"""Default value of ahead_of_remote=0 preserves backward compat."""
r = self._make_report([])
assert _report_needs_action(r, behind_remote=False) is False
def test_both_behind_and_ahead_triggers(self):
"""Diverged repo (both behind and ahead) needs action."""
r = self._make_report([])
assert _report_needs_action(r, behind_remote=True, ahead_of_remote=5) is True
Reference in New Issue View Git Blame Copy Permalink