Refresh public update status after completed transaction recovery

2026-10-13 21:46:45 +03:00
parent 653388f0d7
commit 8512d1d10b
3 changed files with 40 additions and 2 deletions
@@ -429,3 +429,36 @@ CALL
        updater.recover()
        self.assertEqual(updater.status()['highest_sequence'], 1)
        self.assertEqual(self.target.read_bytes(), b'old runtime\n')
+
+    def test_fresh_recovery_reconciles_public_status_after_terminal_journal_before_publish(self):
+        self.package()
+        updater = self.updater()
+        from unittest.mock import patch
+        original_publish = updater.publish
+        def interrupted_publish():
+            if not updater.pending():
+                raise OSError('power lost after terminal journal before public status')
+            return original_publish()
+        for operation in ('apply', 'rollback'):
+            with self.subTest(operation=operation):
+                with patch.object(updater, 'publish', side_effect=interrupted_publish):
+                    with self.assertRaisesRegex(OSError, 'terminal journal'):
+                        if operation == 'apply':
+                            updater.apply(self.artifacts)
+                        else:
+                            updater.rollback(updater.state()['current']['transaction'])
+                authoritative = updater.status()
+                self.assertEqual(authoritative['pending'], [])
+                stale = json.loads(updater.public_status.read_text())
+                self.assertNotEqual(stale, authoritative)
+                result = subprocess.run([sys.executable, '-I', str(ENGINE.parent.parent / 'bin/hydra-update-recover'),
+                                         '--root', str(self.root), '--public-key', str(self.pub)], capture_output=True, text=True)
+                self.assertEqual(result.returncode, 0, result.stderr)
+                self.assertEqual(json.loads(result.stdout), [])
+                self.assertEqual(json.loads(updater.public_status.read_text()), authoritative)
+                self.assertEqual(self.target.read_bytes(), b'new runtime\n' if operation == 'apply' else b'old runtime\n')
+                updater.base.chmod(0)
+                try:
+                    self.assertEqual(updater.status(), authoritative)
+                finally:
+                    updater.base.chmod(0o700)