95 lines
3.6 KiB
Python
95 lines
3.6 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""测试用的共同脚手架。"""
|
|
|
|
import os
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
|
|
class _QueueLogRedirect:
|
|
"""把队列执行记录挪到临时目录，别让测试写进用户的真实记录里。
|
|
|
|
机器人在好几条路径上都会记一步队列流水，而测试里的机器人多半是
|
|
`__new__` 出来的半成品，谁也不会想到去挡这个写入。不挡的话跑一次测试就往
|
|
真实的 `queue_events.json` 里灌进几十条「会话 abababab 跳过」，把用户真正
|
|
需要的排查线索挤出去——记录只留最近 500 条。
|
|
"""
|
|
|
|
def __init__(self):
|
|
self._directory = None
|
|
self._original = None
|
|
|
|
def start(self) -> None:
|
|
from queue_log import QueueLog
|
|
from wechat_bot import WeChatBot
|
|
|
|
self._directory = tempfile.TemporaryDirectory()
|
|
self._original = WeChatBot.queue_log
|
|
log = QueueLog(os.path.join(self._directory.name, "queue_events.json"))
|
|
WeChatBot.queue_log = property(lambda _self, _log=log: _log)
|
|
|
|
def stop(self) -> None:
|
|
from wechat_bot import WeChatBot
|
|
|
|
if self._original is not None:
|
|
WeChatBot.queue_log = self._original
|
|
self._original = None
|
|
if self._directory is not None:
|
|
self._directory.cleanup()
|
|
self._directory = None
|
|
|
|
|
|
queue_log_redirect = _QueueLogRedirect()
|
|
|
|
|
|
class _LocalStateRedirect:
|
|
"""把桌面端的本机配置挪到临时目录，别让测试读到开发机的真实设置。
|
|
|
|
这是踩过一次才补的。`ai_chat.current_provider()` 会去读
|
|
`backend_connection.json` 判断有没有配模型网关——配了就走网关，没配才用本机
|
|
单模型。于是只要开发机上碰巧连过一次后台（那次同步把网关地址写进了配置），
|
|
一大片和网关毫无关系的测试就会集体改变行为：视觉、Dify、诊断脱敏、出口解析
|
|
全都开始走网关分支。
|
|
|
|
症状极具误导性：报的是"视觉请求失败，降级为文字回复"，看不出跟配置文件有
|
|
任何关系。同一份代码在两台机器上结果不同，谁也说不清是谁的问题。
|
|
|
|
RUNTIME_FILE 也要一起挡：不挡的话，开发机上碰巧有个本地后台在跑，
|
|
`discover_local_server_url()` 就会读到真实的同机凭证。
|
|
"""
|
|
|
|
def __init__(self):
|
|
self._directory = None
|
|
self._saved = {}
|
|
|
|
def start(self) -> None:
|
|
import ai_config
|
|
import backend_client
|
|
|
|
self._directory = tempfile.TemporaryDirectory()
|
|
root = self._directory.name
|
|
self._saved = {
|
|
"CONNECTION_FILE": backend_client.CONNECTION_FILE,
|
|
"RUNTIME_FILE": backend_client.RUNTIME_FILE,
|
|
"_SETTINGS_FILE": ai_config._SETTINGS_FILE,
|
|
}
|
|
backend_client.CONNECTION_FILE = Path(root) / "backend_connection.json"
|
|
backend_client.RUNTIME_FILE = Path(root) / "backend_runtime.json"
|
|
ai_config._SETTINGS_FILE = os.path.join(root, "ai_settings.json")
|
|
|
|
def stop(self) -> None:
|
|
import ai_config
|
|
import backend_client
|
|
|
|
if self._saved:
|
|
backend_client.CONNECTION_FILE = self._saved["CONNECTION_FILE"]
|
|
backend_client.RUNTIME_FILE = self._saved["RUNTIME_FILE"]
|
|
ai_config._SETTINGS_FILE = self._saved["_SETTINGS_FILE"]
|
|
self._saved = {}
|
|
if self._directory is not None:
|
|
self._directory.cleanup()
|
|
self._directory = None
|
|
|
|
|
|
local_state_redirect = _LocalStateRedirect()
|