86 lines
3.1 KiB
Python
86 lines
3.1 KiB
Python
"""Unit tests for batch upload and raw file saving (#89)."""
|
|
|
|
import base64
|
|
from pathlib import Path
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from backend.indexer import get_vault_data
|
|
from backend.main import app
|
|
from backend.services.errors import ServiceError
|
|
from backend.services.mutations import (
|
|
batch_upload_files,
|
|
save_raw_file,
|
|
)
|
|
|
|
|
|
def test_save_raw_file_text_and_binary(client):
|
|
vault_root = Path(get_vault_data("TestVault")["path"])
|
|
|
|
# Save a markdown file
|
|
res = save_raw_file("TestVault", "folder/note_test.md", b"# Hello World\n")
|
|
assert res["success"] is True
|
|
assert res["path"] == "folder/note_test.md"
|
|
assert (vault_root / "folder" / "note_test.md").read_text(encoding="utf-8") == "# Hello World\n"
|
|
|
|
# Save a png image
|
|
img_bytes = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR"
|
|
res_img = save_raw_file("TestVault", "assets/logo.png", img_bytes)
|
|
assert res_img["success"] is True
|
|
assert res_img["path"] == "assets/logo.png"
|
|
assert (vault_root / "assets" / "logo.png").read_bytes() == img_bytes
|
|
|
|
|
|
def test_save_raw_file_unsupported_ext(client):
|
|
with pytest.raises(ServiceError) as exc_info:
|
|
save_raw_file("TestVault", "danger.exe", b"binary")
|
|
assert exc_info.value.code == "unsupported_extension"
|
|
|
|
|
|
def test_batch_upload_files_recursive_structure(client):
|
|
vault_root = Path(get_vault_data("TestVault")["path"])
|
|
|
|
files = [
|
|
{"path": "empty_dir", "is_dir": True},
|
|
{"path": "doc1.md", "content": b"# Doc 1"},
|
|
{"path": "sub/doc2.md", "content": b"# Doc 2"},
|
|
{"path": "sub/deep/doc3.md", "content": b"# Doc 3"},
|
|
{"path": "assets/img.png", "content": b"\x89PNG\r\n\x1a\n"},
|
|
{"path": "invalid.exe", "content": b"bad"},
|
|
]
|
|
|
|
res = batch_upload_files("TestVault", "target_folder", files, overwrite=True)
|
|
assert res["vault"] == "TestVault"
|
|
assert res["target_dir"] == "target_folder"
|
|
assert len(res["uploaded"]) == 4
|
|
assert len(res["created_dirs"]) == 1
|
|
assert len(res["errors"]) == 1
|
|
assert res["errors"][0]["path"] == "target_folder/invalid.exe"
|
|
|
|
# Verify disk files
|
|
assert (vault_root / "target_folder" / "empty_dir").is_dir()
|
|
assert (vault_root / "target_folder" / "doc1.md").read_text(encoding="utf-8") == "# Doc 1"
|
|
assert (vault_root / "target_folder" / "sub" / "deep" / "doc3.md").read_text(encoding="utf-8") == "# Doc 3"
|
|
|
|
|
|
def test_api_batch_upload(client):
|
|
vault_root = Path(get_vault_data("TestVault")["path"])
|
|
|
|
content_b64 = base64.b64encode(b"# Uploaded via API").decode("utf-8")
|
|
payload = {
|
|
"target_dir": "api_import",
|
|
"files": [
|
|
{"path": "api_doc.md", "content": content_b64},
|
|
{"path": "nested/folder", "is_dir": True},
|
|
],
|
|
"overwrite": True,
|
|
}
|
|
|
|
resp = client.post("/api/vault/TestVault/batch-upload", json=payload)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["success"] is True
|
|
assert "api_import/api_doc.md" in data["uploaded"]
|
|
assert "api_import/nested/folder" in data["created_dirs"]
|
|
assert (vault_root / "api_import" / "api_doc.md").read_text(encoding="utf-8") == "# Uploaded via API"
|