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"]