diff --git a/bot_bottle/cli/__init__.py b/bot_bottle/cli/__init__.py index 1dcb525..ae421fc 100644 --- a/bot_bottle/cli/__init__.py +++ b/bot_bottle/cli/__init__.py @@ -38,6 +38,13 @@ COMMANDS = { "supervise": cmd_supervise, } +# Commands that manage host prerequisites (or are otherwise store-free) and +# must run before — or without — a migrated DB. `backend` provisions/probes +# the host (TAP pool, /dev/kvm, firecracker) and never opens the store, so +# gating it on the schema breaks preflight on a fresh CI runner where stdin +# isn't a TTY and the migration prompt can't be answered. +NO_MIGRATION_COMMANDS = frozenset({"backend"}) + def usage() -> None: sys.stderr.write(f"usage: {PROG} [args...]\n\n") @@ -80,7 +87,7 @@ def main(argv: list[str] | None = None) -> int: usage() die(f"unknown command: {command}") mgr = StoreManager.instance() - if not mgr.is_migrated(): + if command not in NO_MIGRATION_COMMANDS and not mgr.is_migrated(): sys.stderr.write("bot-bottle: database schema is out of date\n") sys.stderr.write("Migrate now? [y/N] ") sys.stderr.flush() diff --git a/tests/unit/test_cli_dispatch.py b/tests/unit/test_cli_dispatch.py index 64cf655..2981f73 100644 --- a/tests/unit/test_cli_dispatch.py +++ b/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()