Merge branch 'main' into copilot/fix-agent-config-repository-error

This commit is contained in:
Wenjie Zhang 2026-03-30 15:41:07 +08:00 committed by GitHub
commit 97c585f6c6
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
57 changed files with 1596 additions and 1339 deletions

View File

@ -2,16 +2,14 @@ from dotenv import load_dotenv
load_dotenv(".env", override=True)
import os # noqa: E402
from concurrent.futures import ThreadPoolExecutor # noqa: E402
from yuxi.config import config as config # noqa: E402
__version__ = "0.6.0.beta2"
if os.getenv("YUXI_SKIP_APP_INIT") != "1":
from yuxi.knowledge import graph_base as graph_base # noqa: E402
from yuxi.knowledge import knowledge_base as knowledge_base # noqa: E402
from yuxi.knowledge import graph_base as graph_base # noqa: E402
from yuxi.knowledge import knowledge_base as knowledge_base # noqa: E402
executor = ThreadPoolExecutor() # noqa: E402

View File

@ -6,6 +6,7 @@ from .implementations.dify import DifyKB
from .manager import KnowledgeBaseManager
_LITE_MODE = os.environ.get("LITE_MODE", "").lower() in ("true", "1")
_SKIP_APP_INIT = os.environ.get("YUXI_SKIP_APP_INIT") == "1"
if not _LITE_MODE:
from .graphs.upload_graph_service import UploadGraphService
@ -23,7 +24,7 @@ work_dir = os.path.join(config.save_dir, "knowledge_base_data")
knowledge_base = KnowledgeBaseManager(work_dir)
# 创建图数据库实例
if _LITE_MODE:
if _LITE_MODE or _SKIP_APP_INIT:
from ..utils import logger
class _LiteGraphStub:
@ -38,7 +39,10 @@ if _LITE_MODE:
graph_base = _LiteGraphStub()
# 向后兼容
GraphDatabase = _LiteGraphStub
logger.info("LITE_MODE enabled, knowledge graph services disabled")
if _LITE_MODE:
logger.info("LITE_MODE enabled, knowledge graph services disabled")
else:
logger.info("YUXI_SKIP_APP_INIT enabled, knowledge graph services disabled for current process")
else:
graph_base = UploadGraphService()
# 向后兼容:让 GraphDatabase 指向 UploadGraphService

View File

@ -1,3 +1,4 @@
import asyncio
import os
import traceback
from functools import partial
@ -34,6 +35,9 @@ class LightRagKB(KnowledgeBase):
# 存储 LightRAG 实例映射 {db_id: LightRAG}
self.instances: dict[str, LightRAG] = {}
self._db_write_locks: dict[str, asyncio.Lock] = {}
self._db_instance_locks: dict[str, asyncio.Lock] = {}
self._lock_guard = asyncio.Lock()
logger.info("LightRagKB initialized")
@ -197,20 +201,34 @@ class LightRagKB(KnowledgeBase):
if db_id not in self.databases_meta:
return None
try:
# 创建实例
rag = await self._create_kb_instance(db_id, {})
instance_lock = await self._get_db_instance_lock(db_id)
async with instance_lock:
if db_id in self.instances:
logger.info(f"Using cached LightRAG instance for {db_id}")
return self.instances[db_id]
# 异步初始化存储
await self._initialize_kb_instance(rag)
try:
# 创建实例
rag = await self._create_kb_instance(db_id, {})
self.instances[db_id] = rag
return rag
# 异步初始化存储
await self._initialize_kb_instance(rag)
except Exception as e:
logger.error(f"Failed to create LightRAG instance for {db_id}: {e}")
logger.error(f"Traceback: {traceback.format_exc()}")
return None
self.instances[db_id] = rag
return rag
except Exception as e:
logger.error(f"Failed to create LightRAG instance for {db_id}: {e}")
logger.error(f"Traceback: {traceback.format_exc()}")
return None
async def _get_db_write_lock(self, db_id: str) -> asyncio.Lock:
async with self._lock_guard:
return self._db_write_locks.setdefault(db_id, asyncio.Lock())
async def _get_db_instance_lock(self, db_id: str) -> asyncio.Lock:
async with self._lock_guard:
return self._db_instance_locks.setdefault(db_id, asyncio.Lock())
def _get_llm_func(self, llm_info: dict):
"""获取 LLM 函数"""
@ -299,163 +317,73 @@ class LightRagKB(KnowledgeBase):
if db_id not in self.databases_meta:
raise ValueError(f"Database {db_id} not found")
rag = await self._get_lightrag_instance(db_id)
if not rag:
raise ValueError(f"Failed to get LightRAG instance for {db_id}")
db_write_lock = await self._get_db_write_lock(db_id)
async with db_write_lock:
rag = await self._get_lightrag_instance(db_id)
if not rag:
raise ValueError(f"Failed to get LightRAG instance for {db_id}")
# Get file meta
if file_id not in self.files_meta:
raise ValueError(f"File {file_id} not found")
file_meta = self.files_meta[file_id]
# Validate current status - only allow indexing from these states
current_status = file_meta.get("status")
allowed_statuses = {
FileStatus.PARSED,
FileStatus.ERROR_INDEXING,
FileStatus.INDEXED, # For re-indexing
"done", # Legacy status
}
if current_status not in allowed_statuses:
raise ValueError(
f"Cannot index file with status '{current_status}'. "
f"File must be parsed first (status should be one of: {', '.join(allowed_statuses)})"
)
# Check markdown file exists
if not file_meta.get("markdown_file"):
raise ValueError("File has not been parsed yet (no markdown_file)")
# Clear previous error if any
if "error" in file_meta:
self.files_meta[file_id].pop("error", None)
# Update status and add to processing queue
self.files_meta[file_id]["status"] = FileStatus.INDEXING
self.files_meta[file_id]["updated_at"] = utc_isoformat()
if operator_id:
self.files_meta[file_id]["updated_by"] = operator_id
await self._persist_file(file_id)
# Add to processing queue
self._add_to_processing_queue(file_id)
try:
# Read markdown
markdown_content = await self._read_markdown_from_minio(file_meta["markdown_file"])
file_path = file_meta.get("path")
filename = file_meta.get("filename") or file_id
processing_params = resolve_chunk_processing_params(
kb_additional_params=self.databases_meta.get(db_id, {}).get("metadata"),
file_processing_params=file_meta.get("processing_params"),
)
self.files_meta[file_id]["processing_params"] = processing_params
await self._save_metadata()
chunks = chunk_markdown(markdown_content, file_id, filename, processing_params)
chunk_input, split_by_character, split_by_character_only = self._prepare_lightrag_insert_payload(chunks)
if not chunk_input:
chunk_input = markdown_content
# Clean up existing chunks if any (for re-indexing)
await self.delete_file_chunks_only(db_id, file_id)
# Insert
await rag.ainsert(
input=chunk_input,
ids=file_id,
file_paths=file_path,
split_by_character=split_by_character,
split_by_character_only=split_by_character_only,
)
await self._ensure_doc_processed(rag, file_id)
logger.info(
f"Indexed file {file_id} into LightRAG with {len(chunks)} chunks, "
f"chunk_preset_id={processing_params.get('chunk_preset_id')}"
)
# Update status
self.files_meta[file_id]["status"] = FileStatus.INDEXED
self.files_meta[file_id]["updated_at"] = utc_isoformat()
if operator_id:
self.files_meta[file_id]["updated_by"] = operator_id
await self._persist_file(file_id)
return self.files_meta[file_id]
except Exception as e:
logger.error(f"Indexing failed for {file_id}: {e}")
self.files_meta[file_id]["status"] = FileStatus.ERROR_INDEXING
self.files_meta[file_id]["error"] = str(e)
self.files_meta[file_id]["updated_at"] = utc_isoformat()
if operator_id:
self.files_meta[file_id]["updated_by"] = operator_id
await self._persist_file(file_id)
raise
finally:
# Remove from processing queue
self._remove_from_processing_queue(file_id)
async def update_content(self, db_id: str, file_ids: list[str], params: dict | None = None) -> list[dict]:
"""更新内容 - 根据file_ids重新解析文件并更新向量库"""
if db_id not in self.databases_meta:
raise ValueError(f"Database {db_id} not found")
rag = await self._get_lightrag_instance(db_id)
if not rag:
raise ValueError(f"Failed to get LightRAG instance for {db_id}")
# 处理默认参数
if params is None:
params = {}
processed_items_info = []
for file_id in file_ids:
# 从元数据中获取文件信息
# Get file meta
if file_id not in self.files_meta:
logger.warning(f"File {file_id} not found in metadata, skipping")
continue
raise ValueError(f"File {file_id} not found")
file_meta = self.files_meta[file_id]
file_path = file_meta.get("path")
if not file_path:
logger.warning(f"File path not found for {file_id}, skipping")
continue
# Validate current status - only allow indexing from these states
current_status = file_meta.get("status")
allowed_statuses = {
FileStatus.PARSED,
FileStatus.ERROR_INDEXING,
FileStatus.INDEXED, # For re-indexing
"done", # Legacy status
}
# 添加到处理队列
if current_status not in allowed_statuses:
raise ValueError(
f"Cannot index file with status '{current_status}'. "
f"File must be parsed first (status should be one of: {', '.join(allowed_statuses)})"
)
# Check markdown file exists
if not file_meta.get("markdown_file"):
raise ValueError("File has not been parsed yet (no markdown_file)")
# Clear previous error if any
if "error" in file_meta:
self.files_meta[file_id].pop("error", None)
# Update status and add to processing queue
self.files_meta[file_id]["status"] = FileStatus.INDEXING
self.files_meta[file_id]["updated_at"] = utc_isoformat()
if operator_id:
self.files_meta[file_id]["updated_by"] = operator_id
await self._persist_file(file_id)
# Add to processing queue
self._add_to_processing_queue(file_id)
try:
# 更新状态为处理中
resolved_params = resolve_chunk_processing_params(
kb_additional_params=self.databases_meta.get(db_id, {}).get("metadata"),
file_processing_params=self.files_meta[file_id].get("processing_params"),
request_params=params,
)
self.files_meta[file_id]["processing_params"] = resolved_params
self.files_meta[file_id]["status"] = "processing"
await self._persist_file(file_id)
# 重新解析文件为 markdown
params["image_bucket"] = "public"
params["image_prefix"] = f"{db_id}/kb-images"
markdown_content = await Parser.aparse(source=file_path, params=params)
markdown_content_lines = markdown_content[:100].replace("\n", " ")
logger.info(f"Markdown content: {markdown_content_lines}...")
# Read markdown
markdown_content = await self._read_markdown_from_minio(file_meta["markdown_file"])
file_path = file_meta.get("path")
filename = file_meta.get("filename") or file_id
chunks = chunk_markdown(markdown_content, file_id, filename, resolved_params)
chunk_input, split_by_character, split_by_character_only = self._prepare_lightrag_insert_payload(chunks)
processing_params = resolve_chunk_processing_params(
kb_additional_params=self.databases_meta.get(db_id, {}).get("metadata"),
file_processing_params=file_meta.get("processing_params"),
)
self.files_meta[file_id]["processing_params"] = processing_params
await self._save_metadata()
chunks = chunk_markdown(markdown_content, file_id, filename, processing_params)
chunk_input, split_by_character, split_by_character_only = self._prepare_lightrag_insert_payload(
chunks
)
if not chunk_input:
chunk_input = markdown_content
# 先删除现有的 LightRAG 数据仅删除chunks保留元数据
# Clean up existing chunks if any (for re-indexing)
await self.delete_file_chunks_only(db_id, file_id)
# 使用 LightRAG 重新插入内容
# Insert
await rag.ainsert(
input=chunk_input,
ids=file_id,
@ -465,39 +393,137 @@ class LightRagKB(KnowledgeBase):
)
await self._ensure_doc_processed(rag, file_id)
logger.info(f"Updated file {file_path} in LightRAG. Done.")
logger.info(
f"Indexed file {file_id} into LightRAG with {len(chunks)} chunks, "
f"chunk_preset_id={processing_params.get('chunk_preset_id')}"
)
# 更新元数据状态
self.files_meta[file_id]["status"] = "done"
# Update status
self.files_meta[file_id]["status"] = FileStatus.INDEXED
self.files_meta[file_id]["updated_at"] = utc_isoformat()
if operator_id:
self.files_meta[file_id]["updated_by"] = operator_id
await self._persist_file(file_id)
# 从处理队列中移除
self._remove_from_processing_queue(file_id)
# 返回更新后的文件信息
updated_file_meta = file_meta.copy()
updated_file_meta["status"] = "done"
updated_file_meta["file_id"] = file_id
processed_items_info.append(updated_file_meta)
return self.files_meta[file_id]
except Exception as e:
error_msg = str(e)
logger.error(f"更新file {file_path} 失败: {error_msg}, {traceback.format_exc()}")
self.files_meta[file_id]["status"] = "failed"
self.files_meta[file_id]["error"] = error_msg
logger.error(f"Indexing failed for {file_id}: {e}")
self.files_meta[file_id]["status"] = FileStatus.ERROR_INDEXING
self.files_meta[file_id]["error"] = str(e)
self.files_meta[file_id]["updated_at"] = utc_isoformat()
if operator_id:
self.files_meta[file_id]["updated_by"] = operator_id
await self._persist_file(file_id)
raise
# 从处理队列中移除
finally:
# Remove from processing queue
self._remove_from_processing_queue(file_id)
# 返回失败的文件信息
failed_file_meta = file_meta.copy()
failed_file_meta["status"] = "failed"
failed_file_meta["file_id"] = file_id
failed_file_meta["error"] = error_msg
processed_items_info.append(failed_file_meta)
async def update_content(self, db_id: str, file_ids: list[str], params: dict | None = None) -> list[dict]:
"""更新内容 - 根据file_ids重新解析文件并更新向量库"""
if db_id not in self.databases_meta:
raise ValueError(f"Database {db_id} not found")
return processed_items_info
db_write_lock = await self._get_db_write_lock(db_id)
async with db_write_lock:
rag = await self._get_lightrag_instance(db_id)
if not rag:
raise ValueError(f"Failed to get LightRAG instance for {db_id}")
# 处理默认参数
if params is None:
params = {}
processed_items_info = []
for file_id in file_ids:
# 从元数据中获取文件信息
if file_id not in self.files_meta:
logger.warning(f"File {file_id} not found in metadata, skipping")
continue
file_meta = self.files_meta[file_id]
file_path = file_meta.get("path")
if not file_path:
logger.warning(f"File path not found for {file_id}, skipping")
continue
# 添加到处理队列
self._add_to_processing_queue(file_id)
try:
# 更新状态为处理中
resolved_params = resolve_chunk_processing_params(
kb_additional_params=self.databases_meta.get(db_id, {}).get("metadata"),
file_processing_params=self.files_meta[file_id].get("processing_params"),
request_params=params,
)
self.files_meta[file_id]["processing_params"] = resolved_params
self.files_meta[file_id]["status"] = "processing"
await self._persist_file(file_id)
# 重新解析文件为 markdown
params["image_bucket"] = "public"
params["image_prefix"] = f"{db_id}/kb-images"
markdown_content = await Parser.aparse(source=file_path, params=params)
markdown_content_lines = markdown_content[:100].replace("\n", " ")
logger.info(f"Markdown content: {markdown_content_lines}...")
filename = file_meta.get("filename") or file_id
chunks = chunk_markdown(markdown_content, file_id, filename, resolved_params)
chunk_input, split_by_character, split_by_character_only = self._prepare_lightrag_insert_payload(
chunks
)
if not chunk_input:
chunk_input = markdown_content
# 先删除现有的 LightRAG 数据仅删除chunks保留元数据
await self.delete_file_chunks_only(db_id, file_id)
# 使用 LightRAG 重新插入内容
await rag.ainsert(
input=chunk_input,
ids=file_id,
file_paths=file_path,
split_by_character=split_by_character,
split_by_character_only=split_by_character_only,
)
await self._ensure_doc_processed(rag, file_id)
logger.info(f"Updated file {file_path} in LightRAG. Done.")
# 更新元数据状态
self.files_meta[file_id]["status"] = "done"
await self._persist_file(file_id)
# 从处理队列中移除
self._remove_from_processing_queue(file_id)
# 返回更新后的文件信息
updated_file_meta = file_meta.copy()
updated_file_meta["status"] = "done"
updated_file_meta["file_id"] = file_id
processed_items_info.append(updated_file_meta)
except Exception as e:
error_msg = str(e)
logger.error(f"更新file {file_path} 失败: {error_msg}, {traceback.format_exc()}")
self.files_meta[file_id]["status"] = "failed"
self.files_meta[file_id]["error"] = error_msg
await self._persist_file(file_id)
# 从处理队列中移除
self._remove_from_processing_queue(file_id)
# 返回失败的文件信息
failed_file_meta = file_meta.copy()
failed_file_meta["status"] = "failed"
failed_file_meta["file_id"] = file_id
failed_file_meta["error"] = error_msg
processed_items_info.append(failed_file_meta)
return processed_items_info
async def aquery(self, query_text: str, db_id: str, agent_call: bool = False, **kwargs) -> str:
"""异步查询知识库"""

View File

@ -74,7 +74,7 @@ async def _convert_upload_to_markdown(upload: UploadFile) -> ConversionResult:
file_name = Path(upload.filename).name
suffix = Path(file_name).suffix.lower()
if suffix not in ATTACHMENT_ALLOWED_EXTENSIONS:
if ATTACHMENT_ALLOWED_EXTENSIONS and suffix not in ATTACHMENT_ALLOWED_EXTENSIONS:
allowed = ", ".join(ATTACHMENT_ALLOWED_EXTENSIONS)
raise ValueError(f"不支持的文件类型: {suffix or '未知'},当前仅支持 {allowed}")

View File

@ -109,12 +109,14 @@ members = [
[tool.pytest.ini_options]
addopts = "-v --tb=short"
testpaths = ["test"]
testpaths = ["test/unit", "test/integration", "test/e2e"]
pythonpath = ["."]
markers = [
"unit: marks tests that run without live services",
"auth: marks tests that require authentication",
"slow: marks tests as slow",
"integration: marks tests as integration tests"
"integration: marks tests as integration tests",
"e2e: marks tests as end-to-end tests"
]
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "function"

View File

@ -392,6 +392,7 @@ async def add_documents(
try:
# 2. Parse file (PARSING -> PARSED)
file_meta = await knowledge_base.parse_file(db_id, file_id, operator_id=current_user.user_id)
added_files[item] = (file_id, file_meta)
processed_items.append(file_meta)
parse_success_count += 1
except Exception as parse_error:

View File

@ -1,235 +0,0 @@
"""通过 Agent Runs API 验证文件创建与结果落盘的端到端脚本。
流程
1. 使用环境变量中的管理员账号登录
2. 创建 thread
3. 调用 Agent要求其创建冒泡排序脚本并执行
4. 通过线程文件 API 验证脚本文件与结果文件
运行
docker compose exec api uv run python test/api/test_agent_bubble_sort_e2e.py
"""
from __future__ import annotations
import asyncio
import json
import os
import sys
import uuid
from pathlib import Path
import httpx
from dotenv import load_dotenv
_root = Path(__file__).resolve().parents[2]
if str(_root) not in sys.path:
sys.path.insert(0, str(_root))
load_dotenv(".env", override=False)
load_dotenv("test/.env.test", override=False)
API_BASE_URL = os.getenv("TEST_BASE_URL", os.getenv("API_BASE_URL", "http://localhost:5050")).rstrip("/")
USERNAME = os.getenv("TEST_USERNAME") or os.getenv("E2E_USERNAME")
PASSWORD = os.getenv("TEST_PASSWORD") or os.getenv("E2E_PASSWORD")
TIMEOUT = httpx.Timeout(300.0, connect=10.0)
POLL_INTERVAL_SECONDS = float(os.getenv("E2E_RUN_POLL_INTERVAL_SECONDS", "2"))
RUN_TIMEOUT_SECONDS = int(os.getenv("E2E_RUN_TIMEOUT_SECONDS", "240"))
SCRIPT_PATH = "/home/gem/user-data/workspace/bubble_sort.py"
RESULT_PATH = "/home/gem/user-data/outputs/bubble_sort_result.txt"
EXPECTED_OUTPUT = "[1, 2, 4, 5, 8]"
class AgentBubbleSortE2ETester:
def __init__(self):
if not USERNAME or not PASSWORD:
raise RuntimeError("TEST_USERNAME / TEST_PASSWORD 未配置")
self.client = httpx.AsyncClient(base_url=API_BASE_URL, timeout=TIMEOUT, follow_redirects=True)
self.headers: dict[str, str] | None = None
self.agent_id: str | None = None
self.agent_config_id: int | None = None
self.thread_id: str | None = None
async def close(self):
await self.client.aclose()
async def login(self) -> None:
response = await self.client.post("/api/auth/token", data={"username": USERNAME, "password": PASSWORD})
if response.status_code != 200:
raise RuntimeError(f"login failed: {response.status_code} {response.text}")
token = response.json().get("access_token")
if not token:
raise RuntimeError("login succeeded but access_token is missing")
self.headers = {"Authorization": f"Bearer {token}"}
async def pick_agent(self) -> None:
assert self.headers
default_response = await self.client.get("/api/chat/default_agent", headers=self.headers)
if default_response.status_code == 200:
default_agent_id = default_response.json().get("default_agent_id")
if default_agent_id:
self.agent_id = str(default_agent_id)
return
response = await self.client.get("/api/chat/agent", headers=self.headers)
if response.status_code != 200:
raise RuntimeError(f"list agents failed: {response.status_code} {response.text}")
agents = response.json().get("agents") or []
if not agents:
raise RuntimeError("no available agents")
self.agent_id = str(agents[0]["id"])
async def create_thread(self) -> None:
assert self.headers and self.agent_id
response = await self.client.post(
"/api/chat/thread",
json={
"agent_id": self.agent_id,
"title": f"agent-bubble-sort-e2e-{uuid.uuid4().hex[:8]}",
"metadata": {},
},
headers=self.headers,
)
if response.status_code != 200:
raise RuntimeError(f"create thread failed: {response.status_code} {response.text}")
payload = response.json()
self.thread_id = str(payload.get("thread_id") or payload.get("id"))
if not self.thread_id:
raise RuntimeError(f"thread id missing: {payload}")
async def pick_agent_config(self) -> None:
assert self.headers and self.agent_id
response = await self.client.get(f"/api/chat/agent/{self.agent_id}/configs", headers=self.headers)
if response.status_code != 200:
raise RuntimeError(f"list configs failed: {response.status_code} {response.text}")
configs = response.json().get("configs") or []
if not configs:
raise RuntimeError("no available agent configs")
config_id = configs[0].get("id")
if not config_id:
raise RuntimeError(f"config id missing: {response.text}")
self.agent_config_id = int(config_id)
async def create_run(self, query: str) -> str:
assert self.headers and self.agent_config_id and self.thread_id
request_id = f"agent-bubble-sort-{uuid.uuid4()}"
response = await self.client.post(
"/api/chat/runs",
json={
"query": query,
"agent_config_id": self.agent_config_id,
"thread_id": self.thread_id,
"meta": {"request_id": request_id},
},
headers=self.headers,
)
if response.status_code != 200:
raise RuntimeError(f"create run failed: {response.status_code} {response.text}")
run_id = response.json().get("run_id")
if not run_id:
raise RuntimeError(f"run id missing: {response.text}")
return str(run_id)
async def wait_for_run(self, run_id: str) -> dict:
assert self.headers
deadline = asyncio.get_running_loop().time() + RUN_TIMEOUT_SECONDS
last_payload: dict | None = None
while asyncio.get_running_loop().time() < deadline:
response = await self.client.get(f"/api/chat/runs/{run_id}", headers=self.headers)
if response.status_code != 200:
raise RuntimeError(f"get run failed: {response.status_code} {response.text}")
last_payload = response.json().get("run") or {}
status = str(last_payload.get("status") or "")
if status in {"completed", "failed", "cancelled", "interrupted"}:
return last_payload
await asyncio.sleep(POLL_INTERVAL_SECONDS)
raise RuntimeError("run timeout: " + json.dumps(last_payload or {}, ensure_ascii=False))
async def list_thread_files(self, path: str) -> list[dict]:
assert self.headers and self.thread_id
response = await self.client.get(
f"/api/chat/thread/{self.thread_id}/files",
params={"path": path, "recursive": "true"},
headers=self.headers,
)
if response.status_code != 200:
raise RuntimeError(f"list thread files failed: {response.status_code} {response.text}")
files = response.json().get("files")
if not isinstance(files, list):
raise RuntimeError(f"invalid files response: {response.text}")
return files
async def read_thread_file(self, path: str) -> str:
assert self.headers and self.thread_id
response = await self.client.get(
f"/api/chat/thread/{self.thread_id}/files/content",
params={"path": path},
headers=self.headers,
)
if response.status_code != 200:
raise RuntimeError(f"read thread file failed: {response.status_code} {response.text}")
lines = response.json().get("content")
if not isinstance(lines, list):
raise RuntimeError(f"invalid file content response: {response.text}")
return "\n".join(str(line) for line in lines)
async def run_case(self) -> None:
await self.login()
await self.pick_agent()
await self.pick_agent_config()
await self.create_thread()
query = (
"在工作区创建一个 Python 冒泡排序脚本,并执行它。"
f"脚本必须保存到 {SCRIPT_PATH}"
"脚本内容要求:对列表 [5, 1, 4, 2, 8] 使用冒泡排序,并打印排序后的结果。"
f"然后执行该脚本,并将标准输出保存到 {RESULT_PATH}"
"不要写入其他路径。完成后简单回复这两个文件路径。"
)
run_id = await self.create_run(query)
run = await self.wait_for_run(run_id)
if run.get("status") != "completed":
raise RuntimeError("run did not complete successfully: " + json.dumps(run, ensure_ascii=False))
file_entries = await self.list_thread_files("/home/gem/user-data")
file_paths = {str(entry.get("path") or "") for entry in file_entries}
if SCRIPT_PATH not in file_paths:
raise RuntimeError("script file missing: " + json.dumps(sorted(file_paths), ensure_ascii=False))
if RESULT_PATH not in file_paths:
raise RuntimeError("result file missing: " + json.dumps(sorted(file_paths), ensure_ascii=False))
script_content = await self.read_thread_file(SCRIPT_PATH)
if "bubble" not in script_content.lower() and "for" not in script_content:
raise RuntimeError(f"unexpected script content: {script_content}")
result_content = await self.read_thread_file(RESULT_PATH)
if EXPECTED_OUTPUT not in result_content:
raise RuntimeError(f"unexpected result content: {result_content}")
print("[PASS] Agent bubble sort E2E completed")
print(f"thread_id={self.thread_id}")
print(f"run_id={run_id}")
print(f"script_path={SCRIPT_PATH}")
print(f"result_path={RESULT_PATH}")
print(f"result_content={result_content}")
async def main() -> None:
tester = AgentBubbleSortE2ETester()
try:
await tester.run_case()
finally:
await tester.close()
if __name__ == "__main__":
asyncio.run(main())

View File

@ -1,310 +0,0 @@
"""
测试附件上传和 agent state 获取的 API 脚本
使用方式:
cd /home/zwj/workspace/Yuxi
docker compose exec api uv run python test/api/test_attachment_and_agent_state.py
或者本地运行:
python test/api/test_attachment_and_agent_state.py
"""
import asyncio
import contextlib
import os
import sys
import uuid
from pathlib import Path
import httpx
from dotenv import load_dotenv
# 添加项目根目录到 Python 路径
PROJECT_ROOT = Path(__file__).parent.parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
load_dotenv(PROJECT_ROOT / ".env")
# API 配置
API_BASE_URL = os.getenv("API_BASE_URL", "http://localhost:5050")
# 测试账户配置
USERNAME = os.getenv("YUXI_SUPER_ADMIN_NAME", "zwj")
PASSWORD = os.getenv("YUXI_SUPER_ADMIN_PASSWORD", "zwj12138")
# 默认 Agent ID (需要根据实际情况修改,使用类名)
DEFAULT_AGENT_ID = "ChatbotAgent"
class APITester:
def __init__(self, base_url: str, username: str, password: str):
self.base_url = base_url
self.username = username
self.password = password
self.token: str | None = None
self.user_id: str | None = None
self.headers: dict | None = None
@contextlib.asynccontextmanager
async def _client(self, timeout: float = 30.0):
"""获取 HTTP 客户端"""
client = httpx.AsyncClient(timeout=timeout)
try:
yield client
finally:
await client.aclose()
async def login(self) -> bool:
"""登录获取 token"""
print(f"\n{'=' * 60}")
print(f"1. 正在登录: {self.username}")
print(f"{'=' * 60}")
async with self._client() as client:
response = await client.post(
f"{self.base_url}/api/auth/token",
data={"username": self.username, "password": self.password},
)
if response.status_code == 200:
data = response.json()
self.token = data.get("access_token")
self.user_id = str(data.get("user_id"))
self.headers = {"Authorization": f"Bearer {self.token}"}
print(f" ✓ 登录成功! user_id: {self.user_id}")
print(f" ✓ Token: {self.token[:50]}...")
return True
else:
print(f" ✗ 登录失败: {response.status_code} - {response.text}")
return False
async def create_thread(self, agent_id: str) -> str:
"""创建对话线程"""
print(f"\n{'=' * 60}")
print(f"2. 创建对话线程 (agent_id: {agent_id})")
print(f"{'=' * 60}")
async with self._client() as client:
response = await client.post(
f"{self.base_url}/api/chat/thread",
json={"agent_id": agent_id, "title": "API 测试对话"},
headers=self.headers,
)
if response.status_code == 200:
thread = response.json()
thread_id = thread.get("id")
print(f" ✓ 创建成功! thread_id: {thread_id}")
return thread_id
else:
print(f" ✗ 创建失败: {response.status_code} - {response.text}")
return ""
async def upload_attachment(self, thread_id: str, file_path: str) -> dict | None:
"""上传附件"""
print(f"\n{'=' * 60}")
print(f"3. 上传附件: {file_path}")
print(f"{'=' * 60}")
if not os.path.exists(file_path):
print(f" ✗ 文件不存在: {file_path}")
return None
async with self._client(timeout=60.0) as client:
with open(file_path, "rb") as f:
files = {"file": (os.path.basename(file_path), f)}
response = await client.post(
f"{self.base_url}/api/chat/thread/{thread_id}/attachments",
files=files,
headers=self.headers,
)
if response.status_code == 200:
attachment = response.json()
print(" ✓ 上传成功!")
print(f" file_id: {attachment.get('file_id')}")
print(f" file_name: {attachment.get('file_name')}")
print(f" status: {attachment.get('status')}")
return attachment
else:
print(f" ✗ 上传失败: {response.status_code} - {response.text}")
return None
async def list_attachments(self, thread_id: str) -> list[dict]:
"""列出附件"""
print(f"\n{'=' * 60}")
print(f"4. 列出附件 (thread_id: {thread_id})")
print(f"{'=' * 60}")
async with self._client() as client:
response = await client.get(
f"{self.base_url}/api/chat/thread/{thread_id}/attachments",
headers=self.headers,
)
if response.status_code == 200:
data = response.json()
attachments = data.get("attachments", [])
print(f" ✓ 获取到 {len(attachments)} 个附件:")
for att in attachments:
print(f" - {att.get('file_name')}: {att.get('status')}")
return attachments
else:
print(f" ✗ 获取失败: {response.status_code} - {response.text}")
return []
async def get_agent_state(self, agent_id: str, thread_id: str) -> dict | None:
"""获取 agent state"""
print(f"\n{'=' * 60}")
print(f"5. 获取 Agent State (agent_id: {agent_id}, thread_id: {thread_id})")
print(f"{'=' * 60}")
async with self._client() as client:
response = await client.get(
f"{self.base_url}/api/chat/agent/{agent_id}/state",
params={"thread_id": thread_id},
headers=self.headers,
)
if response.status_code == 200:
state = response.json()
agent_state = state.get("agent_state", {})
print(" ✓ 获取成功!")
print(f" files: {len(agent_state.get('files', {}))}")
print(f" todos: {len(agent_state.get('todos', []))}")
if agent_state.get("files"):
print(" 文件列表:")
for path in agent_state["files"]:
file_info = agent_state["files"][path]
print(f" - {path}: {len(file_info.get('content', []))}")
return state
else:
print(f" ✗ 获取失败: {response.status_code} - {response.text}")
return None
async def get_agent_config_id(self, agent_id: str) -> int | None:
"""获取指定 Agent 的一个配置 ID"""
async with self._client() as client:
response = await client.get(
f"{self.base_url}/api/chat/agent/{agent_id}/configs",
headers=self.headers,
)
if response.status_code != 200:
print(f" ✗ 获取 Agent 配置失败: {response.status_code} - {response.text}")
return None
configs = response.json().get("configs", [])
if not configs:
print(" ✗ 当前 Agent 没有可用配置")
return None
return configs[0].get("id")
async def send_chat_message(self, agent_id: str, thread_id: str, query: str) -> bool:
"""发送聊天消息(流式)"""
print(f"\n{'=' * 60}")
print("6. 发送聊天消息")
print(f"{'=' * 60}")
print(f" Query: {query}")
print(f" Thread ID: {thread_id}")
agent_config_id = await self.get_agent_config_id(agent_id)
if not agent_config_id:
return False
async with self._client(timeout=120.0) as client:
async with client.stream(
"POST",
f"{self.base_url}/api/chat/agent",
json={
"query": query,
"agent_config_id": agent_config_id,
"thread_id": thread_id,
},
headers=self.headers,
) as response:
print(f"\n 响应状态: {response.status_code}")
print(" 响应内容:")
async for chunk in response.aiter_lines():
if chunk:
print(f" {chunk[:150]}...")
return response.status_code == 200
async def main():
"""主测试流程"""
print("\n" + "=" * 60)
print(" 附件上传与 Agent State API 测试")
print("=" * 60)
print(f"\nAPI 地址: {API_BASE_URL}")
print(f"测试账户: {USERNAME}")
tester = APITester(API_BASE_URL, USERNAME, PASSWORD)
# 1. 登录
if not await tester.login():
print("\n!!! 登录失败,测试终止 !!!")
return
# 2. 创建线程(使用默认 agent_id
agent_id = DEFAULT_AGENT_ID
thread_id = await tester.create_thread(agent_id)
if not thread_id:
print("\n!!! 创建线程失败,测试终止 !!!")
return
# 3. 创建测试文件
test_content = """# 测试文档
这是一个用于 API 测试的 Markdown 文件
## 主要内容
- 第一点
- 第二点
- 第三点
```python
def hello():
print("Hello, World!")
```
"""
test_file_path = f"/tmp/test_attachment_{uuid.uuid4().hex[:8]}.md"
with open(test_file_path, "w", encoding="utf-8") as f:
f.write(test_content)
print(f"\n 测试文件已创建: {test_file_path}")
# 4. 上传附件
attachment = await tester.upload_attachment(thread_id, test_file_path)
# 5. 列出附件
await tester.list_attachments(thread_id)
# 6. 获取 agent state (验证附件是否在 state 中)
if attachment:
print("\n 等待后端处理...")
await asyncio.sleep(2)
await tester.get_agent_state(agent_id, thread_id)
# 7. 发送聊天消息测试
await tester.send_chat_message(agent_id, thread_id, "你好,请简单介绍一下你自己。")
# 8. 再次获取 agent state (验证 todos 等状态)
await asyncio.sleep(1)
await tester.get_agent_state(agent_id, thread_id)
# 清理测试文件
if os.path.exists(test_file_path):
os.remove(test_file_path)
print(f"\n 测试文件已清理: {test_file_path}")
print(f"\n{'=' * 60}")
print(" 测试完成!")
print(f"{'=' * 60}\n")
if __name__ == "__main__":
asyncio.run(main())

View File

@ -1,206 +0,0 @@
"""lite 模式下 viewer-oriented filesystem 端到端验证。"""
from __future__ import annotations
import asyncio
import os
import sys
import uuid
from pathlib import Path
import httpx
_root = Path(__file__).resolve().parents[2]
if str(_root) not in sys.path:
sys.path.insert(0, str(_root))
from yuxi.agents.backends.sandbox import ( # noqa: E402
ensure_thread_dirs,
sandbox_outputs_dir,
sandbox_uploads_dir,
sandbox_user_data_dir,
sandbox_workspace_dir,
)
API_BASE_URL = os.getenv("API_BASE_URL", "http://localhost:5050").rstrip("/")
USERNAME = os.getenv("E2E_USERNAME", "zwj")
PASSWORD = os.getenv("E2E_PASSWORD", "zwj12138")
TIMEOUT = httpx.Timeout(180.0, connect=10.0)
class ViewerFilesystemE2ETester:
def __init__(self):
self.client = httpx.AsyncClient(base_url=API_BASE_URL, timeout=TIMEOUT, follow_redirects=True)
self.headers: dict[str, str] | None = None
self.agent_id: str | None = None
self.thread_id: str | None = None
self.other_thread_id: str | None = None
async def close(self):
await self.client.aclose()
async def login(self) -> None:
resp = await self.client.post("/api/auth/token", data={"username": USERNAME, "password": PASSWORD})
if resp.status_code != 200:
raise RuntimeError(f"login failed: {resp.status_code} {resp.text}")
self.headers = {"Authorization": f"Bearer {resp.json()['access_token']}"}
async def pick_agent(self) -> None:
assert self.headers
default_resp = await self.client.get("/api/chat/default_agent", headers=self.headers)
if default_resp.status_code == 200 and default_resp.json().get("default_agent_id"):
self.agent_id = str(default_resp.json()["default_agent_id"])
return
agents_resp = await self.client.get("/api/chat/agent", headers=self.headers)
if agents_resp.status_code != 200:
raise RuntimeError(f"failed to list agents: {agents_resp.status_code} {agents_resp.text}")
agents = agents_resp.json().get("agents", [])
if not agents:
raise RuntimeError("no available agents")
self.agent_id = str(agents[0]["id"])
async def create_thread(self) -> None:
assert self.headers and self.agent_id
resp = await self.client.post(
"/api/chat/thread",
json={"agent_id": self.agent_id, "title": f"viewer-fs-e2e-{uuid.uuid4().hex[:8]}", "metadata": {}},
headers=self.headers,
)
if resp.status_code != 200:
raise RuntimeError(f"create thread failed: {resp.status_code} {resp.text}")
payload = resp.json()
self.thread_id = str(payload.get("thread_id") or payload.get("id"))
async def create_other_thread(self) -> None:
assert self.headers and self.agent_id
resp = await self.client.post(
"/api/chat/thread",
json={"agent_id": self.agent_id, "title": f"viewer-fs-e2e-{uuid.uuid4().hex[:8]}", "metadata": {}},
headers=self.headers,
)
if resp.status_code != 200:
raise RuntimeError(f"create second thread failed: {resp.status_code} {resp.text}")
payload = resp.json()
self.other_thread_id = str(payload.get("thread_id") or payload.get("id"))
async def tree(self, path: str, *, thread_id: str | None = None) -> list[dict]:
assert self.headers and self.thread_id and self.agent_id
target_thread_id = thread_id or self.thread_id
resp = await self.client.get(
"/api/viewer/filesystem/tree",
params={"thread_id": target_thread_id, "path": path, "agent_id": self.agent_id},
headers=self.headers,
)
if resp.status_code != 200:
raise RuntimeError(f"viewer tree failed on {path}: {resp.status_code} {resp.text}")
return list(resp.json().get("entries") or [])
async def file(self, path: str, *, thread_id: str | None = None) -> dict:
assert self.headers and self.thread_id and self.agent_id
target_thread_id = thread_id or self.thread_id
resp = await self.client.get(
"/api/viewer/filesystem/file",
params={"thread_id": target_thread_id, "path": path, "agent_id": self.agent_id},
headers=self.headers,
)
if resp.status_code != 200:
raise RuntimeError(f"viewer file failed on {path}: {resp.status_code} {resp.text}")
return dict(resp.json())
async def download(self, path: str, *, thread_id: str | None = None) -> tuple[str, bytes]:
assert self.headers and self.thread_id and self.agent_id
target_thread_id = thread_id or self.thread_id
resp = await self.client.get(
"/api/viewer/filesystem/download",
params={"thread_id": target_thread_id, "path": path, "agent_id": self.agent_id},
headers=self.headers,
)
if resp.status_code != 200:
raise RuntimeError(f"viewer download failed on {path}: {resp.status_code} {resp.text}")
return resp.headers.get("content-disposition", ""), resp.content
async def run_case(self) -> None:
await self.login()
await self.pick_agent()
await self.create_thread()
await self.create_other_thread()
assert self.thread_id and self.other_thread_id
ensure_thread_dirs(self.thread_id)
ensure_thread_dirs(self.other_thread_id)
(sandbox_user_data_dir(self.thread_id) / "root-note.txt").write_text("root-visible\n", encoding="utf-8")
(sandbox_workspace_dir(self.thread_id) / "demo.py").write_text("print(42)\n", encoding="utf-8")
uploads_dir = sandbox_uploads_dir(self.thread_id) / "attachments"
uploads_dir.mkdir(parents=True, exist_ok=True)
(uploads_dir / "thread1.txt").write_text("thread-one-upload\n", encoding="utf-8")
(sandbox_outputs_dir(self.thread_id) / "result.txt").write_text("viewer-output\n", encoding="utf-8")
root_paths = {str(e.get("path", "")) for e in await self.tree("/")}
if "/home/gem/user-data/" not in root_paths:
raise RuntimeError(f"viewer root missing user-data: {sorted(root_paths)}")
user_data_paths = {str(e.get("path", "")) for e in await self.tree("/home/gem/user-data")}
expected_root_paths = {
"/home/gem/user-data/workspace/",
"/home/gem/user-data/uploads/",
"/home/gem/user-data/outputs/",
"/home/gem/user-data/root-note.txt",
}
if not expected_root_paths.issubset(user_data_paths):
raise RuntimeError(f"viewer user-data root mismatch: {sorted(user_data_paths)}")
workspace_paths = {str(e.get("path", "")) for e in await self.tree("/home/gem/user-data/workspace")}
if "/home/gem/user-data/workspace/demo.py" not in workspace_paths:
raise RuntimeError(f"viewer workspace missing demo.py: {sorted(workspace_paths)}")
other_workspace_paths = {
str(e.get("path", ""))
for e in await self.tree(
"/home/gem/user-data/workspace",
thread_id=self.other_thread_id,
)
}
if "/home/gem/user-data/workspace/demo.py" not in other_workspace_paths:
raise RuntimeError(f"shared workspace missing in second thread: {sorted(other_workspace_paths)}")
other_upload_paths = {
str(e.get("path", ""))
for e in await self.tree(
"/home/gem/user-data/uploads",
thread_id=self.other_thread_id,
)
}
if "/home/gem/user-data/uploads/attachments/" in other_upload_paths:
raise RuntimeError(f"thread-local uploads leaked to second thread: {sorted(other_upload_paths)}")
file_payload = await self.file("/home/gem/user-data/workspace/demo.py")
if file_payload.get("content") != "print(42)\n":
raise RuntimeError(f"unexpected viewer file content: {file_payload!r}")
if file_payload.get("preview_type") != "text" or file_payload.get("supported") is not True:
raise RuntimeError(f"unexpected viewer file preview metadata: {file_payload!r}")
other_file_payload = await self.file("/home/gem/user-data/workspace/demo.py", thread_id=self.other_thread_id)
if other_file_payload.get("content") != "print(42)\n":
raise RuntimeError(f"unexpected shared workspace content: {other_file_payload!r}")
content_disposition, payload = await self.download("/home/gem/user-data/outputs/result.txt")
if "result.txt" not in content_disposition:
raise RuntimeError(f"unexpected content-disposition: {content_disposition}")
if payload != b"viewer-output\n":
raise RuntimeError(f"unexpected download payload: {payload!r}")
print("[PASS] Viewer filesystem E2E completed")
print(f"thread_id={self.thread_id}")
async def main() -> None:
tester = ViewerFilesystemE2ETester()
try:
await tester.run_case()
finally:
await tester.close()
if __name__ == "__main__":
asyncio.run(main())

View File

@ -1,304 +1,26 @@
"""
Shared pytest fixtures for exercising FastAPI routers over the running API service.
"""
# ruff: noqa: E402
from __future__ import annotations
import os
import json
import subprocess
import sys
import uuid
from collections.abc import AsyncGenerator
from pathlib import Path
# Ensure /app is on sys.path so 'server' and 'yuxi' modules resolve correctly.
_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if _root not in sys.path:
sys.path.insert(0, _root)
import anyio
import httpx
import pytest
import pytest_asyncio
from dotenv import load_dotenv
# Load project and test specific environment variables.
load_dotenv(".env", override=False)
load_dotenv("test/.env.test", override=False)
PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
API_BASE_URL = os.getenv("TEST_BASE_URL", "http://localhost:5050").rstrip("/")
ADMIN_LOGIN = os.getenv("TEST_USERNAME")
ADMIN_PASSWORD = os.getenv("TEST_PASSWORD")
assert ADMIN_LOGIN, "TEST_USERNAME is not set"
assert ADMIN_PASSWORD, "TEST_PASSWORD is not set"
_ADMIN_TOKEN_CACHE: str | None = None
HTTP_TIMEOUT = httpx.Timeout(30.0, connect=5.0)
SANDBOX_CONTAINER_PREFIX = os.getenv("YUXI_SANDBOX_CONTAINER_PREFIX", "yuxi-sandbox")
@pytest_asyncio.fixture(scope="function")
async def test_client() -> AsyncGenerator[httpx.AsyncClient, None]:
"""Async HTTP client bound to the live API base URL."""
async with httpx.AsyncClient(base_url=API_BASE_URL, timeout=HTTP_TIMEOUT, follow_redirects=True) as client:
yield client
@pytest_asyncio.fixture(scope="function")
async def admin_token() -> str:
"""Authenticate with super admin credentials and cache the bearer token."""
global _ADMIN_TOKEN_CACHE
if _ADMIN_TOKEN_CACHE:
return _ADMIN_TOKEN_CACHE
if not ADMIN_LOGIN or not ADMIN_PASSWORD:
pytest.skip("Admin credentials are not configured via environment variables.")
async with httpx.AsyncClient(
base_url=API_BASE_URL, timeout=HTTP_TIMEOUT, follow_redirects=True
) as bootstrap_client:
response = await bootstrap_client.post(
"/api/auth/token", data={"username": ADMIN_LOGIN, "password": ADMIN_PASSWORD}
)
if response.status_code == 401:
first_run_response = await bootstrap_client.get("/api/auth/check-first-run")
if first_run_response.status_code == 200 and first_run_response.json().get("first_run", False):
pytest.fail(
"Super admin account has not been initialized. Please complete `/api/auth/initialize` "
"before running router tests."
)
if response.status_code != 200:
pytest.fail(f"Failed to authenticate as admin (status={response.status_code}): {response.text}")
token = response.json().get("access_token")
if not token:
pytest.fail("Admin authentication did not return an access token.")
_ADMIN_TOKEN_CACHE = token
return token
@pytest.fixture(scope="function")
def admin_headers(admin_token: str) -> dict[str, str]:
"""Authorization headers for the super admin user."""
return {"Authorization": f"Bearer {admin_token}"}
@pytest.fixture(scope="session", autouse=True)
def cleanup_test_knowledge_databases():
"""
Best-effort cleanup for leftover test knowledge databases (e.g. pytest_* / py_test*).
"""
async def run_cleanup() -> None:
global _ADMIN_TOKEN_CACHE
if not ADMIN_LOGIN or not ADMIN_PASSWORD:
return
if not _ADMIN_TOKEN_CACHE:
async with httpx.AsyncClient(
base_url=API_BASE_URL, timeout=HTTP_TIMEOUT, follow_redirects=True
) as bootstrap_client:
response = await bootstrap_client.post(
"/api/auth/token", data={"username": ADMIN_LOGIN, "password": ADMIN_PASSWORD}
)
if response.status_code != 200:
return
token = response.json().get("access_token")
if not token:
return
_ADMIN_TOKEN_CACHE = token
headers = {"Authorization": f"Bearer {_ADMIN_TOKEN_CACHE}"}
async with httpx.AsyncClient(base_url=API_BASE_URL, timeout=HTTP_TIMEOUT, follow_redirects=True) as client:
try:
list_response = await client.get("/api/knowledge/databases", headers=headers)
except Exception as e:
print(f"Warning: Failed to list knowledge databases for cleanup: {e}")
return
if list_response.status_code != 200:
return
databases = list_response.json().get("databases", [])
prefixes = ("pytest_", "py_test")
for entry in databases:
name = entry.get("name") or ""
db_id = entry.get("db_id")
if not db_id or not isinstance(name, str) or not name.startswith(prefixes):
continue
try:
delete_response = await client.delete(f"/api/knowledge/databases/{db_id}", headers=headers)
if delete_response.status_code not in (200, 404):
print(f"Warning: Failed to cleanup knowledge database {db_id}: {delete_response.text}")
except Exception as e:
print(f"Warning: Exception during cleanup of {db_id}: {e}")
try:
anyio.run(run_cleanup)
except Exception as e:
print(f"Warning: Exception during session cleanup startup: {e}")
yield
try:
anyio.run(run_cleanup)
except Exception as e:
print(f"Warning: Exception during session cleanup teardown: {e}")
def _docker_api_request(method: str, path: str) -> list[dict] | dict:
cmd = [
"curl",
"-sS",
"--unix-socket",
os.getenv("YUXI_DOCKER_API_SOCKET", "/var/run/docker.sock"),
"-X",
method,
f"{os.getenv('YUXI_DOCKER_API_BASE', 'http://localhost').rstrip('/')}{path}",
]
result = subprocess.run(cmd, capture_output=True, text=True, timeout=15, check=False)
if result.returncode != 0:
raise RuntimeError(result.stderr.strip() or "Docker API request failed")
text = (result.stdout or "").strip()
if not text:
return {}
return json.loads(text)
def _cleanup_sandbox_containers() -> None:
try:
containers = _docker_api_request("GET", "/containers/json?all=true")
except Exception as exc:
print(f"Warning: Failed to list sandbox containers for cleanup: {exc}")
return
for container in containers if isinstance(containers, list) else []:
names = container.get("Names") or []
if not any(name.lstrip("/").startswith(f"{SANDBOX_CONTAINER_PREFIX}-") for name in names):
continue
container_id = container.get("Id")
if not container_id:
continue
try:
_docker_api_request("POST", f"/containers/{container_id}/stop?t=2")
except Exception:
pass
try:
_docker_api_request("DELETE", f"/containers/{container_id}?force=true")
except Exception as exc:
print(f"Warning: Failed to cleanup sandbox container {container_id[:12]}: {exc}")
@pytest.fixture(scope="session", autouse=True)
def cleanup_test_sandboxes():
_cleanup_sandbox_containers()
yield
_cleanup_sandbox_containers()
@pytest_asyncio.fixture(scope="function")
async def standard_user(test_client: httpx.AsyncClient, admin_headers: dict[str, str]) -> dict:
"""
Provision a temporary standard user for permission checks.
Yields a dictionary with `user`, `password`, and `headers` keys.
"""
username = f"pytest_user_{uuid.uuid4().hex[:8]}"
password = f"Pw!{uuid.uuid4().hex[:8]}"
response = await test_client.post(
"/api/auth/users",
json={"username": username, "password": password, "role": "user"},
headers=admin_headers,
)
if response.status_code != 200:
pytest.fail(f"Failed to create standard user (status={response.status_code}): {response.text}")
user_payload = response.json()
login_response = await test_client.post(
"/api/auth/token",
data={"username": user_payload["user_id"], "password": password},
)
if login_response.status_code != 200:
pytest.fail(
f"Failed to authenticate as standard user (status={login_response.status_code}): {login_response.text}"
)
access_token = login_response.json().get("access_token")
if not access_token:
pytest.fail("Standard user login succeeded but no access token was returned.")
try:
yield {
"user": user_payload,
"password": password,
"headers": {"Authorization": f"Bearer {access_token}"},
}
finally:
response = await test_client.delete(f"/api/auth/users/{user_payload['id']}", headers=admin_headers)
assert response.status_code == 200, f"Failed to cleanup test user {user_payload['user_id']}: {response.text}"
@pytest_asyncio.fixture(scope="function")
async def knowledge_database(test_client: httpx.AsyncClient, admin_headers: dict[str, str]) -> dict:
"""
Create a temporary knowledge database for tests that need LightRAG metadata.
"""
import time
# 使用UUID作为数据库名称的一部分确保唯一性
unique_id = uuid.uuid4().hex
timestamp = int(time.time() * 1000000) # 微秒级时间戳
db_name = f"pytest_kb_{timestamp}_{unique_id}"
db_id = None
try:
create_response = await test_client.post(
"/api/knowledge/databases",
json={
"database_name": db_name,
"description": "Pytest managed knowledge base",
"embed_model_name": "siliconflow/BAAI/bge-m3",
"kb_type": "lightrag",
"additional_params": {},
},
headers=admin_headers,
)
if create_response.status_code == 200:
db_payload = create_response.json()
db_id = db_payload["db_id"]
elif create_response.status_code == 409:
error_detail = create_response.json().get("detail", "")
pytest.fail(f"Knowledge database name conflict: {error_detail}. Please clean up old test databases first.")
else:
pytest.fail(
f"Failed to create knowledge database (status={create_response.status_code}): {create_response.text}"
)
yield db_payload if db_id else {"db_id": db_id, "name": db_name}
finally:
# 确保清理,即使测试失败
if db_id:
try:
delete_response = await test_client.delete(f"/api/knowledge/databases/{db_id}", headers=admin_headers)
if delete_response.status_code != 200:
print(f"Warning: Failed to cleanup knowledge database {db_id}: {delete_response.text}")
except Exception as e:
print(f"Warning: Exception during cleanup of {db_id}: {e}")
# Avoid package-level knowledge graph initialization during pytest collection.
os.environ.setdefault("YUXI_SKIP_APP_INIT", "1")
def pytest_configure(config: pytest.Config) -> None:
"""Register commonly used custom markers."""
"""Register shared markers without binding every test to a live environment."""
config.addinivalue_line("markers", "unit: marks tests that run without live services")
config.addinivalue_line("markers", "auth: marks tests that require authentication")
config.addinivalue_line("markers", "integration: marks tests that hit the live API service")
config.addinivalue_line("markers", "e2e: marks tests that exercise an end-to-end workflow")
config.addinivalue_line("markers", "slow: marks tests as slow")
pytest_plugins = ["pytest_asyncio"]

View File

@ -0,0 +1,92 @@
from __future__ import annotations
import os
import sys
from collections.abc import AsyncGenerator
from pathlib import Path
import httpx
import pytest
import pytest_asyncio
from dotenv import load_dotenv
PROJECT_ROOT = Path(__file__).resolve().parents[2]
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
load_dotenv(PROJECT_ROOT / ".env", override=False)
load_dotenv(PROJECT_ROOT / "test/.env.test", override=False)
E2E_BASE_URL = os.getenv("TEST_BASE_URL", os.getenv("API_BASE_URL", "http://localhost:5050")).rstrip("/")
E2E_USERNAME = os.getenv("E2E_USERNAME") or os.getenv("TEST_USERNAME")
E2E_PASSWORD = os.getenv("E2E_PASSWORD") or os.getenv("TEST_PASSWORD")
E2E_TIMEOUT = httpx.Timeout(300.0, connect=10.0)
def _require_e2e_credentials() -> tuple[str, str]:
if not E2E_USERNAME or not E2E_PASSWORD:
pytest.skip(
"E2E credentials are not configured via E2E_USERNAME / E2E_PASSWORD "
"or TEST_USERNAME / TEST_PASSWORD."
)
return E2E_USERNAME, E2E_PASSWORD
@pytest.fixture(scope="session")
def e2e_base_url() -> str:
return E2E_BASE_URL
@pytest_asyncio.fixture(scope="function")
async def e2e_client(e2e_base_url: str) -> AsyncGenerator[httpx.AsyncClient, None]:
async with httpx.AsyncClient(base_url=e2e_base_url, timeout=E2E_TIMEOUT, follow_redirects=True) as client:
yield client
@pytest_asyncio.fixture(scope="function")
async def e2e_headers(e2e_client: httpx.AsyncClient) -> dict[str, str]:
username, password = _require_e2e_credentials()
response = await e2e_client.post("/api/auth/token", data={"username": username, "password": password})
if response.status_code != 200:
pytest.fail(f"E2E login failed (status={response.status_code}): {response.text}")
access_token = response.json().get("access_token")
if not access_token:
pytest.fail("E2E login succeeded but no access token was returned.")
return {"Authorization": f"Bearer {access_token}"}
@pytest_asyncio.fixture(scope="function")
async def e2e_agent_context(e2e_client: httpx.AsyncClient, e2e_headers: dict[str, str]) -> dict[str, str | int]:
default_response = await e2e_client.get("/api/chat/default_agent", headers=e2e_headers)
default_agent_id = None
if default_response.status_code == 200:
default_agent_id = default_response.json().get("default_agent_id")
if default_agent_id:
agent_id = str(default_agent_id)
else:
response = await e2e_client.get("/api/chat/agent", headers=e2e_headers)
if response.status_code != 200:
pytest.fail(f"Failed to list agents for E2E tests (status={response.status_code}): {response.text}")
agents = response.json().get("agents") or []
if not agents:
pytest.fail("No agents are available for E2E tests.")
agent_id = str(agents[0]["id"])
config_response = await e2e_client.get(f"/api/chat/agent/{agent_id}/configs", headers=e2e_headers)
if config_response.status_code != 200:
pytest.fail(
"Failed to list agent configs for E2E tests "
f"(status={config_response.status_code}): {config_response.text}"
)
configs = config_response.json().get("configs") or []
if not configs:
pytest.fail(f"No agent configs are available for E2E tests under agent {agent_id}.")
config_id = configs[0].get("id")
if not config_id:
pytest.fail(f"Agent config payload missing id field for agent {agent_id}.")
return {"agent_id": agent_id, "agent_config_id": int(config_id)}

View File

@ -0,0 +1,144 @@
from __future__ import annotations
import asyncio
import json
import os
import uuid
import httpx
import pytest
pytestmark = [pytest.mark.asyncio, pytest.mark.e2e, pytest.mark.slow]
POLL_INTERVAL_SECONDS = float(os.getenv("E2E_RUN_POLL_INTERVAL_SECONDS", "2"))
RUN_TIMEOUT_SECONDS = int(os.getenv("E2E_RUN_TIMEOUT_SECONDS", "240"))
SCRIPT_PATH = "/home/gem/user-data/workspace/bubble_sort.py"
RESULT_PATH = "/home/gem/user-data/outputs/bubble_sort_result.txt"
EXPECTED_OUTPUT = "[1, 2, 4, 5, 8]"
async def _create_thread(client: httpx.AsyncClient, headers: dict[str, str], agent_id: str) -> str:
response = await client.post(
"/api/chat/thread",
json={
"agent_id": agent_id,
"title": f"agent-bubble-sort-e2e-{uuid.uuid4().hex[:8]}",
"metadata": {},
},
headers=headers,
)
assert response.status_code == 200, response.text
payload = response.json()
thread_id = payload.get("thread_id") or payload.get("id")
assert thread_id, payload
return str(thread_id)
async def _create_run(
client: httpx.AsyncClient,
headers: dict[str, str],
*,
agent_config_id: int,
thread_id: str,
query: str,
) -> str:
request_id = f"agent-bubble-sort-{uuid.uuid4()}"
response = await client.post(
"/api/chat/runs",
json={
"query": query,
"agent_config_id": agent_config_id,
"thread_id": thread_id,
"meta": {"request_id": request_id},
},
headers=headers,
)
assert response.status_code == 200, response.text
run_id = response.json().get("run_id")
assert run_id, response.text
return str(run_id)
async def _wait_for_run(client: httpx.AsyncClient, headers: dict[str, str], run_id: str) -> dict:
deadline = asyncio.get_running_loop().time() + RUN_TIMEOUT_SECONDS
last_payload: dict | None = None
while asyncio.get_running_loop().time() < deadline:
response = await client.get(f"/api/chat/runs/{run_id}", headers=headers)
assert response.status_code == 200, response.text
last_payload = response.json().get("run") or {}
status = str(last_payload.get("status") or "")
if status in {"completed", "failed", "cancelled", "interrupted"}:
return last_payload
await asyncio.sleep(POLL_INTERVAL_SECONDS)
pytest.fail("Run timed out: " + json.dumps(last_payload or {}, ensure_ascii=False))
async def _list_thread_files(
client: httpx.AsyncClient,
headers: dict[str, str],
thread_id: str,
path: str,
) -> list[dict]:
response = await client.get(
f"/api/chat/thread/{thread_id}/files",
params={"path": path, "recursive": "true"},
headers=headers,
)
assert response.status_code == 200, response.text
files = response.json().get("files")
assert isinstance(files, list), response.text
return files
async def _read_thread_file(client: httpx.AsyncClient, headers: dict[str, str], thread_id: str, path: str) -> str:
response = await client.get(
f"/api/chat/thread/{thread_id}/files/content",
params={"path": path},
headers=headers,
)
assert response.status_code == 200, response.text
lines = response.json().get("content")
assert isinstance(lines, list), response.text
return "\n".join(str(line) for line in lines)
async def test_agent_bubble_sort_run_creates_expected_artifacts(
e2e_client: httpx.AsyncClient,
e2e_headers: dict[str, str],
e2e_agent_context: dict[str, str | int],
):
thread_id = await _create_thread(e2e_client, e2e_headers, str(e2e_agent_context["agent_id"]))
query = (
"在工作区创建一个 Python 冒泡排序脚本,并执行它。"
f"脚本必须保存到 {SCRIPT_PATH}"
"脚本内容要求:对列表 [5, 1, 4, 2, 8] 使用冒泡排序,并打印排序后的结果。"
f"然后执行该脚本,并将标准输出保存到 {RESULT_PATH}"
"不要写入其他路径。完成后简单回复这两个文件路径。"
)
run_id = await _create_run(
e2e_client,
e2e_headers,
agent_config_id=int(e2e_agent_context["agent_config_id"]),
thread_id=thread_id,
query=query,
)
run_payload = await _wait_for_run(e2e_client, e2e_headers, run_id)
assert run_payload.get("status") == "completed", run_payload
file_entries = await _list_thread_files(e2e_client, e2e_headers, thread_id, "/home/gem/user-data")
file_paths = {str(entry.get("path") or "") for entry in file_entries}
assert SCRIPT_PATH in file_paths, sorted(file_paths)
assert RESULT_PATH in file_paths, sorted(file_paths)
script_content = await _read_thread_file(e2e_client, e2e_headers, thread_id, SCRIPT_PATH)
assert "bubble" in script_content.lower() or "for" in script_content, script_content
result_content = await _read_thread_file(e2e_client, e2e_headers, thread_id, RESULT_PATH)
assert EXPECTED_OUTPUT in result_content, result_content

View File

@ -0,0 +1,134 @@
from __future__ import annotations
import asyncio
import uuid
from pathlib import Path
import httpx
import pytest
pytestmark = [pytest.mark.asyncio, pytest.mark.e2e, pytest.mark.slow]
async def _create_thread(client: httpx.AsyncClient, headers: dict[str, str], agent_id: str) -> str:
response = await client.post(
"/api/chat/thread",
json={"agent_id": agent_id, "title": f"attachment-state-e2e-{uuid.uuid4().hex[:8]}", "metadata": {}},
headers=headers,
)
assert response.status_code == 200, response.text
payload = response.json()
thread_id = payload.get("thread_id") or payload.get("id")
assert thread_id, payload
return str(thread_id)
async def _upload_attachment(
client: httpx.AsyncClient,
headers: dict[str, str],
*,
thread_id: str,
file_path: Path,
) -> dict:
with file_path.open("rb") as handle:
response = await client.post(
f"/api/chat/thread/{thread_id}/attachments",
files={"file": (file_path.name, handle)},
headers=headers,
)
assert response.status_code == 200, response.text
return dict(response.json())
async def _list_attachments(client: httpx.AsyncClient, headers: dict[str, str], *, thread_id: str) -> list[dict]:
response = await client.get(f"/api/chat/thread/{thread_id}/attachments", headers=headers)
assert response.status_code == 200, response.text
return list(response.json().get("attachments") or [])
async def _get_agent_state(
client: httpx.AsyncClient,
headers: dict[str, str],
*,
thread_id: str,
) -> dict:
response = await client.get(f"/api/chat/thread/{thread_id}/state", headers=headers)
assert response.status_code == 200, response.text
return dict(response.json())
async def _send_chat_message(
client: httpx.AsyncClient,
headers: dict[str, str],
*,
thread_id: str,
agent_config_id: int,
query: str,
) -> None:
async with client.stream(
"POST",
"/api/chat/agent",
json={
"query": query,
"agent_config_id": agent_config_id,
"thread_id": thread_id,
},
headers=headers,
) as response:
assert response.status_code == 200, response.text
lines = [line async for line in response.aiter_lines() if line]
assert lines, "Streaming chat response should not be empty."
async def test_attachment_upload_is_reflected_in_agent_state(
tmp_path: Path,
e2e_client: httpx.AsyncClient,
e2e_headers: dict[str, str],
e2e_agent_context: dict[str, str | int],
):
agent_id = str(e2e_agent_context["agent_id"])
agent_config_id = int(e2e_agent_context["agent_config_id"])
thread_id = await _create_thread(e2e_client, e2e_headers, agent_id)
test_file = tmp_path / "attachment-state.md"
test_file.write_text(
"# 测试文档\n\n这是一个用于附件状态验证的 Markdown 文件。\n\n- 第一点\n- 第二点\n",
encoding="utf-8",
)
attachment_payload = await _upload_attachment(
e2e_client,
e2e_headers,
thread_id=thread_id,
file_path=test_file,
)
attachments = await _list_attachments(e2e_client, e2e_headers, thread_id=thread_id)
attachment_names = {item.get("file_name") for item in attachments}
assert test_file.name in attachment_names, attachments
assert attachment_payload.get("file_name") == test_file.name, attachment_payload
await asyncio.sleep(2)
state_payload = await _get_agent_state(
e2e_client,
e2e_headers,
thread_id=thread_id,
)
agent_state = state_payload.get("agent_state") or {}
assert {"files", "todos", "artifacts"}.issubset(agent_state.keys()), agent_state
await _send_chat_message(
e2e_client,
e2e_headers,
thread_id=thread_id,
agent_config_id=agent_config_id,
query="你好,请简单介绍一下你自己。",
)
await asyncio.sleep(1)
state_after_chat = await _get_agent_state(
e2e_client,
e2e_headers,
thread_id=thread_id,
)
assert "agent_state" in state_after_chat, state_after_chat

View File

@ -0,0 +1,195 @@
from __future__ import annotations
import uuid
import httpx
import pytest
from yuxi.agents.backends.sandbox import (
ensure_thread_dirs,
sandbox_outputs_dir,
sandbox_uploads_dir,
sandbox_user_data_dir,
sandbox_workspace_dir,
)
pytestmark = [pytest.mark.asyncio, pytest.mark.e2e, pytest.mark.slow]
async def _create_thread(client: httpx.AsyncClient, headers: dict[str, str], agent_id: str) -> str:
response = await client.post(
"/api/chat/thread",
json={"agent_id": agent_id, "title": f"viewer-fs-e2e-{uuid.uuid4().hex[:8]}", "metadata": {}},
headers=headers,
)
assert response.status_code == 200, response.text
payload = response.json()
thread_id = payload.get("thread_id") or payload.get("id")
assert thread_id, payload
return str(thread_id)
async def _tree(
client: httpx.AsyncClient,
headers: dict[str, str],
*,
agent_id: str,
thread_id: str,
path: str,
) -> list[dict]:
response = await client.get(
"/api/viewer/filesystem/tree",
params={"thread_id": thread_id, "path": path, "agent_id": agent_id},
headers=headers,
)
assert response.status_code == 200, response.text
return list(response.json().get("entries") or [])
async def _file(
client: httpx.AsyncClient,
headers: dict[str, str],
*,
agent_id: str,
thread_id: str,
path: str,
) -> dict:
response = await client.get(
"/api/viewer/filesystem/file",
params={"thread_id": thread_id, "path": path, "agent_id": agent_id},
headers=headers,
)
assert response.status_code == 200, response.text
return dict(response.json())
async def _download(
client: httpx.AsyncClient,
headers: dict[str, str],
*,
agent_id: str,
thread_id: str,
path: str,
) -> tuple[str, bytes]:
response = await client.get(
"/api/viewer/filesystem/download",
params={"thread_id": thread_id, "path": path, "agent_id": agent_id},
headers=headers,
)
assert response.status_code == 200, response.text
return response.headers.get("content-disposition", ""), response.content
async def test_viewer_filesystem_e2e_respects_workspace_sharing_and_thread_local_uploads(
e2e_client: httpx.AsyncClient,
e2e_headers: dict[str, str],
e2e_agent_context: dict[str, str | int],
):
agent_id = str(e2e_agent_context["agent_id"])
thread_id = await _create_thread(e2e_client, e2e_headers, agent_id)
other_thread_id = await _create_thread(e2e_client, e2e_headers, agent_id)
ensure_thread_dirs(thread_id)
ensure_thread_dirs(other_thread_id)
(sandbox_user_data_dir(thread_id) / "root-note.txt").write_text("root-visible\n", encoding="utf-8")
(sandbox_workspace_dir(thread_id) / "demo.py").write_text("print(42)\n", encoding="utf-8")
uploads_dir = sandbox_uploads_dir(thread_id) / "attachments"
uploads_dir.mkdir(parents=True, exist_ok=True)
(uploads_dir / "thread1.txt").write_text("thread-one-upload\n", encoding="utf-8")
(sandbox_outputs_dir(thread_id) / "result.txt").write_text("viewer-output\n", encoding="utf-8")
root_entries = await _tree(
e2e_client,
e2e_headers,
agent_id=agent_id,
thread_id=thread_id,
path="/",
)
root_paths = {str(entry.get("path", "")) for entry in root_entries}
assert "/home/gem/user-data/" in root_paths, sorted(root_paths)
user_data_paths = {
str(entry.get("path", ""))
for entry in await _tree(
e2e_client,
e2e_headers,
agent_id=agent_id,
thread_id=thread_id,
path="/home/gem/user-data",
)
}
expected_root_paths = {
"/home/gem/user-data/workspace/",
"/home/gem/user-data/uploads/",
"/home/gem/user-data/outputs/",
"/home/gem/user-data/root-note.txt",
}
assert expected_root_paths.issubset(user_data_paths), sorted(user_data_paths)
workspace_paths = {
str(entry.get("path", ""))
for entry in await _tree(
e2e_client,
e2e_headers,
agent_id=agent_id,
thread_id=thread_id,
path="/home/gem/user-data/workspace",
)
}
assert "/home/gem/user-data/workspace/demo.py" in workspace_paths, sorted(workspace_paths)
other_workspace_paths = {
str(entry.get("path", ""))
for entry in await _tree(
e2e_client,
e2e_headers,
agent_id=agent_id,
thread_id=other_thread_id,
path="/home/gem/user-data/workspace",
)
}
assert "/home/gem/user-data/workspace/demo.py" in other_workspace_paths, sorted(other_workspace_paths)
other_upload_paths = {
str(entry.get("path", ""))
for entry in await _tree(
e2e_client,
e2e_headers,
agent_id=agent_id,
thread_id=other_thread_id,
path="/home/gem/user-data/uploads",
)
}
assert "/home/gem/user-data/uploads/attachments/" not in other_upload_paths, sorted(other_upload_paths)
file_payload = await _file(
e2e_client,
e2e_headers,
agent_id=agent_id,
thread_id=thread_id,
path="/home/gem/user-data/workspace/demo.py",
)
assert file_payload.get("content") == "print(42)\n", file_payload
assert file_payload.get("preview_type") == "text", file_payload
assert file_payload.get("supported") is True, file_payload
other_file_payload = await _file(
e2e_client,
e2e_headers,
agent_id=agent_id,
thread_id=other_thread_id,
path="/home/gem/user-data/workspace/demo.py",
)
assert other_file_payload.get("content") == "print(42)\n", other_file_payload
content_disposition, payload = await _download(
e2e_client,
e2e_headers,
agent_id=agent_id,
thread_id=thread_id,
path="/home/gem/user-data/outputs/result.txt",
)
assert "result.txt" in content_disposition, content_disposition
assert payload == b"viewer-output\n", payload

View File

@ -422,7 +422,7 @@ async def test_get_supported_file_types(test_client, admin_headers):
async def test_markdown_endpoint_parses_uploaded_text_file(test_client, admin_headers):
"""测试 /files/markdown 能解析上传文件并返回 markdown。"""
data_dir = Path(__file__).resolve().parents[1] / "data"
data_dir = Path(__file__).resolve().parents[2] / "data"
test_file = data_dir / "A_Dream_of_Red_Mansions_10hui.txt"
assert test_file.exists(), f"测试文件不存在: {test_file}"

View File

@ -30,10 +30,10 @@ async def test_get_graphs_list(test_client, admin_headers):
graphs = payload["data"]
assert isinstance(graphs, list)
# Check for Neo4j default graph
# Check for the default upload graph
neo4j_graph = next((g for g in graphs if g["id"] == "neo4j"), None)
assert neo4j_graph is not None
assert neo4j_graph["type"] == "neo4j"
assert neo4j_graph["type"] == "upload"
# Note: LightRAG graphs might be empty if none created, but we check structure
@ -74,9 +74,11 @@ async def test_get_stats_neo4j(test_client, admin_headers):
payload = response.json()
assert payload["success"] is True
data = payload["data"]
assert "total_nodes" in data
assert "total_edges" in data
assert "entity_types" in data
assert isinstance(data, dict)
if data:
assert "total_nodes" in data
assert "total_edges" in data
assert "entity_types" in data
async def test_get_stats_lightrag(test_client, admin_headers, knowledge_database):
@ -123,8 +125,11 @@ async def test_deprecated_neo4j_endpoints(test_client, admin_headers):
response = await test_client.get(
"/api/graph/neo4j/node", params={"entity_name": "NonExistentEntity"}, headers=admin_headers
)
assert response.status_code == 200
payload = response.json()
assert payload["success"] is True
assert "result" in payload
assert payload["message"] == "success"
if response.status_code == 200:
payload = response.json()
assert payload["success"] is True
assert "result" in payload
assert payload["message"] == "success"
else:
assert response.status_code == 500
assert "向量索引不存在" in response.text or "未上传任何三元组" in response.text

View File

@ -226,7 +226,7 @@ async def test_viewer_file_returns_raw_content_without_line_numbers(test_client,
assert response.status_code == 200, response.text
assert "alpha" in response.json()["content"]
assert "beta" in response.json()["content"]
assert response.json()["preview_type"] == "text"
assert response.json()["preview_type"] == "markdown"
assert response.json()["supported"] is True
@ -334,25 +334,10 @@ async def test_viewer_download_returns_full_file_for_large_user_data_content(tes
assert response.content == large_content.encode("utf-8")
async def test_viewer_tree_root_lists_kbs_namespace_when_visible(test_client, standard_user, monkeypatch):
async def test_viewer_tree_root_hides_kbs_namespace_when_no_database_is_visible(test_client, standard_user):
headers = standard_user["headers"]
thread_id = await _create_thread_for_user(test_client, headers)
async def _fake_list_viewer_filesystem_tree(**kwargs):
return {
"entries": [
{"path": "/home/gem/user-data/", "name": "user-data", "is_dir": True, "size": 0, "modified_at": ""},
{"path": "/home/gem/kbs/", "name": "kbs", "is_dir": True, "size": 0, "modified_at": ""},
]
}
router_module = importlib.import_module("server.routers.filesystem_router")
monkeypatch.setitem(
router_module.get_viewer_tree.__globals__,
"list_viewer_filesystem_tree",
_fake_list_viewer_filesystem_tree,
)
response = await test_client.get(
"/api/viewer/filesystem/tree",
params={"thread_id": thread_id, "path": "/"},
@ -362,10 +347,11 @@ async def test_viewer_tree_root_lists_kbs_namespace_when_visible(test_client, st
entries = response.json().get("entries", [])
paths = {entry.get("path") for entry in entries}
assert "/home/gem/kbs/" in paths
assert "/home/gem/user-data/" in paths
assert "/home/gem/kbs/" not in paths
async def test_viewer_can_list_and_read_kbs_namespace(test_client, standard_user):
async def test_viewer_kbs_namespace_is_empty_when_no_database_is_visible(test_client, standard_user):
headers = standard_user["headers"]
thread_id = await _create_thread_for_user(test_client, headers)
@ -376,4 +362,4 @@ async def test_viewer_can_list_and_read_kbs_namespace(test_client, standard_user
)
assert tree_response.status_code == 200, tree_response.text
entries = tree_response.json().get("entries", [])
assert any(str(entry.get("path", "")).startswith("/home/gem/kbs/") for entry in entries)
assert entries == []

View File

@ -0,0 +1,287 @@
"""
Shared pytest fixtures for integration tests that exercise the live API service.
"""
from __future__ import annotations
import json
import os
import subprocess
import sys
import uuid
from collections.abc import AsyncGenerator
from pathlib import Path
import anyio
import httpx
import pytest
import pytest_asyncio
from dotenv import load_dotenv
PROJECT_ROOT = Path(__file__).resolve().parents[2]
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
load_dotenv(PROJECT_ROOT / ".env", override=False)
load_dotenv(PROJECT_ROOT / "test/.env.test", override=False)
API_BASE_URL = os.getenv("TEST_BASE_URL", "http://localhost:5050").rstrip("/")
ADMIN_LOGIN = os.getenv("TEST_USERNAME")
ADMIN_PASSWORD = os.getenv("TEST_PASSWORD")
_ADMIN_TOKEN_CACHE: str | None = None
HTTP_TIMEOUT = httpx.Timeout(30.0, connect=5.0)
SANDBOX_CONTAINER_PREFIX = os.getenv("YUXI_SANDBOX_CONTAINER_PREFIX", "yuxi-sandbox")
def _require_admin_credentials() -> tuple[str, str]:
if not ADMIN_LOGIN or not ADMIN_PASSWORD:
pytest.skip("Integration credentials are not configured via TEST_USERNAME / TEST_PASSWORD.")
return ADMIN_LOGIN, ADMIN_PASSWORD
@pytest_asyncio.fixture(scope="function")
async def test_client() -> AsyncGenerator[httpx.AsyncClient, None]:
async with httpx.AsyncClient(base_url=API_BASE_URL, timeout=HTTP_TIMEOUT, follow_redirects=True) as client:
yield client
@pytest_asyncio.fixture(scope="function")
async def admin_token() -> str:
global _ADMIN_TOKEN_CACHE
if _ADMIN_TOKEN_CACHE:
return _ADMIN_TOKEN_CACHE
username, password = _require_admin_credentials()
async with httpx.AsyncClient(
base_url=API_BASE_URL,
timeout=HTTP_TIMEOUT,
follow_redirects=True,
) as bootstrap_client:
response = await bootstrap_client.post(
"/api/auth/token",
data={"username": username, "password": password},
)
if response.status_code == 401:
first_run_response = await bootstrap_client.get("/api/auth/check-first-run")
if first_run_response.status_code == 200 and first_run_response.json().get("first_run", False):
pytest.fail(
"Super admin account has not been initialized. Complete `/api/auth/initialize` before "
"running integration tests."
)
if response.status_code != 200:
pytest.fail(f"Failed to authenticate as admin (status={response.status_code}): {response.text}")
token = response.json().get("access_token")
if not token:
pytest.fail("Admin authentication did not return an access token.")
_ADMIN_TOKEN_CACHE = token
return token
@pytest.fixture(scope="function")
def admin_headers(admin_token: str) -> dict[str, str]:
return {"Authorization": f"Bearer {admin_token}"}
@pytest.fixture(scope="session", autouse=True)
def cleanup_test_knowledge_databases():
async def run_cleanup() -> None:
global _ADMIN_TOKEN_CACHE
if not ADMIN_LOGIN or not ADMIN_PASSWORD:
return
if not _ADMIN_TOKEN_CACHE:
async with httpx.AsyncClient(
base_url=API_BASE_URL,
timeout=HTTP_TIMEOUT,
follow_redirects=True,
) as bootstrap_client:
response = await bootstrap_client.post(
"/api/auth/token",
data={"username": ADMIN_LOGIN, "password": ADMIN_PASSWORD},
)
if response.status_code != 200:
return
token = response.json().get("access_token")
if not token:
return
_ADMIN_TOKEN_CACHE = token
headers = {"Authorization": f"Bearer {_ADMIN_TOKEN_CACHE}"}
async with httpx.AsyncClient(base_url=API_BASE_URL, timeout=HTTP_TIMEOUT, follow_redirects=True) as client:
try:
list_response = await client.get("/api/knowledge/databases", headers=headers)
except Exception as exc:
print(f"Warning: Failed to list knowledge databases for cleanup: {exc}")
return
if list_response.status_code != 200:
return
databases = list_response.json().get("databases", [])
prefixes = ("pytest_", "py_test")
for entry in databases:
name = entry.get("name") or ""
db_id = entry.get("db_id")
if not db_id or not isinstance(name, str) or not name.startswith(prefixes):
continue
try:
delete_response = await client.delete(f"/api/knowledge/databases/{db_id}", headers=headers)
if delete_response.status_code not in (200, 404):
print(f"Warning: Failed to cleanup knowledge database {db_id}: {delete_response.text}")
except Exception as exc:
print(f"Warning: Exception during cleanup of {db_id}: {exc}")
try:
anyio.run(run_cleanup)
except Exception as exc:
print(f"Warning: Exception during session cleanup startup: {exc}")
yield
try:
anyio.run(run_cleanup)
except Exception as exc:
print(f"Warning: Exception during session cleanup teardown: {exc}")
def _docker_api_request(method: str, path: str) -> list[dict] | dict:
cmd = [
"curl",
"-sS",
"--unix-socket",
os.getenv("YUXI_DOCKER_API_SOCKET", "/var/run/docker.sock"),
"-X",
method,
f"{os.getenv('YUXI_DOCKER_API_BASE', 'http://localhost').rstrip('/')}{path}",
]
result = subprocess.run(cmd, capture_output=True, text=True, timeout=15, check=False)
if result.returncode != 0:
raise RuntimeError(result.stderr.strip() or "Docker API request failed")
text = (result.stdout or "").strip()
if not text:
return {}
return json.loads(text)
def _cleanup_sandbox_containers() -> None:
try:
containers = _docker_api_request("GET", "/containers/json?all=true")
except Exception as exc:
print(f"Warning: Failed to list sandbox containers for cleanup: {exc}")
return
for container in containers if isinstance(containers, list) else []:
names = container.get("Names") or []
if not any(name.lstrip("/").startswith(f"{SANDBOX_CONTAINER_PREFIX}-") for name in names):
continue
container_id = container.get("Id")
if not container_id:
continue
try:
_docker_api_request("POST", f"/containers/{container_id}/stop?t=2")
except Exception:
pass
try:
_docker_api_request("DELETE", f"/containers/{container_id}?force=true")
except Exception as exc:
print(f"Warning: Failed to cleanup sandbox container {container_id[:12]}: {exc}")
@pytest.fixture(scope="session", autouse=True)
def cleanup_test_sandboxes():
_cleanup_sandbox_containers()
yield
_cleanup_sandbox_containers()
@pytest_asyncio.fixture(scope="function")
async def standard_user(test_client: httpx.AsyncClient, admin_headers: dict[str, str]) -> AsyncGenerator[dict, None]:
username = f"pytest_user_{uuid.uuid4().hex[:8]}"
password = f"Pw!{uuid.uuid4().hex[:8]}"
response = await test_client.post(
"/api/auth/users",
json={"username": username, "password": password, "role": "user"},
headers=admin_headers,
)
if response.status_code != 200:
pytest.fail(f"Failed to create standard user (status={response.status_code}): {response.text}")
user_payload = response.json()
login_response = await test_client.post(
"/api/auth/token",
data={"username": user_payload["user_id"], "password": password},
)
if login_response.status_code != 200:
pytest.fail(
f"Failed to authenticate as standard user (status={login_response.status_code}): {login_response.text}"
)
access_token = login_response.json().get("access_token")
if not access_token:
pytest.fail("Standard user login succeeded but no access token was returned.")
try:
yield {
"user": user_payload,
"password": password,
"headers": {"Authorization": f"Bearer {access_token}"},
}
finally:
response = await test_client.delete(f"/api/auth/users/{user_payload['id']}", headers=admin_headers)
assert response.status_code == 200, f"Failed to cleanup test user {user_payload['user_id']}: {response.text}"
@pytest_asyncio.fixture(scope="function")
async def knowledge_database(
test_client: httpx.AsyncClient,
admin_headers: dict[str, str],
) -> AsyncGenerator[dict, None]:
import time
unique_id = uuid.uuid4().hex
timestamp = int(time.time() * 1000000)
db_name = f"pytest_kb_{timestamp}_{unique_id}"
db_id = None
try:
create_response = await test_client.post(
"/api/knowledge/databases",
json={
"database_name": db_name,
"description": "Pytest managed knowledge base",
"embed_model_name": "siliconflow/BAAI/bge-m3",
"kb_type": "lightrag",
"additional_params": {},
},
headers=admin_headers,
)
if create_response.status_code == 200:
db_payload = create_response.json()
db_id = db_payload["db_id"]
elif create_response.status_code == 409:
error_detail = create_response.json().get("detail", "")
pytest.fail(f"Knowledge database name conflict: {error_detail}. Please clean up old test databases first.")
else:
pytest.fail(
f"Failed to create knowledge database (status={create_response.status_code}): {create_response.text}"
)
yield db_payload if db_id else {"db_id": db_id, "name": db_name}
finally:
if db_id:
try:
delete_response = await test_client.delete(f"/api/knowledge/databases/{db_id}", headers=admin_headers)
if delete_response.status_code != 200:
print(f"Warning: Failed to cleanup knowledge database {db_id}: {delete_response.text}")
except Exception as exc:
print(f"Warning: Exception during cleanup of {db_id}: {exc}")

116
backend/test/run_tests.sh Executable file → Normal file
View File

@ -1,96 +1,78 @@
#!/bin/bash
# 测试脚本运行器
# 提供快速运行不同类型测试的方法
set -euo pipefail
echo "Yuxi API 测试运行器"
echo "Yuxi 测试运行器"
echo "========================"
# 检查测试服务是否运行
PYTEST_CMD=("docker" "compose" "exec" "api" "uv" "run" "--group" "test" "pytest")
check_server() {
echo "检查测试服务器状态..."
curl -s http://localhost:5050/api/system/health > /dev/null 2>&1
if [ $? -eq 0 ]; then
if curl -s http://localhost:5050/api/system/health > /dev/null 2>&1; then
echo "✓ 测试服务器运行正常"
return 0
else
echo "✗ 警告: 测试服务器未运行或无法访问"
echo " 请确保服务器在 http://localhost:5050 运行"
return 1
fi
echo "✗ 警告: 测试服务器未运行或无法访问"
echo " 请先执行 docker compose up -d 并确认 api-dev 健康"
return 1
}
run_unit_tests() {
echo "运行单元测试..."
"${PYTEST_CMD[@]}" test/unit -m "not slow"
}
run_integration_tests() {
echo "运行集成测试..."
check_server
"${PYTEST_CMD[@]}" test/integration
}
run_e2e_tests() {
echo "运行端到端测试..."
check_server
"${PYTEST_CMD[@]}" test/e2e -m e2e
}
# 运行所有测试
run_all_tests() {
echo "运行所有API测试..."
uv run pytest test/api/ -v
echo "运行全部测试..."
check_server
"${PYTEST_CMD[@]}" test
}
# 运行认证测试
run_auth_tests() {
echo "运行认证API测试..."
uv run pytest test/api/test_auth_api.py -v
}
# 运行系统测试
run_system_tests() {
echo "运行系统API测试..."
uv run pytest test/api/test_system_api.py -v
}
# 运行对话测试
run_chat_tests() {
echo "运行对话API测试..."
uv run pytest test/api/test_chat_api.py -v
}
# 运行快速测试(只测试基础功能)
run_quick_tests() {
echo "运行快速测试(基础功能)..."
uv run pytest test/api/ -v -m "not slow"
}
# 显示帮助
show_help() {
echo "用法: $0 [选项]"
echo ""
echo "选项:"
echo " all - 运行所有测试"
echo " auth - 运行认证测试"
echo " system - 运行系统测试"
echo " chat - 运行对话测试"
echo " quick - 运行快速测试(排除慢速测试)"
echo " check - 检查服务器状态"
echo " help - 显示此帮助"
echo " unit - 运行单元测试"
echo " integration - 运行集成测试"
echo " e2e - 运行端到端测试"
echo " all - 运行全部测试"
echo " check - 检查测试服务"
echo " help - 显示此帮助"
echo ""
echo "示例:"
echo " $0 all # 运行所有测试"
echo " $0 quick # 快速测试"
echo " $0 check # 检查服务器"
echo " $0 unit"
echo " $0 integration"
echo " $0 e2e"
echo " $0 all"
}
# 主逻辑
case "${1:-all}" in
"unit")
run_unit_tests
;;
"integration")
run_integration_tests
;;
"e2e")
run_e2e_tests
;;
"all")
check_server
run_all_tests
;;
"auth")
check_server
run_auth_tests
;;
"system")
check_server
run_system_tests
;;
"chat")
check_server
run_chat_tests
;;
"quick")
check_server
run_quick_tests
;;
"check")
check_server
;;
@ -102,4 +84,4 @@ case "${1:-all}" in
show_help
exit 1
;;
esac
esac

View File

@ -0,0 +1,200 @@
from __future__ import annotations
import asyncio
from types import SimpleNamespace
import pytest
from server.routers import knowledge_router
from yuxi.knowledge.base import FileStatus
from yuxi.knowledge.implementations import lightrag as lightrag_module
from yuxi.knowledge.implementations.lightrag import LightRagKB
pytestmark = pytest.mark.asyncio
def _build_kb_file(path: str) -> dict:
return {
"status": FileStatus.PARSED,
"markdown_file": "mock://parsed.md",
"path": path,
"filename": path.rsplit("/", 1)[-1],
"processing_params": {},
}
class _FakeDocStatus:
async def get_by_id(self, _file_id: str) -> dict:
return {"status": "processed"}
@pytest.fixture
def light_rag_kb(tmp_path: pytest.TempPathFactory, monkeypatch: pytest.MonkeyPatch) -> LightRagKB:
kb = LightRagKB(str(tmp_path))
monkeypatch.setattr(kb, "_save_metadata", _async_noop)
monkeypatch.setattr(kb, "_persist_file", _async_noop)
monkeypatch.setattr(kb, "_read_markdown_from_minio", _fake_read_markdown)
monkeypatch.setattr(kb, "delete_file_chunks_only", _async_noop)
monkeypatch.setattr(
lightrag_module,
"chunk_markdown",
lambda markdown, file_id, filename, params: [{"content": markdown or f"{file_id}:{filename}:{params}"}],
)
return kb
async def _async_noop(*_args, **_kwargs) -> None:
return None
async def _fake_read_markdown(*_args, **_kwargs) -> str:
return "mock markdown"
async def test_index_file_serializes_writes_within_same_database(
light_rag_kb: LightRagKB,
monkeypatch: pytest.MonkeyPatch,
) -> None:
db_id = "kb_same"
light_rag_kb.databases_meta[db_id] = {"metadata": {}}
light_rag_kb.files_meta["file-1"] = _build_kb_file("/tmp/file-1.md")
light_rag_kb.files_meta["file-2"] = _build_kb_file("/tmp/file-2.md")
started_first = asyncio.Event()
release_first = asyncio.Event()
started_second = asyncio.Event()
call_order: list[str] = []
async def fake_ainsert(*, ids: str, **_kwargs) -> None:
if ids == "file-1":
call_order.append("start:file-1")
started_first.set()
await release_first.wait()
call_order.append("end:file-1")
return
call_order.append("start:file-2")
started_second.set()
call_order.append("end:file-2")
rag = SimpleNamespace(ainsert=fake_ainsert, doc_status=_FakeDocStatus())
monkeypatch.setattr(light_rag_kb, "_get_lightrag_instance", _make_async_return(rag))
task1 = asyncio.create_task(light_rag_kb.index_file(db_id, "file-1"))
await asyncio.wait_for(started_first.wait(), timeout=1)
task2 = asyncio.create_task(light_rag_kb.index_file(db_id, "file-2"))
await asyncio.sleep(0.05)
assert not started_second.is_set()
release_first.set()
await asyncio.gather(task1, task2)
assert call_order == ["start:file-1", "end:file-1", "start:file-2", "end:file-2"]
async def test_index_file_allows_parallel_writes_for_different_databases(
light_rag_kb: LightRagKB,
monkeypatch: pytest.MonkeyPatch,
) -> None:
light_rag_kb.databases_meta["kb_a"] = {"metadata": {}}
light_rag_kb.databases_meta["kb_b"] = {"metadata": {}}
light_rag_kb.files_meta["file-a"] = _build_kb_file("/tmp/file-a.md")
light_rag_kb.files_meta["file-b"] = _build_kb_file("/tmp/file-b.md")
started_a = asyncio.Event()
started_b = asyncio.Event()
release_both = asyncio.Event()
async def fake_ainsert(*, ids: str, **_kwargs) -> None:
if ids == "file-a":
started_a.set()
elif ids == "file-b":
started_b.set()
await release_both.wait()
rag = SimpleNamespace(ainsert=fake_ainsert, doc_status=_FakeDocStatus())
monkeypatch.setattr(light_rag_kb, "_get_lightrag_instance", _make_async_return(rag))
task_a = asyncio.create_task(light_rag_kb.index_file("kb_a", "file-a"))
task_b = asyncio.create_task(light_rag_kb.index_file("kb_b", "file-b"))
await asyncio.wait_for(started_a.wait(), timeout=1)
await asyncio.wait_for(started_b.wait(), timeout=1)
release_both.set()
await asyncio.gather(task_a, task_b)
async def test_add_documents_auto_index_uses_latest_parsed_metadata(monkeypatch: pytest.MonkeyPatch) -> None:
calls: list[tuple[str, str]] = []
class FakeKnowledgeBase:
async def get_database_info(self, _db_id: str) -> dict:
return {"name": "pytest-db"}
async def add_file_record(
self,
_db_id: str,
_item: str,
params: dict | None = None,
operator_id: str | None = None,
) -> dict:
return {"file_id": "file-1", "status": FileStatus.UPLOADED, "params": params, "operator_id": operator_id}
async def parse_file(self, _db_id: str, file_id: str, operator_id: str | None = None) -> dict:
calls.append(("parse", file_id))
return {"file_id": file_id, "status": FileStatus.PARSED, "operator_id": operator_id}
async def update_file_params(
self, _db_id: str, file_id: str, params: dict, operator_id: str | None = None
) -> None:
calls.append(("update_params", file_id))
async def index_file(self, _db_id: str, file_id: str, operator_id: str | None = None) -> dict:
calls.append(("index", file_id))
return {"file_id": file_id, "status": FileStatus.INDEXED, "operator_id": operator_id}
class FakeTaskContext:
async def set_message(self, _message: str) -> None:
return None
async def set_progress(self, _progress: float, _message: str | None = None) -> None:
return None
async def set_result(self, _result) -> None:
return None
async def raise_if_cancelled(self) -> None:
return None
class FakeTasker:
async def enqueue(self, *, coroutine, **_kwargs):
await coroutine(FakeTaskContext())
return SimpleNamespace(id="task-1")
async def fake_ensure_database_not_dify(_db_id: str, _operation: str) -> None:
return None
monkeypatch.setattr("yuxi.knowledge.utils.kb_utils.validate_file_path", lambda _item, _db_id: None)
monkeypatch.setattr(knowledge_router, "knowledge_base", FakeKnowledgeBase())
monkeypatch.setattr(knowledge_router, "tasker", FakeTasker())
monkeypatch.setattr(knowledge_router, "_ensure_database_not_dify", fake_ensure_database_not_dify)
current_user = SimpleNamespace(user_id="user-1", id="user-1")
result = await knowledge_router.add_documents(
"kb_auto_index",
items=["/tmp/example.md"],
params={"content_type": "file", "auto_index": True},
current_user=current_user,
)
assert result["status"] == "queued"
assert calls == [("parse", "file-1"), ("update_params", "file-1"), ("index", "file-1")]
def _make_async_return(value):
async def _inner(*_args, **_kwargs):
return value
return _inner

View File

@ -8,7 +8,10 @@ from yuxi.agents.backends.skills_backend import SelectedSkillsReadonlyBackend
def test_skills_backend_read_outside_root_returns_error_message(monkeypatch) -> None:
def _fake_read(self, file_path: str, offset: int = 0, limit: int = 2000):
raise ValueError("Path:/app/package/yuxi/agents/skills/buildin/reporter outside root directory: /app/saves/skills")
raise ValueError(
"Path:/app/package/yuxi/agents/skills/buildin/reporter "
"outside root directory: /app/saves/skills"
)
monkeypatch.setattr(FilesystemBackend, "read", _fake_read)

View File

@ -12,7 +12,7 @@ from PIL import Image
from yuxi.plugins.parser import Parser
from yuxi.plugins.parser.factory import DocumentProcessorFactory
DATA_DIR = Path(__file__).parent / "data"
DATA_DIR = Path(__file__).resolve().parents[2] / "data"
def _build_pdf(file_path: Path, text: str) -> None:

View File

@ -16,7 +16,7 @@ from yuxi.utils.datetime_utils import utc_now_naive
# Router Tests
# =============================================================================
from fastapi import FastAPI, HTTPException
from fastapi import FastAPI
from fastapi.testclient import TestClient
from server.routers.subagent_router import subagents_router
@ -429,8 +429,8 @@ class TestSubAgentService:
def __init__(self, session):
pass
async def exists_name(self, name):
return False
async def get_by_name(self, name):
return None
async def create(self, **kwargs):
created_agents.append(kwargs)
@ -438,7 +438,14 @@ class TestSubAgentService:
@asynccontextmanager
async def mock_session_context(*args, **kwargs):
yield MagicMock()
session = MagicMock()
session.commit = MagicMock()
async def _commit():
return None
session.commit = _commit
yield session
class MockPgManager:
get_async_session_context = mock_session_context

View File

@ -142,7 +142,7 @@ async def test_create_agent_run_commits_before_enqueue(monkeypatch: pytest.Monke
async def get_conversation_by_thread_id(self, thread_id: str):
del thread_id
return SimpleNamespace(user_id="1", status="active", department_id=1)
return SimpleNamespace(user_id="1", status="active", department_id=1, extra_metadata={"agent_config_id": 1})
class Queue:
async def enqueue_job(self, job_name: str, run_id: str, _job_id: str):
@ -227,7 +227,7 @@ async def test_create_agent_run_handles_integrity_error_with_same_user_existing(
async def get_conversation_by_thread_id(self, thread_id: str):
del thread_id
return SimpleNamespace(user_id="1", status="active", department_id=1)
return SimpleNamespace(user_id="1", status="active", department_id=1, extra_metadata={"agent_config_id": 1})
async def fake_get_arq_pool():
raise AssertionError("should not enqueue on integrity fallback")
@ -300,7 +300,7 @@ async def test_create_agent_run_integrity_error_returns_409_for_other_user(monke
async def get_conversation_by_thread_id(self, thread_id: str):
del thread_id
return SimpleNamespace(user_id="1", status="active", department_id=1)
return SimpleNamespace(user_id="1", status="active", department_id=1, extra_metadata={"agent_config_id": 1})
monkeypatch.setattr(agent_run_service.agent_manager, "get_agent", lambda agent_id: object())
monkeypatch.setattr(agent_run_service, "AgentConfigRepository", FakeConfigRepo)

View File

@ -11,6 +11,8 @@ from yuxi.services import chat_service as svc
class _FakeConvRepo:
def __init__(self, _db):
self.saved_messages: list[dict] = []
self.bound_agent_configs: list[tuple[str, int]] = []
self.conversations: dict[str, SimpleNamespace] = {}
async def add_message_by_thread_id(
self,
@ -34,6 +36,27 @@ class _FakeConvRepo:
)
return SimpleNamespace(id=1)
async def get_conversation_by_thread_id(self, thread_id: str):
return self.conversations.get(thread_id)
async def create_conversation(self, *, user_id: str, agent_id: str, thread_id: str):
conversation = SimpleNamespace(
user_id=user_id,
agent_id=agent_id,
thread_id=thread_id,
extra_metadata={},
)
self.conversations[thread_id] = conversation
return conversation
async def bind_agent_config(self, thread_id: str, agent_config_id: int):
conversation = self.conversations.setdefault(
thread_id,
SimpleNamespace(user_id="user-1", agent_id="test-agent", thread_id=thread_id, extra_metadata={}),
)
conversation.extra_metadata["agent_config_id"] = agent_config_id
self.bound_agent_configs.append((thread_id, agent_config_id))
@pytest.mark.asyncio
async def test_agent_chat_uses_invoke_messages_and_persists_langgraph_state(monkeypatch: pytest.MonkeyPatch):
@ -92,7 +115,7 @@ async def test_agent_chat_uses_invoke_messages_and_persists_langgraph_state(monk
assert result["response"] == "Hi from invoke"
assert result["thread_id"] == "thread-1"
assert result["request_id"] == "req-1"
assert result["agent_state"] == {"todos": ["todo-1"], "files": {}}
assert result["agent_state"] == {"todos": ["todo-1"], "files": {}, "artifacts": []}
invoke_messages = calls["invoke_messages"]
assert isinstance(invoke_messages, list)
@ -102,6 +125,7 @@ async def test_agent_chat_uses_invoke_messages_and_persists_langgraph_state(monk
assert calls["invoke_input_context"] == {"temperature": 0.1, "user_id": "user-1", "thread_id": "thread-1"}
assert calls["saved_state"]["thread_id"] == "thread-1"
assert calls["saved_state"]["config_dict"] == {"configurable": {"thread_id": "thread-1", "user_id": "user-1"}}
assert calls["saved_state"]["conv_repo"].bound_agent_configs == [("thread-1", 123)]
@pytest.mark.asyncio
@ -153,3 +177,4 @@ async def test_agent_chat_sync_returns_finished_even_when_state_has_interrupt(mo
assert result["status"] == "finished"
assert result["response"] == "Need input later"
assert result["thread_id"] == "thread-2"
assert result["request_id"] == "req-2"

View File

@ -1,6 +1,5 @@
"""测试 chat_service 中的 interrupt 相关函数"""
import pytest
import sys
import os
@ -212,7 +211,3 @@ class TestCoerceInterruptPayload:
def test_none_input(self):
result = _coerce_interrupt_payload(None)
assert isinstance(result, dict)
if __name__ == "__main__":
pytest.main([__file__, "-v"])

View File

@ -146,7 +146,8 @@ async def test_convert_upload_to_markdown_truncates_content(
@pytest.mark.asyncio
async def test_convert_upload_to_markdown_rejects_unsupported_extension():
async def test_convert_upload_to_markdown_rejects_unsupported_extension(monkeypatch: pytest.MonkeyPatch):
monkeypatch.setattr(svc, "ATTACHMENT_ALLOWED_EXTENSIONS", (".md",))
upload = _DummyUpload(filename="note.pdf", content_type="application/pdf", data=b"pdf")
with pytest.raises(ValueError, match="不支持的文件类型"):

View File

@ -49,6 +49,7 @@ services:
- ./saves:/app/saves
- ./backend/test:/app/test
- ./backend/scripts:/app/scripts
- ./docker/sandbox_provisioner:/app/docker/sandbox_provisioner:ro
- ./.env:/app/.env
- ${MODEL_DIR:-./models}:/models # 使用默认值处理未定义的环境变量
- /var/run/docker.sock:/var/run/docker.sock
@ -95,6 +96,7 @@ services:
- ./saves:/app/saves
- ./backend/test:/app/test
- ./backend/scripts:/app/scripts
- ./docker/sandbox_provisioner:/app/docker/sandbox_provisioner:ro
- ./.env:/app/.env
- ./backend/pyproject.toml:/app/pyproject.toml
- ./backend/uv.lock:/app/uv.lock

View File

@ -66,6 +66,7 @@ export default defineConfig({
{ text: '参与贡献', link: '/develop-guides/contributing' },
{ text: '开发路线图', link: '/develop-guides/roadmap' },
{ text: '界面设计规范', link: '/develop-guides/design' },
{ text: '测试规范', link: '/develop-guides/testing-guidelines' },
]
}
],

View File

@ -49,6 +49,7 @@
- 新增知识库 PDF、图片的预览功能
- 优化扩展页工具列表筛选区:将“全部分类”筛选收纳为搜索框右侧的紧凑下拉入口,并复用扩展页侧栏工具条样式,避免影响其他管理组件布局
- 重构扩展管理中的 SubAgent 与 MCP 交互,统一为类似 Skills 的“已添加 / 可添加”列表,不再使用服务器级开关切换,为 SubAgent 增加 `enabled` 状态,并让运行时只加载已添加项,调整内置 SubAgent / MCP 的启动同步逻辑,使用代码中的最新定义覆盖数据库展示字段,同时保留启用状态与 MCP 工具禁用状态;统一扩展详情区胶囊操作按钮样式到公共 `extensions.less`,并将内置 MCP 在详情页中的危险操作从禁用“删除”改为可执行的“移除”
- 重构后端测试目录结构:按 `unit / integration / e2e` 分层迁移现有测试,拆分全局 `conftest.py`,统一测试入口为 `uv run --group test pytest`,并新增独立测试规范文档 `docs/vibe/testing-guidelines.md`
<!-- 添加到这里 -->
@ -65,6 +66,7 @@
- 修复 GitHub Pages 文档部署工作流失败:移除 `actions/setup-node@v4` 对不存在 `docs/package-lock.json` 的缓存依赖,并将 `docs` 目录安装命令从 `npm ci` 调整为 `npm install`,避免因未提交锁文件导致 CI 在依赖缓存和安装阶段直接失败
- 修正沙盒 provisioner backend 命名与配置说明:统一对外使用 `docker` / `kubernetes`,保留 `local` 作为兼容别名;同步清理 compose 中未生效的 provisioner 环境变量、补齐 K8s 相关变量注释,并更新沙盒架构文档中的默认模式与 backend 描述
- 修复智能体配置列表接口在“无配置自动创建默认配置”路径下的参数缺失:补齐 `get_or_create_default``agent_id` 入参,避免 `/api/chat/agent/{agent_id}/configs` 返回 500
- 修复 LightRAG 同库写入并发导致的入库失败:为 `index_file` / `update_content` 增加按知识库维度的串行锁,并补齐 `documents` 接口 `auto_index` 阶段对最新解析状态的回写与回归测试,避免长时间入库任务进行中再次选择同库文件时直接并发写入报错
<!-- 添加到这里 -->

View File

@ -0,0 +1,194 @@
# 测试规范与工作流
本文档用于指导 Yuxi 后续如何创建测试文件、修改测试文件,以及如何验证项目功能。目标是务实、稳定、可执行,不追求过度设计。
## 1. 测试分层
当前测试统一分为三层:
- `backend/test/unit`
- 纯单元测试
- 不依赖运行中的 Docker 服务
- 优先使用 `monkeypatch`、fake repo、stub、`tmp_path`
- `backend/test/integration`
- 真实 API 集成测试
- 依赖 `docker compose up -d` 后的运行环境
- 统一通过真实 HTTP 接口验证认证、权限、参数和副作用
- `backend/test/e2e`
- 关键链路端到端测试
- 覆盖 run、viewer、附件、文件落盘等完整流程
- 默认数量少、执行更慢
## 2. 新增测试时怎么选目录
新增测试前先判断:
1. 只测 Python 逻辑,不需要真实服务
放到 `unit`
2. 需要请求真实接口
放到 `integration/api`
3. 需要验证从入口到最终结果的完整链路
放到 `e2e`
不要再默认把测试直接丢到 `backend/test/` 根目录。
## 3. 文件和命名规范
文件名:
- 使用 `test_<domain>_<target>.py`
- 一个文件只测一个明确主题
函数名:
- 使用 `test_<行为>_<预期结果>`
- 名称直接表达业务语义
示例:
- `test_create_agent_run_commits_before_enqueue`
- `test_viewer_download_returns_attachment_response`
- `test_agent_bubble_sort_run_creates_expected_artifacts`
## 4. 写测试的基本要求
每个测试尽量保持三段式:
1. Arrange准备数据、打桩、创建资源
2. Act调用被测行为
3. Assert断言结果
要求:
- 不要只断言 `status_code == 200`
- 要断言关键业务字段和副作用
- 失败信息要能帮助定位问题
## 5. fixture 规范
原则:
- 同一个文件内复用,优先写本地 helper
- 多个文件复用,再提取到对应层级的 `conftest.py`
- 根 `backend/test/conftest.py` 只保留通用 marker不绑定真实环境
当前约定:
- `backend/test/integration/conftest.py`
- 管理 `test_client`、`admin_headers`、`standard_user`、`knowledge_database`
- `backend/test/e2e/conftest.py`
- 管理 `e2e_client`、`e2e_headers`、`e2e_agent_context`
## 6. 允许与禁止
允许:
- 在单元测试里使用 `monkeypatch`
- 在集成测试里通过 fixture 创建测试资源
- 在 E2E 中使用轮询等待最终状态
禁止:
- 在测试文件里硬编码真实账号密码
- 在单元测试里请求真实 HTTP 服务
- 在根 `conftest.py` 里继续添加重环境依赖
- 写 `if __name__ == "__main__":` 作为测试入口
- 用 `print` 作为通过/失败判断手段
- 因为系统里没有默认数据就直接 `skip`
## 7. skip 的使用规则
只在下面两类场景允许 `pytest.skip`
1. 外部可选能力不可用
例如 OCR 服务、外部模型服务未启动
2. E2E 环境变量未配置
例如没有配置专用测试账号
不允许把“系统里没有 agent / config / 预置数据”当成正常 skip 条件。
这类情况应优先改为 fixture 显式准备资源,或者直接 fail 暴露环境问题。
## 8. 修改测试文件时的规则
如果是修 bug
1. 先补一个能稳定复现 bug 的测试
2. 再修代码
3. 先跑最小相关测试集
4. 再跑相关层级回归
如果是改已有功能:
- 行为变了,就更新断言
- 文件职责混乱,就顺手拆分或迁移目录
- 依赖现成系统状态的测试,优先改成 fixture 建资源
## 9. 运行方式
启动环境:
```bash
docker compose up -d
docker ps
docker logs api-dev --tail 100
```
运行单元测试:
```bash
docker compose exec api uv run --group test pytest test/unit -m "not slow"
```
运行集成测试:
```bash
docker compose exec api uv run --group test pytest test/integration
```
运行 E2E
```bash
docker compose exec api uv run --group test pytest test/e2e -m e2e
```
运行全部测试:
```bash
docker compose exec api uv run --group test pytest test
```
也可以使用:
```bash
backend/test/run_tests.sh unit
backend/test/run_tests.sh integration
backend/test/run_tests.sh e2e
backend/test/run_tests.sh all
```
## 10. 推荐的日常开发流程
建议顺序:
1. 本地改代码
2. 先跑相关单元测试
3. 涉及接口时跑相关集成测试
4. 涉及关键主链路时补跑对应 E2E
5. 提交前至少完成“检查 -> 测试 -> Lint”
## 11. 当前落地原则
这套规范的重点不是一步到位重写所有旧测试,而是:
- 新增测试必须按新目录落位
- 改到旧测试时顺手迁移
- 优先保持测试可执行和可信
- 优先减少假绿和环境耦合
对当前 Yuxi 来说,这就是最务实、也最容易持续执行的测试标准。