DocHub/backend/tests/test_routers_documents.py

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,
    InvalidPathError,
    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)
        if ".." in key.split("/"):
            raise InvalidPathError(key)
        return {"key": key, "content": "# 架构设计\n\n内容", "contentType": "text/markdown"}

    def save_document(self, key, content, content_type):
        if ".." in key.split("/"):
            raise InvalidPathError(key)
        self.saved.append((key, content, content_type))
        return {"key": key, "updatedAt": "2026-08-01T10:05:00Z"}

    def delete_document(self, key):
        if ".." in key.split("/"):
            raise InvalidPathError(key)
        self.deleted.append(key)

    def upload_document(self, key, data, content_type):
        if ".." in key.split("/"):
            raise InvalidPathError(key)
        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"]


def test_get_document_returns_400_for_path_traversal_key(client):
    response = client.get("/api/documents/..%2F..%2Fetc%2Fpasswd")
    assert response.status_code == 400


def test_put_document_returns_400_for_path_traversal_key(client):
    response = client.put(
        "/api/documents/..%2F..%2Fetc%2Fpasswd",
        json={"content": "x", "contentType": "text/plain"},
    )
    assert response.status_code == 400


def test_delete_document_returns_400_for_path_traversal_key(client):
    response = client.delete("/api/documents/..%2F..%2Fetc%2Fpasswd")
    assert response.status_code == 400


def test_upload_document_returns_400_for_path_traversal_in_path(client):
    response = client.post(
        "/api/upload",
        data={"path": "../../etc"},
        files={"file": ("passwd.md", b"data", "text/markdown")},
    )
    assert response.status_code == 400