mirror of
https://github.com/0xWheatyz/handler.git
synced 2026-08-30 09:46:24 +00:00
0xWheatyz
6fb26115ce
Phase 2 configures forge for the agents (operator only sets a credential_ref + optional version pin) and lets them drive a junior→senior→deploy workflow: - Credential resolution/injection (control/credentials.py): credential_ref pointers (env:/file:/cmd:) resolved only at spawn, injected as FORGE_TOKEN + host var, with a forge-host-scoped git credential helper reading the token from env (never on disk / in the DB). Resolution is a fail-fast spawn gate. - Role-based forge skills committed into the managed repo (control/skills_gen.py, `handler forge-init`): forge-junior/senior/deploy + a workflow overview. - Hard approval gate (hooks/gate.py, approvals table, migration 0002): merge/deploy — and direct pushes to protected branches — are denied unless a DIFFERENT agent has an `approved` record for the branch, pinned to the reviewed commit (approved_sha). Senior records verdicts via `handler approve`/`reject`. - forge/git seams (control/forge.py, control/gitops.py) matching the Phase 1 seam pattern. - CI status poller (control/poller.py, `handler poll-ci [--watch]`) backfilling ci_status/ci_checked_at via `forge ci list`. - Fix: migrations/env.py commits explicitly after run_migrations — pysqlite on Py 3.12+ was rolling back the final migration's DDL + alembic_version stamp (latent in Phase 1). Reviewed via a separate code-reviewer pass; gate-bypass and credential-scoping findings addressed. 106 tests, ruff clean, verified end-to-end against real git + migrations.
94 lines
3.1 KiB
Python
94 lines
3.1 KiB
Python
"""CI backfill poller: run classification + a full sweep against a faked forge."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from handler.control import poller
|
|
from handler.db import repository as repo
|
|
|
|
|
|
def test_classify_no_runs_is_pending():
|
|
assert poller.classify([]) == "pending"
|
|
|
|
|
|
def test_classify_any_failure_is_fail():
|
|
runs = [{"conclusion": "success"}, {"conclusion": "failure"}]
|
|
assert poller.classify(runs) == "fail"
|
|
|
|
|
|
def test_classify_all_success_is_pass():
|
|
runs = [{"status": "completed", "conclusion": "success"}]
|
|
assert poller.classify(runs) == "pass"
|
|
|
|
|
|
def test_classify_running_stays_pending():
|
|
runs = [{"status": "in_progress", "conclusion": None}]
|
|
assert poller.classify(runs) == "pending"
|
|
|
|
|
|
def test_classify_tolerates_alternate_spellings():
|
|
assert poller.classify([{"conclusion": "succeeded"}]) == "pass"
|
|
assert poller.classify([{"conclusion": "canceled"}]) == "fail"
|
|
|
|
|
|
def test_classify_action_required_stays_pending_not_pass():
|
|
# A terminal-but-non-success conclusion must NOT be reported as a pass.
|
|
runs = [{"status": "completed", "conclusion": "action_required"}]
|
|
assert poller.classify(runs) == "pending"
|
|
|
|
|
|
def test_classify_status_only_forge_success():
|
|
# No conclusion field at all: fall back to the status.
|
|
assert poller.classify([{"status": "completed"}]) == "pass"
|
|
|
|
|
|
def _seed_pending(conn):
|
|
repo.create_project(conn, "p", "/tmp/p")
|
|
a = repo.create_agent(conn, "p", "a", "/tmp/p/a")
|
|
return repo.insert_log_entry(
|
|
conn, a["id"], status="working", push_sha="sha1", ci_status="pending"
|
|
)
|
|
|
|
|
|
def test_sweep_backfills_pass(engine, fake_forge):
|
|
with engine.begin() as conn:
|
|
entry_id = _seed_pending(conn)
|
|
fake_forge["runs"] = [{"status": "completed", "conclusion": "success"}]
|
|
|
|
summary = poller.sweep()
|
|
assert summary == {"checked": 1, "resolved": 1, "pending": 0}
|
|
with engine.begin() as conn:
|
|
assert repo.get_pending_ci_entries(conn) == []
|
|
row = [e for e in repo.get_log(conn, 1) if e["id"] == entry_id][0]
|
|
assert row["ci_status"] == "pass"
|
|
assert row["ci_checked_at"] is not None
|
|
|
|
|
|
def test_sweep_leaves_pending_when_unresolved(engine, fake_forge):
|
|
with engine.begin() as conn:
|
|
_seed_pending(conn)
|
|
fake_forge["runs"] = [{"status": "in_progress"}]
|
|
summary = poller.sweep()
|
|
assert summary["resolved"] == 0
|
|
with engine.begin() as conn:
|
|
assert len(repo.get_pending_ci_entries(conn)) == 1
|
|
|
|
|
|
def test_sweep_leaves_pending_when_forge_unavailable(engine, fake_forge):
|
|
with engine.begin() as conn:
|
|
_seed_pending(conn)
|
|
fake_forge["ci_ok"] = False
|
|
summary = poller.sweep()
|
|
assert summary["resolved"] == 0
|
|
with engine.begin() as conn:
|
|
assert len(repo.get_pending_ci_entries(conn)) == 1
|
|
|
|
|
|
def test_watch_runs_bounded_iterations(engine, fake_forge):
|
|
with engine.begin() as conn:
|
|
_seed_pending(conn)
|
|
fake_forge["runs"] = [{"conclusion": "failure"}]
|
|
summaries = list(poller.watch(iterations=1, interval=0))
|
|
assert len(summaries) == 1
|
|
with engine.begin() as conn:
|
|
assert repo.get_log(conn, 1)[-1]["ci_status"] == "fail"
|