RuoYi-Vue3-FastAPI/ruoyi-fastapi-backend/tests/cli/runtime/test_app_runtime.py

from types import SimpleNamespace

from fastapi import APIRouter, FastAPI
from pytest import MonkeyPatch

from cli.groups.app.controller import AppCommandController
from cli.runtime.app import AppRuntimeService
from cli.runtime.app.gateway import AppInfrastructureGateway
from cli.runtime.app.support import AppSnapshotSupport
from cli.runtime.base import RuntimeEnvironmentService

REDIS_PORT = 6379


def _build_app_with_included_router() -> FastAPI:
    app = FastAPI()
    router = APIRouter()

    @router.get(
        '/visible',
        name='visible_route',
        summary='Visible route',
        operation_id='visibleRoute',
        tags=['route'],
    )
    async def visible_route() -> dict[str, bool]:
        return {'ok': True}

    @router.get('/hidden', include_in_schema=False)
    async def hidden_route() -> dict[str, bool]:
        return {'ok': True}

    app.include_router(router, prefix='/api', tags=['included'])
    return app


def _expected_visible_route() -> dict[str, object]:
    return {
        'path': '/api/visible',
        'methods': ['GET'],
        'name': 'visible_route',
        'summary': 'Visible route',
        'operationId': 'visibleRoute',
        'tags': ['included', 'route'],
        'includeInSchema': True,
    }


class FakeRuntimeEnvironment(RuntimeEnvironmentService):
    """
    模拟运行时环境服务。
    """

    @staticmethod
    def get_backend_dir() -> str:
        """
        返回固定后端目录。

        :return: 固定目录
        """
        return '/tmp/ruoyi-backend'

    @staticmethod
    def get_python_executable() -> str:
        """
        返回固定 Python 可执行文件。

        :return: Python 可执行文件路径
        """
        return '/usr/bin/python3'


def test_app_snapshot_support_builds_config_snapshot() -> None:
    """
    校验应用快照支持对象会构建应用配置快照。

    :return: None
    """
    gateway = AppInfrastructureGateway()
    support = AppSnapshotSupport(gateway, FakeRuntimeEnvironment())

    fake_env_module = SimpleNamespace(
        AppConfig=SimpleNamespace(
            app_env='dev',
            app_name='ruoyi',
            app_host='127.0.0.1',
            app_port=8080,
            app_root_path='/api',
            app_reload=True,
            app_workers=1,
            app_disable_swagger=False,
            app_disable_redoc=False,
        ),
        DataBaseConfig=SimpleNamespace(
            default_source=SimpleNamespace(
                db_type='mysql',
                db_host='127.0.0.1',
                db_port=3306,
                db_database='ruoyi',
            )
        ),
        RedisConfig=SimpleNamespace(redis_host='127.0.0.1', redis_port=REDIS_PORT),
        LogConfig=SimpleNamespace(loguru_level='INFO'),
        TransportCryptoConfig=SimpleNamespace(
            transport_crypto_enabled=True,
            transport_crypto_mode='strict',
        ),
    )

    def _fake_get_env_module() -> SimpleNamespace:
        return fake_env_module

    object.__setattr__(gateway, 'get_env_module', _fake_get_env_module)

    payload = support.build_app_config_snapshot()

    assert payload['env'] == 'dev'
    assert payload['dbType'] == 'mysql'
    assert payload['redisPort'] == REDIS_PORT
    assert payload['transportCryptoMode'] == 'strict'


def test_app_snapshot_support_builds_env_snapshot(monkeypatch: MonkeyPatch) -> None:
    """
    校验应用快照支持对象会构建环境解析快照。

    :param monkeypatch: pytest monkeypatch 工具
    :return: None
    """
    gateway = AppInfrastructureGateway()
    support = AppSnapshotSupport(gateway, FakeRuntimeEnvironment())
    fake_env_module = SimpleNamespace(
        AppConfig=SimpleNamespace(app_env='prod'),
    )

    def _fake_get_env_module() -> SimpleNamespace:
        return fake_env_module

    object.__setattr__(gateway, 'get_env_module', _fake_get_env_module)
    monkeypatch.setenv('APP_ENV', 'test')

    payload = support.build_app_env_snapshot()

    assert payload == {
        'cliEnv': 'test',
        'configEnv': 'prod',
        'appEnv': 'test',
        'envFile': '.env.test',
        'envFilePath': '/tmp/ruoyi-backend/.env.test',
        'envFileExists': False,
        'backendDir': '/tmp/ruoyi-backend',
        'pythonExecutable': '/usr/bin/python3',
    }


def test_app_runtime_service_builds_app_instance() -> None:
    """
    校验应用运行时 facade 会通过基础设施网关构建应用实例。

    :return: None
    """
    gateway = AppInfrastructureGateway()
    service = AppRuntimeService(
        runtime_environment=FakeRuntimeEnvironment(),
        infrastructure_gateway=gateway,
    )

    class FakeServerModule:
        """
        模拟 server 模块。
        """

        @staticmethod
        def create_app() -> dict[str, str]:
            """
            返回模拟应用实例。

            :return: 模拟应用实例
            """
            return {'app': 'ok'}

        @staticmethod
        def _register_application_routers(app: dict[str, str]) -> None:
            """
            模拟不存在于当前 server 模块的旧私有扩展点。

            :param app: 模拟应用实例
            :return: None
            """
            raise AssertionError('不应调用私有路由注册扩展点')

    def _fake_get_server_module() -> FakeServerModule:
        return FakeServerModule()

    object.__setattr__(gateway, 'get_server_module', _fake_get_server_module)

    payload = service.build_app_instance()

    assert payload == {'app': 'ok'}


def test_app_runtime_service_expands_included_router_routes() -> None:
    app = _build_app_with_included_router()
    gateway = AppInfrastructureGateway()
    service = AppRuntimeService(
        runtime_environment=FakeRuntimeEnvironment(),
        infrastructure_gateway=gateway,
    )
    fake_server_module = SimpleNamespace(create_app=lambda: app)

    def _fake_get_server_module() -> SimpleNamespace:
        return fake_server_module

    object.__setattr__(gateway, 'get_server_module', _fake_get_server_module)

    payload = service.get_app_routes_snapshot('dev')
    hidden_payload = service.get_app_routes_snapshot('dev', include_hidden=True)

    assert payload['routes'] == [_expected_visible_route()]
    assert payload['count'] == 1
    assert [route['path'] for route in hidden_payload['routes']] == ['/api/hidden', '/api/visible']
    assert [route['includeInSchema'] for route in hidden_payload['routes']] == [False, True]


def test_app_command_controller_serializes_included_router_routes() -> None:
    app = _build_app_with_included_router()

    routes = AppCommandController._serialize_routes(app)
    hidden_routes = AppCommandController._serialize_routes(app, include_hidden=True)

    assert routes == [_expected_visible_route()]
    assert [route['path'] for route in hidden_routes] == ['/api/hidden', '/api/visible']
    assert [route['includeInSchema'] for route in hidden_routes] == [False, True]