Notiz-Medien & Sprachnachrichten: Fotos/Videos/Dateien + Audio an Notizen
Wiederverwendbarer UI.noteMediaAttacher für beide Notiz-Stellen (UI.noteModal
+ Notizblock-Seite). note_media-Tabelle + POST/DELETE /api/notes/{id}/media
(vor der gierigen /{parent_type}/{parent_id}-Route). Audio per MediaRecorder,
serverseitig nach m4a/AAC transkodiert (ffmpeg) — iOS spielt Chrome-Opus-webm
nicht ab. UI.lightbox global eingeführt. Mikrofon-Policy microphone=(self) +
CSP media-src 'self' blob:, Datenschutz v6. Disk-Cleanup für note_media bei
Notiz-, Account- und Admin-User-Delete. Reine Medien-Notiz ohne Text erlaubt.
noteModal-Bug gefixt: notes.get() liefert Array -> existing[0] statt
existing?.id (verhinderte Bearbeiten, erzeugte Duplikate). 12 neue Tests.
admin.py enthält außerdem KI-Vision-Statusfelder aus paralleler Arbeit
(nicht sauber trennbar ohne interaktives Staging).
This commit is contained in:
parent
203da50e1d
commit
e86d89f3d9
12 changed files with 947 additions and 59 deletions
|
|
@ -59,3 +59,30 @@ def test_delete_account_minimal_user(client):
|
|||
assert resp.status_code == 200, resp.text
|
||||
with db() as conn:
|
||||
assert conn.execute("SELECT 1 FROM users WHERE id=?", (uid,)).fetchone() is None
|
||||
|
||||
|
||||
def test_delete_account_purges_note_media(client):
|
||||
"""Account-Löschung entfernt Notiz-Medien — DB-Zeilen UND Dateien auf Disk."""
|
||||
import io, os
|
||||
from database import db
|
||||
from PIL import Image
|
||||
|
||||
uid, headers = _make_user(client)
|
||||
nid = client.post("/api/notes/diary/1", headers=headers,
|
||||
json={"text": "Mit Foto", "parent_label": "X"}).json()["id"]
|
||||
|
||||
buf = io.BytesIO(); Image.new("RGB", (10, 10), (1, 2, 3)).save(buf, format="JPEG")
|
||||
up = client.post(f"/api/notes/{nid}/media", headers=headers,
|
||||
files={"file": ("f.jpg", buf.getvalue(), "image/jpeg")})
|
||||
assert up.status_code == 200, up.text
|
||||
url = up.json()["url"]
|
||||
fpath = os.path.join(os.getenv("MEDIA_DIR", "/data/media"), url[len("/media/"):])
|
||||
assert os.path.exists(fpath)
|
||||
|
||||
resp = client.delete("/api/profile/account", headers=headers)
|
||||
assert resp.status_code == 200, resp.text
|
||||
|
||||
with db() as conn:
|
||||
assert conn.execute("SELECT COUNT(*) c FROM note_media WHERE note_id=?", (nid,)).fetchone()["c"] == 0
|
||||
assert conn.execute("SELECT COUNT(*) c FROM notes WHERE user_id=?", (uid,)).fetchone()["c"] == 0
|
||||
assert not os.path.exists(fpath), "note_media-Datei blieb als Leiche auf Disk"
|
||||
|
|
|
|||
129
tests/test_notes_media.py
Normal file
129
tests/test_notes_media.py
Normal file
|
|
@ -0,0 +1,129 @@
|
|||
"""Tests für Notiz-Medien: Bild-/Audio-Upload, GET liefert media_items,
|
||||
Löschen entfernt DB-Zeile + Datei, Notiz-Delete räumt Dateien, Validierung."""
|
||||
|
||||
import io
|
||||
import os
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
def _jpeg_bytes(color=(200, 100, 50)):
|
||||
from PIL import Image
|
||||
buf = io.BytesIO()
|
||||
Image.new("RGB", (12, 12), color).save(buf, format="JPEG")
|
||||
return buf.getvalue()
|
||||
|
||||
|
||||
# Minimaler MP4/M4A-Header (ftyp-Box) — genügt für validate_audio(audio/mp4).
|
||||
_M4A_BYTES = b"\x00\x00\x00\x18ftypM4A \x00\x00\x00\x00M4A mp42isom" + b"\x00" * 32
|
||||
|
||||
|
||||
def _create_note(client, user, text="Notiz mit Medien"):
|
||||
r = client.post("/api/notes/diary/1", headers=user["headers"],
|
||||
json={"text": text, "parent_label": "Testobjekt"})
|
||||
assert r.status_code == 201, r.text
|
||||
return r.json()
|
||||
|
||||
|
||||
def _media_path(url):
|
||||
media_dir = os.getenv("MEDIA_DIR", "/data/media")
|
||||
rel = url[len("/media/"):] if url.startswith("/media/") else url.lstrip("/")
|
||||
return os.path.join(media_dir, rel)
|
||||
|
||||
|
||||
def test_upload_image_to_note(client, user):
|
||||
note = _create_note(client, user)
|
||||
r = client.post(
|
||||
f"/api/notes/{note['id']}/media",
|
||||
headers=user["headers"],
|
||||
files={"file": ("foto.jpg", _jpeg_bytes(), "image/jpeg")},
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
m = r.json()
|
||||
assert m["media_type"] == "image"
|
||||
assert m["url"].startswith("/media/notes/")
|
||||
assert os.path.exists(_media_path(m["url"]))
|
||||
|
||||
|
||||
def test_note_get_includes_media_items(client, user):
|
||||
note = _create_note(client, user)
|
||||
client.post(f"/api/notes/{note['id']}/media", headers=user["headers"],
|
||||
files={"file": ("a.jpg", _jpeg_bytes(), "image/jpeg")})
|
||||
r = client.get("/api/notes/diary/1", headers=user["headers"])
|
||||
assert r.status_code == 200
|
||||
target = next(n for n in r.json() if n["id"] == note["id"])
|
||||
assert len(target["media_items"]) == 1
|
||||
assert target["media_items"][0]["media_type"] == "image"
|
||||
|
||||
|
||||
def test_upload_audio_to_note(client, user):
|
||||
# Reine Sprachnotiz: leerer Text ist erlaubt, das Medium trägt die Notiz.
|
||||
note = _create_note(client, user, text="")
|
||||
r = client.post(
|
||||
f"/api/notes/{note['id']}/media",
|
||||
headers=user["headers"],
|
||||
files={"file": ("sprachnachricht.m4a", _M4A_BYTES, "audio/mp4")},
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
m = r.json()
|
||||
assert m["media_type"] == "audio"
|
||||
assert m["url"].endswith(".m4a")
|
||||
assert os.path.exists(_media_path(m["url"]))
|
||||
|
||||
|
||||
def test_delete_media_removes_row_and_file(client, user):
|
||||
note = _create_note(client, user)
|
||||
up = client.post(f"/api/notes/{note['id']}/media", headers=user["headers"],
|
||||
files={"file": ("x.jpg", _jpeg_bytes(), "image/jpeg")}).json()
|
||||
path = _media_path(up["url"])
|
||||
assert os.path.exists(path)
|
||||
|
||||
r = client.delete(f"/api/notes/{note['id']}/media/{up['id']}", headers=user["headers"])
|
||||
assert r.status_code == 204
|
||||
assert not os.path.exists(path)
|
||||
|
||||
g = client.get("/api/notes/diary/1", headers=user["headers"]).json()
|
||||
target = next(n for n in g if n["id"] == note["id"])
|
||||
assert target["media_items"] == []
|
||||
|
||||
|
||||
def test_delete_note_removes_media_files(client, user):
|
||||
note = _create_note(client, user)
|
||||
up = client.post(f"/api/notes/{note['id']}/media", headers=user["headers"],
|
||||
files={"file": ("y.jpg", _jpeg_bytes(), "image/jpeg")}).json()
|
||||
path = _media_path(up["url"])
|
||||
assert os.path.exists(path)
|
||||
|
||||
r = client.delete(f"/api/notes/{note['id']}", headers=user["headers"])
|
||||
assert r.status_code == 204
|
||||
assert not os.path.exists(path)
|
||||
|
||||
|
||||
def test_upload_rejects_corrupt_image(client, user):
|
||||
note = _create_note(client, user)
|
||||
r = client.post(
|
||||
f"/api/notes/{note['id']}/media",
|
||||
headers=user["headers"],
|
||||
files={"file": ("fake.jpg", b"this is not a jpeg", "image/jpeg")},
|
||||
)
|
||||
assert r.status_code == 415
|
||||
|
||||
|
||||
def test_upload_media_requires_own_note(client, user):
|
||||
r = client.post(
|
||||
"/api/notes/999999/media",
|
||||
headers=user["headers"],
|
||||
files={"file": ("z.jpg", _jpeg_bytes(), "image/jpeg")},
|
||||
)
|
||||
assert r.status_code == 404
|
||||
|
||||
|
||||
def test_audio_utils_unit():
|
||||
"""to_m4a reicht bereits-AAC durch; validate_audio prüft Magic-Bytes."""
|
||||
from media_utils import to_m4a, validate_audio
|
||||
data, ext = to_m4a(_M4A_BYTES, ".m4a")
|
||||
assert ext == ".m4a"
|
||||
assert data == _M4A_BYTES # schon AAC → keine Transkodierung
|
||||
validate_audio(_M4A_BYTES, "audio/mp4") # kein Raise
|
||||
with pytest.raises(ValueError):
|
||||
validate_audio(b"xxxx", "audio/mp4")
|
||||
Loading…
Add table
Add a link
Reference in a new issue