diff --git a/fastapi_startkit/src/fastapi_startkit/console/run_command.py b/fastapi_startkit/src/fastapi_startkit/console/run_command.py new file mode 100644 index 00000000..4b3a9c5c --- /dev/null +++ b/fastapi_startkit/src/fastapi_startkit/console/run_command.py @@ -0,0 +1,35 @@ +import shlex + +from cleo.helpers import argument + +from fastapi_startkit.console.command import Command + + +class RunCommand(Command): + name = "run" + description = "Run another registered console command programmatically." + + arguments = [ + argument( + "command_name", + description="The name of the command to run (e.g. db:migrate).", + ), + argument( + "args", + description="Arguments to forward to the command. Prefix forwarded options with -- " + "(e.g. run db:seed -- --force).", + optional=True, + multiple=True, + ), + ] + + def handle(self) -> int: + command = self.argument("command_name") + forwarded = self.argument("args") or [] + + # Cleo's `call` re-tokenizes the args string and binds it against the + # target command merged with the application definition, whose first + # positional is the command name. The name must therefore lead the + # string, otherwise the first forwarded argument is swallowed by it. + # shlex.join keeps tokens containing spaces intact. + return self.call(command, shlex.join([command, *forwarded])) diff --git a/fastapi_startkit/src/fastapi_startkit/providers/app_provider.py b/fastapi_startkit/src/fastapi_startkit/providers/app_provider.py index f6b984ce..94083a96 100644 --- a/fastapi_startkit/src/fastapi_startkit/providers/app_provider.py +++ b/fastapi_startkit/src/fastapi_startkit/providers/app_provider.py @@ -1,4 +1,5 @@ from fastapi_startkit.console.publish_command import PublishCommand +from fastapi_startkit.console.run_command import RunCommand from fastapi_startkit.providers import Provider @@ -7,4 +8,4 @@ def register(self) -> None: pass def boot(self) -> None: - self.commands([PublishCommand]) + self.commands([PublishCommand, RunCommand]) 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_run_command.py b/fastapi_startkit/tests/console/test_run_command.py new file mode 100644 index 00000000..e5c4443c --- /dev/null +++ b/fastapi_startkit/tests/console/test_run_command.py @@ -0,0 +1,80 @@ +"""Tests for RunCommand. + +RunCommand is a thin wrapper that invokes another registered command via +Cleo's ``Command.call`` and propagates its exit code. A dummy command is +registered alongside RunCommand on a real Cleo application so that +``self.call`` can resolve it. +""" + +from __future__ import annotations + +from cleo.application import Application +from cleo.commands.command import Command as CleoCommand +from cleo.helpers import argument, option +from cleo.testers.command_tester import CommandTester + +from fastapi_startkit.console.run_command import RunCommand + + +class _DummyCommand(CleoCommand): + 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 + + +def _tester(exit_code: int = 0) -> CommandTester: + _DummyCommand.exit_code = exit_code + _DummyCommand.received = {} + + app = Application() + app.add(_DummyCommand()) + app.add(RunCommand()) + + return CommandTester(app.find("run")) + + +def test_forwards_positional_argument() -> None: + tester = _tester() + + status = tester.execute("dummy:do hello") + + assert status == 0 + assert _DummyCommand.received["name"] == "hello" + + +def test_propagates_non_zero_exit_code() -> None: + tester = _tester(exit_code=5) + + status = tester.execute("dummy:do hello") + + assert status == 5 + + +def test_forwards_options_after_separator() -> None: + tester = _tester() + + status = tester.execute("dummy:do hello -- --force") + + assert status == 0 + assert _DummyCommand.received["force"] is True + + +def test_runs_command_without_extra_args() -> None: + tester = _tester() + + status = tester.execute("dummy:do") + + assert status == 0 + assert _DummyCommand.received["name"] is None