justin
34d30e3134
Media model + migration; an ObjectStore interface with an S3/MinIO (boto3) implementation behind the service layer. Upload (multipart) stores bytes in object storage + a metadata row (checksum, size, content-type, optional attach to person/event/source); list returns presigned URLs; delete is soft. Editor-gated, privacy-filtered, audited. 24 tests pass (object store faked). Introduces the worker container (same image, 'python -m app.worker'): its first job is the scheduled 30-day soft-delete purge across tables + media object cleanup. Compose gains worker + S3 env on backend/worker; dev override builds the worker too. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Signed-off-by: Justin Paul <justin@jpaul.me>
46 lines
1.6 KiB
Python
46 lines
1.6 KiB
Python
"""Media upload/list/delete through the API (object store faked in conftest)."""
|
|
|
|
from tests.conftest import auth, register
|
|
|
|
|
|
async def _tree(client, email):
|
|
h = auth(await register(client, email))
|
|
tree_id = (await client.post("/api/v1/trees", json={"name": "M"}, headers=h)).json()["id"]
|
|
return h, tree_id
|
|
|
|
|
|
async def test_media_upload_list_delete(client):
|
|
h, tree_id = await _tree(client, "media1@example.com")
|
|
|
|
resp = await client.post(
|
|
f"/api/v1/trees/{tree_id}/media",
|
|
files={"file": ("scan.txt", b"hello world", "text/plain")},
|
|
data={"title": "A scan"},
|
|
headers=h,
|
|
)
|
|
assert resp.status_code == 201, resp.text
|
|
body = resp.json()
|
|
assert body["original_filename"] == "scan.txt"
|
|
assert body["byte_size"] == 11
|
|
assert body["url"].startswith("https://objects.test/")
|
|
media_id = body["id"]
|
|
|
|
listed = await client.get(f"/api/v1/trees/{tree_id}/media", headers=h)
|
|
assert listed.status_code == 200
|
|
assert len(listed.json()) == 1
|
|
|
|
resp = await client.delete(f"/api/v1/trees/{tree_id}/media/{media_id}", headers=h)
|
|
assert resp.status_code == 204
|
|
assert len((await client.get(f"/api/v1/trees/{tree_id}/media", headers=h)).json()) == 0
|
|
|
|
|
|
async def test_non_member_cannot_upload(client):
|
|
h, tree_id = await _tree(client, "media2@example.com")
|
|
other = auth(await register(client, "media-intruder@example.com"))
|
|
resp = await client.post(
|
|
f"/api/v1/trees/{tree_id}/media",
|
|
files={"file": ("x.txt", b"x", "text/plain")},
|
|
headers=other,
|
|
)
|
|
assert resp.status_code == 403
|