282 lines
10 KiB
Python
282 lines
10 KiB
Python
import hashlib
|
|
from urllib.parse import urlsplit
|
|
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import select
|
|
|
|
from memrelay.models import CurationChange
|
|
from tests.conftest import csrf_headers
|
|
|
|
|
|
def local_url(url: str) -> str:
|
|
parsed = urlsplit(url)
|
|
return f"{parsed.path}?{parsed.query}"
|
|
|
|
|
|
def upload_file(client: TestClient, path: str, content: bytes, **metadata) -> dict:
|
|
prepared = client.post(
|
|
"/api/v1/files/upload-prepare",
|
|
json={
|
|
"path": path,
|
|
"size": len(content),
|
|
"sha256": hashlib.sha256(content).hexdigest(),
|
|
**metadata,
|
|
},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert prepared.status_code == 201, prepared.text
|
|
upload_url = local_url(prepared.json()["upload_url"])
|
|
uploaded = client.put(upload_url, content=content)
|
|
assert uploaded.status_code == 200, uploaded.text
|
|
return {**uploaded.json(), "upload_url": upload_url, "task_id": prepared.json()["id"]}
|
|
|
|
|
|
def test_upload_catalog_search_range_and_one_time_url(initialized_client: TestClient) -> None:
|
|
client = initialized_client
|
|
content = b"0123456789"
|
|
uploaded = upload_file(
|
|
client,
|
|
"packages/tool.bin",
|
|
content,
|
|
description="Reusable build tool",
|
|
version="1.2.3",
|
|
purpose="build",
|
|
tags=["cache", "tool"],
|
|
)
|
|
assert uploaded["checksum_sha256"] == hashlib.sha256(content).hexdigest()
|
|
assert uploaded["version"] == "1.2.3"
|
|
assert client.put(uploaded["upload_url"], content=content).status_code == 409
|
|
root_items = client.get("/api/v1/files", params={"parent": ""}).json()
|
|
assert [(item["path"], item["is_directory"]) for item in root_items] == [("packages", True)]
|
|
|
|
task = client.get(f"/api/v1/files/uploads/{uploaded['task_id']}").json()
|
|
assert task["status"] == "completed"
|
|
assert task["file_id"] == uploaded["id"]
|
|
assert (
|
|
client.get("/api/v1/files", params={"query": "Reusable"}).json()[0]["id"] == uploaded["id"]
|
|
)
|
|
assert client.get("/api/v1/files", params={"query": "1.2.3"}).json()[0]["id"] == uploaded["id"]
|
|
|
|
download = client.get(f"/api/v1/files/{uploaded['id']}/download").json()["url"]
|
|
full = client.get(local_url(download))
|
|
assert full.status_code == 200
|
|
assert full.content == content
|
|
partial = client.get(local_url(download), headers={"Range": "bytes=2-5"})
|
|
assert partial.status_code == 206
|
|
assert partial.content == b"2345"
|
|
assert partial.headers["content-range"] == "bytes 2-5/10"
|
|
|
|
|
|
def test_path_safety_size_hash_and_overwrite(initialized_client: TestClient) -> None:
|
|
client = initialized_client
|
|
traversal = client.post(
|
|
"/api/v1/files/upload-prepare",
|
|
json={"path": "../outside.txt", "size": 1},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert traversal.status_code == 422
|
|
|
|
prepared = client.post(
|
|
"/api/v1/files/upload-prepare",
|
|
json={"path": "bad-hash.txt", "size": 3, "sha256": "0" * 64},
|
|
headers=csrf_headers(client),
|
|
).json()
|
|
response = client.put(local_url(prepared["upload_url"]), content=b"abc")
|
|
assert response.status_code == 422
|
|
assert response.json()["error"]["code"] == "UPLOAD_HASH_MISMATCH"
|
|
assert client.get(f"/api/v1/files/uploads/{prepared['id']}").json()["status"] == "failed"
|
|
|
|
upload_file(client, "existing.txt", b"first")
|
|
duplicate = client.post(
|
|
"/api/v1/files/upload-prepare",
|
|
json={"path": "existing.txt", "size": 6},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert duplicate.status_code == 409
|
|
replaced = upload_file(client, "existing.txt", b"second", overwrite=True)
|
|
assert replaced["size"] == 6
|
|
|
|
|
|
def test_external_scan_metadata_move_and_confirmed_delete(initialized_client: TestClient) -> None:
|
|
client = initialized_client
|
|
files_dir = client.app.state.settings.files_dir
|
|
external = files_dir / "external" / "manual.pdf"
|
|
external.parent.mkdir(parents=True)
|
|
external.write_bytes(b"pdf-data")
|
|
|
|
scan = client.post("/api/v1/files/scan", headers=csrf_headers(client))
|
|
assert scan.status_code == 200
|
|
assert scan.json()["added"] == 2
|
|
record = client.get("/api/v1/files", params={"query": "manual.pdf"}).json()[0]
|
|
with client.app.state.session_factory() as db:
|
|
scanned_change = db.scalar(
|
|
select(CurationChange).where(
|
|
CurationChange.source_type == "file",
|
|
CurationChange.source_id == record["id"],
|
|
CurationChange.change_type == "create",
|
|
)
|
|
)
|
|
assert scanned_change is not None
|
|
assert scanned_change.origin == "user"
|
|
assert scanned_change.usage_profile_id is not None
|
|
|
|
updated = client.patch(
|
|
f"/api/v1/files/{record['id']}",
|
|
json={"description": "Manual", "version": "2026", "tags": ["docs"]},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert updated.json()["description"] == "Manual"
|
|
moved = client.post(
|
|
f"/api/v1/files/{record['id']}/move",
|
|
json={"path": "documents/manual.pdf"},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert moved.json()["path"] == "documents/manual.pdf"
|
|
assert not external.exists()
|
|
root_items = client.get("/api/v1/files", params={"parent": ""}).json()
|
|
documents = next(item for item in root_items if item["path"] == "documents")
|
|
assert documents["is_directory"] is True
|
|
|
|
assert (
|
|
client.delete(f"/api/v1/files/{record['id']}", headers=csrf_headers(client)).status_code
|
|
== 422
|
|
)
|
|
assert (
|
|
client.delete(
|
|
f"/api/v1/files/{record['id']}",
|
|
params={"confirmed": "true"},
|
|
headers=csrf_headers(client),
|
|
).status_code
|
|
== 204
|
|
)
|
|
assert client.get(f"/api/v1/files/{record['id']}").status_code == 404
|
|
|
|
|
|
def test_metadata_patch_preserves_omitted_fields_and_allows_explicit_clear(
|
|
initialized_client: TestClient,
|
|
) -> None:
|
|
client = initialized_client
|
|
uploaded = upload_file(
|
|
client,
|
|
"documents/metadata.txt",
|
|
b"metadata",
|
|
description="Original description",
|
|
version="1.0",
|
|
purpose="reference",
|
|
tags=["docs", "stable"],
|
|
)
|
|
|
|
partial = client.patch(
|
|
f"/api/v1/files/{uploaded['id']}",
|
|
json={"description": "Updated description"},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert partial.status_code == 200, partial.text
|
|
assert partial.json()["description"] == "Updated description"
|
|
assert partial.json()["version"] == "1.0"
|
|
assert partial.json()["purpose"] == "reference"
|
|
assert partial.json()["tags"] == ["docs", "stable"]
|
|
|
|
cleared = client.patch(
|
|
f"/api/v1/files/{uploaded['id']}",
|
|
json={"version": None, "tags": []},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert cleared.status_code == 200, cleared.text
|
|
assert cleared.json()["description"] == "Updated description"
|
|
assert cleared.json()["version"] is None
|
|
assert cleared.json()["purpose"] == "reference"
|
|
assert cleared.json()["tags"] == []
|
|
|
|
|
|
def test_scan_marks_missing_file(initialized_client: TestClient) -> None:
|
|
client = initialized_client
|
|
uploaded = upload_file(client, "temporary.txt", b"temporary")
|
|
(client.app.state.settings.files_dir / "temporary.txt").unlink()
|
|
scan = client.post("/api/v1/files/scan", headers=csrf_headers(client)).json()
|
|
assert scan["missing"] == 1
|
|
assert client.get(f"/api/v1/files/{uploaded['id']}").json()["status"] == "missing"
|
|
|
|
|
|
def test_directory_create_browse_move_scan_and_delete(initialized_client: TestClient) -> None:
|
|
client = initialized_client
|
|
created = client.post(
|
|
"/api/v1/files/directories",
|
|
json={"path": "documents"},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert created.status_code == 201
|
|
directory = created.json()
|
|
assert directory["is_directory"] is True
|
|
assert directory["mime_type"] == "inode/directory"
|
|
|
|
nested = client.post(
|
|
"/api/v1/files/directories",
|
|
json={"path": "documents/guides"},
|
|
headers=csrf_headers(client),
|
|
).json()
|
|
root_items = client.get("/api/v1/files", params={"parent": ""}).json()
|
|
assert [item["path"] for item in root_items] == ["documents"]
|
|
child_items = client.get("/api/v1/files", params={"parent": "documents"}).json()
|
|
assert [item["path"] for item in child_items] == ["documents/guides"]
|
|
|
|
non_empty = client.delete(
|
|
f"/api/v1/files/{directory['id']}",
|
|
params={"confirmed": "true"},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert non_empty.status_code == 409
|
|
assert non_empty.json()["error"]["code"] == "DIRECTORY_NOT_EMPTY"
|
|
|
|
moved = client.post(
|
|
f"/api/v1/files/{nested['id']}/move",
|
|
json={"path": "documents/manuals"},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert moved.status_code == 200
|
|
assert moved.json()["path"] == "documents/manuals"
|
|
assert (
|
|
client.delete(
|
|
f"/api/v1/files/{nested['id']}",
|
|
params={"confirmed": "true"},
|
|
headers=csrf_headers(client),
|
|
).status_code
|
|
== 204
|
|
)
|
|
assert (
|
|
client.delete(
|
|
f"/api/v1/files/{directory['id']}",
|
|
params={"confirmed": "true"},
|
|
headers=csrf_headers(client),
|
|
).status_code
|
|
== 204
|
|
)
|
|
|
|
external_empty = client.app.state.settings.files_dir / "external-empty"
|
|
external_empty.mkdir()
|
|
scanned = client.post("/api/v1/files/scan", headers=csrf_headers(client)).json()
|
|
assert scanned["added"] == 1
|
|
result = client.get("/api/v1/files", params={"query": "external-empty"}).json()[0]
|
|
assert result["is_directory"] is True
|
|
|
|
|
|
def test_nested_directory_creation_indexes_each_parent(initialized_client: TestClient) -> None:
|
|
client = initialized_client
|
|
created = client.post(
|
|
"/api/v1/files/directories",
|
|
json={"path": "packages/tools/cache"},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert created.status_code == 201
|
|
assert created.json()["path"] == "packages/tools/cache"
|
|
assert [item["path"] for item in client.get("/api/v1/files", params={"parent": ""}).json()] == [
|
|
"packages"
|
|
]
|
|
assert [
|
|
item["path"] for item in client.get("/api/v1/files", params={"parent": "packages"}).json()
|
|
] == ["packages/tools"]
|
|
assert [
|
|
item["path"]
|
|
for item in client.get("/api/v1/files", params={"parent": "packages/tools"}).json()
|
|
] == ["packages/tools/cache"]
|