193 lines
7.5 KiB
Python
193 lines
7.5 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_permanent() -> None:
|
|
with _client() as client:
|
|
kb_id, token = _setup_kb(client)
|
|
resp = client.get(f"/k/{token}")
|
|
assert resp.status_code == 200
|
|
# link 接口返回长期有效
|
|
resp = client.get(f"/api/knowledge-bases/{kb_id}/link")
|
|
assert resp.json()["expires_at"] is 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})
|
|
# 重新生成 → 新链接长期有效
|
|
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 None
|
|
|
|
|
|
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_removes_files() -> 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 = resp.json()["items"][0]
|
|
storage_path = doc["storage_path"] if "storage_path" in doc else None
|
|
doc_id = doc["id"]
|
|
|
|
# 删除文档
|
|
resp = client.delete(f"/api/documents/{doc_id}")
|
|
assert resp.status_code == 204
|
|
|
|
# 验证物理文件已删除
|
|
from app.core.config import get_settings
|
|
from pathlib import Path
|
|
# 通过数据库查询 storage_path（响应里没有）
|
|
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
|
|
if storage_path:
|
|
full = Path(get_settings().storage_root_path) / storage_path
|
|
assert not full.exists(), f"文件未被清理: {full}"
|
|
|
|
|
|
def test_kb_delete_removes_doc_files() -> 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 app.core.db import get_session_factory
|
|
from app.models.document import Document
|
|
from app.core.config import get_settings
|
|
from pathlib import Path
|
|
|
|
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 not (root / p).exists(), f"文件未被清理: {p}" |