Refactor to reuse code and increase test coverage

Author: mlasevich

src/fastapi_cli/discover.py

@@ -4,7 +4,7 @@
 from dataclasses import dataclass
 from logging import getLogger
 from pathlib import Path
-from typing import Iterator, Union
+from typing import Iterator, Tuple, Union
 
 from rich import print
 from rich.padding import Padding
@@ -149,9 +149,9 @@ def get_app_name(*, mod_data: ModuleData, app_name: Union[str, None] = None) ->
     raise FastAPICLIException("Could not find FastAPI app in module, try using --app")
 
 
-def get_import_string(
+def get_import_string_parts(
     *, path: Union[Path, None] = None, app_name: Union[str, None] = None
-) -> str:
+) -> Tuple[ModuleData, str]:
     if not path:
         path = get_default_path()
     logger.info(f"Using path [blue]{path}[/blue]")
@@ -161,6 +161,15 @@ def get_import_string(
     mod_data = get_module_data_from_path(path)
     sys.path.insert(0, str(mod_data.extra_sys_path))
     use_app_name = get_app_name(mod_data=mod_data, app_name=app_name)
+
+    return mod_data, use_app_name
+
+
+def get_import_string(
+    *, path: Union[Path, None] = None, app_name: Union[str, None] = None
+) -> str:
+    mod_data, use_app_name = get_import_string_parts(path=path, app_name=app_name)
+    import_string = f"{mod_data.module_import_str}:{use_app_name}"
     import_example = Syntax(
         f"from {mod_data.module_import_str} import {use_app_name}", "python"
     )
@@ -175,57 +184,17 @@ def get_import_string(
     )
     logger.info("Found importable FastAPI app")
     print(import_panel)
-    import_string = f"{mod_data.module_import_str}:{use_app_name}"
+
     logger.info(f"Using import string [b green]{import_string}[/b green]")
     return import_string
 
 
 def get_app(
     *, path: Union[Path, None] = None, app_name: Union[str, None] = None
 ) -> FastAPI:
-    if not path:
-        path = get_default_path()
-    logger.debug(f"Using path [blue]{path}[/blue]")
-    logger.debug(f"Resolved absolute path {path.resolve()}")
-    if not path.exists():
-        raise FastAPICLIException(f"Path does not exist {path}")
-    mod_data = get_module_data_from_path(path)
-    try:
-        with mod_data.sys_path():
-            mod = importlib.import_module(mod_data.module_import_str)
-    except (ImportError, ValueError) as e:
-        logger.error(f"Import error: {e}")
-        logger.warning(
-            "Ensure all the package directories have an [blue]__init__.py["
-            "/blue] file"
-        )
-        raise
-    if not FastAPI:  # type: ignore[truthy-function]
-        raise FastAPICLIException(
-            "Could not import FastAPI, try running 'pip install fastapi'"
-        ) from None
-    object_names = dir(mod)
-    object_names_set = set(object_names)
-    if app_name:
-        if app_name not in object_names_set:
-            raise FastAPICLIException(
-                f"Could not find app name {app_name} in "
-                f"{mod_data.module_import_str}"
-            )
-        app = getattr(mod, app_name)
-        if not isinstance(app, FastAPI):
-            raise FastAPICLIException(
-                f"The app name {app_name} in {mod_data.module_import_str} "
-                f"doesn't seem to be a FastAPI app"
-            )
-        return app
-    for preferred_name in ["app", "api"]:
-        if preferred_name in object_names_set:
-            obj = getattr(mod, preferred_name)
-            if isinstance(obj, FastAPI):
-                return obj
-    for name in object_names:
-        obj = getattr(mod, name)
-        if isinstance(obj, FastAPI):
-            return obj
-    raise FastAPICLIException("Could not find FastAPI app in module, try using --app")
+    mod_data, use_app_name = get_import_string_parts(path=path, app_name=app_name)
+    with mod_data.sys_path():
+        mod = importlib.import_module(mod_data.module_import_str)
+    app = getattr(mod, use_app_name)
+    ## get_import_string_parts guarantees app is FastAPI object
+    return app  # type: ignore[no-any-return]

tests/test_cli.py

@@ -1,10 +1,12 @@
+import os
 import subprocess
 import sys
 from pathlib import Path
 from unittest.mock import patch
 
 import uvicorn
 from fastapi_cli.cli import app
+from fastapi_cli.exceptions import FastAPICLIException
 from typer.testing import CliRunner
 
 from tests.utils import changing_dir
@@ -14,6 +16,13 @@
 assets_path = Path(__file__).parent / "assets"
 
 
+def read_file(filename: str, strip: bool = True) -> str:
+    """Read file and return content as string"""
+    with open("openapi.json") as stream:
+        data = stream.read()
+    return data.strip() if data and strip else data
+
+
 def test_dev() -> None:
     with changing_dir(assets_path):
         with patch.object(uvicorn, "run") as mock_run:
@@ -190,6 +199,41 @@ def test_schema() -> None:
         assert expected in result.output, result.output
 
 
+def test_schema_file() -> None:
+    with changing_dir(assets_path):
+        filename = "unit-test.json"
+        expected = read_file("openapi.json", strip=True)
+        assert expected != "", "Failed to read expected result"
+        result = runner.invoke(
+            app, ["schema", "single_file_app.py", "--output", filename]
+        )
+        assert os.path.isfile(filename)
+        actual = read_file(filename, strip=True)
+        os.remove(filename)
+        assert result.exit_code == 0, result.output
+        assert expected == actual
+
+
+def test_schema_invalid_path() -> None:
+    with changing_dir(assets_path):
+        result = runner.invoke(app, ["schema", "invalid/single_file_app.py"])
+        assert result.exit_code == 1, result.output
+        assert isinstance(result.exception, FastAPICLIException)
+        assert "Path does not exist invalid/single_file_app.py" in str(result.exception)
+
+
+#
+#
+# def test_schema_invalid_package() -> None:
+#     with changing_dir(assets_path):
+#         result = runner.invoke(
+#             app, ["schema", "broken_package/mod/app.py"]
+#         )
+#         assert result.exit_code == 1, result.output
+#         assert isinstance(result.exception, ImportError)
+#         assert "attempted relative import beyond top-level package" in str(result.exception)
+
+
 def test_run_help() -> None:
     result = runner.invoke(app, ["run", "--help"])
     assert result.exit_code == 0, result.output

tests/test_utils_package.py

@@ -1,7 +1,7 @@
 from pathlib import Path
 
 import pytest
-from fastapi_cli.discover import get_import_string
+from fastapi_cli.discover import get_app, get_import_string
 from fastapi_cli.exceptions import FastAPICLIException
 from pytest import CaptureFixture
 
@@ -427,6 +427,17 @@ def test_broken_package_dir(capsys: CaptureFixture[str]) -> None:
     assert "Ensure all the package directories have an __init__.py file" in captured.out
 
 
+def test_get_app_broken_package_dir(capsys: CaptureFixture[str]) -> None:
+    with changing_dir(assets_path):
+        # TODO (when deprecating Python 3.8): remove ValueError
+        with pytest.raises((ImportError, ValueError)):
+            get_app(path=Path("broken_package/mod/app.py"))
+
+    captured = capsys.readouterr()
+    assert "Import error:" in captured.out
+    assert "Ensure all the package directories have an __init__.py file" in captured.out
+
+
 def test_package_dir_no_app() -> None:
     with changing_dir(assets_path):
         with pytest.raises(FastAPICLIException) as e: