251 lines
10 KiB
Python
251 lines
10 KiB
Python
"""链接有效期功能测试。"""
|
|
|
|
import io
|
|
import zipfile
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.main import app
|
|
|
|
|
|
def _client() -> TestClient:
|
|
return TestClient(app, raise_server_exceptions=False)
|
|
|
|
|
|
def _setup_kb(client: TestClient) -> tuple[str, str]:
|
|
"""注册+创建知识库，返回 (kb_id, token)。"""
|
|
client.post("/api/auth/register", json={
|
|
"username": "expiry_user", "email": "expiry@example.com", "password": "password123",
|
|
})
|
|
resp = client.post("/api/knowledge-bases", json={"name": "Expiry Test KB"})
|
|
kb_id = resp.json()["id"]
|
|
resp = client.get(f"/api/knowledge-bases/{kb_id}/link")
|
|
return kb_id, resp.json()["token"]
|
|
|
|
|
|
def test_default_30min() -> None:
|
|
"""新建知识库链接默认 30 分钟有效期。"""
|
|
with _client() as client:
|
|
kb_id, token = _setup_kb(client)
|
|
resp = client.get(f"/k/{token}")
|
|
assert resp.status_code == 200
|
|
resp = client.get(f"/api/knowledge-bases/{kb_id}/link")
|
|
assert resp.json()["expires_at"] is not None
|
|
assert resp.json()["is_expired"] is False
|
|
|
|
|
|
def test_set_expiry_future() -> None:
|
|
with _client() as client:
|
|
kb_id, token = _setup_kb(client)
|
|
resp = client.post(f"/api/knowledge-bases/{kb_id}/set-expiry",
|
|
json={"expires_in_minutes": 10})
|
|
assert resp.status_code == 200
|
|
# 未过期仍可访问
|
|
assert client.get(f"/k/{token}").status_code == 200
|
|
# link 接口返回过期时间
|
|
resp = client.get(f"/api/knowledge-bases/{kb_id}/link")
|
|
assert resp.json()["expires_at"] is not None
|
|
assert resp.json()["is_expired"] is False
|
|
|
|
|
|
def test_expired_link_html() -> None:
|
|
with _client() as client:
|
|
kb_id, token = _setup_kb(client)
|
|
# 设为过去时间（负数分钟）→ 立即过期
|
|
client.post(f"/api/knowledge-bases/{kb_id}/set-expiry",
|
|
json={"expires_in_minutes": -1})
|
|
resp = client.get(f"/k/{token}")
|
|
assert resp.status_code == 410
|
|
assert "已失效" in resp.text
|
|
assert "noindex" in resp.text
|
|
|
|
|
|
def test_expired_link_json_md_txt() -> None:
|
|
with _client() as client:
|
|
kb_id, token = _setup_kb(client)
|
|
client.post(f"/api/knowledge-bases/{kb_id}/set-expiry",
|
|
json={"expires_in_minutes": -1})
|
|
|
|
resp = client.get(f"/k/{token}.json")
|
|
assert resp.status_code == 410
|
|
assert resp.json()["code"] == "LINK_EXPIRED"
|
|
|
|
resp = client.get(f"/k/{token}.md")
|
|
assert resp.status_code == 410
|
|
assert "已失效" in resp.text
|
|
|
|
resp = client.get(f"/k/{token}.txt")
|
|
assert resp.status_code == 410
|
|
|
|
|
|
def test_expired_doc_page() -> None:
|
|
"""过期后文档页也不可访问。"""
|
|
with _client() as client:
|
|
kb_id, token = _setup_kb(client)
|
|
client.post(f"/api/knowledge-bases/{kb_id}/set-expiry",
|
|
json={"expires_in_minutes": -1})
|
|
resp = client.get(f"/k/{token}/doc/whatever-token")
|
|
assert resp.status_code == 410
|
|
|
|
|
|
def test_restore_permanent() -> None:
|
|
"""过期后重新设为长期有效可恢复访问。"""
|
|
with _client() as client:
|
|
kb_id, token = _setup_kb(client)
|
|
client.post(f"/api/knowledge-bases/{kb_id}/set-expiry",
|
|
json={"expires_in_minutes": -1})
|
|
assert client.get(f"/k/{token}").status_code == 410
|
|
# 恢复
|
|
client.post(f"/api/knowledge-bases/{kb_id}/set-expiry",
|
|
json={"expires_in_minutes": None})
|
|
assert client.get(f"/k/{token}").status_code == 200
|
|
|
|
|
|
def test_regenerate_resets_expiry() -> None:
|
|
with _client() as client:
|
|
kb_id, old_token = _setup_kb(client)
|
|
client.post(f"/api/knowledge-bases/{kb_id}/set-expiry",
|
|
json={"expires_in_minutes": -1})
|
|
# 重新生成 → 新链接默认 30 分钟有效期
|
|
resp = client.post(f"/api/knowledge-bases/{kb_id}/regenerate-token")
|
|
new_token = resp.json()["token"]
|
|
assert new_token != old_token
|
|
assert client.get(f"/k/{new_token}").status_code == 200
|
|
resp = client.get(f"/api/knowledge-bases/{kb_id}/link")
|
|
assert resp.json()["expires_at"] is not None
|
|
assert resp.json()["is_expired"] is False
|
|
|
|
|
|
def _make_docx_bytes() -> bytes:
|
|
buf = io.BytesIO()
|
|
with zipfile.ZipFile(buf, "w") as zf:
|
|
zf.writestr("[Content_Types].xml", '<?xml version="1.0"?><Types></Types>')
|
|
zf.writestr("_rels/.rels", '<?xml version="1.0"?><Relationships></Relationships>')
|
|
zf.writestr("word/document.xml", '<?xml version="1.0"?><w:document xmlns:w="http://schemas.openxmlformats.org/wordprocessingml/2006/main"><w:body><w:p><w:r><w:t>Test</w:t></w:r></w:p></w:body></w:document>')
|
|
return buf.getvalue()
|
|
|
|
|
|
def test_doc_delete_goes_to_recycle_bin() -> None:
|
|
"""删除文档 → 进回收站，文件保留；彻底删除后文件清理。"""
|
|
with _client() as client:
|
|
client.post("/api/auth/register", json={
|
|
"username": "cleanup_user", "email": "cleanup@example.com", "password": "password123",
|
|
})
|
|
resp = client.post("/api/knowledge-bases", json={"name": "Cleanup KB"})
|
|
kb_id = resp.json()["id"]
|
|
|
|
files = {"file": ("t.docx", _make_docx_bytes(),
|
|
"application/vnd.openxmlformats-officedocument.wordprocessingml.document")}
|
|
client.post("/api/documents/upload", data={"kb_id": kb_id}, files=files)
|
|
|
|
resp = client.get(f"/api/documents?kb_id={kb_id}")
|
|
doc_id = resp.json()["items"][0]["id"]
|
|
|
|
from pathlib import Path
|
|
|
|
from app.core.config import get_settings
|
|
from app.core.db import get_session_factory
|
|
from app.models.document import Document
|
|
|
|
factory = get_session_factory()
|
|
with factory() as session:
|
|
d = session.get(Document, doc_id)
|
|
storage_path = d.storage_path
|
|
|
|
# 删除 → 进回收站，文件仍存在
|
|
resp = client.delete(f"/api/documents/{doc_id}")
|
|
assert resp.status_code == 204
|
|
full = Path(get_settings().storage_root_path) / storage_path
|
|
assert full.exists(), "回收站阶段文件不应被删除"
|
|
|
|
# 回收站列表可见
|
|
resp = client.get("/api/recycle-bin")
|
|
assert any(item["id"] == doc_id for item in resp.json()["documents"])
|
|
|
|
# 彻底删除 → 文件清理
|
|
resp = client.delete(f"/api/recycle-bin/documents/{doc_id}")
|
|
assert resp.status_code == 204
|
|
assert not full.exists(), "彻底删除后文件应被清理"
|
|
|
|
|
|
def test_kb_delete_goes_to_recycle_bin() -> None:
|
|
"""删除知识库 → 进回收站；彻底删除后文件清理。"""
|
|
with _client() as client:
|
|
client.post("/api/auth/register", json={
|
|
"username": "kb_cleanup_user", "email": "kbcleanup@example.com", "password": "password123",
|
|
})
|
|
resp = client.post("/api/knowledge-bases", json={"name": "KB Cleanup KB"})
|
|
kb_id = resp.json()["id"]
|
|
|
|
files = {"file": ("t.docx", _make_docx_bytes(),
|
|
"application/vnd.openxmlformats-officedocument.wordprocessingml.document")}
|
|
client.post("/api/documents/upload", data={"kb_id": kb_id}, files=files)
|
|
|
|
from pathlib import Path
|
|
|
|
from app.core.config import get_settings
|
|
from app.core.db import get_session_factory
|
|
from app.models.document import Document
|
|
|
|
factory = get_session_factory()
|
|
with factory() as session:
|
|
docs = list(session.query(Document).filter_by(knowledge_base_id=kb_id).all())
|
|
paths = [d.storage_path for d in docs if d.storage_path]
|
|
|
|
# 删除知识库 → 回收站，文件保留
|
|
resp = client.delete(f"/api/knowledge-bases/{kb_id}")
|
|
assert resp.status_code == 204
|
|
root = Path(get_settings().storage_root_path)
|
|
for p in paths:
|
|
assert (root / p).exists(), "回收站阶段文件不应被删除"
|
|
|
|
# 回收站可见
|
|
resp = client.get("/api/recycle-bin")
|
|
assert any(item["id"] == kb_id for item in resp.json()["knowledge_bases"])
|
|
|
|
# 彻底删除 → 文件清理
|
|
resp = client.delete(f"/api/recycle-bin/knowledge-bases/{kb_id}")
|
|
assert resp.status_code == 204
|
|
for p in paths:
|
|
assert not (root / p).exists(), "彻底删除后文件应被清理"
|
|
|
|
|
|
def test_category_delete_and_restore() -> None:
|
|
"""删除目录 → 目录+子目录+文档进回收站；恢复后全部回来。"""
|
|
with _client() as client:
|
|
client.post("/api/auth/register", json={
|
|
"username": "cat_rb_user", "email": "catrb@example.com", "password": "password123",
|
|
})
|
|
resp = client.post("/api/knowledge-bases", json={"name": "Cat RB KB"})
|
|
kb_id = resp.json()["id"]
|
|
|
|
# 在"01 公司层"下创建文本文档
|
|
cats = client.get(f"/api/knowledge-bases/{kb_id}/categories").json()
|
|
target = next(c for c in cats if c["name"] == "公司基本信息")
|
|
|
|
client.post("/api/documents/create-text", json={
|
|
"kb_id": kb_id, "title": "Cat Doc", "content": "hello",
|
|
"category_id": target["id"],
|
|
})
|
|
|
|
# 删除目录
|
|
resp = client.delete(f"/api/knowledge-bases/{kb_id}/categories/{target['id']}")
|
|
assert resp.status_code == 204
|
|
|
|
# 文档不可见
|
|
docs = client.get(f"/api/documents?kb_id={kb_id}").json()
|
|
assert docs["total"] == 0
|
|
|
|
# 回收站里有目录和文档
|
|
rb = client.get("/api/recycle-bin").json()
|
|
assert any(c["name"] == "公司基本信息" for c in rb["categories"])
|
|
assert any(d["name"] == "Cat Doc" for d in rb["documents"])
|
|
|
|
# 恢复目录 → 同批文档恢复
|
|
cat_id = next(c["id"] for c in rb["categories"] if c["name"] == "公司基本信息")
|
|
resp = client.post(f"/api/recycle-bin/categories/{cat_id}/restore")
|
|
assert resp.status_code == 204
|
|
|
|
docs = client.get(f"/api/documents?kb_id={kb_id}").json()
|
|
assert docs["total"] == 1
|
|
assert docs["items"][0]["title"] == "Cat Doc" |