feat(forge): report unreadable repositories as unreadable (STATE-WP-0084-T01)
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s
Build and Publish Multi-Context Image / build-and-push (push) Successful in 25s

A private repository failed derivation the same way a broken one did, so
"cannot read" and "does not exist" were indistinguishable from outside.
They authorise opposite things: only the second can justify retiring a
record.

- ForgeUnreadableError (a ForgeDeriveError, so old callers still catch it)
  for permission-shaped clone failures, including Forgejo's 404 for an
  unauthenticated private repo — indistinguishable here, and the safe
  reading of an ambiguous answer cannot destroy a record.
- GIT_TERMINAL_PROMPT=0: an unattended pass must fail, not block on a
  username prompt. Failing is what makes the case observable.
- DerivedProjection.retirement_eligible separates "no records found" from
  "no records exist". A checkout with no workplans/ directory cannot
  evidence an absence — the empty-clone path that would have proposed
  every record in a repository for retirement.
- Retirement from an ineligible source is refused even when acknowledged.
- Fleet keeps unreadable out of the error bucket.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 3377672@bnt-lap001
Assistant-Session: 15463ccf-238f-4e13-b163-93aa25c6d166
This commit is contained in:
tegwick 2026-08-26 21:45:03 +02:00
parent 8b38f815e5
commit 85181cd3e4
3 changed files with 294 additions and 6 deletions

View file

@ -417,3 +417,118 @@ class TestSlugCollisionRefusal:
assert out.status == "refused"
assert out.refused[0]["reason"].startswith("slug already belongs")
assert session.added == []
class TestUnreadableIsNotMissing:
"""STATE-WP-0084-T01.
A repository central is not permitted to read and a repository whose
records no longer derive authorise opposite things. Every test here exists
to keep the retirement path unreachable from an answer that cannot support
it by construction, not by the reset happening to fail first.
"""
@pytest.mark.parametrize(
"stderr",
[
"fatal: could not read Username for 'https://forgejo.coulomb.social'",
"remote: Invalid username or password.\nfatal: Authentication failed",
"fatal: could not read Username for 'https://f': terminal prompts disabled",
"fatal: repository 'https://forgejo.coulomb.social/rapp-openbao.git' not found",
"fatal: unable to access '...': The requested URL returned error: 403",
],
)
def test_permission_shaped_failures_are_classified_unreadable(self, stderr, monkeypatch):
def boom(*a, **k):
raise fp.ForgeDeriveError(stderr)
monkeypatch.setattr(fp, "_run_git", boom)
with pytest.raises(fp.ForgeUnreadableError):
fp.derive_from_forge("rapp-openbao")
def test_a_genuine_fault_stays_a_plain_error(self, monkeypatch):
def boom(*a, **k):
raise fp.ForgeDeriveError("fatal: early EOF\nfatal: index-pack failed")
monkeypatch.setattr(fp, "_run_git", boom)
with pytest.raises(fp.ForgeDeriveError) as exc:
fp.derive_from_forge("demo")
assert not isinstance(exc.value, fp.ForgeUnreadableError)
def test_unreadable_is_a_derive_error_so_old_callers_still_catch_it(self):
assert issubclass(fp.ForgeUnreadableError, fp.ForgeDeriveError)
def test_a_checkout_without_workplans_cannot_evidence_absence(self, tmp_path):
(tmp_path / "bare").mkdir()
p = fp.derive_from_checkout(tmp_path / "bare", "bare", "abc")
assert p.workplans == []
assert p.records_source_present is False
assert p.retirement_eligible is False
def test_a_real_checkout_can(self, tmp_path):
p = fp.derive_from_checkout(_repo(tmp_path), "demo", "abc")
assert p.records_source_present is True
assert p.retirement_eligible is True
def test_the_diff_withholds_stale_rather_than_computing_it(self):
derived = fp.DerivedProjection(
repo_slug="demo", commit="c0ffee", records_source_present=False
)
hub = [{"id": "11111111-1111-1111-1111-111111111111", "slug": "demo-wp-0001",
"status": "active", "backing_relative_path": "workplans/a.md"}]
d = fp.diff_against_hub(derived, hub, {})
assert d.stale == []
assert d.would_remove == 0
assert d.stale_withheld
@pytest.mark.asyncio
async def test_an_empty_source_cannot_retire_even_when_acknowledged(self):
derived = fp.DerivedProjection(
repo_slug="demo", commit="c0ffee", records_source_present=False
)
row = _Row(slug="demo-wp-0001", status="active", path="workplans/a.md")
session = _FakeSession(repo=_Repo(), rows=[row])
out = await fp.reset_repository_projection(
session, "demo", derived=derived, acknowledge_retirements=True
)
assert out.status == "refused"
assert out.retired == []
assert row.projection_retired_at is None
assert out.refused[0]["reason"].startswith("source produced no records")
@pytest.mark.asyncio
async def test_an_unreadable_repository_reports_unreadable_not_error(self, monkeypatch):
def boom(*a, **k):
raise fp.ForgeUnreadableError("rapp-openbao could not be read from the forge")
monkeypatch.setattr(fp, "derive_from_forge", boom)
session = _FakeSession(repo=_Repo(), rows=[])
out = await fp.reset_repository_projection(session, "rapp-openbao")
assert out.status == "unreadable"
assert out.retired == [] and out.created == [] and out.updated == []
@pytest.mark.asyncio
async def test_the_fleet_keeps_unreadable_out_of_the_error_bucket(self, monkeypatch):
async def fake(session, slug, **kw):
if slug == "private":
out = fp.ResetOutcome(repo_slug=slug, commit="", status="unreadable")
out.refused.append({"reason": "repository could not be read from the forge",
"slug": slug, "detail": "could not read Username"})
return out
if slug == "broken":
raise RuntimeError("index-pack failed")
out = fp.ResetOutcome(repo_slug=slug, commit="c0ffee", status="applied")
out.updated.append("A-WP-0001")
return out
monkeypatch.setattr(fp, "reset_repository_projection", fake)
sessions = [_FakeSession(repo=_Repo(), rows=[]) for _ in range(3)]
outcome = await fp.reset_fleet_projection(
TestFleetReset._Factory(sessions), ["ok", "private", "broken"]
)
assert list(outcome.unreadable) == ["private"]
assert list(outcome.errors) == ["broken"]
assert "private" not in outcome.results
d = outcome.to_dict()
assert d["unreadable_count"] == 1 and d["errored"] == 1
assert d["repositories"] == 3