Tianyang
13789cc0b4
- main.py: remove duplicate /health endpoint definition - text_extract.py: HTML-escape extracted plain text before indexing so raw markup in uploaded documents can't render as live HTML via search snippets (stored XSS fix) - document_service.py: base64-encode content when UTF-8 decoding fails instead of raising UnicodeDecodeError (500) on binary files; add upload validation for file extension (allowlist) and size (10MB max) - routers/documents.py: map new validation errors to HTTP 400 with a clear detail message instead of letting them 500 - add/extend tests covering escaping, binary get_document, and upload validation rejection + acceptance paths
122 lines
4.3 KiB
Python
122 lines
4.3 KiB
Python
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.main import app
|
|
from app.dependencies import get_document_service
|
|
from app.document_service import DocumentNotFoundError, FileTooLargeError, UnsupportedFileTypeError
|
|
|
|
|
|
class FakeDocumentService:
|
|
def __init__(self):
|
|
self.saved = []
|
|
self.deleted = []
|
|
self.uploaded = []
|
|
|
|
def list_documents(self, path):
|
|
return {
|
|
"folders": ["产品文档", "团队规范"],
|
|
"files": [
|
|
{"key": "架构设计.md", "name": "架构设计.md", "size": 512, "modifiedAt": "2026-08-01T10:00:00Z"},
|
|
],
|
|
}
|
|
|
|
def get_document(self, key):
|
|
if key == "不存在/文档.md":
|
|
raise DocumentNotFoundError(key)
|
|
return {"key": key, "content": "# 架构设计\n\n内容", "contentType": "text/markdown"}
|
|
|
|
def save_document(self, key, content, content_type):
|
|
self.saved.append((key, content, content_type))
|
|
return {"key": key, "updatedAt": "2026-08-01T10:05:00Z"}
|
|
|
|
def delete_document(self, key):
|
|
self.deleted.append(key)
|
|
|
|
def upload_document(self, key, data, content_type):
|
|
if key.endswith(".exe"):
|
|
raise UnsupportedFileTypeError(".exe")
|
|
if len(data) > 1024:
|
|
raise FileTooLargeError(1024)
|
|
self.uploaded.append((key, data, content_type))
|
|
return {"key": key, "name": key.rsplit("/", 1)[-1], "size": len(data)}
|
|
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
fake_service = FakeDocumentService()
|
|
app.dependency_overrides[get_document_service] = lambda: fake_service
|
|
with TestClient(app) as test_client:
|
|
test_client.fake_service = fake_service
|
|
yield test_client
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
def test_list_documents_returns_folders_and_files(client):
|
|
response = client.get("/api/documents", params={"path": "产品文档"})
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["folders"] == ["产品文档", "团队规范"]
|
|
assert body["files"][0]["key"] == "架构设计.md"
|
|
|
|
|
|
def test_get_document_returns_content(client):
|
|
response = client.get("/api/documents/产品文档/架构设计.md")
|
|
assert response.status_code == 200
|
|
assert response.json()["contentType"] == "text/markdown"
|
|
|
|
|
|
def test_get_document_returns_404_for_missing_key(client):
|
|
response = client.get("/api/documents/不存在/文档.md")
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_put_document_saves_content(client):
|
|
response = client.put(
|
|
"/api/documents/产品文档/架构设计.md",
|
|
json={"content": "# 更新后的内容", "contentType": "text/markdown"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["key"] == "产品文档/架构设计.md"
|
|
assert client.fake_service.saved[0][0] == "产品文档/架构设计.md"
|
|
assert client.fake_service.saved[0][1] == "# 更新后的内容".encode("utf-8")
|
|
|
|
|
|
def test_delete_document_returns_204(client):
|
|
response = client.delete("/api/documents/产品文档/架构设计.md")
|
|
assert response.status_code == 204
|
|
assert response.content == b""
|
|
assert client.fake_service.deleted == ["产品文档/架构设计.md"]
|
|
|
|
|
|
def test_upload_document_returns_201_with_metadata(client):
|
|
response = client.post(
|
|
"/api/upload",
|
|
data={"path": "产品文档"},
|
|
files={"file": ("新文档.md", b"# new content", "text/markdown")},
|
|
)
|
|
assert response.status_code == 201
|
|
body = response.json()
|
|
assert body["key"] == "产品文档/新文档.md"
|
|
assert body["name"] == "新文档.md"
|
|
assert body["size"] == len(b"# new content")
|
|
|
|
|
|
def test_upload_document_returns_400_for_unsupported_file_type(client):
|
|
response = client.post(
|
|
"/api/upload",
|
|
data={"path": "产品文档"},
|
|
files={"file": ("恶意程序.exe", b"binary", "application/octet-stream")},
|
|
)
|
|
assert response.status_code == 400
|
|
assert "unsupported file type" in response.json()["detail"]
|
|
|
|
|
|
def test_upload_document_returns_400_for_oversized_file(client):
|
|
response = client.post(
|
|
"/api/upload",
|
|
data={"path": "产品文档"},
|
|
files={"file": ("大文件.md", b"x" * 2000, "text/markdown")},
|
|
)
|
|
assert response.status_code == 400
|
|
assert "file too large" in response.json()["detail"]
|