""" Upload spool (node-26#44): a failed upload must never delete the recording. Before the spool, upload_recording unlinked the file in a `finally`, so any C2 outage, 5xx or timeout lost that call's audio for good. """ import asyncio import json from pathlib import Path import pytest from app.config import settings from app.internal import upload_spool from app.internal.call_recorder import CallRecorder from app.internal.upload_spool import OK, REJECT, RETRY, UploadMeta, UploadResult, UploadSpool, classify class Clock: def __init__(self, t: float = 1_000_000.0): self.t = t def __call__(self) -> float: return self.t class FakeUploader: def __init__(self, *results: UploadResult): self.results = list(results) self.calls = [] async def __call__(self, path: Path, meta: UploadMeta) -> UploadResult: assert path.exists(), "the uploader must be handed a file that exists" self.calls.append((path.name, meta)) return self.results.pop(0) if self.results else UploadResult(OK, url="gs://x") def _recording(tmp_path: Path, name: str = "20261027_call-1.flac", size: int = 100) -> Path: p = tmp_path / "rec" / name p.parent.mkdir(parents=True, exist_ok=True) p.write_bytes(b"f" * size) return p def _spool(tmp_path: Path, uploader, clock=None, **kw) -> UploadSpool: args = dict(max_bytes=10_000, max_age_seconds=3600, backoff_min=10, backoff_max=80) args.update(kw) return UploadSpool(tmp_path / "spool", uploader, clock=clock or Clock(), **args) # --------------------------------------------------------------------------- # Classification # --------------------------------------------------------------------------- @pytest.mark.parametrize("status,url,kind", [ (200, "gs://b/o", OK), (200, None, RETRY), # C2's GCS write failed: nothing stored (401, None, RETRY), # a reissued key is on its way (408, None, RETRY), (429, None, RETRY), (500, None, RETRY), (503, None, RETRY), (400, None, REJECT), (403, None, RETRY), # not approved yet: approval may still come (413, None, REJECT), ]) def test_classify(status, url, kind): assert classify(status, url) == kind # --------------------------------------------------------------------------- # upload_recording: delete only on success # --------------------------------------------------------------------------- @pytest.fixture def recorder(tmp_path, monkeypatch): monkeypatch.setattr(settings, "c2_url", "https://c2.example") rec = CallRecorder() rec.spool = _spool(tmp_path, rec._post) return rec def _patch_post(monkeypatch, result: UploadResult): async def fake(path, meta, mime, timeout=None): return result monkeypatch.setattr(upload_spool, "post_recording", fake) async def test_failed_upload_keeps_the_file_in_the_spool(tmp_path, recorder, monkeypatch): _patch_post(monkeypatch, UploadResult(RETRY, detail="ConnectError: refused")) path = _recording(tmp_path) url = await recorder.upload_recording(path, "call-1", talkgroup_id=101, talkgroup_name="Dispatch", system_id="sys-1", audio_start_epoch=1.5, audio_end_epoch=9.25) assert url is None assert not path.exists() spooled = tmp_path / "spool" / path.name assert spooled.read_bytes() == b"f" * 100 meta = json.loads((tmp_path / "spool" / (path.name + ".json")).read_text()) assert meta["call_id"] == "call-1" and meta["talkgroup_id"] == 101 and meta["system_id"] == "sys-1" assert meta["audio_start_epoch"] == 1.5 and meta["audio_end_epoch"] == 9.25 assert meta["last_error"] == "ConnectError: refused" assert recorder.spool.stats()["files"] == 1 async def test_successful_upload_deletes_the_file(tmp_path, recorder, monkeypatch): _patch_post(monkeypatch, UploadResult(OK, url="gs://b/call-1.flac", status=200)) path = _recording(tmp_path) assert await recorder.upload_recording(path, "call-1") == "gs://b/call-1.flac" assert not path.exists() assert recorder.spool.stats()["files"] == 0 async def test_rejected_upload_is_not_spooled(tmp_path, recorder, monkeypatch): _patch_post(monkeypatch, UploadResult(REJECT, status=413, detail="HTTP 413: File too large")) path = _recording(tmp_path) assert await recorder.upload_recording(path, "call-1") is None assert not path.exists() assert recorder.spool.stats()["files"] == 0 assert recorder.spool.last_result["kind"] == REJECT async def test_real_post_network_error_is_retryable(tmp_path, monkeypatch): monkeypatch.setattr(settings, "c2_url", "http://127.0.0.1:9") # discard port: refused path = _recording(tmp_path) result = await upload_spool.post_recording(path, UploadMeta(call_id="c"), "audio/flac") assert result.kind == RETRY and result.status is None assert path.exists(), "post_recording must never touch the file" # --------------------------------------------------------------------------- # Retry, startup drain, caps # --------------------------------------------------------------------------- async def test_retry_backs_off_then_uploads(tmp_path): clock = Clock() up = FakeUploader(UploadResult(RETRY, status=503, detail="HTTP 503"), UploadResult(OK, url="gs://x")) sp = _spool(tmp_path, up, clock=clock) sp.add(_recording(tmp_path), UploadMeta(call_id="c1"), UploadResult(RETRY, detail="first")) assert await sp.drain_once() == 0 and up.calls == [] # not due yet clock.t += 10 assert await sp.drain_once() == 0 and len(up.calls) == 1 # retry 2 fails data = json.loads(next((tmp_path / "spool").glob("*.json")).read_text()) assert data["attempts"] == 2 and data["next_attempt_at"] == clock.t + 20 clock.t += 20 assert await sp.drain_once() == 1 assert list((tmp_path / "spool").iterdir()) == [] async def test_startup_drain_uploads_what_survived_a_restart(tmp_path): clock = Clock() first = _spool(tmp_path, FakeUploader(), clock=clock) first.add(_recording(tmp_path, "a_c1.flac"), UploadMeta(call_id="c1", talkgroup_id=7), UploadResult(RETRY)) first.add(_recording(tmp_path, "b_c2.flac"), UploadMeta(call_id="c2"), UploadResult(RETRY)) # A new process: same directory, nothing due by the sidecar timers. up = FakeUploader() restarted = _spool(tmp_path, up, clock=clock, poll_seconds=0.01) restarted.start() for _ in range(100): if len(up.calls) == 2: break await asyncio.sleep(0.01) await restarted.stop() assert [m.call_id for _, m in up.calls] == ["c1", "c2"] assert up.calls[0][1].talkgroup_id == 7 assert list((tmp_path / "spool").iterdir()) == [] async def test_rejected_on_retry_is_deleted(tmp_path): clock = Clock() sp = _spool(tmp_path, FakeUploader(UploadResult(REJECT, status=413)), clock=clock) sp.add(_recording(tmp_path), UploadMeta(call_id="c1"), UploadResult(RETRY)) await sp.drain_once(force=True) assert sp.stats()["files"] == 0 async def test_age_cap_drops_expired_entries(tmp_path): clock = Clock() sp = _spool(tmp_path, FakeUploader(), clock=clock, max_age_seconds=3600) sp.add(_recording(tmp_path, "old_c1.flac"), UploadMeta(call_id="c1"), UploadResult(RETRY)) clock.t += 1800 sp.add(_recording(tmp_path, "new_c2.flac"), UploadMeta(call_id="c2"), UploadResult(RETRY)) clock.t += 1801 assert sp.enforce_caps() == 1 assert [p.name for p in (tmp_path / "spool").glob("*.flac")] == ["new_c2.flac"] async def test_size_cap_evicts_oldest_first(tmp_path): clock = Clock() sp = _spool(tmp_path, FakeUploader(), clock=clock, max_bytes=250) for i in range(3): sp.add(_recording(tmp_path, f"{i}_c{i}.flac", size=100), UploadMeta(call_id=f"c{i}"), UploadResult(RETRY)) clock.t += 1 assert sp.enforce_caps() == 1 assert sorted(p.name for p in (tmp_path / "spool").glob("*.flac")) == ["1_c1.flac", "2_c2.flac"] assert sp.stats()["bytes"] == 200 async def test_half_written_pairs_are_cleaned(tmp_path): sp = _spool(tmp_path, FakeUploader()) d = tmp_path / "spool" d.mkdir() (d / "lost.flac.json").write_text(json.dumps({"call_id": "x", "spooled_at": 0})) # no audio (d / "orphan.flac").write_bytes(b"a") # no sidecar, no call_id in name assert sp.stats()["files"] == 0 assert list(d.iterdir()) == [] async def test_lost_or_corrupt_sidecar_is_rebuilt_from_the_filename(tmp_path): """A power cut can zero a sidecar; the audio must survive it.""" up = FakeUploader() sp = _spool(tmp_path, up) d = tmp_path / "spool" d.mkdir() (d / "20261027_120000_call-a.flac").write_bytes(b"a") (d / "20261027_120001_call_b.flac").write_bytes(b"b") (d / "20261027_120001_call_b.flac.json").write_text("") assert sp.stats()["files"] == 2 await sp.drain_once() assert sorted(m.call_id for _, m in up.calls) == ["call-a", "call_b"] async def test_wake_makes_everything_due_now(tmp_path): clock = Clock() up = FakeUploader() sp = _spool(tmp_path, up, clock=clock) sp.add(_recording(tmp_path), UploadMeta(call_id="c1"), UploadResult(RETRY, status=401)) sp.wake() assert await sp.drain_once() == 1 def test_form_carries_the_radio_list_and_omits_it_when_empty(): """C2 starts correlation from /upload, so the radio list rides on it (server-26#304).""" radios = json.dumps([{"radio": 555, "start": 0.0, "end": 1.0}]) assert upload_spool._form(UploadMeta(call_id="c1", radios=radios))["radios"] == radios assert "radios" not in upload_spool._form(UploadMeta(call_id="c1"))