Merge remote-tracking branch 'origin/main' into ci-kvm-runner

tests/unit/test_cli_dispatch.py

@@ -95,5 +95,60 @@ class TestMainDispatch(unittest.TestCase):
             self.assertEqual(130, main(["x"]))
 
 
+class TestMigrationGate(unittest.TestCase):
+    """The dispatcher's schema-migration gate (cli/__init__.py)."""
+
+    def setUp(self) -> None:
+        # Force the "schema out of date" branch for every test here.
+        patcher = patch.object(StoreManager, "is_migrated", return_value=False)
+        patcher.start()
+        self.addCleanup(patcher.stop)
+        self.addCleanup(StoreManager.reset)
+
+    def test_store_command_blocks_when_stdin_cannot_confirm(self) -> None:
+        # Non-TTY stdin at EOF (as in CI): the [y/N] prompt reads "" and the
+        # command is refused rather than migrating silently.
+        ran: list[bool] = []
+
+        def handler(_rest: list[str]) -> int:
+            ran.append(True)
+            return 0
+
+        with patch.dict(climod.COMMANDS, {"list": handler}), \
+                patch("sys.stdin", io.StringIO("")), \
+                patch("sys.stderr", io.StringIO()):
+            self.assertEqual(1, main(["list"]))
+        self.assertEqual([], ran, "gated command must not dispatch")
+
+    def test_backend_command_skips_gate(self) -> None:
+        # `backend` provisions/probes the host and never opens the store, so
+        # it must run even on an unmigrated DB with unanswerable stdin.
+        ran: list[bool] = []
+
+        def handler(_rest: list[str]) -> int:
+            ran.append(True)
+            return 0
+
+        with patch.dict(climod.COMMANDS, {"backend": handler}), \
+                patch("sys.stdin", io.StringIO("")), \
+                patch("sys.stderr", io.StringIO()):
+            self.assertEqual(0, main(["backend", "status"]))
+        self.assertEqual([True], ran, "exempt command must dispatch")
+
+    def test_store_command_migrates_on_confirmation(self) -> None:
+        migrated: list[bool] = []
+
+        def handler(_rest: list[str]) -> int:
+            return 0
+
+        with patch.dict(climod.COMMANDS, {"list": handler}), \
+                patch.object(StoreManager, "migrate",
+                             side_effect=lambda: migrated.append(True)), \
+                patch("sys.stdin", io.StringIO("y\n")), \
+                patch("sys.stderr", io.StringIO()):
+            self.assertEqual(0, main(["list"]))
+        self.assertEqual([True], migrated, "confirmed gate must migrate")
+
+
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_git_gate.py

@@ -168,16 +168,18 @@ class TestHookRender(unittest.TestCase):
         # Stdin is buffered to a tempfile so both phases can re-read.
         self.assertIn("refs_file=$(mktemp)", hook)
 
-    def test_new_ref_scan_scoped_to_incoming_commits(self):
-        # A new branch (old=all-zeros) must scan only commits new to the
-        # gate, not the full ancestry — otherwise historical findings
-        # block every new-branch push (PRD 0028 / issue #106).
+    def test_scan_scoped_to_incoming_commits(self):
+        # Every non-delete push scans only commits new to the gate, not
+        # the full ancestry and not the `$old..$new` delta — otherwise
+        # historical fixtures block new-branch pushes (PRD 0028 / #106)
+        # and a rebase/force-push onto an advanced main drags in main's
+        # history incl. the sandbox-escape fixtures (#346).
         hook = git_gate_render_hook()
         self.assertIn('log_opts="$new --not --all"', hook)
-        # The old over-broad full-ancestry range must be gone.
+        # Neither the full-ancestry range nor the ancestry-blind delta
+        # range may survive.
         self.assertNotIn('log_opts="$new"', hook)
-        # Existing-branch delta scan is unchanged.
-        self.assertIn('log_opts="$old..$new"', hook)
+        self.assertNotIn('log_opts="$old..$new"', hook)
 
     def test_forward_ssh_is_non_interactive_and_bounded(self):
         # No prompt (BatchMode) and a connect timeout, so an unreachable