From cfbc4e4c6b173f48195752de6a8af42c0661622c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=AC=A7=E9=98=B3=E6=B4=8B?= <2443479321@qq.com> Date: Tue, 8 Sep 2026 19:46:02 +0800 Subject: [PATCH] =?UTF-8?q?test:=E5=88=A0=E9=99=A4=E6=B5=8B=E8=AF=95?= =?UTF-8?q?=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- test_auth.py | 67 ---------------------------- test_llm.py | 118 ------------------------------------------------- test_main.http | 11 ----- test_repo.py | 44 ------------------ 4 files changed, 240 deletions(-) delete mode 100644 test_auth.py delete mode 100644 test_llm.py delete mode 100644 test_main.http delete mode 100644 test_repo.py diff --git a/test_auth.py b/test_auth.py deleted file mode 100644 index 7cc58d2..0000000 --- a/test_auth.py +++ /dev/null @@ -1,67 +0,0 @@ -"""认证 E2E：登录取 token → /auth/me 依赖注入回查 sys_user → 反例（错密码/坏 token/已删用户）。 - -全程跑在同一个事件循环（httpx.ASGITransport），避免单例 async engine 跨循环。 -""" -import asyncio - -import httpx - -import main as appmod -from config.database.mysql import get_engine, get_session_factory -from model.sys_user import SysUser -from repositories.sys_user import SysUserRepo -from service.auth import hash_password - -USER = "auth_test_u" -PWD = "Test@1234" - - -async def seed() -> int: - async with get_session_factory()() as s: - repo = SysUserRepo(s) - existing = await repo.get_by_username(USER) - if existing: - return existing.id - u = SysUser(username=USER, password_hash=hash_password(PWD), phone="13800001111", - user_type="EMPLOYEE", employee_role="理财顾问", status="正常") - return (await repo.add(u)).id - - -async def cleanup(uid: int) -> None: - async with get_session_factory()() as s: - await SysUserRepo(s).delete(uid) - - -async def run(): - uid = await seed() - try: - transport = httpx.ASGITransport(app=appmod.app) - async with httpx.AsyncClient(transport=transport, base_url="http://test") as c: - # 错密码 → 400 - r = await c.post("/api/auth/login", json={"username": USER, "password": "wrong"}) - assert r.status_code == 400 and r.json()["code"] == 400, r.text - # 正常登录 → token - r = await c.post("/api/auth/login", json={"username": USER, "password": PWD}) - assert r.status_code == 200, r.text - token = r.json()["data"]["token"] - assert token - # /auth/me：依赖注入按 id 回查 sys_user - r = await c.get("/api/auth/me", headers={"Authorization": f"Bearer {token}"}) - me = r.json()["data"]["user"] - assert r.status_code == 200 and me["id"] == uid and me["employee_role"] == "理财顾问", r.text - # 缺 token / 坏 token → 401 - assert (await c.get("/api/auth/me")).status_code == 401 - assert (await c.get("/api/auth/me", headers={"Authorization": "Bearer abc.def.ghi"})).status_code == 401 - # 删除用户后 token 回查失败 → 401 - await cleanup(uid) - assert (await c.get("/api/auth/me", headers={"Authorization": f"Bearer {token}"})).status_code == 401 - # 不存在用户登录 → 400 - r = await c.post("/api/auth/login", json={"username": "no_such_user", "password": PWD}) - assert r.status_code == 400 - print("AUTH TESTS PASSED") - finally: - await get_engine().dispose() - - -if __name__ == "__main__": - asyncio.run(run()) \ No newline at end of file diff --git a/test_llm.py b/test_llm.py deleted file mode 100644 index 7b13813..0000000 --- a/test_llm.py +++ /dev/null @@ -1,118 +0,0 @@ -"""tool.llm 自检：本地 Mock OpenAI 兼容服务，验证双模式 URL、重试、备用模型降级、embedding。 - -运行：python test_llm.py（纯 assert，无框架；可通过 exitcode 判断） -""" -import asyncio -import http.server -import json -import threading - -from config.settings import settings -from tool.llm import LLMFailError, LLMClient - -INDEX = json.dumps({"index": 0}).encode() - - -class MockLLMServer(http.server.BaseHTTPRequestHandler): - """OpenAI 兼容 Mock：chat/embeddings/tags。model=boom:1b 返回 500 用于测降级。""" - - def log_message(self, *args): - pass - - def _reply(self, code: int, obj: dict): - body = json.dumps(obj).encode() - self.send_response(code) - self.send_header("Content-Type", "application/json") - self.send_header("Content-Length", str(len(body))) - self.end_headers() - self.wfile.write(body) - - def do_GET(self): # ollama /api/tags 健康探测 - self._reply(200, {"models": [{"name": "qwen2.5:7b"}]}) - - def do_POST(self): - n = int(self.headers.get("Content-Length", 0)) - req = json.loads(self.rfile.read(n)) - if self.path == "/v1/chat/completions": - model = req["model"] - if model.startswith("boom"): - self._reply(500, {"error": "model overloaded"}) - return - self._reply(200, {"choices": [{"message": {"content": f"reply:{model}"}}]}) - elif self.path == "/v1/embeddings": - texts = req["input"] - self._reply(200, {"data": [{"embedding": [float(i + 1)] * 4} for i in range(len(texts))]}) - else: - self._reply(404, {"error": "not found"}) - - -def start_server() -> tuple[threading.Thread, int]: - srv = http.server.ThreadingHTTPServer(("127.0.0.1", 0), MockLLMServer) - port = srv.server_address[1] - t = threading.Thread(target=srv.serve_forever, daemon=True) - t.start() - return t, port - - -def cfg(base_url: str, **kw): - """从 .env 加载的 settings.llm 派生测试配置（只覆盖用例所需字段）。""" - kw.setdefault("mode", "ollama") - kw.setdefault("ollama_base", base_url) - kw.setdefault("max_retries", 2) - kw.setdefault("retry_backoff_sec", 0) - return settings.llm.model_copy(update=kw) - - -async def achat(c: LLMClient, text: str = "hi") -> str: - return await c.chat([{"role": "user", "content": text}]) - - -def main(): - _, port = start_server() - base = f"http://127.0.0.1:{port}" - - # 1. ollama 模式 URL + 正常 chat - c = LLMClient(cfg(base, ollama_chat_model="ok:1b")) - assert c.is_ollama and c.base_url == f"{base}/v1", c.base_url - out = asyncio.run(achat(c)) - assert out == "reply:ok:1b", out - - # 2. 备用模型降级：主模型 500 → 自动切 fallback_chat_model - c2 = LLMClient(cfg(base, ollama_chat_model="boom:1b", ollama_embed_model="bge-m3", - fallback_chat_model="ok:1b")) - out = asyncio.run(achat(c2)) - assert out == "reply:ok:1b", out - - # 3. 全部失败 → LLMFailError（重试耗尽） - c3 = LLMClient(cfg(base, ollama_chat_model="boom:1b", max_retries=1)) - try: - asyncio.run(achat(c3)) - raise AssertionError("should have raised LLMFailError") - except LLMFailError: - pass - - # 4. API 模式：端点/鉴权头/模型选择 - c4 = LLMClient(settings.llm.model_copy(update={ - "mode": "api", "api_base": base + "/v1", "api_key": "sk-test", - "api_chat_model": "gpt-x", "max_retries": 1, - })) - assert not c4.is_ollama and c4.base_url == f"{base}/v1" - assert c4._headers()["Authorization"] == "Bearer sk-test" - assert c4.chat_model == "gpt-x" - out = asyncio.run(achat(c4)) - assert out == "reply:gpt-x", out - - # 5. embedding 批量 - c5 = LLMClient(cfg(base)) - vecs = asyncio.run(c5.embed(["基金", "债券"])) - assert len(vecs) == 2 and len(vecs[0]) == 4, vecs - assert asyncio.run(c5.embed_one("货币")) == [1.0] * 4 - - # 6. 健康探测（ollama /api/tags） - asyncio.run(c5.check_health()) - - print("ALL LLM TESTS PASSED") - - -if __name__ == "__main__": - main() \ No newline at end of file diff --git a/test_main.http b/test_main.http deleted file mode 100644 index a2d81a9..0000000 --- a/test_main.http +++ /dev/null @@ -1,11 +0,0 @@ -# Test your FastAPI endpoints - -GET http://127.0.0.1:8000/ -Accept: application/json - -### - -GET http://127.0.0.1:8000/hello/User -Accept: application/json - -### diff --git a/test_repo.py b/test_repo.py deleted file mode 100644 index 9c17af6..0000000 --- a/test_repo.py +++ /dev/null @@ -1,44 +0,0 @@ -"""数据访问层自检：MySQL 可达且表已建则跑真实 CRUD；否则 SKIP（exit 0）。""" -import asyncio - -from sqlalchemy import text - -from config.database.mysql import get_engine, get_session_factory -from repositories.sys_config import SysConfigRepo - - -async def _db_ok() -> bool: - """MySQL 可达且 sys_config 表已建（schema.sql 已导入）才算就绪。""" - try: - async with get_engine().connect() as conn: - await conn.execute(text("SELECT 1")) - async with get_session_factory()() as session: - await session.execute(text("SELECT 1 FROM sys_config LIMIT 1")) - return True - except Exception: - return False - - -async def main(): - try: - if not await _db_ok(): - print("SKIP: MySQL 不可达或 sys_config 未建表，CRUD 检查延后（先启动 MySQL 并导入 sql/schema.sql）") - return - async with get_session_factory()() as session: - repo = SysConfigRepo(session) - row = await repo.set_value("test.hnw", "3000000", "自检") - assert row.config_key == "test.hnw" - assert await repo.get_value("test.hnw") == "3000000" - assert await repo.count() > 0 - rows = await repo.list(where=(repo.model.config_key == "test.hnw",)) - assert len(rows) == 1 - assert await repo.delete(rows[0].id) - assert await repo.get_value("test.hnw") is None # 幂等清理 - print("REPO TESTS PASSED") - finally: - import config.database as _db - await _db.dispose() # 干净收尾，避免连接在事件循环关闭后被 GC - - -if __name__ == "__main__": - asyncio.run(main()) \ No newline at end of file -- 2.54.0