181 lines
8.4 KiB
Python
181 lines
8.4 KiB
Python
import sqlite3
|
|
|
|
import pytest
|
|
|
|
from src.human_gate_store import GateConflict, GateValidationError, HumanGateStore
|
|
|
|
|
|
def candidate(hash_value="abc123", *, priority=2, check_state="success"):
|
|
return {
|
|
"source": "release-bot",
|
|
"project": "stackchain/dashboard",
|
|
"candidate_hash": hash_value,
|
|
"title": "Dashboard candidate",
|
|
"priority": priority,
|
|
"artifacts": [{"name": "manifest", "url": "https://forge.example/artifacts/manifest.json"}],
|
|
"links": [{"label": "change", "url": "https://forge.example/pulls/1415"}],
|
|
"checks": [{"name": "browser", "state": check_state, "required": True}],
|
|
"score": {"value": 92, "provenance": "release-evaluator/v2"},
|
|
"provenance": {"producer": "release-bot", "run_id": "run-9"},
|
|
}
|
|
|
|
|
|
def checklist():
|
|
return {"exact_hash": True, "artifacts_reviewed": True, "provenance_reviewed": True}
|
|
|
|
|
|
def test_intake_is_account_bound_idempotent_and_survives_restart(tmp_path):
|
|
path = tmp_path / "gates.sqlite3"
|
|
store = HumanGateStore(path, clock=lambda: 100)
|
|
first = store.intake("timmy", candidate(), idempotency_key="producer-9")
|
|
repeated = store.intake("timmy", candidate(), idempotency_key="producer-9")
|
|
|
|
restarted = HumanGateStore(path, clock=lambda: 101)
|
|
|
|
assert repeated == first
|
|
assert restarted.detail("timmy", first["id"])["candidate_hash"] == "abc123"
|
|
assert restarted.list("alex")["pending_count"] == 0
|
|
with sqlite3.connect(path) as connection:
|
|
assert connection.execute("PRAGMA journal_mode").fetchone()[0] == "wal"
|
|
|
|
|
|
def test_new_hash_supersedes_only_older_pending_candidate_without_losing_audit(tmp_path):
|
|
store = HumanGateStore(tmp_path / "gates.sqlite3", clock=iter([100, 101]).__next__)
|
|
old = store.intake("timmy", candidate("abc123"), idempotency_key="run-1")
|
|
new = store.intake("timmy", candidate("def456"), idempotency_key="run-2")
|
|
|
|
old_detail = store.detail("timmy", old["id"])
|
|
queue = store.list("timmy")
|
|
|
|
assert old_detail["state"] == "superseded"
|
|
assert old_detail["superseded_by"] == new["id"]
|
|
assert old_detail["history"][-1]["action"] == "superseded"
|
|
assert queue["pending_count"] == 1
|
|
assert queue["items"][0]["candidate_hash"] == "def456"
|
|
|
|
|
|
def test_pending_queue_orders_highest_priority_then_oldest(tmp_path):
|
|
clock = iter([100, 101, 102]).__next__
|
|
store = HumanGateStore(tmp_path / "gates.sqlite3", clock=clock)
|
|
low = store.intake("timmy", {**candidate("a1", priority=1), "project": "p/one"}, idempotency_key="1")
|
|
oldest_high = store.intake("timmy", {**candidate("b2", priority=5), "project": "p/two"}, idempotency_key="2")
|
|
newest_high = store.intake("timmy", {**candidate("c3", priority=5), "project": "p/three"}, idempotency_key="3")
|
|
|
|
assert [item["id"] for item in store.list("timmy")["items"]] == [oldest_high["id"], newest_high["id"], low["id"]]
|
|
|
|
|
|
def test_decision_history_pages_are_bounded_stable_and_not_hidden_by_pending_work(tmp_path):
|
|
tick = iter(range(1000)).__next__
|
|
store = HumanGateStore(tmp_path / "gates.sqlite3", clock=tick)
|
|
decided = []
|
|
for index in range(3):
|
|
gate = store.intake(
|
|
"timmy",
|
|
{**candidate(f"decided-{index}"), "project": f"decided/{index}"},
|
|
idempotency_key=f"decided-{index}",
|
|
)
|
|
store.decide(
|
|
"timmy", gate["id"], expected_revision=1, decision="release",
|
|
reason="", override_reason="", checklist=checklist(),
|
|
idempotency_key=f"decision-{index}",
|
|
)
|
|
decided.append(gate["id"])
|
|
for index in range(105):
|
|
store.intake(
|
|
"timmy",
|
|
{**candidate(f"pending-{index}"), "project": f"pending/{index}"},
|
|
idempotency_key=f"pending-{index}",
|
|
)
|
|
|
|
first = store.list("timmy", state="history", limit=2)
|
|
second = store.list("timmy", state="history", limit=2, cursor=first["next_cursor"])
|
|
|
|
assert first["pending_count"] == 105
|
|
assert [item["id"] for item in first["items"]] == list(reversed(decided[1:]))
|
|
assert first["next_cursor"]
|
|
assert [item["id"] for item in second["items"]] == [decided[0]]
|
|
assert second["next_cursor"] is None
|
|
assert store.list("timmy", state="history", limit=2, cursor=first["next_cursor"]) == second
|
|
with pytest.raises(GateValidationError, match="cursor"):
|
|
store.list("alex", state="history", cursor=first["next_cursor"])
|
|
with pytest.raises(GateValidationError, match="cursor"):
|
|
store.list("timmy", state="history", cursor="not-a-cursor")
|
|
|
|
|
|
def test_decision_checks_revision_rules_and_returns_durable_idempotent_receipt(tmp_path):
|
|
path = tmp_path / "gates.sqlite3"
|
|
store = HumanGateStore(path, clock=iter([100, 101]).__next__)
|
|
gate = store.intake("timmy", candidate(), idempotency_key="run-1")
|
|
|
|
receipt = store.decide(
|
|
"timmy", gate["id"], expected_revision=gate["revision"], decision="release",
|
|
reason="", override_reason="", checklist=checklist(), idempotency_key="decision-1",
|
|
)
|
|
repeated = store.decide(
|
|
"timmy", gate["id"], expected_revision=gate["revision"], decision="release",
|
|
reason="", override_reason="", checklist=checklist(), idempotency_key="decision-1",
|
|
)
|
|
|
|
assert repeated == receipt
|
|
assert receipt["state"] == "released"
|
|
assert receipt["candidate_hash"] == "abc123"
|
|
assert HumanGateStore(path, clock=lambda: 999).receipt("timmy", receipt["receipt_id"]) == receipt
|
|
with pytest.raises(GateConflict):
|
|
store.decide("timmy", gate["id"], expected_revision=gate["revision"], decision="hold", reason="later", override_reason="", checklist=checklist(), idempotency_key="decision-2")
|
|
|
|
|
|
def test_hold_requires_reason_and_release_requires_checklist_and_override_for_unmet_checks(tmp_path):
|
|
store = HumanGateStore(tmp_path / "gates.sqlite3", clock=lambda: 100)
|
|
gate = store.intake("timmy", candidate(check_state="failure"), idempotency_key="run-1")
|
|
|
|
with pytest.raises(GateValidationError, match="Hold reason"):
|
|
store.decide("timmy", gate["id"], expected_revision=1, decision="hold", reason="", override_reason="", checklist={}, idempotency_key="d1")
|
|
with pytest.raises(GateValidationError, match="checklist"):
|
|
store.decide("timmy", gate["id"], expected_revision=1, decision="release", reason="", override_reason="needed", checklist={"exact_hash": True}, idempotency_key="d2")
|
|
with pytest.raises(GateValidationError, match="override reason"):
|
|
store.decide("timmy", gate["id"], expected_revision=1, decision="release", reason="", override_reason="", checklist=checklist(), idempotency_key="d3")
|
|
|
|
receipt = store.decide("timmy", gate["id"], expected_revision=1, decision="hold", reason="Awaiting owner", override_reason="", checklist={}, idempotency_key="d4")
|
|
assert receipt["state"] == "held"
|
|
assert receipt["checklist"] == {}
|
|
|
|
|
|
def test_same_hash_update_coalesces_one_card_and_preserves_history(tmp_path):
|
|
store = HumanGateStore(tmp_path / "gates.sqlite3", clock=iter([100, 101]).__next__)
|
|
original = store.intake("timmy", candidate(check_state="pending"), idempotency_key="run-1")
|
|
|
|
updated = store.intake("timmy", candidate(check_state="success"), idempotency_key="run-2")
|
|
|
|
assert updated["id"] == original["id"]
|
|
assert updated["revision"] == 2
|
|
assert updated["checks"][0]["state"] == "success"
|
|
assert updated["history"][-1]["action"] == "updated"
|
|
assert store.list("timmy")["pending_count"] == 1
|
|
|
|
|
|
def test_updated_checks_reopen_a_released_hash_for_review(tmp_path):
|
|
store = HumanGateStore(tmp_path / "gates.sqlite3", clock=iter([100, 101, 102]).__next__)
|
|
gate = store.intake("timmy", candidate(), idempotency_key="run-1")
|
|
store.decide(
|
|
"timmy", gate["id"], expected_revision=1, decision="release",
|
|
reason="", override_reason="", checklist=checklist(), idempotency_key="decision-1",
|
|
)
|
|
|
|
reopened = store.intake(
|
|
"timmy", candidate(check_state="failure"), idempotency_key="run-2",
|
|
)
|
|
|
|
assert reopened["state"] == "pending"
|
|
assert reopened["revision"] == 3
|
|
assert reopened["checks"][0]["state"] == "failure"
|
|
assert reopened["history"][-1]["action"] == "reopened"
|
|
assert store.list("timmy")["pending_count"] == 1
|
|
|
|
|
|
def test_same_hash_identity_facts_cannot_be_redefined(tmp_path):
|
|
store = HumanGateStore(tmp_path / "gates.sqlite3", clock=lambda: 100)
|
|
store.intake("timmy", candidate(), idempotency_key="run-1")
|
|
|
|
with pytest.raises(GateConflict, match="hash"):
|
|
store.intake("timmy", {**candidate(), "title": "Changed facts"}, idempotency_key="run-2")
|