This commit is contained in:
amb
2026-09-01 13:00:36 +08:00
parent 1d8621717a
commit dfd38c99a0
35 changed files with 3141 additions and 21 deletions
+171
View File
@@ -0,0 +1,171 @@
"""Phase 4 知识库 CRUD + Token 管理测试。"""
from fastapi.testclient import TestClient
from app.main import app
def _client() -> TestClient:
return TestClient(app, raise_server_exceptions=False)
def _register_and_login(client: TestClient, username: str = "testuser") -> None:
client.post("/api/auth/register", json={
"username": username,
"email": f"{username}@example.com",
"password": "password123",
})
# --- 创建 ---
def test_create_knowledge_base() -> None:
with _client() as client:
_register_and_login(client)
resp = client.post("/api/knowledge-bases", json={
"name": "公司知识库",
"description": "公司相关文档",
})
assert resp.status_code == 201
body = resp.json()
assert body["name"] == "公司知识库"
assert body["description"] == "公司相关文档"
assert body["enabled"] is True
assert body["token_hint"] is not None
assert body["ai_url"].startswith("/k/")
def test_create_kb_name_required() -> None:
with _client() as client:
_register_and_login(client)
resp = client.post("/api/knowledge-bases", json={"name": ""})
assert resp.status_code == 422
# --- 列表 ---
def test_list_knowledge_bases() -> None:
with _client() as client:
_register_and_login(client)
client.post("/api/knowledge-bases", json={"name": "KB1"})
client.post("/api/knowledge-bases", json={"name": "KB2"})
resp = client.get("/api/knowledge-bases")
assert resp.status_code == 200
body = resp.json()
assert body["total"] == 2
assert len(body["items"]) == 2
def test_list_kb_pagination() -> None:
with _client() as client:
_register_and_login(client)
for i in range(5):
client.post("/api/knowledge-bases", json={"name": f"KB{i}"})
resp = client.get("/api/knowledge-bases?page=1&page_size=2")
assert resp.status_code == 200
body = resp.json()
assert body["total"] == 5
assert len(body["items"]) == 2
# --- 详情 ---
def test_get_knowledge_base() -> None:
with _client() as client:
_register_and_login(client)
create_resp = client.post("/api/knowledge-bases", json={"name": "详情测试"})
kb_id = create_resp.json()["id"]
resp = client.get(f"/api/knowledge-bases/{kb_id}")
assert resp.status_code == 200
assert resp.json()["name"] == "详情测试"
def test_get_kb_not_found() -> None:
with _client() as client:
_register_and_login(client)
resp = client.get("/api/knowledge-bases/nonexistent")
assert resp.status_code == 404
# --- 编辑 ---
def test_update_knowledge_base() -> None:
with _client() as client:
_register_and_login(client)
create_resp = client.post("/api/knowledge-bases", json={"name": "旧名称"})
kb_id = create_resp.json()["id"]
resp = client.put(f"/api/knowledge-bases/{kb_id}", json={"name": "新名称"})
assert resp.status_code == 200
assert resp.json()["name"] == "新名称"
# --- 删除 ---
def test_delete_knowledge_base() -> None:
with _client() as client:
_register_and_login(client)
create_resp = client.post("/api/knowledge-bases", json={"name": "待删除"})
kb_id = create_resp.json()["id"]
# 删除
resp = client.delete(f"/api/knowledge-bases/{kb_id}")
assert resp.status_code == 204
# 删除后对自己也查不到(status=DELETED
resp = client.get(f"/api/knowledge-bases/{kb_id}")
assert resp.status_code == 404
# --- Token 管理 ---
def test_regenerate_token() -> None:
with _client() as client:
_register_and_login(client)
create_resp = client.post("/api/knowledge-bases", json={"name": "Token 测试"})
kb_id = create_resp.json()["id"]
old_hint = create_resp.json()["token_hint"]
resp = client.post(f"/api/knowledge-bases/{kb_id}/regenerate-token")
assert resp.status_code == 200
body = resp.json()
assert body["ai_url"].startswith("/k/")
assert body["token"] != ""
# hint 可能相同(概率极低)但 token 应该不同
def test_enable_disable() -> None:
with _client() as client:
_register_and_login(client)
create_resp = client.post("/api/knowledge-bases", json={"name": "开关测试"})
kb_id = create_resp.json()["id"]
# 禁用
resp = client.post(f"/api/knowledge-bases/{kb_id}/disable")
assert resp.status_code == 200
assert resp.json()["enabled"] is False
# 启用
resp = client.post(f"/api/knowledge-bases/{kb_id}/enable")
assert resp.status_code == 200
assert resp.json()["enabled"] is True
def test_get_link() -> None:
with _client() as client:
_register_and_login(client)
create_resp = client.post("/api/knowledge-bases", json={"name": "链接测试"})
kb_id = create_resp.json()["id"]
resp = client.get(f"/api/knowledge-bases/{kb_id}/link")
assert resp.status_code == 200
body = resp.json()
assert body["ai_url"].startswith("/k/")
assert len(body["token"]) > 0
# --- IDOR 防护 ---
def test_user_a_cannot_access_user_b_kb() -> None:
with _client() as client:
# 用户 A 创建知识库
_register_and_login(client, "user_a")
create_resp = client.post("/api/knowledge-bases", json={"name": "A 的知识库"})
kb_id = create_resp.json()["id"]
# 用户 B 登录
_register_and_login(client, "user_b")
resp = client.get(f"/api/knowledge-bases/{kb_id}")
assert resp.status_code == 404 # 对 B 来说"不存在"