yuxuanhui
45 lines
1.6 KiB
Python
45 lines
1.6 KiB
Python
"""Synthetic PostgreSQL-backed server for container proxy acceptance only."""
|
|
|
|
from contextlib import asynccontextmanager
|
|
|
|
import httpx
|
|
from fastapi import Depends
|
|
|
|
from app.config import Settings
|
|
from app.main import create_app
|
|
from app.mcp_api.auth import SCOPES, create_token
|
|
from app.models import Account
|
|
from app.security import cipher, require_auth
|
|
from app.worldquant import WqClient
|
|
from tests.backtest_fake import Platform
|
|
|
|
|
|
def create_test_app():
|
|
settings = Settings(_env_file=None, enable_runner=False, mcp_enabled=True)
|
|
if not settings.database_url.endswith("/wq_mcp_test"):
|
|
raise RuntimeError("Disposable test database required")
|
|
platform = Platform()
|
|
app = create_app(settings, WqClient(settings, transport=httpx.MockTransport(platform)))
|
|
original = app.router.lifespan_context
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(application):
|
|
async with original(application):
|
|
async with app.state.sessions.begin() as db:
|
|
account = await db.get(Account, 1)
|
|
account.email = "synthetic@example.com"
|
|
account.wq_user_id = "TEST_USER"
|
|
account.connection_status = "connected"
|
|
account.password_encrypted = cipher(settings).encrypt(platform.secret.encode()).decode()
|
|
yield
|
|
|
|
app.router.lifespan_context = lifespan
|
|
|
|
@app.post("/api/v1/test/mcp-token", dependencies=[Depends(require_auth)])
|
|
async def issue_fixture():
|
|
async with app.state.sessions.begin() as db:
|
|
_, token = await create_token(db, "container-fixture", SCOPES)
|
|
return {"token": token}
|
|
|
|
return app
|