From 224c9ce46cfe9bc9ba8c66df6f7f0b258a9d8d09 Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Mon, 29 Jun 2026 13:59:00 -0700 Subject: [PATCH 1/6] feat(core): add Application.run() to invoke registered commands programmatically Expose a public run() method on the core Application that executes an already-registered console command by name from Python, without going through sys.argv, and returns the command exit code. app.run("db:migrate") app.run("db:seed", "--force") app.run("db:seed", ["--force"]) Reuses the existing ConsoleApplication wiring (mirrors handle_command), dispatching via Cleo with a constructed StringInput and auto_exit disabled so the code is returned rather than terminating the host process. --- .../src/fastapi_startkit/application.py | 23 ++++++ fastapi_startkit/tests/console/__init__.py | 0 .../tests/console/test_application_run.py | 72 +++++++++++++++++++ 3 files changed, 95 insertions(+) create mode 100644 fastapi_startkit/tests/console/__init__.py create mode 100644 fastapi_startkit/tests/console/test_application_run.py diff --git a/fastapi_startkit/src/fastapi_startkit/application.py b/fastapi_startkit/src/fastapi_startkit/application.py index 175aa753..4ce5ecaa 100644 --- a/fastapi_startkit/src/fastapi_startkit/application.py +++ b/fastapi_startkit/src/fastapi_startkit/application.py @@ -1,4 +1,5 @@ import os +import shlex from fastapi_startkit.providers.app_provider import AppProvider from pathlib import Path from typing import TYPE_CHECKING, Optional @@ -225,3 +226,25 @@ def handle_command(self): from fastapi_startkit.console import ConsoleApplication ConsoleApplication(self).handle() + + def run(self, command: str, args: "str | list[str] | None" = None) -> int: + """Run a registered console command by name from Python and return its exit code. + + Unlike :meth:`handle_command`, this does not read ``sys.argv``: the command and + its arguments are passed explicitly, e.g. ``app.run("db:migrate")`` or + ``app.run("db:seed", "--force")`` / ``app.run("db:seed", ["--force"])``. + """ + from cleo.io.inputs.string_input import StringInput + + from fastapi_startkit.console import ConsoleApplication + + if isinstance(args, (list, tuple)): + args = shlex.join(str(arg) for arg in args) + + command_line = f"{command} {args}".strip() if args else command + + console = ConsoleApplication(self) + # Return the exit code instead of terminating the host process. + console.auto_exits(False) + + return console.run(StringInput(command_line)) diff --git a/fastapi_startkit/tests/console/__init__.py b/fastapi_startkit/tests/console/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/fastapi_startkit/tests/console/test_application_run.py b/fastapi_startkit/tests/console/test_application_run.py new file mode 100644 index 00000000..af97e210 --- /dev/null +++ b/fastapi_startkit/tests/console/test_application_run.py @@ -0,0 +1,72 @@ +"""Tests for Application.run() — invoking a registered console command from Python. + +A dummy command is registered on a fresh Application so the programmatic API can +resolve and execute it without touching ``sys.argv``. +""" + +from __future__ import annotations + +import pytest +from cleo.helpers import argument, option + +from fastapi_startkit.application import Application +from fastapi_startkit.console import Command +from fastapi_startkit.container import Container + + +class DummyCommand(Command): + name = "dummy:do" + description = "Records what it received and returns a configurable exit code." + + arguments = [argument("name", optional=True)] + options = [option("force", "f", description="Force flag.", flag=True)] + + exit_code = 0 + received: dict = {} + + def handle(self) -> int: + type(self).received = { + "name": self.argument("name"), + "force": self.option("force"), + } + return type(self).exit_code + + +@pytest.fixture +def app(): + DummyCommand.exit_code = 0 + DummyCommand.received = {} + + previous = Container._instance + application = Application(env="testing") + application.add_commands([DummyCommand]) + yield application + Container.set_instance(previous) + + +def test_run_returns_zero_exit_code(app): + assert app.run("dummy:do") == 0 + + +def test_run_propagates_non_zero_exit_code(app): + DummyCommand.exit_code = 3 + + assert app.run("dummy:do") == 3 + + +def test_run_without_args(app): + app.run("dummy:do") + + assert DummyCommand.received == {"name": None, "force": False} + + +def test_run_forwards_string_args(app): + app.run("dummy:do", "hello --force") + + assert DummyCommand.received == {"name": "hello", "force": True} + + +def test_run_forwards_list_args(app): + app.run("dummy:do", ["hello", "--force"]) + + assert DummyCommand.received == {"name": "hello", "force": True} From dec32216f114c1a4e3364fc616ea7858ff3c7d65 Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Mon, 29 Jun 2026 14:24:17 -0700 Subject: [PATCH 2/6] test(core): expand Application.run() coverage Add cases for exit-code type, unknown command, None/empty args, short options, list/tuple args, space-preserving and stringified values, and repeat invocation. --- .../tests/console/test_application_run.py | 97 ++++++++++++++++--- 1 file changed, 85 insertions(+), 12 deletions(-) diff --git a/fastapi_startkit/tests/console/test_application_run.py b/fastapi_startkit/tests/console/test_application_run.py index af97e210..90472ce9 100644 --- a/fastapi_startkit/tests/console/test_application_run.py +++ b/fastapi_startkit/tests/console/test_application_run.py @@ -1,7 +1,9 @@ -"""Tests for Application.run() — invoking a registered console command from Python. +"""Unit tests for ``Application.run()``. -A dummy command is registered on a fresh Application so the programmatic API can -resolve and execute it without touching ``sys.argv``. +``Application.run()`` invokes an already-registered console command by name from +Python code — without going through ``sys.argv`` — and returns the command's +exit code. A dummy command is registered on a fresh ``Application`` so the +programmatic API has something concrete to resolve and execute. """ from __future__ import annotations @@ -34,6 +36,12 @@ def handle(self) -> int: @pytest.fixture def app(): + """A booted Application with DummyCommand registered. + + The container is a process-wide singleton, so the previous instance is + restored on teardown to keep these tests isolated from the rest of the + suite. + """ DummyCommand.exit_code = 0 DummyCommand.received = {} @@ -44,29 +52,94 @@ def app(): Container.set_instance(previous) -def test_run_returns_zero_exit_code(app): - assert app.run("dummy:do") == 0 +# --------------------------------------------------------------------------- +# Exit code +# --------------------------------------------------------------------------- + + +def test_returns_zero_exit_code_as_int(app): + result = app.run("dummy:do") + + assert result == 0 + assert isinstance(result, int) -def test_run_propagates_non_zero_exit_code(app): +def test_propagates_non_zero_exit_code(app): DummyCommand.exit_code = 3 assert app.run("dummy:do") == 3 -def test_run_without_args(app): +def test_unknown_command_returns_error_code(app): + # Cleo reports the error through the application instead of raising; the + # call still returns a non-zero exit code rather than terminating the host. + assert app.run("does:not-exist") == 1 + + +# --------------------------------------------------------------------------- +# Argument handling +# --------------------------------------------------------------------------- + + +def test_runs_without_args(app): app.run("dummy:do") assert DummyCommand.received == {"name": None, "force": False} -def test_run_forwards_string_args(app): - app.run("dummy:do", "hello --force") +def test_none_args_is_equivalent_to_no_args(app): + app.run("dummy:do", None) - assert DummyCommand.received == {"name": "hello", "force": True} + assert DummyCommand.received == {"name": None, "force": False} + + +def test_empty_string_args_is_equivalent_to_no_args(app): + app.run("dummy:do", "") + + assert DummyCommand.received == {"name": None, "force": False} -def test_run_forwards_list_args(app): - app.run("dummy:do", ["hello", "--force"]) +@pytest.mark.parametrize( + "args", + [ + "hello --force", + "hello -f", + ["hello", "--force"], + ("hello", "-f"), + ], +) +def test_forwards_string_and_sequence_args(app, args): + app.run("dummy:do", args) assert DummyCommand.received == {"name": "hello", "force": True} + + +def test_forwards_positional_arg_only(app): + app.run("dummy:do", "hello") + + assert DummyCommand.received == {"name": "hello", "force": False} + + +def test_list_arg_preserves_value_with_spaces(app): + app.run("dummy:do", ["hello world"]) + + assert DummyCommand.received["name"] == "hello world" + + +def test_non_string_list_args_are_stringified(app): + app.run("dummy:do", [123]) + + assert DummyCommand.received["name"] == "123" + + +# --------------------------------------------------------------------------- +# Re-entrancy +# --------------------------------------------------------------------------- + + +def test_can_be_invoked_repeatedly(app): + assert app.run("dummy:do", "first") == 0 + assert DummyCommand.received["name"] == "first" + + assert app.run("dummy:do", "second --force") == 0 + assert DummyCommand.received == {"name": "second", "force": True} From 92afff48a1ecc3558dabb0ee64273f89ad9e28a6 Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Mon, 29 Jun 2026 14:25:03 -0700 Subject: [PATCH 3/6] test(core): rewrite Application.run() tests with unittest.TestCase --- .../tests/console/test_application_run.py | 139 ++++++++---------- 1 file changed, 58 insertions(+), 81 deletions(-) diff --git a/fastapi_startkit/tests/console/test_application_run.py b/fastapi_startkit/tests/console/test_application_run.py index 90472ce9..2a90e84c 100644 --- a/fastapi_startkit/tests/console/test_application_run.py +++ b/fastapi_startkit/tests/console/test_application_run.py @@ -8,7 +8,8 @@ from __future__ import annotations -import pytest +import unittest + from cleo.helpers import argument, option from fastapi_startkit.application import Application @@ -34,112 +35,88 @@ def handle(self) -> int: return type(self).exit_code -@pytest.fixture -def app(): - """A booted Application with DummyCommand registered. - - The container is a process-wide singleton, so the previous instance is - restored on teardown to keep these tests isolated from the rest of the - suite. - """ - DummyCommand.exit_code = 0 - DummyCommand.received = {} - - previous = Container._instance - application = Application(env="testing") - application.add_commands([DummyCommand]) - yield application - Container.set_instance(previous) - - -# --------------------------------------------------------------------------- -# Exit code -# --------------------------------------------------------------------------- - - -def test_returns_zero_exit_code_as_int(app): - result = app.run("dummy:do") - - assert result == 0 - assert isinstance(result, int) - - -def test_propagates_non_zero_exit_code(app): - DummyCommand.exit_code = 3 - - assert app.run("dummy:do") == 3 - +class ApplicationRunTest(unittest.TestCase): + def setUp(self) -> None: + DummyCommand.exit_code = 0 + DummyCommand.received = {} -def test_unknown_command_returns_error_code(app): - # Cleo reports the error through the application instead of raising; the - # call still returns a non-zero exit code rather than terminating the host. - assert app.run("does:not-exist") == 1 + # The container is a process-wide singleton; remember the current one + # so it can be restored in tearDown and keep tests isolated. + self._previous = Container._instance + self.app = Application(env="testing") + self.app.add_commands([DummyCommand]) + def tearDown(self) -> None: + Container.set_instance(self._previous) -# --------------------------------------------------------------------------- -# Argument handling -# --------------------------------------------------------------------------- + # -- exit code --------------------------------------------------------- + def test_returns_zero_exit_code_as_int(self): + result = self.app.run("dummy:do") -def test_runs_without_args(app): - app.run("dummy:do") + self.assertEqual(result, 0) + self.assertIsInstance(result, int) - assert DummyCommand.received == {"name": None, "force": False} + def test_propagates_non_zero_exit_code(self): + DummyCommand.exit_code = 3 + self.assertEqual(self.app.run("dummy:do"), 3) -def test_none_args_is_equivalent_to_no_args(app): - app.run("dummy:do", None) + def test_unknown_command_returns_error_code(self): + # Cleo reports the error through the application instead of raising; the + # call still returns a non-zero exit code rather than terminating the host. + self.assertEqual(self.app.run("does:not-exist"), 1) - assert DummyCommand.received == {"name": None, "force": False} + # -- argument handling ------------------------------------------------- + def test_runs_without_args(self): + self.app.run("dummy:do") -def test_empty_string_args_is_equivalent_to_no_args(app): - app.run("dummy:do", "") + self.assertEqual(DummyCommand.received, {"name": None, "force": False}) - assert DummyCommand.received == {"name": None, "force": False} + def test_none_args_is_equivalent_to_no_args(self): + self.app.run("dummy:do", None) + self.assertEqual(DummyCommand.received, {"name": None, "force": False}) -@pytest.mark.parametrize( - "args", - [ - "hello --force", - "hello -f", - ["hello", "--force"], - ("hello", "-f"), - ], -) -def test_forwards_string_and_sequence_args(app, args): - app.run("dummy:do", args) + def test_empty_string_args_is_equivalent_to_no_args(self): + self.app.run("dummy:do", "") - assert DummyCommand.received == {"name": "hello", "force": True} + self.assertEqual(DummyCommand.received, {"name": None, "force": False}) + def test_forwards_string_and_sequence_args(self): + for args in ("hello --force", "hello -f", ["hello", "--force"], ("hello", "-f")): + with self.subTest(args=args): + DummyCommand.received = {} -def test_forwards_positional_arg_only(app): - app.run("dummy:do", "hello") + self.app.run("dummy:do", args) - assert DummyCommand.received == {"name": "hello", "force": False} + self.assertEqual(DummyCommand.received, {"name": "hello", "force": True}) + def test_forwards_positional_arg_only(self): + self.app.run("dummy:do", "hello") -def test_list_arg_preserves_value_with_spaces(app): - app.run("dummy:do", ["hello world"]) + self.assertEqual(DummyCommand.received, {"name": "hello", "force": False}) - assert DummyCommand.received["name"] == "hello world" + def test_list_arg_preserves_value_with_spaces(self): + self.app.run("dummy:do", ["hello world"]) + self.assertEqual(DummyCommand.received["name"], "hello world") -def test_non_string_list_args_are_stringified(app): - app.run("dummy:do", [123]) + def test_non_string_list_args_are_stringified(self): + self.app.run("dummy:do", [123]) - assert DummyCommand.received["name"] == "123" + self.assertEqual(DummyCommand.received["name"], "123") + # -- re-entrancy ------------------------------------------------------- -# --------------------------------------------------------------------------- -# Re-entrancy -# --------------------------------------------------------------------------- + def test_can_be_invoked_repeatedly(self): + self.assertEqual(self.app.run("dummy:do", "first"), 0) + self.assertEqual(DummyCommand.received["name"], "first") + self.assertEqual(self.app.run("dummy:do", "second --force"), 0) + self.assertEqual(DummyCommand.received, {"name": "second", "force": True}) -def test_can_be_invoked_repeatedly(app): - assert app.run("dummy:do", "first") == 0 - assert DummyCommand.received["name"] == "first" - assert app.run("dummy:do", "second --force") == 0 - assert DummyCommand.received == {"name": "second", "force": True} +if __name__ == "__main__": + unittest.main() From 5a7f57474284662c7940e71cb90fa30cebb21737 Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Mon, 29 Jun 2026 14:25:36 -0700 Subject: [PATCH 4/6] test(core): drop comments and __main__ block from Application.run() tests --- .../tests/console/test_application_run.py | 22 ------------------- 1 file changed, 22 deletions(-) diff --git a/fastapi_startkit/tests/console/test_application_run.py b/fastapi_startkit/tests/console/test_application_run.py index 2a90e84c..3d640d1c 100644 --- a/fastapi_startkit/tests/console/test_application_run.py +++ b/fastapi_startkit/tests/console/test_application_run.py @@ -1,11 +1,3 @@ -"""Unit tests for ``Application.run()``. - -``Application.run()`` invokes an already-registered console command by name from -Python code — without going through ``sys.argv`` — and returns the command's -exit code. A dummy command is registered on a fresh ``Application`` so the -programmatic API has something concrete to resolve and execute. -""" - from __future__ import annotations import unittest @@ -40,8 +32,6 @@ def setUp(self) -> None: DummyCommand.exit_code = 0 DummyCommand.received = {} - # The container is a process-wide singleton; remember the current one - # so it can be restored in tearDown and keep tests isolated. self._previous = Container._instance self.app = Application(env="testing") self.app.add_commands([DummyCommand]) @@ -49,8 +39,6 @@ def setUp(self) -> None: def tearDown(self) -> None: Container.set_instance(self._previous) - # -- exit code --------------------------------------------------------- - def test_returns_zero_exit_code_as_int(self): result = self.app.run("dummy:do") @@ -63,12 +51,8 @@ def test_propagates_non_zero_exit_code(self): self.assertEqual(self.app.run("dummy:do"), 3) def test_unknown_command_returns_error_code(self): - # Cleo reports the error through the application instead of raising; the - # call still returns a non-zero exit code rather than terminating the host. self.assertEqual(self.app.run("does:not-exist"), 1) - # -- argument handling ------------------------------------------------- - def test_runs_without_args(self): self.app.run("dummy:do") @@ -108,15 +92,9 @@ def test_non_string_list_args_are_stringified(self): self.assertEqual(DummyCommand.received["name"], "123") - # -- re-entrancy ------------------------------------------------------- - def test_can_be_invoked_repeatedly(self): self.assertEqual(self.app.run("dummy:do", "first"), 0) self.assertEqual(DummyCommand.received["name"], "first") self.assertEqual(self.app.run("dummy:do", "second --force"), 0) self.assertEqual(DummyCommand.received, {"name": "second", "force": True}) - - -if __name__ == "__main__": - unittest.main() From 02c794a05a24298e55a967264ecac8d3d54b3dc7 Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Mon, 29 Jun 2026 14:26:09 -0700 Subject: [PATCH 5/6] refactor(core): drop docstring and comment from Application.run() --- fastapi_startkit/src/fastapi_startkit/application.py | 7 ------- 1 file changed, 7 deletions(-) diff --git a/fastapi_startkit/src/fastapi_startkit/application.py b/fastapi_startkit/src/fastapi_startkit/application.py index 4ce5ecaa..0778e067 100644 --- a/fastapi_startkit/src/fastapi_startkit/application.py +++ b/fastapi_startkit/src/fastapi_startkit/application.py @@ -228,12 +228,6 @@ def handle_command(self): ConsoleApplication(self).handle() def run(self, command: str, args: "str | list[str] | None" = None) -> int: - """Run a registered console command by name from Python and return its exit code. - - Unlike :meth:`handle_command`, this does not read ``sys.argv``: the command and - its arguments are passed explicitly, e.g. ``app.run("db:migrate")`` or - ``app.run("db:seed", "--force")`` / ``app.run("db:seed", ["--force"])``. - """ from cleo.io.inputs.string_input import StringInput from fastapi_startkit.console import ConsoleApplication @@ -244,7 +238,6 @@ def run(self, command: str, args: "str | list[str] | None" = None) -> int: command_line = f"{command} {args}".strip() if args else command console = ConsoleApplication(self) - # Return the exit code instead of terminating the host process. console.auto_exits(False) return console.run(StringInput(command_line)) From a7f6f49f557cd3d433fdf41633eadcb7244f278a Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Mon, 29 Jun 2026 14:50:40 -0700 Subject: [PATCH 6/6] fix(core): raise on unknown command in Application.run() Resolve the command via ConsoleApplication.find() before dispatch so an unknown name raises CleoCommandNotFoundError that callers can catch, instead of being swallowed into exit code 1 by Cleo's non-auto-exit run loop. Known-command behavior (returned exit code, no process exit) is unchanged. --- fastapi_startkit/src/fastapi_startkit/application.py | 2 ++ fastapi_startkit/tests/console/test_application_run.py | 6 ++++-- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/fastapi_startkit/src/fastapi_startkit/application.py b/fastapi_startkit/src/fastapi_startkit/application.py index 0778e067..44932dc2 100644 --- a/fastapi_startkit/src/fastapi_startkit/application.py +++ b/fastapi_startkit/src/fastapi_startkit/application.py @@ -240,4 +240,6 @@ def run(self, command: str, args: "str | list[str] | None" = None) -> int: console = ConsoleApplication(self) console.auto_exits(False) + console.find(command) + return console.run(StringInput(command_line)) diff --git a/fastapi_startkit/tests/console/test_application_run.py b/fastapi_startkit/tests/console/test_application_run.py index 3d640d1c..3c5321d0 100644 --- a/fastapi_startkit/tests/console/test_application_run.py +++ b/fastapi_startkit/tests/console/test_application_run.py @@ -2,6 +2,7 @@ import unittest +from cleo.exceptions import CleoCommandNotFoundError from cleo.helpers import argument, option from fastapi_startkit.application import Application @@ -50,8 +51,9 @@ def test_propagates_non_zero_exit_code(self): self.assertEqual(self.app.run("dummy:do"), 3) - def test_unknown_command_returns_error_code(self): - self.assertEqual(self.app.run("does:not-exist"), 1) + def test_unknown_command_raises(self): + with self.assertRaises(CleoCommandNotFoundError): + self.app.run("does:not:exist") def test_runs_without_args(self): self.app.run("dummy:do")