From 35ccfbc82d4e4b79ca950130c4020d64fb181a36 Mon Sep 17 00:00:00 2001 From: Wenjie Zhang Date: Sat, 21 Mar 2026 00:03:33 +0800 Subject: [PATCH] =?UTF-8?q?test:=20=E4=BF=AE=E6=95=B4=E6=B5=8B=E8=AF=95?= =?UTF-8?q?=E6=96=87=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../yuxi/repositories/subagent_repository.py | 5 +- backend/pyproject.toml | 1 + backend/test/bruteforce_simulation.py | 122 ---------- backend/test/compare_kb_metadata_with_db.py | 158 ------------- backend/test/conftest.py | 6 + backend/test/test_concurrency.py | 136 ----------- backend/test/test_kb_minio_cleanup.py | 132 ----------- backend/test/test_manual_eval.py | 212 ------------------ backend/test/test_mysql_connection.py | 206 ----------------- backend/test/test_mysql_import.py | 24 -- backend/test/test_neo4j.py | 37 --- backend/test/test_subagent.py | 63 ------ 12 files changed, 10 insertions(+), 1092 deletions(-) delete mode 100644 backend/test/bruteforce_simulation.py delete mode 100644 backend/test/compare_kb_metadata_with_db.py delete mode 100644 backend/test/test_concurrency.py delete mode 100644 backend/test/test_kb_minio_cleanup.py delete mode 100644 backend/test/test_manual_eval.py delete mode 100644 backend/test/test_mysql_connection.py delete mode 100644 backend/test/test_mysql_import.py delete mode 100755 backend/test/test_neo4j.py diff --git a/backend/package/yuxi/repositories/subagent_repository.py b/backend/package/yuxi/repositories/subagent_repository.py index 69861ab1..73a91fae 100644 --- a/backend/package/yuxi/repositories/subagent_repository.py +++ b/backend/package/yuxi/repositories/subagent_repository.py @@ -81,11 +81,12 @@ class SubAgentRepository: "system_prompt": system_prompt, "tools": tools, } - if model_provided: - updates["model"] = model for field, value in updates.items(): if value is not None: setattr(item, field, value) + # model_provided=True 时显式设置 model 值（包括 None），用于清空字段 + if model_provided: + item.model = model item.updated_by = updated_by item.updated_at = utc_now_naive() await self.db.commit() diff --git a/backend/pyproject.toml b/backend/pyproject.toml index 78b1cc78..20f715ee 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -106,6 +106,7 @@ members = [ [tool.pytest.ini_options] addopts = "-v --tb=short" testpaths = ["test"] +pythonpath = ["."] markers = [ "auth: marks tests that require authentication", "slow: marks tests as slow", diff --git a/backend/test/bruteforce_simulation.py b/backend/test/bruteforce_simulation.py deleted file mode 100644 index 4a367106..00000000 --- a/backend/test/bruteforce_simulation.py +++ /dev/null @@ -1,122 +0,0 @@ -""" -Quick script to hammer the login endpoint with invalid credentials and observe rate limiting. - -Usage: - uv run python test/bruteforce_simulation.py --username demo --attempts 20 -""" - -from __future__ import annotations - -import argparse -import asyncio -import os -import random -import string -import sys -import time -from collections import Counter - -import httpx - - -def parse_args() -> argparse.Namespace: - parser = argparse.ArgumentParser(description="Simulate brute-force login attempts.") - parser.add_argument("--base-url", default=os.getenv("TEST_BASE_URL", "http://localhost:5050"), help="API base URL") - parser.add_argument("--username", default=os.getenv("TEST_USERNAME", "admin"), help="Login identifier to attack") - parser.add_argument("--attempts", type=int, default=20, help="Total number of attempts to issue (default: 20)") - parser.add_argument( - "--concurrency", - type=int, - default=4, - help="Concurrent request limit (default: 4)", - ) - parser.add_argument( - "--delay", - type=float, - default=0.05, - help="Delay in seconds between scheduling attempts (default: 0.05)", - ) - parser.add_argument( - "--password", - default=None, - help="Explicit password to reuse for each request; random values are used when omitted", - ) - return parser.parse_args() - - -def build_payload(username: str, password: str) -> dict[str, str]: - return {"username": username, "password": password} - - -def random_password() -> str: - base = string.ascii_letters + string.digits + "!@#$%^&*" - return "".join(random.choices(base, k=12)) - - -async def attempt_login( - client: httpx.AsyncClient, - semaphore: asyncio.Semaphore, - attempt_no: int, - username: str, - static_password: str | None, -) -> tuple[int, float]: - async with semaphore: - password = static_password or random_password() - payload = build_payload(username, password) - started = time.perf_counter() - response = await client.post("/api/auth/token", data=payload) - elapsed = time.perf_counter() - started - detail = ( - response.json().get("detail") - if response.headers.get("content-type", "").startswith("application/json") - else response.text - ) - print( - f"[{attempt_no:02d}] {response.status_code} in {elapsed * 1000:.1f} ms (pwd={password!r}) detail={detail!r}" - ) - return response.status_code, elapsed - - -async def run_simulation(args: argparse.Namespace) -> int: - timeout = httpx.Timeout(10.0, connect=3.0) - limits = httpx.Limits(max_connections=args.concurrency, max_keepalive_connections=args.concurrency) - semaphore = asyncio.Semaphore(args.concurrency) - status_counts: Counter[int] = Counter() - - async with httpx.AsyncClient(base_url=args.base_url.rstrip("/"), timeout=timeout, limits=limits) as client: - tasks = [] - for attempt_no in range(1, args.attempts + 1): - tasks.append( - asyncio.create_task(attempt_login(client, semaphore, attempt_no, args.username, args.password)) - ) - if args.delay: - await asyncio.sleep(args.delay) - - for task in asyncio.as_completed(tasks): - status_code, _ = await task - status_counts[status_code] += 1 - - print("\nSummary:") - for code, total in sorted(status_counts.items()): - print(f" HTTP {code}: {total} hits") - - if 429 in status_counts: - print("Rate limiting engaged (HTTP 429 encountered).") - return 0 - - print("No rate limiting observed; consider increasing attempt count or reducing delay.") - return 1 - - -def main() -> None: - args = parse_args() - try: - exit_code = asyncio.run(run_simulation(args)) - except KeyboardInterrupt: - print("\nInterrupted.") - exit_code = 130 - sys.exit(exit_code) - - -if __name__ == "__main__": - main() diff --git a/backend/test/compare_kb_metadata_with_db.py b/backend/test/compare_kb_metadata_with_db.py deleted file mode 100644 index 9e6bb631..00000000 --- a/backend/test/compare_kb_metadata_with_db.py +++ /dev/null @@ -1,158 +0,0 @@ -import asyncio -import glob -import json -import os -import sys -from dataclasses import dataclass -from typing import Any - -from sqlalchemy import func, select - -os.environ.setdefault("YUXI_SKIP_APP_INIT", "1") -sys.path.insert(0, os.path.dirname(os.path.dirname(__file__))) - -from yuxi.config import config -from yuxi.storage.postgres.manager import pg_manager -from yuxi.storage.postgres.models_knowledge import ( - EvaluationBenchmark, - EvaluationResult, - EvaluationResultDetail, - KnowledgeBase, - KnowledgeFile, -) - - -def _load_json(path: str) -> dict[str, Any]: - if not os.path.exists(path): - return {} - with open(path, encoding="utf-8") as f: - return json.load(f) - - -@dataclass(frozen=True) -class JsonState: - kb_ids: set[str] - file_ids: set[str] - benchmark_ids: set[str] - result_task_ids: set[str] - result_detail_count: int - - -@dataclass(frozen=True) -class DbState: - kb_ids: set[str] - file_ids: set[str] - benchmark_ids: set[str] - result_task_ids: set[str] - result_detail_count: int - - -def load_json_state() -> JsonState: - base_dir = os.path.join(config.save_dir, "knowledge_base_data") - global_meta = _load_json(os.path.join(base_dir, "global_metadata.json")).get("databases", {}) or {} - - kb_ids: set[str] = set(global_meta.keys()) - file_ids: set[str] = set() - benchmark_ids: set[str] = set() - result_task_ids: set[str] = set() - result_detail_count = 0 - - kb_type_dirs = [ - p for p in glob.glob(os.path.join(base_dir, "*_data")) if os.path.isdir(p) and os.path.basename(p) != "uploads" - ] - - for kb_dir in kb_type_dirs: - kb_type = os.path.basename(kb_dir)[: -len("_data")] - meta_file = os.path.join(kb_dir, f"metadata_{kb_type}.json") - meta = _load_json(meta_file) - - databases_meta: dict[str, Any] = meta.get("databases", {}) or {} - files_meta: dict[str, Any] = meta.get("files", {}) or {} - benchmarks_meta: dict[str, Any] = meta.get("benchmarks", {}) or {} - - kb_ids.update(databases_meta.keys()) - file_ids.update(files_meta.keys()) - - for _db_id, bmap in benchmarks_meta.items(): - if not isinstance(bmap, dict): - continue - benchmark_ids.update(bmap.keys()) - - for db_id in databases_meta.keys(): - result_dir = os.path.join(kb_dir, db_id, "results") - if not os.path.isdir(result_dir): - continue - for result_path in glob.glob(os.path.join(result_dir, "*.json")): - try: - data = _load_json(result_path) - except Exception: - continue - task_id = data.get("task_id") or os.path.splitext(os.path.basename(result_path))[0] - result_task_ids.add(task_id) - interim = data.get("interim_results") or data.get("results") or [] - result_detail_count += len(interim) - - return JsonState( - kb_ids=kb_ids, - file_ids=file_ids, - benchmark_ids=benchmark_ids, - result_task_ids=result_task_ids, - result_detail_count=result_detail_count, - ) - - -async def load_db_state() -> DbState: - async with pg_manager.get_async_session_context() as session: - kb_ids = set((await session.execute(select(KnowledgeBase.db_id))).scalars().all()) - file_ids = set((await session.execute(select(KnowledgeFile.file_id))).scalars().all()) - benchmark_ids = set((await session.execute(select(EvaluationBenchmark.benchmark_id))).scalars().all()) - result_task_ids = set((await session.execute(select(EvaluationResult.task_id))).scalars().all()) - detail_count = (await session.execute(select(func.count(EvaluationResultDetail.id)))).scalar_one() - - return DbState( - kb_ids=kb_ids, - file_ids=file_ids, - benchmark_ids=benchmark_ids, - result_task_ids=result_task_ids, - result_detail_count=int(detail_count or 0), - ) - - -def _diff(name: str, json_set: set[str], db_set: set[str], limit: int = 30) -> list[str]: - missing = sorted(json_set - db_set) - extra = sorted(db_set - json_set) - lines: list[str] = [] - lines.append(f"{name}: json={len(json_set)} db={len(db_set)}") - if missing: - preview = ", ".join(missing[:limit]) - lines.append(f" missing_in_db({len(missing)}): {preview}") - if extra: - preview = ", ".join(extra[:limit]) - lines.append(f" extra_in_db({len(extra)}): {preview}") - return lines - - -async def main() -> None: - engine_url = pg_manager.async_engine.url.render_as_string(hide_password=True) - print(f"db_url={engine_url}") - - json_state = load_json_state() - db_state = await load_db_state() - - for line in _diff("knowledge_bases.db_id", json_state.kb_ids, db_state.kb_ids): - print(line) - for line in _diff("knowledge_files.file_id", json_state.file_ids, db_state.file_ids): - print(line) - for line in _diff("evaluation_benchmarks.benchmark_id", json_state.benchmark_ids, db_state.benchmark_ids): - print(line) - for line in _diff("evaluation_results.task_id", json_state.result_task_ids, db_state.result_task_ids): - print(line) - - print( - "evaluation_result_details: " - f"json_count={json_state.result_detail_count} db_count={db_state.result_detail_count}" - ) - - -if __name__ == "__main__": - asyncio.run(main()) diff --git a/backend/test/conftest.py b/backend/test/conftest.py index 47c6fa21..d59c2ee8 100644 --- a/backend/test/conftest.py +++ b/backend/test/conftest.py @@ -5,9 +5,15 @@ Shared pytest fixtures for exercising FastAPI routers over the running API servi from __future__ import annotations import os +import sys import uuid from collections.abc import AsyncGenerator +# 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 diff --git a/backend/test/test_concurrency.py b/backend/test/test_concurrency.py deleted file mode 100644 index 72c7a54f..00000000 --- a/backend/test/test_concurrency.py +++ /dev/null @@ -1,136 +0,0 @@ -import asyncio -import time - -import aiohttp - - -async def make_request(session: aiohttp.ClientSession, request_id: int) -> dict: - """发送单个请求到API""" - url = "http://localhost:5000/chat/call" - payload = {"query": "写一个冒泡排序", "meta": {}} - - start_time = time.time() - print(f"请求 {request_id} 开始时间: {time.strftime('%H:%M:%S', time.localtime(start_time))}") - try: - async with session.post(url, json=payload) as response: - result = await response.json() - print(f"请求 {request_id} 结果: {result}") - end_time = time.time() - duration = end_time - start_time - print( - f"请求 {request_id} 完成时间: " - f"{time.strftime('%H:%M:%S', time.localtime(end_time))} " - f"(耗时: {duration:.2f}秒)" - ) - return { - "request_id": request_id, - "status": response.status, - "time": duration, - "start_time": start_time, - "end_time": end_time, - "success": True, - } - except Exception as e: - end_time = time.time() - duration = end_time - start_time - print( - f"请求 {request_id} 失败时间: " - f"{time.strftime('%H:%M:%S', time.localtime(end_time))} " - f"(耗时: {duration:.2f}秒)" - ) - return { - "request_id": request_id, - "status": None, - "time": duration, - "start_time": start_time, - "end_time": end_time, - "success": False, - "error": str(e), - } - - -async def run_concurrent_test(num_requests: int = 10) -> list[dict]: - """运行并发测试""" - async with aiohttp.ClientSession() as session: - tasks = [make_request(session, i) for i in range(num_requests)] - return await asyncio.gather(*tasks) - - -def analyze_results(results: list[dict]) -> None: - """分析并打印测试结果""" - total_requests = len(results) - successful_requests = sum(1 for r in results if r["success"]) - failed_requests = total_requests - successful_requests - - response_times = [r["time"] for r in results if r["success"]] - if response_times: - avg_time = sum(response_times) / len(response_times) - max_time = max(response_times) - min_time = min(response_times) - else: - avg_time = max_time = min_time = 0 - - print("\n=== 并发测试结果 ===") - print(f"总请求数: {total_requests}") - print(f"成功请求: {successful_requests}") - print(f"失败请求: {failed_requests}") - print(f"平均响应时间: {avg_time:.2f} 秒") - print(f"最长响应时间: {max_time:.2f} 秒") - print(f"最短响应时间: {min_time:.2f} 秒") - - if failed_requests > 0: - print("\n失败的请求:") - for result in results: - if not result["success"]: - print(f"请求 ID {result['request_id']}: {result.get('error', '未知错误')}") - - # 添加请求时间线分析 - print("\n=== 请求时间线分析 ===") - sorted_results = sorted(results, key=lambda x: x["start_time"]) - test_start_time = sorted_results[0]["start_time"] - test_end_time = sorted_results[-1]["end_time"] - print(f"测试开始时间: {time.strftime('%H:%M:%S', time.localtime(test_start_time))}") - print(f"测试结束时间: {time.strftime('%H:%M:%S', time.localtime(test_end_time))}") - - print("\n时间线详情:") - print("请求ID 开始时间 结束时间 耗时(秒) 重叠请求数") - active_requests = [] - - for result in sorted_results: - # 计算当前时间点的活跃请求数 - start_time = result["start_time"] - end_time = result["end_time"] - - # 清理已完成的请求 - active_requests = [t for t in active_requests if t > start_time] - active_requests.append(end_time) - - print( - f"{result['request_id']:^7} " - f"{time.strftime('%H:%M:%S', time.localtime(start_time))} " - f"{time.strftime('%H:%M:%S', time.localtime(end_time))} " - f"{result['time']:^8.2f} {len(active_requests):^6}" - ) - - # 计算最大并发数 - max_concurrent = 0 - timeline = [] - for r in results: - timeline.append((r["start_time"], 1)) - timeline.append((r["end_time"], -1)) - - timeline.sort(key=lambda x: x[0]) - current_concurrent = 0 - for _, change in timeline: - current_concurrent += change - max_concurrent = max(max_concurrent, current_concurrent) - - print(f"\n最大并发请求数: {max_concurrent}") - - -if __name__ == "__main__": - NUM_REQUESTS = 100 # 设置并发请求数 - - print(f"开始运行 {NUM_REQUESTS} 个并发请求的测试...") - results = asyncio.run(run_concurrent_test(NUM_REQUESTS)) - analyze_results(results) diff --git a/backend/test/test_kb_minio_cleanup.py b/backend/test/test_kb_minio_cleanup.py deleted file mode 100644 index d8121924..00000000 --- a/backend/test/test_kb_minio_cleanup.py +++ /dev/null @@ -1,132 +0,0 @@ -""" -测试知识库删除时 MinIO 文件清理功能 - -测试步骤： -1. 创建测试知识库 -2. 上传测试文件到 MinIO（模拟文件上传流程） -3. 删除知识库 -4. 验证 MinIO 文件是否被清理 -""" - -import asyncio -import os -import sys - -# 添加项目根目录到 Python 路径 -sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) - - -async def test_delete_knowledge_base_cleanup(): - """测试删除知识库时 MinIO 文件清理""" - from yuxi.storage.minio import get_minio_client - from yuxi.knowledge import knowledge_base - - # 初始化 - minio_client = get_minio_client() - kb_manager = knowledge_base - doc_bucket = minio_client.KB_BUCKETS["documents"] - parsed_bucket = minio_client.KB_BUCKETS["parsed"] - - # 测试参数 - test_db_name = "test_cleanup_db" - test_file_content = b"This is test content for cleanup verification" - - try: - # 1. 创建知识库 - print("1. 创建测试知识库...") - kb_info = await kb_manager.create_database( - database_name=test_db_name, - description="Test knowledge base for MinIO cleanup", - kb_type="lightrag", - ) - db_id = kb_info["db_id"] - print(f" 知识库创建成功: db_id={db_id}") - - # 2. 上传测试文件到 MinIO（模拟文件上传） - print("2. 上传测试文件到 MinIO...") - - # 确保 bucket 存在 - minio_client.ensure_bucket_exists(doc_bucket) - minio_client.ensure_bucket_exists(parsed_bucket) - - # 上传到知识库文档路径 (knowledgebases/{db_id}/upload/...) - test_object_name = f"{db_id}/upload/test_file_{db_id[:8]}.txt" - minio_client.upload_file( - bucket_name=doc_bucket, - object_name=test_object_name, - data=test_file_content, - content_type="text/plain", - ) - print(f" 文件上传到 {doc_bucket}: {test_object_name}") - - # 上传到知识库解析路径 (knowledgebases/{db_id}/parsed/...) - test_file_id = f"file_test_{db_id[:8]}" - parsed_object_name = f"{db_id}/parsed/{test_file_id}.md" - parsed_content = b"# Test Parsed Content\n\nThis is parsed markdown." - minio_client.upload_file( - bucket_name=parsed_bucket, - object_name=parsed_object_name, - data=parsed_content, - content_type="text/markdown", - ) - print(f" 文件上传到 {parsed_bucket}: {parsed_object_name}") - - # 3. 验证文件存在 - print("3. 验证文件存在...") - assert minio_client.file_exists(doc_bucket, test_object_name), "原始文件应该存在" - assert minio_client.file_exists(parsed_bucket, parsed_object_name), "解析后文件应该存在" - print(" 文件存在验证通过") - - # 4. 删除知识库 - print("4. 删除知识库...") - result = await kb_manager.delete_database(db_id) - print(f" 删除结果: {result}") - - # 5. 验证 MinIO 文件已清理 - print("5. 验证 MinIO 文件已清理...") - - # 检查文档对象 - doc_exists = minio_client.file_exists(doc_bucket, test_object_name) - print(f" {doc_bucket}/{test_object_name} 存在: {doc_exists}") - - # 检查解析对象 - parsed_exists = minio_client.file_exists(parsed_bucket, parsed_object_name) - print(f" {parsed_bucket}/{parsed_object_name} 存在: {parsed_exists}") - - # 6. 输出测试结果 - print("\n" + "=" * 50) - if not doc_exists and not parsed_exists: - print("✅ 测试通过！MinIO 文件已成功清理") - return True - else: - print("❌ 测试失败！以下文件未被清理:") - if doc_exists: - print(f" - {doc_bucket}/{test_object_name}") - if parsed_exists: - print(f" - {parsed_bucket}/{parsed_object_name}") - return False - - except Exception as e: - print(f"\n❌ 测试异常: {e}") - import traceback - - traceback.print_exc() - return False - - finally: - # 清理测试数据（确保清理干净） - print("\n清理测试数据...") - try: - # 尝试删除可能残留的文件（忽略错误） - for bucket, obj in [(doc_bucket, test_object_name), (parsed_bucket, parsed_object_name)]: - try: - await minio_client.adelete_file(bucket, obj) - except Exception: - pass - except Exception as e: - print(f" 清理时出错: {e}") - - -if __name__ == "__main__": - result = asyncio.run(test_delete_knowledge_base_cleanup()) - sys.exit(0 if result else 1) diff --git a/backend/test/test_manual_eval.py b/backend/test/test_manual_eval.py deleted file mode 100644 index b34109fe..00000000 --- a/backend/test/test_manual_eval.py +++ /dev/null @@ -1,212 +0,0 @@ -import requests -import json -import time -import sys -import os - - -# 添加评估指标测试功能 -def test_evaluation_metrics(): - """测试评估指标计算""" - print("\n" + "=" * 50) - print("测试评估指标计算") - print("=" * 50) - - try: - from yuxi.utils.evaluation_metrics import EvaluationMetricsCalculator - - # 测试检索指标 - retrieved_chunks = [ - {"content": "test1", "metadata": {"chunk_id": "file_bbb147_chunk_0"}}, - {"content": "test2", "metadata": {"chunk_id": "file_bbb147_chunk_1"}}, - {"content": "test3", "metadata": {"chunk_id": "file_bbb147_chunk_2"}}, - ] - gold_chunk_ids = ["file_bbb147_chunk_0", "file_bbb147_chunk_2"] - - print("测试检索指标...") - retrieval_metrics = EvaluationMetricsCalculator.calculate_retrieval_metrics(retrieved_chunks, gold_chunk_ids) - print(f"检索指标结果: {retrieval_metrics}") - - # 测试答案指标 - # generated_answer = "该研究以数据语义化—知识结构化—可信推理的技术主线" - # gold_answer = "该研究以数据语义化—知识结构化—可信推理的技术主线，遵循数据—知识—推理—应用的演化逻辑" - - print("\n测试答案指标（需要Judge LLM，跳过实际LLM调用）...") - # 由于需要judge_llm，这里只测试检索指标 - print("跳过答案指标测试（需要配置Judge LLM）") - - print("评估指标计算测试完成！") - return True - - except Exception as e: - print(f"评估指标测试失败: {e}") - return False - - -BASE_URL = "http://localhost:5050" -USERNAME = "zwj" -PASSWORD = "zwj12138" -DB_ID = "kb_5e343066eb4713959698ae6ca16843a0" - - -def get_token(): - try: - resp = requests.post(f"{BASE_URL}/api/auth/token", data={"username": USERNAME, "password": PASSWORD}) - if resp.status_code != 200: - print(f"Login failed: {resp.text}") - return None - return resp.json()["access_token"] - except Exception as e: - print(f"Connection failed: {e}") - return None - - -def main(): - # 首先测试评估指标计算 - test_evaluation_metrics() - - token = get_token() - if not token: - sys.exit(1) - - headers = {"Authorization": f"Bearer {token}"} - - print(f"\n1. Trying to retrieve a real chunk from {DB_ID}...") - - chunk_content = "This is a fallback test query." - chunk_id = "unknown" - - try: - # 尝试查询 (Fixed payload structure) - resp = requests.post( - f"{BASE_URL}/api/knowledge/databases/{DB_ID}/query", - headers=headers, - json={"query": "人工智能", "meta": {"top_k": 5}}, - ) - - if resp.status_code == 200: - data = resp.json() - # 结果在 result 字段中 - results = data.get("result", []) - - first_chunk = None - if isinstance(results, list) and len(results) > 0: - first_chunk = results[0] - elif isinstance(results, dict) and "retrieved_chunks" in results: - if results["retrieved_chunks"]: - first_chunk = results["retrieved_chunks"][0] - - if first_chunk: - print(f"Found chunk: {str(first_chunk.get('content', ''))[:50]}...") - chunk_content = first_chunk.get("content", chunk_content) - chunk_id = first_chunk.get("chunk_id", chunk_id) or first_chunk.get("id", chunk_id) - else: - print("No chunks found in retrieval results. Using fallback.") - print(f"Raw results: {results}") - else: - print(f"Query failed: {resp.text}") - - except Exception as e: - print(f"Query Exception: {e}") - - print(f"\n2. Creating benchmark with query based on chunk: {chunk_id}") - - # 构造 Benchmark 数据 - benchmark_data = { - "query": chunk_content, # 使用 Chunk 内容作为查询，理论上应该能召回它自己 - "gold_chunk_ids": [str(chunk_id)], # Ensure string - "gold_answer": "This is a gold answer.", - } - - # 生成临时文件 - with open("temp_benchmark.jsonl", "w") as f: - f.write(json.dumps(benchmark_data, ensure_ascii=False) + "\n") - - print("\n3. Uploading benchmark...") - try: - files = {"file": ("temp_benchmark.jsonl", open("temp_benchmark.jsonl", "rb"), "application/jsonlines")} - # Fixed: use params for name/description - params = {"name": "Manual Eval Test", "description": "Test generated from script"} - - resp = requests.post( - f"{BASE_URL}/api/evaluation/databases/{DB_ID}/benchmarks/upload", - headers=headers, - params=params, - files=files, - ) - - if resp.status_code != 200: - print(f"Upload failed: {resp.text}") - sys.exit(1) - - benchmark = resp.json() - # Benchmark response format depends on Service impl. - # My filesystem impl returns the metadata dict directly. - # But wait, KnowledgeRouter might wrap it? - # router returns: return {"message": "上传成功", "data": result} (Assume standard wrapper) - # Let's check router impl. - - # From router code read earlier: - # result = await service.upload_benchmark(...) - # return {"message": "上传成功", "data": result} - - benchmark_id = benchmark["data"]["benchmark_id"] - print(f"Benchmark uploaded: {benchmark_id}") - - except Exception as e: - print(f"Upload Exception: {e}") - sys.exit(1) - - print("\n4. Running evaluation...") - try: - payload = {"benchmark_id": benchmark_id, "retrieval_config": {"top_k": 5}} - - resp = requests.post(f"{BASE_URL}/api/evaluation/databases/{DB_ID}/run", headers=headers, json=payload) - - if resp.status_code != 200: - print(f"Run evaluation failed: {resp.text}") - sys.exit(1) - - task_id = resp.json()["data"]["task_id"] - print(f"Evaluation task started: {task_id}") - - # 轮询状态 - while True: - resp = requests.get(f"{BASE_URL}/api/evaluation/{task_id}/progress", headers=headers) - if resp.status_code != 200: - print(f"Get progress failed: {resp.text}") - break - - progress = resp.json() - # The progress endpoint returns {task_id, status, ...} based on my service impl? - # Router wrapper: return {"message": "success", "data": result} - - data = progress.get("data", progress) # Handle wrapper if exists - status = data["status"] - current_progress = data.get("progress", 0) - - print(f"Status: {status}, Progress: {current_progress}%") - - if status in ["completed", "failed"]: - break - - time.sleep(2) - - if status == "completed": - print("\nEvaluation Completed!") - # 获取结果 - resp = requests.get(f"{BASE_URL}/api/evaluation/{task_id}/results", headers=headers) - print(json.dumps(resp.json(), indent=2, ensure_ascii=False)) - else: - print("\nEvaluation Failed!") - - except Exception as e: - print(f"Evaluation Exception: {e}") - - # 清理 - if os.path.exists("temp_benchmark.jsonl"): - os.remove("temp_benchmark.jsonl") - - -if __name__ == "__main__": - main() diff --git a/backend/test/test_mysql_connection.py b/backend/test/test_mysql_connection.py deleted file mode 100644 index ed030d91..00000000 --- a/backend/test/test_mysql_connection.py +++ /dev/null @@ -1,206 +0,0 @@ -#!/usr/bin/env python3 -""" -MySQL 数据库连接验证脚本 -""" - -import os - -# 添加项目根目录到 Python 路径 -# sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent.parent.parent)) - - -def load_env_file(env_file=".env"): - """加载 .env 文件""" - env_vars = {} - if os.path.exists(env_file): - with open(env_file, encoding="utf-8") as f: - for line in f: - line = line.strip() - if line and not line.startswith("#") and "=" in line: - key, value = line.split("=", 1) - env_vars[key] = value.strip("\"'") - return env_vars - - -def test_mysql_connection(): - """测试 MySQL 连接""" - print("=== MySQL 数据库连接验证 ===\n") - - # 加载环境变量 - env_vars = load_env_file() - - # 检查必需的环境变量 - required_vars = ["MYSQL_HOST", "MYSQL_USER", "MYSQL_PASSWORD", "MYSQL_DATABASE"] - missing_vars = [] - - for var in required_vars: - if var not in env_vars: - missing_vars.append(var) - - if missing_vars: - print(f"❌ 缺少必需的环境变量: {', '.join(missing_vars)}") - print("\n请在 .env 文件中设置以下变量:") - for var in missing_vars: - print(f" {var}=your_value") - return False - - # 显示配置信息（隐藏密码） - print("📋 数据库配置:") - print(f" Host: {env_vars.get('MYSQL_HOST', 'Not set')}") - print(f" User: {env_vars.get('MYSQL_USER', 'Not set')}") - print(f" Database: {env_vars.get('MYSQL_DATABASE', 'Not set')}") - print(f" Port: {env_vars.get('MYSQL_PORT', '3306')}") - print(f" Charset: {env_vars.get('MYSQL_CHARSET', 'utf8mb4')}") - print() - - try: - # 导入 MySQL 连接管理器 - from yuxi.agents.toolkits.mysql.connection import MySQLConnectionManager - - # 创建连接配置 - mysql_config = { - "host": env_vars["MYSQL_HOST"], - "user": env_vars["MYSQL_USER"], - "password": env_vars["MYSQL_PASSWORD"], - "database": env_vars["MYSQL_DATABASE"], - "port": int(env_vars.get("MYSQL_PORT", "3306")), - "charset": env_vars.get("MYSQL_CHARSET", "utf8mb4"), - } - - # 创建连接管理器 - print("🔄 正在连接数据库...") - conn_manager = MySQLConnectionManager(mysql_config) - - # 测试连接 - with conn_manager.get_cursor() as cursor: - # 测试基本连接 - cursor.execute("SELECT 1 as test") - result = cursor.fetchone() - if result and result["test"] == 1: - print("✅ 数据库连接成功！") - - # 获取数据库版本 - cursor.execute("SELECT VERSION() as version") - version_info = cursor.fetchone() - print(f"📊 MySQL 版本: {version_info['version']}") - - # 获取当前数据库 - cursor.execute("SELECT DATABASE() as db_name") - db_info = cursor.fetchone() - print(f"🗄️ 当前数据库: {db_info['db_name']}") - - # 获取表数量 - cursor.execute("SHOW TABLES") - tables = cursor.fetchall() - print(f"📋 表数量: {len(tables)}") - - if tables: - print("\n📝 数据库表列表:") - for i, table in enumerate(tables[:10]): # 只显示前10个表 - table_name = list(table.values())[0] - print(f" {i + 1}. {table_name}") - - if len(tables) > 10: - print(f" ... 还有 {len(tables) - 10} 个表未显示") - - return True - else: - print("❌ 数据库连接测试失败") - return False - - except ImportError as e: - print(f"❌ 导入错误: {e}") - print("请确保已安装 pymysql 依赖") - return False - except Exception as e: - print(f"❌ 连接失败: {e}") - - # 提供故障排除建议 - print("\n💡 故障排除建议:") - print("1. 检查数据库服务是否正在运行") - print("2. 验证主机地址和端口是否正确") - print("3. 确认用户名和密码是否正确") - print("4. 检查数据库是否存在") - print("5. 确认网络连接和防火墙设置") - - return False - - -def test_tools(): - """测试 MySQL 工具是否正常工作""" - print("\n=== MySQL 工具测试 ===\n") - - try: - from yuxi.agents.toolkits.mysql.tools import mysql_list_tables, mysql_describe_table, mysql_query - - print("✅ MySQL 工具导入成功") - - # 测试获取表名 - print("\n🔄 测试获取表名...") - result = mysql_list_tables.invoke({}) - if "失败" not in result and "错误" not in result: - print("✅ 获取表名工具正常") - else: - print(f"❌ 获取表名工具异常: {result}") - return False - - # 如果有表，测试获取表结构 - if "数据库中的表:" in result: - print("\n🔄 测试获取表结构...") - # 提取第一个表名 - lines = result.split("\n") - for line in lines: - if "- " in line and "(" in line: - table_name = line.split("- ")[1].split(" ")[0] - break - else: - table_name = None - - if table_name: - structure_result = mysql_describe_table.invoke({"table_name": table_name}) - if "失败" not in structure_result and "错误" not in structure_result: - print("✅ 获取表结构工具正常") - else: - print(f"❌ 获取表结构工具异常: {structure_result}") - return False - - # 测试简单查询 - print("\n🔄 测试SQL查询...") - query_result = mysql_query.invoke({"sql": f"SELECT COUNT(*) as total FROM `{table_name}`"}) - if "失败" not in query_result and "错误" not in query_result: - print("✅ SQL查询工具正常") - else: - print(f"❌ SQL查询工具异常: {query_result}") - return False - - return True - - except Exception as e: - print(f"❌ 工具测试失败: {e}") - return False - - -def main(): - """主函数""" - print("MySQL 数据库连接和工具验证脚本") - print("=" * 50) - - # 测试连接 - connection_ok = test_mysql_connection() - - if connection_ok: - # 测试工具 - tools_ok = test_tools() - - print("\n" + "=" * 50) - if connection_ok and tools_ok: - print("🎉 所有测试通过！MySQL 工具包可以正常使用") - else: - print("❌ 部分测试失败，请检查配置") - else: - print("\n" + "=" * 50) - print("❌ 数据库连接失败，请检查配置") - - -if __name__ == "__main__": - main() diff --git a/backend/test/test_mysql_import.py b/backend/test/test_mysql_import.py deleted file mode 100644 index c3a5a231..00000000 --- a/backend/test/test_mysql_import.py +++ /dev/null @@ -1,24 +0,0 @@ -#!/usr/bin/env python3 -""" -简单测试 MySQL 工具导入 -""" - -import sys - -sys.path.insert(0, ".") - -try: - from yuxi.agents.toolkits.mysql.tools import mysql_list_tables, mysql_describe_table, mysql_query - - print("✅ MySQL 工具导入成功") - print(f"工具列表: {[tool.name for tool in [mysql_list_tables, mysql_describe_table, mysql_query]]}") - - # 检查工具描述 - for tool in [mysql_list_tables, mysql_describe_table, mysql_query]: - print(f" - {tool.name}: {tool.description[:60]}...") - -except Exception as e: - print(f"❌ 导入失败: {e}") - import traceback - - traceback.print_exc() diff --git a/backend/test/test_neo4j.py b/backend/test/test_neo4j.py deleted file mode 100755 index f4c6e1c7..00000000 --- a/backend/test/test_neo4j.py +++ /dev/null @@ -1,37 +0,0 @@ -from neo4j import GraphDatabase -from neo4j.exceptions import AuthError, ServiceUnavailable - -# sudo ln -s /snap/core22/1586/usr/sbin/iptables /usr/sbin/iptables - - -def check_neo4j_status(uri="bolt://localhost:7687", username="neo4j", password="0123456789"): - """ - 检查 Neo4j 数据库是否可以连接并正常工作。 - - 参数: - uri (str): Neo4j 的 URI，默认为 "bolt://localhost:7687" - username (str): 数据库用户名，默认为 "neo4j" - password (str): 数据库密码，默认为 "0123456789" - - 返回: - str: "OK" 表示连接成功，"UNAVAILABLE" 表示服务不可用，"AUTH_FAILED" 表示认证失败。 - """ - try: - driver = GraphDatabase.driver(uri, auth=(username, password)) - with driver.session() as session: - # 简单的查询来测试连接 - result = session.run("RETURN 1") - if result.single()[0] == 1: - return "OK" - except ServiceUnavailable: - return "UNAVAILABLE" - except AuthError: - return "AUTH_FAILED" - finally: - # 确保关闭驱动 - driver.close() - - -# 测试函数 -status = check_neo4j_status() -print(f"Neo4j status: {status}") diff --git a/backend/test/test_subagent.py b/backend/test/test_subagent.py index 43eded64..cbffa029 100644 --- a/backend/test/test_subagent.py +++ b/backend/test/test_subagent.py @@ -457,26 +457,6 @@ class TestSubAgentService: assert second[0]["tools"] == ["tool_a"] service_module.clear_specs_cache() - def test_resolve_subagent_tools_does_not_mutate_input(self): - from yuxi.services import subagent_service as service_module - - mock_tool = MagicMock() - mock_tool.name = "tool_a" - specs = [ - { - "name": "test-agent", - "description": "Test", - "system_prompt": "You are a test", - "tools": ["tool_a"], - } - ] - - resolved = service_module.resolve_subagent_tools(specs, [mock_tool]) - - assert specs[0]["tools"] == ["tool_a"] - assert resolved[0]["tools"] == [mock_tool] - - # ============================================================================= # Model Tests # ============================================================================= @@ -549,49 +529,6 @@ class TestSubAgentModel: class TestDeepAgentSubagentSelection: - def test_filter_specs_by_names_empty_selection_returns_empty(self): - from yuxi.services.subagent_service import filter_specs_by_names - - specs = [ - {"name": "research-agent", "description": "r"}, - {"name": "critique-agent", "description": "c"}, - ] - - filtered, missing = filter_specs_by_names(specs, []) - - assert filtered == [] - assert missing == [] - - def test_filter_specs_by_names_none_selection_returns_all(self): - from yuxi.services.subagent_service import filter_specs_by_names - - specs = [ - {"name": "research-agent", "description": "r"}, - {"name": "critique-agent", "description": "c"}, - ] - - filtered, missing = filter_specs_by_names(specs, None) - - assert filtered == specs - assert missing == [] - - def test_filter_specs_by_names_returns_subset_and_missing(self): - from yuxi.services.subagent_service import filter_specs_by_names - - specs = [ - {"name": "research-agent", "description": "r"}, - {"name": "critique-agent", "description": "c"}, - {"name": 123, "description": "invalid"}, - ] - - filtered, missing = filter_specs_by_names( - specs, - ["research-agent", "missing-agent", "research-agent", ""], - ) - - assert [item["name"] for item in filtered] == ["research-agent"] - assert missing == ["missing-agent"] - @pytest.mark.asyncio async def test_get_subagents_from_names_filters_and_resolves_tools(self, monkeypatch): from yuxi.services import subagent_service as service_module