feat(core): add Application.run() to invoke registered commands programmatically

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,20 @@ 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:
+        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)
+        console.auto_exits(False)
+
+        console.find(command)
+
+        return console.run(StringInput(command_line))

fastapi_startkit/tests/console/test_application_run.py

@@ -0,0 +1,102 @@
+from __future__ import annotations
+
+import unittest
+
+from cleo.exceptions import CleoCommandNotFoundError
+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
+
+
+class ApplicationRunTest(unittest.TestCase):
+    def setUp(self) -> None:
+        DummyCommand.exit_code = 0
+        DummyCommand.received = {}
+
+        self._previous = Container._instance
+        self.app = Application(env="testing")
+        self.app.add_commands([DummyCommand])
+
+    def tearDown(self) -> None:
+        Container.set_instance(self._previous)
+
+    def test_returns_zero_exit_code_as_int(self):
+        result = self.app.run("dummy:do")
+
+        self.assertEqual(result, 0)
+        self.assertIsInstance(result, int)
+
+    def test_propagates_non_zero_exit_code(self):
+        DummyCommand.exit_code = 3
+
+        self.assertEqual(self.app.run("dummy:do"), 3)
+
+    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")
+
+        self.assertEqual(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})
+
+    def test_empty_string_args_is_equivalent_to_no_args(self):
+        self.app.run("dummy:do", "")
+
+        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 = {}
+
+                self.app.run("dummy:do", args)
+
+                self.assertEqual(DummyCommand.received, {"name": "hello", "force": True})
+
+    def test_forwards_positional_arg_only(self):
+        self.app.run("dummy:do", "hello")
+
+        self.assertEqual(DummyCommand.received, {"name": "hello", "force": False})
+
+    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(self):
+        self.app.run("dummy:do", [123])
+
+        self.assertEqual(DummyCommand.received["name"], "123")
+
+    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})