|
1 | 1 | import sys |
| 2 | +import types |
2 | 3 | from collections.abc import AsyncGenerator, Generator |
3 | 4 | from contextlib import asynccontextmanager |
4 | 5 | from typing import TYPE_CHECKING, Annotated, Callable, Literal, Union, cast |
5 | 6 | from unittest.mock import MagicMock |
6 | 7 |
|
| 8 | +import click as base_click |
7 | 9 | import pytest |
| 10 | +import typer |
8 | 11 | from fastapi import Depends, FastAPI, HTTPException, Request, Response |
9 | 12 | from fastapi.testclient import TestClient |
10 | 13 | from pytest import FixtureRequest |
11 | 14 | from pytest_mock import MockerFixture |
12 | 15 | from sqlalchemy import Engine |
13 | 16 | from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession |
14 | 17 | from sqlalchemy.orm import Session |
| 18 | +from typer.testing import CliRunner |
15 | 19 | from typing_extensions import assert_type |
16 | 20 |
|
17 | 21 | from advanced_alchemy.exceptions import ImproperConfigurationError |
@@ -55,6 +59,77 @@ def alchemy(config: AnyConfig, app: FastAPI) -> AdvancedAlchemy: |
55 | 59 | return AdvancedAlchemy(config, app=app) |
56 | 60 |
|
57 | 61 |
|
| 62 | +def _install_fastapi_cli_app(monkeypatch: pytest.MonkeyPatch, typer_app: typer.Typer) -> None: |
| 63 | + fastapi_cli = types.ModuleType("fastapi_cli") |
| 64 | + fastapi_cli_cli = types.ModuleType("fastapi_cli.cli") |
| 65 | + fastapi_cli_cli.app = typer_app |
| 66 | + setattr(fastapi_cli, "cli", fastapi_cli_cli) |
| 67 | + monkeypatch.setitem(sys.modules, "fastapi_cli", fastapi_cli) |
| 68 | + monkeypatch.setitem(sys.modules, "fastapi_cli.cli", fastapi_cli_cli) |
| 69 | + |
| 70 | + |
| 71 | +def test_assign_cli_group_delegates_database_and_db(monkeypatch: pytest.MonkeyPatch) -> None: |
| 72 | + from advanced_alchemy.extensions.fastapi import extension as fastapi_extension |
| 73 | + from advanced_alchemy.extensions.fastapi.extension import assign_cli_group |
| 74 | + |
| 75 | + seen: list[tuple[FastAPI, str]] = [] |
| 76 | + |
| 77 | + def fake_register_database_commands(app: FastAPI) -> base_click.Group: |
| 78 | + @base_click.group(name="database") |
| 79 | + def database_group() -> None: |
| 80 | + pass |
| 81 | + |
| 82 | + @database_group.command(name="upgrade") |
| 83 | + @base_click.argument("revision", default="head") |
| 84 | + def upgrade_database(revision: str) -> None: |
| 85 | + seen.append((app, revision)) |
| 86 | + |
| 87 | + return database_group |
| 88 | + |
| 89 | + monkeypatch.setattr(fastapi_extension, "register_database_commands", fake_register_database_commands) |
| 90 | + app = FastAPI() |
| 91 | + typer_app = typer.Typer(context_settings={"help_option_names": ["-h", "--help"]}) |
| 92 | + _install_fastapi_cli_app(monkeypatch, typer_app) |
| 93 | + |
| 94 | + assign_cli_group(app) |
| 95 | + |
| 96 | + runner = CliRunner() |
| 97 | + result = runner.invoke(typer_app, ["database", "upgrade", "abc123"]) |
| 98 | + alias_result = runner.invoke(typer_app, ["db", "upgrade"]) |
| 99 | + |
| 100 | + assert result.exit_code == 0, result.output |
| 101 | + assert alias_result.exit_code == 0, alias_result.output |
| 102 | + assert seen == [(app, "abc123"), (app, "head")] |
| 103 | + |
| 104 | + |
| 105 | +def test_assign_cli_group_delegates_database_help(monkeypatch: pytest.MonkeyPatch) -> None: |
| 106 | + from advanced_alchemy.extensions.fastapi import extension as fastapi_extension |
| 107 | + from advanced_alchemy.extensions.fastapi.extension import assign_cli_group |
| 108 | + |
| 109 | + def fake_register_database_commands(app: FastAPI) -> base_click.Group: |
| 110 | + @base_click.group(name="database") |
| 111 | + def database_group() -> None: |
| 112 | + pass |
| 113 | + |
| 114 | + @database_group.command(name="upgrade") |
| 115 | + def upgrade_database() -> None: |
| 116 | + pass |
| 117 | + |
| 118 | + return database_group |
| 119 | + |
| 120 | + monkeypatch.setattr(fastapi_extension, "register_database_commands", fake_register_database_commands) |
| 121 | + app = FastAPI() |
| 122 | + typer_app = typer.Typer(context_settings={"help_option_names": ["-h", "--help"]}) |
| 123 | + _install_fastapi_cli_app(monkeypatch, typer_app) |
| 124 | + assign_cli_group(app) |
| 125 | + |
| 126 | + result = CliRunner().invoke(typer_app, ["database", "--help"]) |
| 127 | + |
| 128 | + assert result.exit_code == 0, result.output |
| 129 | + assert "Commands:" in result.output |
| 130 | + assert "upgrade" in result.output |
| 131 | + |
| 132 | + |
58 | 133 | async def test_infer_types_from_config(async_config: SQLAlchemyAsyncConfig, sync_config: SQLAlchemySyncConfig) -> None: |
59 | 134 | if TYPE_CHECKING: |
60 | 135 | alchemy = AdvancedAlchemy(config=[async_config, sync_config]) |
|
0 commit comments