mirror of
https://github.com/MemeMeow-Studio/MemeMeow.git
synced 2026-10-26 20:12:13 +08:00
74 lines
3.3 KiB
Python
74 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from contextvars import ContextVar
|
|
import os
|
|
from pathlib import Path
|
|
import threading
|
|
from uuid import uuid4
|
|
|
|
from fastapi import FastAPI, HTTPException
|
|
import pytest
|
|
from sqlalchemy import create_engine, delete, event
|
|
from starlette.requests import Request
|
|
|
|
import api
|
|
from backend.config import Settings
|
|
from backend.persistence.models import Scope, ScopeContext
|
|
from backend.persistence.resources import DatabaseResources
|
|
from backend.scope import ScopeServiceFactory
|
|
from backend.search_http import SearchRequest
|
|
|
|
|
|
# 使用真实搜索入口和数据库查询验证线程调度及请求上下文传播。
|
|
def test_search_database_queries_run_in_worker_thread(tmp_path: Path) -> None:
|
|
"""空测试 scope 通过真实缓存查询返回 503，所有查询均在工作线程执行。"""
|
|
url = os.environ.get("MEMEMEOW_TEST_DATABASE_URL")
|
|
if not url:
|
|
pytest.skip("需要显式提供开发测试 PostgreSQL 连接")
|
|
engine = create_engine(url, pool_size=1, max_overflow=0, pool_timeout=3)
|
|
scope = ScopeContext(f"search-threadpool-validation-{uuid4().hex}")
|
|
resources = DatabaseResources(engine, image_root=tmp_path / "images", data_root=tmp_path, require_local_scope=False)
|
|
settings = Settings(data_root=tmp_path, image_root=tmp_path / "images")
|
|
factory = ScopeServiceFactory(resources, settings)
|
|
request_context = ContextVar("search_request_context", default="missing")
|
|
observed: list[tuple[int, str]] = []
|
|
|
|
def record_query(_connection, _cursor, _statement, _parameters, _context, _executemany) -> None:
|
|
"""通过 SQLAlchemy 事件观察真实查询的线程和请求上下文。"""
|
|
observed.append((threading.get_ident(), request_context.get()))
|
|
|
|
try:
|
|
with resources.factory() as session, session.begin():
|
|
session.add(Scope(id=scope.scope_id))
|
|
application = FastAPI()
|
|
application.state.settings = settings
|
|
request = Request({"type": "http", "method": "POST", "path": "/search", "headers": [], "app": application})
|
|
request.state.scope = scope
|
|
request.state.services = factory.for_scope(scope)
|
|
|
|
async def search() -> None:
|
|
"""通过真实 API handler 验证查询线程、上下文和错误传播。"""
|
|
loop_thread = threading.get_ident()
|
|
token = request_context.set(scope.scope_id)
|
|
event.listen(engine, "before_cursor_execute", record_query)
|
|
try:
|
|
with pytest.raises(HTTPException) as captured:
|
|
await api.search_images(request, SearchRequest(query="线程调度测试"))
|
|
assert captured.value.status_code == 503
|
|
assert captured.value.detail["error"] == "cache_not_ready"
|
|
assert observed
|
|
assert all(thread_id != loop_thread for thread_id, _ in observed)
|
|
assert all(context == scope.scope_id for _, context in observed)
|
|
assert engine.pool.checkedout() == 0
|
|
finally:
|
|
event.remove(engine, "before_cursor_execute", record_query)
|
|
request_context.reset(token)
|
|
|
|
asyncio.run(search())
|
|
finally:
|
|
factory.shutdown()
|
|
with resources.factory() as session, session.begin():
|
|
session.execute(delete(Scope).where(Scope.id == scope.scope_id))
|
|
engine.dispose()
|