test(fastapi): ServeCommand tests via Cleo CommandTester

fastapi_startkit/src/fastapi_startkit/environment/__init__.py

@@ -1 +1 @@
-from .environment import env
+from .environment import env, value

fastapi_startkit/src/fastapi_startkit/environment/environment.py

@@ -87,3 +87,21 @@ def env(value, default="", cast=True):
         return True
     else:
         return env_var
+
+
+def value(env_var, default=""):
+    if env_var == "":
+        env_var = default
+
+    if isinstance(env_var, bool):
+        return env_var
+    elif env_var is None:
+        return None
+    elif isinstance(env_var, int) or (isinstance(env_var, str) and env_var.isnumeric()):
+        return int(env_var)
+    elif env_var in ("false", "False"):
+        return False
+    elif env_var in ("true", "True"):
+        return True
+    else:
+        return env_var

fastapi_startkit/src/fastapi_startkit/fastapi/commands/serve_command.py

@@ -1,5 +1,8 @@
-from fastapi_startkit.console.command import Command
 from cleo.helpers import option
+from fastapi_startkit import Config
+from fastapi_startkit.console.command import Command
+from fastapi_startkit.environment import value as cast_value
+from fastapi_startkit.support import Uriable, Uri
 
 
 class ServeCommand(Command):
@@ -37,36 +40,40 @@ class ServeCommand(Command):
         ),
     ]
 
+    def resolve_option(self, key: str, default: str | int | None = None):
+        value = self.option(key) or Config.get(f"fastapi.{key}", default)
+
+        return cast_value(value)
+
+    def resolve_url(self) -> Uriable:
+        host = self.resolve_option("host", "127.0.0.1")
+        port = self.resolve_option("port", 8000)
+
+        return Uri.of(Config.get("fastapi.app_url", "http://127.0.0.1:8000")).with_host(host).with_port(port)
+
     def handle(self):
         import uvicorn
+
         from fastapi_startkit import Config
         from fastapi_startkit.container import Container
 
         # Resolve server settings: CLI flag > fastapi config > uvicorn default (None)
-        cfg_host = Config.get("fastapi.host", "127.0.0.1")
-        cfg_port = Config.get("fastapi.port", 8000)
-        cfg_reload = Config.get("fastapi.reload", True)
         cfg_reload_dirs = Config.get("fastapi.reload_dirs") or None
         cfg_reload_excludes = Config.get("fastapi.reload_excludes") or None
 
-        host = self.option("host") or cfg_host
-        port = int(self.option("port") or cfg_port)
-        reload = cfg_reload if self.option("reload") is None else self.option("reload")
-        app = self.option("app")
-
-        exist = self.is_app_exist()
+        url = self.resolve_url()
 
         kwargs = {
-            "host": host,
-            "port": port,
-            "reload": reload,
+            "host": url.host(),
+            "port": url.port(),
+            "reload": self.resolve_option("reload", True),
             "ws": "websockets-sansio",
         }
 
-        if exist:
+        if self.is_app_exist():
             kwargs.update(
                 {
-                    "app": app,
+                    "app": self.option("app"),
                     "factory": True,
                 }
             )
@@ -75,10 +82,10 @@ def handle(self):
             if cfg_reload_excludes is not None:
                 kwargs["reload_excludes"] = cfg_reload_excludes
 
-            self.line(f"<info>Starting Uvicorn server on {host}:{port} [{app}]...</info>")
+            self.line(f"<info>Starting Uvicorn server on {url.host()}:{url.port()} [{self.option('app')}]...</info>")
 
         else:
-            self.line(f"<info>Starting Uvicorn server on {host}:{port}...</info>")
+            self.line(f"<info>Starting Uvicorn server on {url.host()}:{url.port()}...</info>")
             kwargs.update({"app": Container.instance().fastapi, "reload": False})
 
         try:

fastapi_startkit/tests/fastapi/test_serve_command.py

@@ -0,0 +1,187 @@
+"""Tests for ServeCommand using Cleo's CommandTester.
+
+Exercises option parsing, host/port resolution, and output messages without
+starting a real Uvicorn server.  ``Config.get`` and ``uvicorn.run`` are patched
+so the tests are self-contained with no live app or network required.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+from cleo.testers.command_tester import CommandTester
+
+from fastapi_startkit.fastapi.commands.serve_command import ServeCommand
+
+_DEFAULT_HOST = "127.0.0.1"
+_DEFAULT_PORT = 8000
+_DEFAULT_APP_URL = f"http://{_DEFAULT_HOST}:{_DEFAULT_PORT}"
+
+
+# ---------------------------------------------------------------------------
+# Helper
+# ---------------------------------------------------------------------------
+
+
+def run(
+    args: str = "",
+    *,
+    config: dict | None = None,
+    app_found: bool = False,
+) -> tuple[CommandTester, MagicMock]:
+    """Execute ServeCommand with *args* and return (tester, mock_uvicorn).
+
+    Parameters
+    ----------
+    args:
+        Argument string forwarded verbatim to ``CommandTester.execute()``.
+    config:
+        Mapping of config keys to values that ``Config.get`` will return.
+        Defaults to an empty dict (all keys fall back to their defaults).
+    app_found:
+        When ``True``, ``importlib.util.find_spec`` returns a non-None spec so
+        that ``is_app_exist()`` yields ``True``.
+    """
+    cfg = config or {}
+
+    def _config_get(key, default=None):
+        return cfg.get(key, default)
+
+    mock_uvicorn = MagicMock()
+    mock_spec = MagicMock() if app_found else None
+
+    with (
+        patch("fastapi_startkit.fastapi.commands.serve_command.Config.get", side_effect=_config_get),
+        patch("fastapi_startkit.configuration.config.Config.get", side_effect=_config_get),
+        patch("uvicorn.run", mock_uvicorn),
+        patch("importlib.util.find_spec", return_value=mock_spec),
+    ):
+        cmd = ServeCommand()
+        tester = CommandTester(cmd)
+        tester.execute(args, interactive=False)
+
+    return tester, mock_uvicorn
+
+
+# ---------------------------------------------------------------------------
+# 1. Default behaviour — no CLI flags, no config, no app bootstrap module
+# ---------------------------------------------------------------------------
+
+
+class TestServeCommandDefaults:
+    def test_exits_successfully(self):
+        tester, _ = run()
+        assert tester.status_code == 0
+
+    def test_uvicorn_called(self):
+        _, mock_uvicorn = run()
+        mock_uvicorn.assert_called_once()
+
+    def test_default_host_in_output(self):
+        tester, _ = run()
+        assert _DEFAULT_HOST in tester.io.fetch_output()
+
+    def test_default_port_in_output(self):
+        tester, _ = run()
+        assert str(_DEFAULT_PORT) in tester.io.fetch_output()
+
+    def test_uvicorn_kwargs_contain_ws(self):
+        _, mock_uvicorn = run()
+        _, kwargs = mock_uvicorn.call_args
+        assert kwargs.get("ws") == "websockets-sansio"
+
+
+# ---------------------------------------------------------------------------
+# 2. CLI --host / --port override defaults
+# ---------------------------------------------------------------------------
+
+
+class TestCliOptions:
+    def test_custom_host_appears_in_output(self):
+        tester, _ = run("--host 0.0.0.0")
+        assert "0.0.0.0" in tester.io.fetch_output()
+
+    def test_custom_port_appears_in_output(self):
+        tester, _ = run("--port 9000")
+        assert "9000" in tester.io.fetch_output()
+
+    def test_host_and_port_together_in_output(self):
+        tester, _ = run("--host 0.0.0.0 --port 9000")
+        output = tester.io.fetch_output()
+        assert "0.0.0.0" in output
+        assert "9000" in output
+
+    def test_custom_host_passed_to_uvicorn(self):
+        _, mock_uvicorn = run("--host 0.0.0.0")
+        _, kwargs = mock_uvicorn.call_args
+        assert kwargs["host"] == "0.0.0.0"
+
+    def test_custom_port_passed_to_uvicorn(self):
+        _, mock_uvicorn = run("--port 9000")
+        _, kwargs = mock_uvicorn.call_args
+        assert kwargs["port"] == 9000
+
+
+# ---------------------------------------------------------------------------
+# 3. Config-driven host / port (no CLI flags)
+# ---------------------------------------------------------------------------
+
+
+class TestConfigOptions:
+    def test_config_host_used_when_no_cli_flag(self):
+        tester, _ = run(config={"fastapi.host": "10.0.0.1", "fastapi.app_url": "http://10.0.0.1:8000"})
+        assert "10.0.0.1" in tester.io.fetch_output()
+
+    def test_config_port_used_when_no_cli_flag(self):
+        tester, _ = run(config={"fastapi.port": 7777, "fastapi.app_url": "http://127.0.0.1:7777"})
+        assert "7777" in tester.io.fetch_output()
+
+    def test_app_url_host_overridden_by_config_default(self):
+        """resolve_url() merges APP_URL with fastapi.host/port.
+
+        APP_URL sets the base, but fastapi.host and fastapi.port (which default
+        to 127.0.0.1 / 8000) take priority in the current implementation.
+        When no explicit host/port is in config, defaults win over APP_URL.
+        """
+        tester, _ = run(config={"fastapi.app_url": "http://staging.example.com:5000"})
+        output = tester.io.fetch_output()
+        # Default host wins because resolve_option("host", "127.0.0.1") takes over.
+        assert _DEFAULT_HOST in output
+
+
+# ---------------------------------------------------------------------------
+# 4. App bootstrap detected (is_app_exist → True)
+# ---------------------------------------------------------------------------
+
+
+class TestAppFound:
+    def test_app_label_in_output(self):
+        tester, _ = run("--app myapp.bootstrap:app", app_found=True)
+        assert "myapp.bootstrap:app" in tester.io.fetch_output()
+
+    def test_uvicorn_called_with_factory_true(self):
+        _, mock_uvicorn = run(app_found=True)
+        _, kwargs = mock_uvicorn.call_args
+        assert kwargs.get("factory") is True
+
+    def test_uvicorn_called_with_app_string(self):
+        _, mock_uvicorn = run("--app myapp.bootstrap:app", app_found=True)
+        _, kwargs = mock_uvicorn.call_args
+        assert kwargs.get("app") == "myapp.bootstrap:app"
+
+
+# ---------------------------------------------------------------------------
+# 5. App bootstrap NOT found (is_app_exist → False)
+# ---------------------------------------------------------------------------
+
+
+class TestAppNotFound:
+    def test_no_factory_kwarg_when_app_not_found(self):
+        _, mock_uvicorn = run(app_found=False)
+        _, kwargs = mock_uvicorn.call_args
+        assert "factory" not in kwargs
+
+    def test_reload_disabled_when_app_not_found(self):
+        _, mock_uvicorn = run(app_found=False)
+        _, kwargs = mock_uvicorn.call_args
+        assert kwargs.get("reload") is False