S.H.O.N.A.R._Desktop_Companion/backend/tests/test_recordings.py

"""Upload sessions + recordings CRUD tests (M2).

Uses real WAV magic bytes; validation is byte-level, so fakes would be
testing the wrong thing.
"""

from __future__ import annotations

import struct
import uuid


def wav_bytes(payload_len: int = 64) -> bytes:
    data = bytes(range(payload_len % 256)) * (payload_len // 256 + 1)
    data = data[:payload_len]
    header = (
        b"RIFF" + struct.pack("<I", 36 + len(data)) + b"WAVE"
        + b"fmt " + struct.pack("<IHHIIHH", 16, 1, 1, 8000, 8000, 1, 8)
        + b"data" + struct.pack("<I", len(data))
    )
    return header + data


def mp4_bytes() -> bytes:
    return b"\x00\x00\x00 ftypM4A " + b"\x00" * 64


AUTH = {"email": "m2@example.com",
        "password": "m2-test-" + "passw0rd-123"}


async def user_tokens(client, email=AUTH["email"], password=AUTH["password"]):
    r = await client.post("/api/v1/auth/register", json={"email": email, "password": password})
    assert r.status_code == 201, r.text
    return r.json()["access_token"]


async def auth(token: str) -> dict:
    return {"Authorization": f"Bearer {token}"}


async def upload_full(client, token: str, data: bytes, mime="audio/wav",
                      client_id=None, title=None):
    h = await auth(token)
    r = await client.post(
        "/api/v1/uploads",
        json={"declared_mime_type": mime, "declared_size_bytes": len(data),
              "client_recording_id": client_id, "title": title},
        headers=h,
    )
    assert r.status_code == 201, r.text
    sid = r.json()["id"]
    r = await client.put(f"/api/v1/uploads/{sid}/chunks/0", content=data,
                         headers={**h, "content-type": "application/octet-stream"})
    assert r.status_code == 201, r.text
    r = await client.post(
        f"/api/v1/uploads/{sid}/finalize",
        json={"duration_seconds": 12.5},
        headers=h,
    )
    return sid, r


# --- upload session ----------------------------------------------------------


async def test_upload_happy_path_creates_recording(client):
    token = await user_tokens(client)
    data = wav_bytes()
    sid, r = await upload_full(client, token, data, title="Standup")
    assert r.status_code == 201, r.text
    rec = r.json()
    assert rec["title"] == "Standup"
    assert rec["has_audio"] is True
    # M7: with no AI configured the pipeline marks audio-only explicitly.
    assert rec["processing_status"] == "ai_disabled"
    assert rec["duration_seconds"] == 12.5
    # No storage keys or internals leak.
    assert "storage" not in r.text and "key" not in r.text.lower().replace("chunk", "")


async def test_upload_rejects_bad_mime_declared(client):
    token = await user_tokens(client)
    h = await auth(token)
    r = await client.post("/api/v1/uploads",
                          json={"declared_mime_type": "application/x-msdownload",
                                "declared_size_bytes": 100}, headers=h)
    assert r.status_code == 415


async def test_upload_rejects_oversize(client):
    token = await user_tokens(client)
    h = await auth(token)
    r = await client.post("/api/v1/uploads",
                          json={"declared_mime_type": "audio/wav",
                                "declared_size_bytes": 5 * 1024**3}, headers=h)
    assert r.status_code == 413


async def test_finalize_rejects_bytes_not_matching_mime(client):
    token = await user_tokens(client)
    data = mp4_bytes()
    _sid, r = await upload_full(client, token, data, mime="audio/wav")
    assert r.status_code == 415


async def test_finalize_rejects_size_mismatch(client):
    token = await user_tokens(client)
    h = await auth(token)
    data = wav_bytes()
    r = await client.post("/api/v1/uploads",
                          json={"declared_mime_type": "audio/wav",
                                "declared_size_bytes": len(data) + 10}, headers=h)
    sid = r.json()["id"]
    await client.put(f"/api/v1/uploads/{sid}/chunks/0", content=data,
                     headers={**h, "content-type": "application/octet-stream"})
    r = await client.post(f"/api/v1/uploads/{sid}/finalize", json={}, headers=h)
    assert r.status_code == 422
    detail = r.json()["detail"].lower()
    assert "missing chunks" in detail or "size mismatch" in detail


async def test_chunk_resume_status_and_idempotency(client):
    token = await user_tokens(client)
    h = await auth(token)
    data = wav_bytes()
    r = await client.post("/api/v1/uploads",
                          json={"declared_mime_type": "audio/wav",
                                "declared_size_bytes": len(data)}, headers=h)
    sid = r.json()["id"]
    r = await client.get(f"/api/v1/uploads/{sid}", headers=h)
    assert r.status_code == 200
    assert r.json()["received_chunk_indexes"] == []
    hdr = {**h, "content-type": "application/octet-stream"}
    await client.put(f"/api/v1/uploads/{sid}/chunks/0", content=data, headers=hdr)
    # Duplicate PUT of chunk 0 (retry) must not duplicate or corrupt.
    await client.put(f"/api/v1/uploads/{sid}/chunks/0", content=data, headers=hdr)
    r = await client.get(f"/api/v1/uploads/{sid}", headers=h)
    assert r.json()["received_chunk_indexes"] == [0]
    r = await client.post(f"/api/v1/uploads/{sid}/finalize", json={}, headers=h)
    assert r.status_code == 201


async def test_chunk_checksum_enforced(client):
    token = await user_tokens(client)
    h = await auth(token)
    data = wav_bytes()
    r = await client.post("/api/v1/uploads",
                          json={"declared_mime_type": "audio/wav",
                                "declared_size_bytes": len(data)}, headers=h)
    sid = r.json()["id"]
    r = await client.put(f"/api/v1/uploads/{sid}/chunks/0", content=data,
                         headers={**h, "content-type": "application/octet-stream",
                                  "x-chunk-sha256": "0" * 64})
    assert r.status_code == 422


async def test_finalize_idempotent_per_client_recording_id(client):
    token = await user_tokens(client)
    cid = str(uuid.uuid4())
    _sid1, r1 = await upload_full(client, token, wav_bytes(64), client_id=cid)
    _sid2, r2 = await upload_full(client, token, wav_bytes(64), client_id=cid, title="Renamed")
    assert r1.status_code == 201 and r2.status_code == 201
    # Same recording id, original preserved, metadata updated.
    assert r1.json()["id"] == r2.json()["id"]
    assert r2.json()["title"] == "Renamed"


# --- ownership ---------------------------------------------------------------


async def test_cross_user_isolation(client):
    ta = await user_tokens(client, "a@example.com")
    tb = await user_tokens(client, "b@example.com")
    _sid, r = await upload_full(client, ta, wav_bytes())
    rec_id = r.json()["id"]
    r = await client.get(f"/api/v1/recordings/{rec_id}", headers=await auth(tb))
    assert r.status_code == 404
    r = await client.get(f"/api/v1/recordings/{rec_id}/audio", headers=await auth(tb))
    assert r.status_code == 404
    r = await client.get("/api/v1/recordings", headers=await auth(tb))
    assert r.json()["total"] == 0


async def test_uploads_require_auth(client):
    r = await client.get("/api/v1/recordings")
    assert r.status_code == 401


# --- recordings CRUD -----------------------------------------------------------


async def test_update_metadata_and_tags(client):
    token = await user_tokens(client)
    h = await auth(token)
    _sid, r = await upload_full(client, token, wav_bytes())
    rec_id = r.json()["id"]
    r = await client.patch(f"/api/v1/recordings/{rec_id}",
                           json={"title": "Sync meeting", "notes": "n1",
                                 "tags": ["Work", " meeting ", "work"]}, headers=h)
    assert r.status_code == 200
    body = r.json()
    assert body["title"] == "Sync meeting"
    assert body["tags"] == ["meeting", "work"]  # normalized, deduped, sorted
    # listing shows same
    r = await client.get("/api/v1/recordings", headers=h)
    assert r.json()["total"] == 1
    assert r.json()["items"][0]["tags"] == ["meeting", "work"]


async def test_location_dropped_without_consent(client):
    token = await user_tokens(client)
    h = await auth(token)
    _sid, r = await upload_full(client, token, wav_bytes())
    rec_id = r.json()["id"]
    r = await client.patch(f"/api/v1/recordings/{rec_id}",
                           json={"latitude": 41.8, "longitude": -87.6}, headers=h)
    assert r.json()["latitude"] is None
    # enable consent
    await client.patch("/api/v1/users/me", json={"location_storage_enabled": True}, headers=h)
    _sid2, r2 = await upload_full(client, token, wav_bytes(128), client_id=str(uuid.uuid4()))
    rid2 = r2.json()["id"]
    r = await client.patch(f"/api/v1/recordings/{rid2}",
                           json={"latitude": 41.8, "longitude": -87.6}, headers=h)
    assert r.json()["latitude"] == 41.8


async def test_soft_then_purge_delete(client, storage_root):
    token = await user_tokens(client)
    h = await auth(token)
    _sid, r = await upload_full(client, token, wav_bytes())
    rec_id = r.json()["id"]

    files_before = list(storage_root.rglob("*"))
    assert any(p.is_file() for p in files_before)

    r = await client.delete(f"/api/v1/recordings/{rec_id}", headers=h)
    assert r.status_code == 204
    r = await client.get(f"/api/v1/recordings/{rec_id}", headers=h)
    assert r.status_code == 404

    # Purge deletes rows AND stored files.
    token2 = await user_tokens(client, "p2@example.com")
    h2 = await auth(token2)
    _sid, r = await upload_full(client, token2, wav_bytes(96))
    rec2 = r.json()["id"]
    r = await client.delete(f"/api/v1/recordings/{rec2}?purge=true", headers=h2)
    assert r.status_code == 204
    remaining = [p for p in storage_root.rglob("*") if p.is_file() and f"{rec2}" in str(p)]
    assert remaining == []


async def test_download_audio_roundtrip(client):
    token = await user_tokens(client)
    h = await auth(token)
    data = wav_bytes(128)
    _sid, r = await upload_full(client, token, data)
    rec_id = r.json()["id"]
    r = await client.get(f"/api/v1/recordings/{rec_id}/audio", headers=h)
    assert r.status_code == 200
    assert r.content == data
    assert r.headers["content-type"] == "audio/wav"
    assert "attachment" in r.headers["content-disposition"]
    assert "no-store" in r.headers["cache-control"]