From 280e73e1f2caede4cb4bd7e90778ec999e9d4aae Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Sat, 5 Sep 2026 15:12:05 +0100 Subject: [PATCH] test(delivery): bind recovery receipts to successful submissions Exercise all three reconciliation recovery paths through the mocked transport for accepted and rejected POSTs. Assert rejected submissions raise without emitting a success receipt. Validation: 30 reconciliation tests and 46 release promotion policy tests pass. No remote mutations performed. Change-source: pulse-maintainer --- .../reconcile_release_convergence_test.py | 29 +++++++++++++++++++ 1 file changed, 29 insertions(+) diff --git a/scripts/release_control/reconcile_release_convergence_test.py b/scripts/release_control/reconcile_release_convergence_test.py index 56f6aadfa..1b2653608 100644 --- a/scripts/release_control/reconcile_release_convergence_test.py +++ b/scripts/release_control/reconcile_release_convergence_test.py @@ -461,6 +461,35 @@ class CredentialContainmentTests(unittest.TestCase): class ReconciliationTests(unittest.TestCase): + def test_recovery_receipts_require_successful_submission(self): + cases = [ + ({}, "Dispatched fresh convergence controls"), + ({"current_controls": True}, "Re-ran current-control convergence"), + ({"committed": False}, "Renewed pre-commit convergence owner"), + ] + for options, receipt in cases: + for code in (0, 1): + with self.subTest(options=options, code=code): + github = FakeGitHub(**options) + transport = subject.GitHub(github.repository, "gh") + github.post = transport.post + result = subprocess.CompletedProcess( + [], code, stdout="", stderr="rejected" if code else "" + ) + output = io.StringIO() + with patch.object(subject.subprocess, "run", return_value=result) as command, contextlib.redirect_stdout(output): + if code: + with self.assertRaises(subject.ReconciliationError): + subject.reconcile(github, github.run_id, 5) + else: + subject.reconcile(github, github.run_id, 5) + command.assert_called_once() + if code: + self.assertNotIn(receipt, output.getvalue()) + else: + self.assertIn(receipt, output.getvalue()) + self.assertNotIn("DRY RUN", output.getvalue()) + def test_post_reports_submission_only_after_success(self): for payload in (None, {"ref": "main"}): for code in (0, 1):