Logan CusanoandClaude Opus 5.5
c836986b89
Backend: /media/calls/{id}/audio now honours a single HTTP Range
(bytes=a-b, a-, -n) with 206 + Content-Range, 416 for unsatisfiable,
and a 200 full body (Accept-Ranges: bytes) otherwise. The body is
streamed from GCS in 512 KiB span reads (storage.audio_object_size +
download_audio_range) instead of downloading the whole object first,
so a 600 s FLAC starts after one span and seeks fetch only what they
need. HMAC check and Firestore lookup unchanged and still first.
Frontend: InlinePlayer moved to components/InlinePlayer.tsx and now
drives lib/audioQueue.ts, a useSyncExternalStore store over ONE audio
element (one call plays at a time). Wall spine rows get play buttons
and a "Play incident" queue (oldest first) with the playing row
highlighted; map path-stop popups get a play button and the playing
stop gets an accent ring; rail cards get "play latest". Playback is
gated by canPlayAudio() (BUSINESS_MODEL section 3 / Gate A hook).
Nothing plays until clicked. Spine truncation is now stated
("showing latest 30 of M", show-all toggle, count outside the loaded
window) for server-26#239.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
221 lines
8.1 KiB
Python
221 lines
8.1 KiB
Python
"""
|
|
HTTP Range on /media/calls/{id}/audio (server-26#233).
|
|
|
|
The GCS reads are replaced by an in-memory object: ``audio_object_size`` and
|
|
``download_audio_range`` are patched where routers/media.py looks them up, and
|
|
every span read is recorded so the tests can assert the route never asks GCS
|
|
for more than the requested range (plus at most one STREAM_CHUNK_BYTES span).
|
|
"""
|
|
import time
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.internal import storage
|
|
from app.routers import media
|
|
|
|
app = FastAPI()
|
|
app.include_router(media.router)
|
|
client = TestClient(app)
|
|
|
|
CALL_ID = "call-abc"
|
|
GCS_URI = "gs://bucket/calls/call-abc.flac"
|
|
|
|
|
|
class FakeObject:
|
|
"""A stored object plus a log of every (start, end) span read from it."""
|
|
|
|
def __init__(self, data: bytes):
|
|
self.data = data
|
|
self.reads: list[tuple[int, int]] = []
|
|
self.size_calls = 0
|
|
|
|
async def size(self, gcs_uri):
|
|
self.size_calls += 1
|
|
return len(self.data)
|
|
|
|
async def read(self, gcs_uri, start, end):
|
|
self.reads.append((start, end))
|
|
return self.data[start:end + 1]
|
|
|
|
|
|
@pytest.fixture
|
|
def signed(monkeypatch):
|
|
monkeypatch.setattr(storage.settings, "service_key", "test-key", raising=False)
|
|
exp = int(time.time()) + 600
|
|
return {"exp": exp, "sig": storage.sign_audio_link(CALL_ID, exp)}
|
|
|
|
|
|
def _run(obj: FakeObject, params, headers=None, doc=None):
|
|
doc_get = AsyncMock(return_value=doc if doc is not None else {"call_id": CALL_ID, "audio_gcs_uri": GCS_URI})
|
|
with patch.object(media.fstore, "doc_get", doc_get), \
|
|
patch.object(media, "audio_object_size", obj.size), \
|
|
patch.object(media, "download_audio_range", obj.read):
|
|
resp = client.get(f"/media/calls/{CALL_ID}/audio", params=params, headers=headers or {})
|
|
return resp, doc_get
|
|
|
|
|
|
def test_bounded_range_returns_206_with_exact_bytes(signed):
|
|
obj = FakeObject(bytes(range(256)) * 40) # 10240 bytes
|
|
resp, _ = _run(obj, signed, {"Range": "bytes=100-199"})
|
|
assert resp.status_code == 206
|
|
assert resp.content == obj.data[100:200]
|
|
assert resp.headers["content-range"] == f"bytes 100-199/{len(obj.data)}"
|
|
assert resp.headers["content-length"] == "100"
|
|
assert resp.headers["accept-ranges"] == "bytes"
|
|
assert resp.headers["content-type"] == "audio/flac"
|
|
assert resp.headers["cache-control"] == "private, max-age=3600"
|
|
# Only the requested span was read from GCS — not the whole object.
|
|
assert obj.reads == [(100, 199)]
|
|
|
|
|
|
def test_open_ended_range(signed):
|
|
obj = FakeObject(b"x" * 5000 + b"tail")
|
|
resp, _ = _run(obj, signed, {"Range": "bytes=5000-"})
|
|
assert resp.status_code == 206
|
|
assert resp.content == b"tail"
|
|
assert resp.headers["content-range"] == "bytes 5000-5003/5004"
|
|
|
|
|
|
def test_suffix_range(signed):
|
|
obj = FakeObject(b"0123456789")
|
|
resp, _ = _run(obj, signed, {"Range": "bytes=-3"})
|
|
assert resp.status_code == 206
|
|
assert resp.content == b"789"
|
|
assert resp.headers["content-range"] == "bytes 7-9/10"
|
|
|
|
|
|
def test_suffix_longer_than_object_returns_whole_object(signed):
|
|
obj = FakeObject(b"0123456789")
|
|
resp, _ = _run(obj, signed, {"Range": "bytes=-500"})
|
|
assert resp.status_code == 206
|
|
assert resp.content == obj.data
|
|
assert resp.headers["content-range"] == "bytes 0-9/10"
|
|
|
|
|
|
def test_end_past_object_is_clamped(signed):
|
|
obj = FakeObject(b"0123456789")
|
|
resp, _ = _run(obj, signed, {"Range": "bytes=8-999"})
|
|
assert resp.status_code == 206
|
|
assert resp.content == b"89"
|
|
assert resp.headers["content-range"] == "bytes 8-9/10"
|
|
|
|
|
|
@pytest.mark.parametrize("rng", ["bytes=10-", "bytes=50-60", "bytes=-0"])
|
|
def test_unsatisfiable_range_is_416(signed, rng):
|
|
obj = FakeObject(b"0123456789")
|
|
resp, _ = _run(obj, signed, {"Range": rng})
|
|
assert resp.status_code == 416
|
|
assert resp.headers["content-range"] == "bytes */10"
|
|
assert obj.reads == []
|
|
|
|
|
|
def test_no_range_is_200_full_body_and_advertises_ranges(signed):
|
|
obj = FakeObject(b"abcdef" * 100)
|
|
resp, _ = _run(obj, signed)
|
|
assert resp.status_code == 200
|
|
assert resp.content == obj.data
|
|
assert resp.headers["accept-ranges"] == "bytes"
|
|
assert resp.headers["content-length"] == str(len(obj.data))
|
|
assert "content-range" not in resp.headers
|
|
|
|
|
|
@pytest.mark.parametrize("rng", ["bytes=0-1,4-5", "items=0-5", "bytes=5-2", "garbage"])
|
|
def test_ignored_range_headers_fall_back_to_200(signed, rng):
|
|
obj = FakeObject(b"0123456789")
|
|
resp, _ = _run(obj, signed, {"Range": rng})
|
|
assert resp.status_code == 200
|
|
assert resp.content == obj.data
|
|
|
|
|
|
def test_long_object_is_streamed_in_bounded_spans(signed):
|
|
# A ~13 MB "600 s FLAC": a seek near the end reads only from there on, in
|
|
# spans no larger than STREAM_CHUNK_BYTES, and the first span is all the
|
|
# route waits on before the response starts.
|
|
size = 13 * 1024 * 1024
|
|
obj = FakeObject(bytes(size))
|
|
start = size - 1_500_000
|
|
resp, _ = _run(obj, signed, {"Range": f"bytes={start}-"})
|
|
assert resp.status_code == 206
|
|
assert len(resp.content) == 1_500_000
|
|
assert obj.reads[0][0] == start
|
|
assert all(e - s + 1 <= media.STREAM_CHUNK_BYTES for s, e in obj.reads)
|
|
assert sum(e - s + 1 for s, e in obj.reads) == 1_500_000
|
|
|
|
|
|
def test_bad_signature_is_403_before_any_lookup(signed):
|
|
obj = FakeObject(b"0123456789")
|
|
resp, doc_get = _run(obj, {"exp": signed["exp"], "sig": "0" * 64}, {"Range": "bytes=0-3"})
|
|
assert resp.status_code == 403
|
|
doc_get.assert_not_called()
|
|
assert obj.size_calls == 0
|
|
assert obj.reads == []
|
|
|
|
|
|
def test_expired_link_is_403(signed, monkeypatch):
|
|
exp = int(time.time()) - 5
|
|
obj = FakeObject(b"0123456789")
|
|
resp, doc_get = _run(obj, {"exp": exp, "sig": storage.sign_audio_link(CALL_ID, exp)})
|
|
assert resp.status_code == 403
|
|
doc_get.assert_not_called()
|
|
|
|
|
|
def test_missing_object_is_404(signed):
|
|
obj = FakeObject(b"")
|
|
|
|
async def no_size(gcs_uri):
|
|
return None
|
|
|
|
with patch.object(media.fstore, "doc_get", AsyncMock(return_value={"call_id": CALL_ID, "audio_gcs_uri": GCS_URI})), \
|
|
patch.object(media, "audio_object_size", no_size), \
|
|
patch.object(media, "download_audio_range", obj.read):
|
|
resp = client.get(f"/media/calls/{CALL_ID}/audio", params=signed)
|
|
assert resp.status_code == 404
|
|
|
|
|
|
def test_legacy_mp3_keeps_its_content_type(signed):
|
|
obj = FakeObject(b"ID3" + b"\0" * 50)
|
|
resp, _ = _run(obj, signed, {"Range": "bytes=0-2"}, doc={"call_id": CALL_ID, "audio_gcs_uri": "gs://b/calls/x.mp3"})
|
|
assert resp.status_code == 206
|
|
assert resp.headers["content-type"] == "audio/mpeg"
|
|
assert resp.content == b"ID3"
|
|
|
|
|
|
# ── storage.py: the ranged reads never pull the whole object ────────────────
|
|
def _fake_client(monkeypatch, size=1234, payload=b"span"):
|
|
from unittest.mock import MagicMock
|
|
client = MagicMock()
|
|
blob = client.bucket.return_value.blob.return_value
|
|
blob.size = size
|
|
blob.download_as_bytes.return_value = payload
|
|
monkeypatch.setattr(storage, "_gcs_client_cache", client)
|
|
return client, blob
|
|
|
|
|
|
def test_storage_size_reloads_metadata_only(monkeypatch):
|
|
import asyncio
|
|
client, blob = _fake_client(monkeypatch, size=13_000_000)
|
|
assert asyncio.run(storage.audio_object_size(GCS_URI)) == 13_000_000
|
|
client.bucket.assert_called_with("bucket")
|
|
client.bucket.return_value.blob.assert_called_with("calls/call-abc.flac")
|
|
blob.reload.assert_called_once()
|
|
blob.download_as_bytes.assert_not_called()
|
|
|
|
|
|
def test_storage_range_read_passes_inclusive_bounds(monkeypatch):
|
|
import asyncio
|
|
_, blob = _fake_client(monkeypatch)
|
|
assert asyncio.run(storage.download_audio_range(GCS_URI, 100, 199)) == b"span"
|
|
blob.download_as_bytes.assert_called_once_with(start=100, end=199)
|
|
|
|
|
|
def test_storage_range_read_failure_returns_none(monkeypatch):
|
|
import asyncio
|
|
_, blob = _fake_client(monkeypatch)
|
|
blob.download_as_bytes.side_effect = RuntimeError("gcs down")
|
|
assert asyncio.run(storage.download_audio_range(GCS_URI, 0, 10)) is None
|
|
blob.reload.side_effect = RuntimeError("404")
|
|
assert asyncio.run(storage.audio_object_size(GCS_URI)) is None
|