File size: 2,698 Bytes
0602b26 9fb95ba 0602b26 9fb95ba 0602b26 9fb95ba 0602b26 9fb95ba 0602b26 9fb95ba 0602b26 9fb95ba 0602b26 4294178 9fb95ba 4294178 0602b26 d29db6e 0602b26 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 | """Explicit test composition for the API adapter."""
from collections.abc import Mapping, MutableMapping
from fastapi import FastAPI
from free_claude_code.api.app import create_app
from free_claude_code.api.ports import ApiServices
from free_claude_code.application.connected_accounts import ConnectedAccountPort
from free_claude_code.config.settings import Settings
from free_claude_code.providers.base import BaseProvider
from free_claude_code.providers.runtime import ProviderRuntime
from free_claude_code.runtime.application import ApplicationRuntime, RestartCallback
from free_claude_code.runtime.provider_manager import ProviderRuntimeManager
def create_test_app(
settings: Settings | None = None,
*,
providers: MutableMapping[str, BaseProvider] | None = None,
restart_callback: RestartCallback | None = None,
connected_accounts: Mapping[str, ConnectedAccountPort] | None = None,
) -> FastAPI:
"""Build an API app with explicit in-memory runtime services."""
settings = settings or Settings()
connected_accounts = dict(connected_accounts or {})
def connected_provider_ids() -> tuple[str, ...]:
return tuple(
provider_id
for provider_id, account in connected_accounts.items()
if account.is_connected()
)
if providers is None:
manager = ProviderRuntimeManager(
settings,
connected_provider_ids=connected_provider_ids,
)
else:
manager = ProviderRuntimeManager(
settings,
runtime_factory=lambda snapshot: ProviderRuntime(
snapshot,
dict(providers),
),
connected_provider_ids=connected_provider_ids,
)
runtime = ApplicationRuntime(
manager,
transcriber=None,
restart_callback=restart_callback,
connected_accounts=connected_accounts,
)
return create_app(
ApiServices(
requests=manager,
admin=runtime,
tasks=runtime,
)
)
def runtime_for_app(app: FastAPI) -> ApplicationRuntime:
"""Return the runtime supplied by :func:`create_test_app`."""
runtime = app.state.services.admin
if not isinstance(runtime, ApplicationRuntime):
raise TypeError("Test app does not use ApplicationRuntime")
return runtime
def provider_manager_for_app(app: FastAPI) -> ProviderRuntimeManager:
"""Return the provider manager supplied by :func:`create_test_app`."""
manager = app.state.services.requests
if not isinstance(manager, ProviderRuntimeManager):
raise TypeError("Test app does not use ProviderRuntimeManager")
return manager
|