#!/usr/bin/env python3 import importlib.util import json import tempfile import unittest from pathlib import Path SOURCE = Path(__file__).with_name("platform_infra_gitea_hook_reconciler.py") SPEC = importlib.util.spec_from_file_location("hook_reconciler", SOURCE) MODULE = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(MODULE) class HookRecoveryTest(unittest.TestCase): def setUp(self): self.temporary = tempfile.TemporaryDirectory() self.state_path = str(Path(self.temporary.name) / "hook-recovery" / "ledger.json") self.fixed_seconds = 1_783_840_000 MODULE.time.time = lambda: self.fixed_seconds self.delivery = { "id": 123, "guid": "delivery-guid", "event": "push", "status_code": 502, "delivered_at": "2026-07-12T06:31:12Z", } self.detail = { **self.delivery, "repositoryName": "pikasTech/agentrun", "ref": "refs/heads/v0.2", "requestedCommit": "f" * 40, } self.topology = { "events": ["push"], "reconcile": { "requestTimeoutMs": 15_000, "failedDeliveryRecovery": { "enabled": True, "targetId": "NC01", "statePath": self.state_path, "lookbackSeconds": 259_200, "historyPerHook": 50, "maxRedeliveriesPerCycle": 4, "cooldownMs": 300_000, "maxAttemptsPerDelivery": 3, "retryStatusCodes": [502, 503, 504], "stateRetentionSeconds": 1_209_600, }, }, } self.repository = {"repository": "pikasTech/agentrun"} self.desired_hook = { "targetId": "NC01", "url": "https://example.invalid/_unidesk/github-to-gitea/nc01", "repoKeys": ["agentrun-nc01-v02"], "branches": ["v0.2"], } self.observed_hook = {"id": 77} MODULE.list_hook_deliveries = lambda *_args: [dict(self.delivery)] MODULE.hook_delivery_detail = lambda *_args: dict(self.detail) def tearDown(self): self.temporary.cleanup() def test_write_ahead_cooldown_and_source_commit_closeout(self): ledger = MODULE.load_ledger(self.state_path) calls = [] def request(*args, **_kwargs): persisted = json.loads(Path(self.state_path).read_text()) record = next(iter(persisted["records"].values())) self.assertEqual(record["state"], "attempted") self.assertEqual(record["attempts"], 1) calls.append(args[3]) return 202, None MODULE.request = request budget = {"remaining": 4} self.assertTrue(MODULE.recover_hook_deliveries( self.topology, self.repository, self.desired_hook, self.observed_hook, "token", ledger, set(), budget, )) self.assertEqual(len(calls), 1) self.assertEqual(budget["remaining"], 3) restarted = MODULE.load_ledger(self.state_path) self.assertTrue(MODULE.recover_hook_deliveries( self.topology, self.repository, self.desired_hook, self.observed_hook, "token", restarted, set(), {"remaining": 4}, )) self.assertEqual(len(calls), 1, "restart inside cooldown must not redeliver") committed = {("agentrun-nc01-v02", "delivery-guid")} self.assertTrue(MODULE.recover_hook_deliveries( self.topology, self.repository, self.desired_hook, self.observed_hook, "token", restarted, committed, {"remaining": 4}, )) final = next(iter(MODULE.load_ledger(self.state_path)["records"].values())) self.assertEqual(final["state"], "source-committed") self.assertEqual(final["completionReason"], "source-committed") self.assertEqual(len(calls), 1, "committed source must not redeliver") def test_restart_preserves_max_attempts_and_scope_mismatch_fails_closed(self): ledger = MODULE.load_ledger(self.state_path) key = MODULE.delivery_identity("pikasTech/agentrun", 77, "delivery-guid") now_ms = int(self.fixed_seconds * 1000) ledger["records"][key] = { "repository": "pikasTech/agentrun", "targetId": "NC01", "repoKeys": ["agentrun-nc01-v02"], "hookId": 77, "deliveryId": "delivery-guid", "apiDeliveryId": "123", "state": "attempted", "attempts": 3, "cooldownMs": 300_000, "nextAtEpochMs": now_ms - 1, "nextAt": MODULE.now_iso(now_ms - 1), "updatedAtEpochMs": now_ms - 1, "updatedAt": MODULE.now_iso(now_ms - 1), } MODULE.atomic_write_ledger(self.state_path, ledger) MODULE.request = lambda *_args, **_kwargs: self.fail("maxAttempts must survive restart") self.assertTrue(MODULE.recover_hook_deliveries( self.topology, self.repository, self.desired_hook, self.observed_hook, "token", MODULE.load_ledger(self.state_path), set(), {"remaining": 4}, )) terminal = MODULE.load_ledger(self.state_path)["records"][key] self.assertEqual(terminal["state"], "terminal-exhausted") Path(self.state_path).unlink() MODULE.hook_delivery_detail = lambda *_args: {**self.detail, "ref": "refs/heads/feature"} empty = MODULE.load_ledger(self.state_path) self.assertTrue(MODULE.recover_hook_deliveries( self.topology, self.repository, self.desired_hook, self.observed_hook, "token", empty, set(), {"remaining": 4}, )) self.assertEqual(empty["records"], {}) def test_same_guid_uses_only_latest_attempt_and_terminal_never_regresses(self): ledger = MODULE.load_ledger(self.state_path) key = MODULE.delivery_identity("pikasTech/agentrun", 77, "delivery-guid") now_ms = int(self.fixed_seconds * 1000) ledger["records"][key] = { "repository": "pikasTech/agentrun", "targetId": "NC01", "repoKeys": ["agentrun-nc01-v02"], "hookId": 77, "hookUrlSha256": MODULE.url_hash(self.desired_hook["url"]), "deliveryId": "delivery-guid", "apiDeliveryId": "123", "state": "attempted", "attempts": 1, "cooldownMs": 300_000, "nextAtEpochMs": now_ms - 1, "nextAt": MODULE.now_iso(now_ms - 1), "updatedAtEpochMs": now_ms - 1, "updatedAt": MODULE.now_iso(now_ms - 1), } MODULE.atomic_write_ledger(self.state_path, ledger) old = dict(self.delivery) latest = {**self.delivery, "id": 124, "status_code": 200, "delivered_at": "2026-07-12T06:32:12Z"} MODULE.list_hook_deliveries = lambda *_args: [old, latest] MODULE.hook_delivery_detail = lambda *_args: { **(latest if str(_args[-1]) == "124" else old), "repositoryName": "pikasTech/agentrun", "ref": "refs/heads/v0.2", "requestedCommit": "f" * 40, } MODULE.request = lambda *_args, **_kwargs: self.fail("latest successful attempt must suppress old failed attempt") self.assertTrue(MODULE.recover_hook_deliveries( self.topology, self.repository, self.desired_hook, self.observed_hook, "token", MODULE.load_ledger(self.state_path), set(), {"remaining": 4}, )) complete = MODULE.load_ledger(self.state_path)["records"][key] self.assertEqual(complete["state"], "delivery-succeeded-awaiting-source") self.assertEqual(complete["completionReason"], "github-delivery-succeeded") MODULE.list_hook_deliveries = lambda *_args: [old] self.assertTrue(MODULE.recover_hook_deliveries( self.topology, self.repository, self.desired_hook, self.observed_hook, "token", MODULE.load_ledger(self.state_path), set(), {"remaining": 4}, )) self.assertEqual(MODULE.load_ledger(self.state_path)["records"][key]["state"], "delivery-succeeded-awaiting-source") committed = {("agentrun-nc01-v02", "delivery-guid")} self.assertTrue(MODULE.recover_hook_deliveries( self.topology, self.repository, self.desired_hook, self.observed_hook, "token", MODULE.load_ledger(self.state_path), committed, {"remaining": 4}, )) final = MODULE.load_ledger(self.state_path)["records"][key] self.assertEqual(final["state"], "source-committed") self.assertEqual(final["completionReason"], "source-committed") def test_removed_hook_pending_record_becomes_terminal_without_post(self): ledger = MODULE.load_ledger(self.state_path) key = MODULE.delivery_identity("pikasTech/agentrun", 66, "old-hook-guid") now_ms = int(self.fixed_seconds * 1000) ledger["records"][key] = { "repository": "pikasTech/agentrun", "targetId": "NC01", "repoKeys": ["agentrun-nc01-v02"], "hookId": 66, "hookUrlSha256": "sha256:old", "deliveryId": "old-hook-guid", "apiDeliveryId": "99", "state": "pending", "attempts": 0, "cooldownMs": 300_000, "nextAtEpochMs": now_ms, "nextAt": MODULE.now_iso(now_ms), "updatedAtEpochMs": now_ms, "updatedAt": MODULE.now_iso(now_ms), "lastStatusCode": 502, } MODULE.atomic_write_ledger(self.state_path, ledger) MODULE.request = lambda *_args, **_kwargs: self.fail("removed hook must never be redelivered") MODULE.terminalize_no_longer_desired(ledger, self.topology["reconcile"]["failedDeliveryRecovery"], "pikasTech/agentrun", self.desired_hook, 77) terminal = MODULE.load_ledger(self.state_path)["records"][key] self.assertEqual(terminal["state"], "terminal-exhausted") self.assertEqual(terminal["completionReason"], "hook-no-longer-desired") terminal["state"] = "delivery-succeeded-awaiting-source" terminal["completionReason"] = "github-delivery-succeeded" reloaded = MODULE.load_ledger(self.state_path) reloaded["records"][key] = terminal MODULE.atomic_write_ledger(self.state_path, reloaded) MODULE.close_committed_ledger_records( reloaded, self.topology["reconcile"]["failedDeliveryRecovery"], {("agentrun-nc01-v02", "old-hook-guid")}, ) committed = MODULE.load_ledger(self.state_path)["records"][key] self.assertEqual(committed["state"], "source-committed") self.assertEqual(committed["completionReason"], "source-committed") if __name__ == "__main__": unittest.main()