From d041077bb66a6624e8d1b3909d187071922db65d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Mon, 12 Oct 2026 19:15:46 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E8=A3=85=E9=85=8D=E5=88=87=E6=8D=A2?= =?UTF-8?q?=E2=80=94=E2=80=94main/=E6=9C=8D=E5=8A=A1=E6=9E=84=E9=80=A0?= =?UTF-8?q?=E7=BA=BF/conftest=20=E8=BF=81=E8=87=B3=20Registry=EF=BC=88?= =?UTF-8?q?=E5=90=AB=E9=BB=98=E8=AE=A4=E7=A7=9F=E6=88=B7=E8=BF=87=E6=B8=A1?= =?UTF-8?q?=E5=85=9C=E5=BA=95=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - main.py: 删全局 engine/async_session;lifespan 装配 TenantRegistry (seed_profiles → 仓库根 tenants/,路径锚定),预热默认租户 jiancai 挂 app.state.default_runtime,逐 active 租户(库文件已存在)跑 sweep_stale_pending,退出 dispose_all。app.state.registry 预置即沿用 (测试注入缝) - routes/runtime.py(过渡,Task 7 删): get_runtime——state.tenant 优先, 缺省落 default_runtime;chat.py/chat_rest.py 全部改走它,消除 main↔routes 循环依赖 - ConversationManager 构造线: 新增 runtime 可选参(持 session_factory/ compiled_wall/config 引用;prompts 仍模块级,Task 6;业务接线 Task 9), fire_pending_pushes 随任务传 session_factory(后台写回派发租户的库) - lead_push_service: 去 app.main import;写库入口显式收 session_factory, 模块级 async_session=None 仅作测试注入兜底 - conftest: client 夹具 = 每测试独立 Registry(tmp data_dir)+ 真实 lifespan 在 TestClient 循环内预热;删 _dispose_main_engine; async_db 保持 NullPool(跨 asyncio.run 循环安全,刻意不换 Registry 引擎) - 测试仅动夹具/桩签名(断言零改动): wiring 桩补 session_factory 形参; history-404 用例改经真实 WS 路由造 widget 会话(引擎常驻连接绑定 TestClient 循环,测试线程不得直开 session_factory) Co-Authored-By: Claude Code --- app/engine/routes/chat.py | 9 +- app/engine/routes/chat_rest.py | 21 +++-- app/engine/routes/runtime.py | 26 ++++++ app/engine/services/conversation.py | 22 ++++- app/engine/services/lead_push_service.py | 50 +++++++---- app/main.py | 51 +++++++---- app/platform/registry.py | 13 +++ tests/conftest.py | 105 ++++++++--------------- tests/test_chat_rest_session_history.py | 24 +++--- tests/test_lead_push_service.py | 2 +- tests/test_lead_push_wiring.py | 12 +-- 11 files changed, 198 insertions(+), 137 deletions(-) create mode 100644 app/engine/routes/runtime.py diff --git a/app/engine/routes/chat.py b/app/engine/routes/chat.py index 724d3aa..8509c48 100644 --- a/app/engine/routes/chat.py +++ b/app/engine/routes/chat.py @@ -3,8 +3,8 @@ import logging from fastapi import APIRouter, WebSocket, WebSocketDisconnect -from app.main import async_session from app.engine.models import ConversationSource, ConversationStatus +from app.engine.routes.runtime import get_runtime from app.engine.services.conversation import ConversationManager logger = logging.getLogger(__name__) @@ -17,8 +17,11 @@ async def ws_chat(websocket: WebSocket): session_id = None try: - async with async_session() as db: - manager = ConversationManager(db) + # Task 5b: 租户运行时经 state/默认租户兜底解析(过渡期无中间件); + # manager 由装配线从 runtime 造(db 会话 + runtime 引用) + runtime = get_runtime(websocket) + async with runtime.session_factory() as db: + manager = ConversationManager(db, runtime=runtime) # Collect client info client_ip = websocket.client.host if websocket.client else "" diff --git a/app/engine/routes/chat_rest.py b/app/engine/routes/chat_rest.py index d26b380..4aafced 100644 --- a/app/engine/routes/chat_rest.py +++ b/app/engine/routes/chat_rest.py @@ -17,8 +17,8 @@ from fastapi import APIRouter, HTTPException, Request from pydantic import BaseModel from sqlalchemy import select -from app.main import async_session from app.engine.models import ConversationSource, ConversationStatus, Message +from app.engine.routes.runtime import get_runtime from app.engine.services.conversation import ConversationManager from app.engine.services.rate_limit import check as rate_check @@ -45,8 +45,11 @@ async def chat_session(req: SessionRequest, request: Request): if xf: client_info["ip"] = xf - async with async_session() as db: - manager = ConversationManager(db) + # Task 5b: 租户运行时经 state/默认租户兜底解析(过渡期无中间件); + # manager 由装配线从 runtime 造(db 会话 + runtime 引用) + runtime = get_runtime(request) + async with runtime.session_factory() as db: + manager = ConversationManager(db, runtime=runtime) if req.session_id: conv = await manager.get_by_session(req.session_id) if conv and conv.status == ConversationStatus.active: @@ -68,12 +71,13 @@ async def chat_session(req: SessionRequest, request: Request): @router.get("/{session_id}/messages") -async def chat_history(session_id: str): +async def chat_history(session_id: str, request: Request): """Public history endpoint for session replay. App-source only, to narrow the read surface; an unknown or non-app session yields 404. """ - async with async_session() as db: - manager = ConversationManager(db) + runtime = get_runtime(request) + async with runtime.session_factory() as db: + manager = ConversationManager(db, runtime=runtime) conv = await manager.get_by_session(session_id) if not conv or conv.source != ConversationSource.app: raise HTTPException(status_code=404, detail="session not found") @@ -115,8 +119,9 @@ async def chat_message(req: MessageRequest, request: Request): if not rate_check(ip, dev): raise HTTPException(status_code=429, detail="rate limit exceeded") - async with async_session() as db: - manager = ConversationManager(db) + runtime = get_runtime(request) + async with runtime.session_factory() as db: + manager = ConversationManager(db, runtime=runtime) conv = await manager.get_by_session(req.session_id) if not conv: raise HTTPException(status_code=404, detail="session not found") diff --git a/app/engine/routes/runtime.py b/app/engine/routes/runtime.py new file mode 100644 index 0000000..b0c5426 --- /dev/null +++ b/app/engine/routes/runtime.py @@ -0,0 +1,26 @@ +"""路由取当前租户运行时——Task 5b→7 的过渡兜底。 + +Task 7 的 Host 中间件上线前,没有人为请求挂 ``state.tenant``。本助手的 +兜底规则 = state 里有租户用 state 的(中间件灰度接入即生效);没有则用 +``app.state.default_runtime``(lifespan 预热的默认租户 jiancai),API 测试 +与直连(无 Host 语义)在 5b→7 过渡期不断。 + +已知过渡限制(记录在案,Task 7 一并解决):默认租户经 registry_reload +热切换后 default_runtime 仍指向旧 runtime——过渡期不修,Task 7 中间件 +按请求 Host 重新 resolve 后本模块整体删除。 +""" +from fastapi import Request, WebSocket + +from app.platform.registry import TenantRuntime + + +def get_runtime(request_or_ws: Request | WebSocket) -> TenantRuntime: + """state.tenant 优先,缺省落 app.state.default_runtime(Task 7 删)。""" + runtime = getattr(request_or_ws.state, "tenant", None) + if runtime is not None: + return runtime + runtime = getattr(request_or_ws.app.state, "default_runtime", None) + if runtime is None or runtime.session_factory is None: + raise RuntimeError( + "默认租户运行时未就绪——lifespan 未运行或未预热 default_runtime") + return runtime diff --git a/app/engine/services/conversation.py b/app/engine/services/conversation.py index a97c688..2c1d6a8 100644 --- a/app/engine/services/conversation.py +++ b/app/engine/services/conversation.py @@ -3,6 +3,7 @@ import re import uuid from collections.abc import AsyncGenerator from datetime import datetime, timedelta, timezone +from typing import TYPE_CHECKING from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession @@ -26,6 +27,9 @@ from app.engine.services.contact_pipeline import ( normalize_contacts, resolve_by_recency, scan_user_messages, split_validate, ) +if TYPE_CHECKING: + from app.platform.registry import TenantRuntime + logger = logging.getLogger(__name__) # 职能边界触发词(Task 13.2, 评审#1/#4 修订): 命中即注入红线提醒。精修说明: @@ -323,9 +327,20 @@ def _boundary_rotate() -> int: class ConversationManager: - def __init__(self, db: AsyncSession, ai_service: AIService | None = None): + def __init__(self, db: AsyncSession, ai_service: AIService | None = None, + runtime: "TenantRuntime | None" = None): self.db = db self.ai = ai_service or AIService() + # Task 5b 装配线注入: 持有租户运行时引用(session_factory/compiled_wall/ + # config 就位;prompts 仍为模块级,Task 6 才 PromptProvider 化;email/push + # 等 settings→config 业务接线是 Task 9)。None = 无租户上下文(直连测试)。 + self.runtime = runtime + + @property + def session_factory(self): + """推送后台任务写结果用的工厂(写回派发租户的库);无 runtime 时 + None,lead_push_service 回退模块级兜底。""" + return self.runtime.session_factory if self.runtime is not None else None async def create_conversation( self, @@ -777,8 +792,9 @@ class ConversationManager: await self.db.commit() # 询盘推送(设计 8.1): 主事务提交后才派生后台任务——payload 内存 - # 传递,任务不读库只写结果;fire 为同步函数,纳秒级返回不阻塞 - lead_push_service.fire_pending_pushes(pending_pushes) + # 传递,任务不读库只写结果;fire 为同步函数,纳秒级返回不阻塞。 + # session_factory 随任务传递: 写回本租户的库(Task 5b) + lead_push_service.fire_pending_pushes(pending_pushes, self.session_factory) yield { "type": "evaluated", diff --git a/app/engine/services/lead_push_service.py b/app/engine/services/lead_push_service.py index 8c5b12e..afe9591 100644 --- a/app/engine/services/lead_push_service.py +++ b/app/engine/services/lead_push_service.py @@ -7,7 +7,8 @@ - sweep_stale_pending: lifespan 启动清扫(崩溃遗留的 pending → failed)。 后台任务红线(设计 8.1): 不读库只写结果,payload 内存传递; -结果写入用独立 session(from app.main import async_session,照 chat_rest 模式)。 +结果写入用独立 session——session_factory 由派发方传入(Task 5b 起无全局 +单例,写回派发它的那个租户的库)。 """ import asyncio import logging @@ -18,7 +19,6 @@ from sqlalchemy import func, select, update from sqlalchemy.ext.asyncio import AsyncSession from app.engine.config import settings -from app.main import async_session from app.engine.models import ( Conversation, CustomerProfile, Evaluation, IntentLevel, LeadPushLog, Message, MessageRole, @@ -28,6 +28,12 @@ from app.engine.services.ip_service import resolve_country logger = logging.getLogger(__name__) +# Task 5b 装配切换: app.main 全局单例已删,写库入口改为显式传 session_factory +# (租户正确: 后台任务写结果必须落回派发租户的库)。模块级 async_session 仅作 +# 无工厂上下文时的兜底(测试直接驱动 _run_push/sweep_stale_pending 时注入), +# 生产装配不设值——装配线(main.lifespan/ConversationManager)总是显式传。 +async_session = None + # 对方契约(docs/AI_INQUIRY_PUSH_API.md): phone/whatsapp/email 至少一个有效 _TRIO = ("phone", "whatsapp", "email") _CST = timezone(timedelta(hours=8)) @@ -207,9 +213,10 @@ def _retryable(resp: httpx.Response | None) -> bool: return resp is None or resp.status_code >= 500 -async def _update_log(push_log_id: str, **fields) -> None: - """后台任务写结果——独立 session + 小事务(设计 6.3)。""" - async with async_session() as session: +async def _update_log(session_factory, push_log_id: str, **fields) -> None: + """后台任务写结果——独立 session + 小事务(设计 6.3)。 + session_factory: 派发租户的工厂(Task 5b),写回同一租户库。""" + async with session_factory() as session: await session.execute( update(LeadPushLog) .where(LeadPushLog.id == push_log_id) @@ -218,11 +225,14 @@ async def _update_log(push_log_id: str, **fields) -> None: await session.commit() -async def _run_push(push_log_id: str, payload: dict) -> None: +async def _run_push(push_log_id: str, payload: dict, + session_factory=None) -> None: """后台任务体: 分类重试 + 200 逐条结果解析(设计 6.3/§10)。 红线: 全套 try/except,异常只记日志绝不逃逸(fire-and-forget)。 + session_factory 缺省回退模块级 async_session(测试注入缝)。 """ + factory = session_factory or async_session try: attempts = 0 while True: @@ -230,7 +240,7 @@ async def _run_push(push_log_id: str, payload: dict) -> None: resp, transport_error = await _do_post(payload) if _retryable(resp) and attempts < settings.PUSH_MAX_ATTEMPTS: await _update_log( - push_log_id, attempts=attempts, + factory, push_log_id, attempts=attempts, error_message=transport_error or (f"http {resp.status_code}" if resp is not None else "unknown"), ) @@ -245,7 +255,8 @@ async def _run_push(push_log_id: str, payload: dict) -> None: except ValueError: # 非 JSON 的 200(如网关重定向到 HTML 登录页)—— # 无法确认对方已接受,诚实落 failed(幂等键保证补推安全) - await _update_log(push_log_id, attempts=attempts, status="failed", + await _update_log(factory, push_log_id, attempts=attempts, + status="failed", response_status=200, error_message="non-JSON 200 body") return @@ -257,7 +268,7 @@ async def _run_push(push_log_id: str, payload: dict) -> None: ] if failed_mine: await _update_log( - push_log_id, attempts=attempts, status="failed", + factory, push_log_id, attempts=attempts, status="failed", response_status=200, error_message=str(failed_mine[0].get("reason") or "rejected"), ) @@ -266,7 +277,7 @@ async def _run_push(push_log_id: str, payload: dict) -> None: outcome = ("duplicated" if (data.get("duplicated") or 0) > 0 else "accepted") await _update_log( - push_log_id, attempts=attempts, status="success", + factory, push_log_id, attempts=attempts, status="success", response_status=200, outcome=outcome, ) return @@ -277,7 +288,7 @@ async def _run_push(push_log_id: str, payload: dict) -> None: "密钥/报文问题需人工介入,不重试", push_log_id, resp.status_code) await _update_log( - push_log_id, attempts=attempts, status="failed", + factory, push_log_id, attempts=attempts, status="failed", response_status=resp.status_code if resp is not None else None, error_message=transport_error or (f"http {resp.status_code}" if resp is not None else "unknown"), @@ -290,27 +301,34 @@ async def _run_push(push_log_id: str, payload: dict) -> None: _tasks: set[asyncio.Task] = set() -def fire_pending_pushes(pushes: list[LeadPushLog]) -> None: +def fire_pending_pushes(pushes: list[LeadPushLog], session_factory=None) -> None: """主事务 commit 之后调用(设计 8.1): payload 内存传递,任务不读库。 - 同步函数、纳秒级返回——这就是"不阻塞"的派发点。 + 同步函数、纳秒级返回——这就是"不阻塞"的派发点。session_factory = + 派发租户的工厂,随任务传递(后台写结果落回同一租户库);缺省回退 + 模块级 async_session(测试注入缝)。 """ + factory = session_factory or async_session for log in pushes: if log.status != "pending": continue - task = asyncio.create_task(_run_push(log.id, log.request_payload)) + task = asyncio.create_task( + _run_push(log.id, log.request_payload, factory)) _tasks.add(task) task.add_done_callback(_tasks.discard) -async def sweep_stale_pending(stale_minutes: int = 30) -> int: +async def sweep_stale_pending(stale_minutes: int = 30, session_factory=None) -> int: """启动清扫(设计 8.2): 崩溃遗留的 pending → failed。 + session_factory: 目标租户的工厂(lifespan 逐租户调用);缺省回退 + 模块级 async_session(测试注入缝)。 已知限制(设计 §14): 推送成功但结果写回失败也会被标 failed—— 幂等键保证将来补推安全(对方返回 duplicated),概率极低,接受。 """ + factory = session_factory or async_session cutoff = datetime.now(timezone.utc) - timedelta(minutes=stale_minutes) - async with async_session() as session: + async with factory() as session: result = await session.execute( update(LeadPushLog).where( LeadPushLog.status == "pending", diff --git a/app/main.py b/app/main.py index 44ba3aa..ed119ca 100644 --- a/app/main.py +++ b/app/main.py @@ -1,30 +1,49 @@ from contextlib import asynccontextmanager +from pathlib import Path from fastapi import FastAPI from fastapi.staticfiles import StaticFiles -from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine -from app.engine.config import settings, ensure_dirs, STATIC_DIR -from app.engine.models import ensure_schema +from app.engine.config import STATIC_DIR +from app.platform.registry import TenantRegistry - -# 后台推送任务的独立连接写结果可能撞上 SMTP 窗口内持有的写锁,默认 5s -# 忙等上限低于 SMTP 最坏时长,抬到 30s;等待发生在 aiosqlite 工作线程,不阻塞事件循环 -engine = create_async_engine(settings.DATABASE_URL, echo=settings.APP_DEBUG, - connect_args={"timeout": 30}) -async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) +# 路径锚定(不随 CWD 漂移):seed_profiles 与 tenants 都在仓库根 +BASE_DIR = Path(__file__).resolve().parents[1] +SEED_DIR = BASE_DIR / "seed_profiles" +TENANTS_DIR = BASE_DIR / "tenants" # .gitignore 已忽略 tenants/*.db +# 过渡兜底(Task 5b→7,中间件上线后随 routes/runtime.py 一并删除): +# Host 中间件落地前请求 state 里没有租户,一律落默认租户 jiancai +DEFAULT_TENANT_HOST = "chat2.agent.hnxinyouhe.com" @asynccontextmanager async def lifespan(app: FastAPI): - ensure_dirs() - await ensure_schema(engine) - # 询盘推送(设计 8.2): 清扫崩溃遗留的 stale pending → failed - # 函数内导入: lead_push_service 顶部 from app.main import async_session, - # 顶层互相导入会循环 + # 测试注入缝:conftest 预置过 app.state.registry(临时 data_dir 的独立 + # Registry)则沿用,不再加载仓库 seed——测试租户库与生产互不污染 + registry = getattr(app.state, "registry", None) + if registry is None: + registry = TenantRegistry.load_from_dir(SEED_DIR, data_dir=TENANTS_DIR) + app.state.registry = registry + + # 预热默认租户(过渡兜底读 app.state.default_runtime)。engine 建在本 + # lifespan 的事件循环里——连接与 loop 绑定,别处另起 asyncio.run 复用 + # 同一 engine 会跨 loop 死锁(测试直连造数走 conftest.async_db) + app.state.default_runtime = await registry.resolve(DEFAULT_TENANT_HOST) + + # 询盘推送(设计 8.2): 清扫崩溃遗留的 stale pending → failed。逐租户、 + # 用该租户自己的 session_factory;db 文件不存在的租户跳过——全新库 + # 无遗留行,保持 engine 懒建纪律(预热的默认租户文件已建,照常扫)。 from app.engine.services.lead_push_service import sweep_stale_pending - await sweep_stale_pending() - yield + for rt in registry.runtimes(): + if rt.config.status != "active" or not rt.db_path.exists(): + continue + ready = await registry.ensure_ready(rt) + await sweep_stale_pending(session_factory=ready.session_factory) + + try: + yield + finally: + await registry.dispose_all() # Windows 文件锁: 不关则库文件删不掉 app = FastAPI(title="HESUPIPE AI 客服", lifespan=lifespan) diff --git a/app/platform/registry.py b/app/platform/registry.py index 746711f..9e19622 100644 --- a/app/platform/registry.py +++ b/app/platform/registry.py @@ -216,6 +216,19 @@ class TenantRegistry: """resolve_site 的便捷形式(双 API 一层实现)。""" return (await self.resolve_site(host))[0] + # ---- 装配期辅助(Task 5b: main.lifespan 逐租户启动清扫)------------------ + + def runtimes(self) -> tuple[TenantRuntime, ...]: + """全部租户运行时的加载期快照(engine 仍懒建)——装配期遍历用 + (lifespan 启动清扫逐租户跑)。返回快照 tuple:遍历中 registry_reload + 换条目不影响本轮迭代。""" + return tuple(self._by_slug.values()) + + async def ensure_ready(self, rt: TenantRuntime) -> TenantRuntime: + """懒建的公开入口(装配期批处理用)——与 resolve 走同一条就绪路径, + 现势检查/双检/建库失败半开语义完全一致。""" + return await self._ensure_ready(rt) + # ---- 懒建 / 热加载 / 释放 --------------------------------------------- async def _ensure_ready(self, rt: TenantRuntime) -> TenantRuntime: diff --git a/tests/conftest.py b/tests/conftest.py index 90ed83d..4b6dc52 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,27 +1,23 @@ -import os -import tempfile import asyncio +import os +import shutil +import tempfile +from pathlib import Path import pytest -from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from sqlalchemy.pool import NullPool from app.engine.models import Base -# `chat_rest` is created in Task A4. Tolerate its absence so earlier tasks -# (e.g. A0's /health smoke) can use the `client` fixture without it. -# Import `app.main` FIRST so the natural main -> chat_rest load order resolves -# (chat_rest does `from app.main import async_session`; if conftest imports -# chat_rest before main is loaded, main's `from chat_rest import router` hits -# a partially-initialized module -> circular ImportError). -_HAS_CHAT_REST = False -chat_rest_mod = None -try: - import app.main # noqa: F401 - import app.engine.routes.chat_rest as chat_rest_mod - _HAS_CHAT_REST = True -except ModuleNotFoundError: - pass +# Task 5b 装配切换: app 不再有全局 engine/async_session——旧 conftest 里 +# "先 import app.main 再 import chat_rest"的循环导入次序坑随全局单例一并 +# 消失(routes 不再反向 import app.main),直接导入即可。 +from app.main import app +from app.platform.registry import TenantRegistry + +REPO_ROOT = Path(__file__).resolve().parents[1] +SEED_DIR = REPO_ROOT / "seed_profiles" async def _create_all(engine): @@ -29,34 +25,15 @@ async def _create_all(engine): await conn.run_sync(Base.metadata.create_all) -@pytest.fixture(scope="session", autouse=True) -def _dispose_main_engine(): - """Dispose `app.main.engine` at session end. - - Best-effort cleanup: explicitly disposing the app-level engine before - interpreter shutdown reduces (but does NOT eliminate) the noisy - `Windows fatal exception: access violation` traces that aiosqlite worker - threads emit when interrupted mid-cleanup at interpreter exit. - - Reality check on Windows: - - All tests pass and pytest exits 0; the traces are written to stderr - AFTER the run completes and do not affect pass/fail or the exit code. - - They originate from aiosqlite background threads being torn down by the - interpreter and are harmless. - - They do not occur on Linux/CI. - """ - yield - from app.main import engine as main_engine - try: - asyncio.run(main_engine.dispose()) - except Exception: - pass - - @pytest.fixture def async_db(): """Fresh temp-file SQLite sessionmaker for direct async-DB tests (e.g. testing ConversationManager without HTTP). Isolated per test. + + 刻意不走 Registry 引擎: 本夹具被各测试反复 asyncio.run(每次全新事件 + 循环),NullPool 连接即借即还才能跨循环安全;Registry 的 pool=2 常驻 + 连接与创建它的循环绑定,只能活在单一事件循环里(client 夹具的 + TestClient 循环,见下)。 """ fd, path = tempfile.mkstemp(suffix=".db") os.close(fd) @@ -84,37 +61,25 @@ def _reset_rate_limit(): @pytest.fixture -def client(monkeypatch): - """Temporary file-SQLite TestClient with `chat_rest.async_session` overridden. +def client(tmp_path): + """TestClient + 每测试独立 Registry(tmp_path 作 data_dir)。 - Design notes: - - File DB (not :memory:) avoids the StaticPool cross-loop reuse pitfall. - - `NullPool` closes each connection when the session returns it, so no - aiosqlite worker thread outlives the event loop that created it. - Together with `_dispose_main_engine`, this REDUCES but does NOT - eliminate the harmless Windows access-violation stderr traces described - above; it has no effect on test correctness or on Linux/CI. - - Monkeypatching the module-level `async_session` binding in `chat_rest` - works because the route handlers resolve it as a module global at call. + - 预置 app.state.registry(测试注入缝): lifespan 见 registry 已挂则 + 沿用,不再加载仓库 seed_profiles/tenants——每测试拿到全新租户库, + 与生产数据互不污染。 + - 默认租户(jiancai)的 engine 由 lifespan 在 TestClient 的 portal 事件 + 循环内预热,后续请求也在同一循环处理(连接/循环绑定安全)。测试 + 线程需要直接造数时不得用该 engine——asyncio.run 另起循环,跨循环 + 复用常驻连接会死锁;直连断言走 async_db,或经真实路由(WS/REST)。 + - 退出: lifespan finally 里 registry.dispose_all()(Windows 文件锁, + 不关则 tmp 目录删不掉),随后清理 tmp 目录。 """ from fastapi.testclient import TestClient - from app.main import app - fd, path = tempfile.mkstemp(suffix=".db") - os.close(fd) - engine = create_async_engine( - f"sqlite+aiosqlite:///{path}", poolclass=NullPool - ) - asyncio.run(_create_all(engine)) - sm = async_sessionmaker(engine, expire_on_commit=False) - if _HAS_CHAT_REST: - monkeypatch.setattr(chat_rest_mod, "async_session", sm) - - with TestClient(app) as c: - yield c - - asyncio.run(engine.dispose()) + registry = TenantRegistry.load_from_dir(SEED_DIR, data_dir=tmp_path) + app.state.registry = registry try: - os.remove(path) - except PermissionError: - pass + with TestClient(app) as c: + yield c + finally: + shutil.rmtree(tmp_path, ignore_errors=True) diff --git a/tests/test_chat_rest_session_history.py b/tests/test_chat_rest_session_history.py index 0ef70bc..b9a479f 100644 --- a/tests/test_chat_rest_session_history.py +++ b/tests/test_chat_rest_session_history.py @@ -1,4 +1,7 @@ -# `client` fixture from conftest (temp SQLite + overrides chat_rest.async_session). +# `client` fixture from conftest (per-test Registry on tmp data_dir; default +# tenant runtime pre-warmed by the app lifespan inside the TestClient loop). + +import json def test_session_creates_new(client): @@ -26,17 +29,10 @@ def test_history_404_unknown_sid(client): def test_history_rejects_non_app_session(client): # widget-source conversations are not readable via this public endpoint - # (narrow the read surface to app only). - import asyncio - from app.engine.routes import chat_rest - from app.engine.services.conversation import ConversationManager - from app.engine.models import ConversationSource - - async def _make_widget(): - async with chat_rest.async_session() as db: - mgr = ConversationManager(db) - c = await mgr.create_conversation(source=ConversationSource.widget) - return c.session_id - - sid = asyncio.run(_make_widget()) + # (narrow the read surface to app only). 经真实 WS 路由造 widget 会话: + # 租户 engine 的常驻连接绑定 TestClient 的事件循环,测试线程不得另起 + # asyncio.run 直开其 session_factory(跨循环复用会死锁)。 + with client.websocket_connect("/ws/chat") as ws: + ws.send_text(json.dumps({"source": "widget"})) + sid = ws.receive_json()["session_id"] assert client.get(f"/api/chat/{sid}/messages").status_code == 404 diff --git a/tests/test_lead_push_service.py b/tests/test_lead_push_service.py index fae8969..e6e9ce1 100644 --- a/tests/test_lead_push_service.py +++ b/tests/test_lead_push_service.py @@ -482,7 +482,7 @@ def test_fire_pending_creates_tasks(monkeypatch): from app.engine.services import lead_push_service as svc pushed = [] - async def fake_run(log_id, payload): + async def fake_run(log_id, payload, session_factory=None): pushed.append((log_id, payload)) monkeypatch.setattr(svc, "_run_push", fake_run) diff --git a/tests/test_lead_push_wiring.py b/tests/test_lead_push_wiring.py index acdb14e..781a544 100644 --- a/tests/test_lead_push_wiring.py +++ b/tests/test_lead_push_wiring.py @@ -50,7 +50,7 @@ def _setup(async_db, monkeypatch, ai, lead_succeeds=True): fired = [] monkeypatch.setattr(lead_push_service, "fire_pending_pushes", - lambda logs: fired.extend(logs)) + lambda logs, session_factory=None: fired.extend(logs)) monkeypatch.setattr(lead_push_service.settings, "PUSH_ENABLED", True) async def run(): @@ -165,7 +165,7 @@ def test_push_disabled_writes_nothing(async_db, monkeypatch): monkeypatch.setattr(email_service, "send_lead_email", fake_lead) fired = [] monkeypatch.setattr(lead_push_service, "fire_pending_pushes", - lambda logs: fired.extend(logs)) + lambda logs, session_factory=None: fired.extend(logs)) async def run(): async with async_db() as db: @@ -183,7 +183,7 @@ def test_fire_end_to_end_single_loop(async_db, monkeypatch): """真 fire + 假 _run_push: 同一事件循环内 stream 完成后任务确实被派发执行。""" pushed = [] - async def fake_run(log_id, payload): + async def fake_run(log_id, payload, session_factory=None): pushed.append(log_id) monkeypatch.setattr(lead_push_service, "_run_push", fake_run) @@ -234,7 +234,7 @@ def test_fire_happens_only_after_commit(async_db, monkeypatch): sent.append("lead") return True - async def fake_run(log_id, payload): + async def fake_run(log_id, payload, session_factory=None): async with async_db() as db: try: row = (await db.execute( @@ -249,9 +249,9 @@ def test_fire_happens_only_after_commit(async_db, monkeypatch): real_fire = lead_push_service.fire_pending_pushes - def recording_fire(logs): + def recording_fire(logs, session_factory=None): fired.append(True) - real_fire(logs) + real_fire(logs, session_factory) monkeypatch.setattr(email_service, "send_lead_email", fake_lead) monkeypatch.setattr(lead_push_service, "_run_push", fake_run)