from pathlib import Path from fastapi.testclient import TestClient from sqlalchemy import select from sqlalchemy.exc import IntegrityError import services.recording_service.app as recording_module from services.recording_service.app import app as recording_app from services.shared.db import get_session from services.shared.security import issue_app_token from services.shared.sql_models import CallRecordingRow from services.voice_adapter_service.app import app as voice_app def _write_audio_fixture(path: Path, size: int = 32) -> Path: path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(b"RIFF" + b"\x00" * size) return path def _supervisor_headers() -> dict[str, str]: return {"X-User": "supervisor", "X-Role": "supervisor"} def _admin_headers() -> dict[str, str]: return {"X-User": "admin", "X-Role": "admin"} def _operator_headers() -> dict[str, str]: return {"X-User": "operator", "X-Role": "operator"} def test_import_from_voice_event_creates_recording_and_is_idempotent(tmp_path, monkeypatch): monkeypatch.setenv("CC_RECORDINGS_DIR", str(tmp_path / "recordings")) voice_client = TestClient(voice_app) recording_client = TestClient(recording_app) source_path = _write_audio_fixture(tmp_path / "fixtures" / "call.wav") event = voice_client.post( "/integrations/voice/events", headers=_admin_headers(), json={ "event_type": "recording.ready", "call_id": f"call_{tmp_path.name}", "interaction_id": f"int_{tmp_path.name}", "payload": { "source_path": str(source_path), "file_name": "call.wav", "mime_type": "audio/wav", "duration_seconds": 1, }, }, ) assert event.status_code == 200 event_id = event.json()["event_id"] created = recording_client.post( f"/recordings/import-from-voice-event/{event_id}", headers=_supervisor_headers(), ) assert created.status_code == 200 payload = created.json() assert payload["recording_id"].startswith("rec_") assert payload["status"] == "ready" assert payload["mime_type"] == "audio/wav" duplicated = recording_client.post( f"/recordings/import-from-voice-event/{event_id}", headers=_supervisor_headers(), ) assert duplicated.status_code == 200 assert duplicated.json()["recording_id"] == payload["recording_id"] def test_register_recording_lists_filters_and_archives(tmp_path, monkeypatch): monkeypatch.setenv("CC_RECORDINGS_DIR", str(tmp_path / "managed")) recording_client = TestClient(recording_app) source_path = _write_audio_fixture(tmp_path / "fixtures" / "register.wav") call_id = f"call_{tmp_path.name}" interaction_id = f"int_{tmp_path.name}" created = recording_client.post( "/recordings/register", headers=_supervisor_headers(), json={ "call_id": call_id, "interaction_id": interaction_id, "source_path": str(source_path), "file_name": "register.wav", "mime_type": "audio/wav", "duration_seconds": 2, }, ) assert created.status_code == 200 recording_id = created.json()["recording_id"] listed = recording_client.get(f"/recordings?call_id={call_id}", headers=_supervisor_headers()) assert listed.status_code == 200 assert any(item["recording_id"] == recording_id for item in listed.json()) filtered = recording_client.get( f"/recordings?interaction_id={interaction_id}", headers=_supervisor_headers(), ) assert filtered.status_code == 200 assert any(item["recording_id"] == recording_id for item in filtered.json()) detail = recording_client.get(f"/recordings/{recording_id}", headers=_supervisor_headers()) assert detail.status_code == 200 assert detail.json()["call_id"] == call_id archived = recording_client.post( f"/recordings/{recording_id}/archive", headers=_supervisor_headers(), ) assert archived.status_code == 200 assert archived.json()["status"] == "archived" def test_recording_content_supports_inline_and_attachment(tmp_path, monkeypatch): monkeypatch.setenv("CC_RECORDINGS_DIR", str(tmp_path / "managed")) recording_client = TestClient(recording_app) source_path = _write_audio_fixture(tmp_path / "fixtures" / "preview.wav") created = recording_client.post( "/recordings/register", headers=_supervisor_headers(), json={ "call_id": f"call_{tmp_path.name}", "source_path": str(source_path), "file_name": "preview.wav", }, ) assert created.status_code == 200 recording_id = created.json()["recording_id"] inline = recording_client.get( f"/recordings/{recording_id}/content?disposition=inline", headers=_supervisor_headers(), ) assert inline.status_code == 200 assert inline.content.startswith(b"RIFF") assert inline.headers["content-type"].startswith("audio/wav") attachment = recording_client.get( f"/recordings/{recording_id}/content?disposition=attachment", headers=_supervisor_headers(), ) assert attachment.status_code == 200 assert "attachment" in attachment.headers["content-disposition"] def test_missing_file_marks_recording_missing(tmp_path, monkeypatch): storage_root = tmp_path / "managed" monkeypatch.setenv("CC_RECORDINGS_DIR", str(storage_root)) recording_client = TestClient(recording_app) source_path = _write_audio_fixture(tmp_path / "fixtures" / "missing.wav") created = recording_client.post( "/recordings/register", headers=_supervisor_headers(), json={ "call_id": f"call_{tmp_path.name}", "source_path": str(source_path), "file_name": "missing.wav", }, ) assert created.status_code == 200 recording_id = created.json()["recording_id"] stored_files = list(storage_root.rglob("*.wav")) assert stored_files stored_files[0].unlink() missing = recording_client.get( f"/recordings/{recording_id}/content", headers=_supervisor_headers(), ) assert missing.status_code == 410 detail = recording_client.get(f"/recordings/{recording_id}", headers=_supervisor_headers()) assert detail.status_code == 200 assert detail.json()["status"] == "missing" def test_recording_import_upload_accepts_multipart(tmp_path, monkeypatch): monkeypatch.setenv("CC_RECORDINGS_DIR", str(tmp_path / "managed")) recording_client = TestClient(recording_app) fixture = _write_audio_fixture(tmp_path / "fixtures" / "upload.wav") with fixture.open("rb") as handle: created = recording_client.post( "/recordings/import-upload", headers=_admin_headers(), data={ "call_id": f"call_upload_{tmp_path.name}", "interaction_id": f"int_upload_{tmp_path.name}", "file_name": "upload.wav", "mime_type": "audio/wav", "duration_seconds": "4", }, files={"file": ("upload.wav", handle, "audio/wav")}, ) assert created.status_code == 200 payload = created.json() assert payload["recording_id"].startswith("rec_") assert payload["call_id"].startswith("call_upload_") assert payload["status"] == "ready" assert payload["mime_type"] == "audio/wav" def test_recording_import_upload_is_idempotent_by_source_event_id(tmp_path, monkeypatch): monkeypatch.setenv("CC_RECORDINGS_DIR", str(tmp_path / "managed")) recording_client = TestClient(recording_app) fixture = _write_audio_fixture(tmp_path / "fixtures" / "upload-idempotent.wav") source_event_id = f"bridge_upload_{tmp_path.name}" form_data = { "call_id": f"call_upload_idempotent_{tmp_path.name}", "interaction_id": f"int_upload_idempotent_{tmp_path.name}", "source_event_id": source_event_id, "file_name": "upload-idempotent.wav", "mime_type": "audio/wav", "duration_seconds": "4", } with fixture.open("rb") as handle: first = recording_client.post( "/recordings/import-upload", headers=_admin_headers(), data=form_data, files={"file": ("upload-idempotent.wav", handle, "audio/wav")}, ) with fixture.open("rb") as handle: second = recording_client.post( "/recordings/import-upload", headers=_admin_headers(), data=form_data, files={"file": ("upload-idempotent.wav", handle, "audio/wav")}, ) assert first.status_code == 200 assert second.status_code == 200 assert first.json()["recording_id"] == second.json()["recording_id"] assert first.json()["source_event_id"] == source_event_id session = get_session() try: rows = session.execute( select(CallRecordingRow).where(CallRecordingRow.source_event_id == source_event_id) ).scalars().all() assert len(rows) == 1 finally: session.close() def test_register_recording_recovers_from_source_event_id_conflict(tmp_path, monkeypatch): monkeypatch.setenv("CC_RECORDINGS_DIR", str(tmp_path / "managed")) source_path = _write_audio_fixture(tmp_path / "fixtures" / "race.wav") source_event_id = f"src_race_{tmp_path.name}" existing_call_id = f"call_race_{tmp_path.name}" session = get_session() original_commit = session.commit commit_attempts = {"count": 0} def _conflicting_commit(): if commit_attempts["count"] == 0: commit_attempts["count"] += 1 other = get_session() try: existing = recording_module._register_recording( other, call_id=existing_call_id, interaction_id=f"int_race_{tmp_path.name}", source_path=source_path.resolve(), file_name="race.wav", mime_type="audio/wav", duration_seconds=3, recorded_at=None, source_event_id=source_event_id, ) assert existing.source_event_id == source_event_id finally: other.close() raise IntegrityError("duplicate source_event_id", params=None, orig=Exception("duplicate")) return original_commit() try: monkeypatch.setattr(session, "commit", _conflicting_commit) row = recording_module._register_recording( session, call_id=existing_call_id, interaction_id=f"int_race_{tmp_path.name}", source_path=source_path.resolve(), file_name="race.wav", mime_type="audio/wav", duration_seconds=3, recorded_at=None, source_event_id=source_event_id, ) assert row.source_event_id == source_event_id verify = get_session() try: rows = verify.execute( select(CallRecordingRow).where(CallRecordingRow.source_event_id == source_event_id) ).scalars().all() assert len(rows) == 1 assert rows[0].recording_id == row.recording_id finally: verify.close() finally: session.close() def test_recording_rejects_unsupported_or_oversized_files_and_operator_access(tmp_path, monkeypatch): monkeypatch.setenv("CC_RECORDINGS_DIR", str(tmp_path / "managed")) monkeypatch.setenv("RECORDING_MAX_BYTES", "1024") recording_client = TestClient(recording_app) unsupported_path = (tmp_path / "fixtures" / "bad.txt") unsupported_path.parent.mkdir(parents=True, exist_ok=True) unsupported_path.write_text("not audio", encoding="utf-8") unsupported = recording_client.post( "/recordings/register", headers=_supervisor_headers(), json={ "call_id": f"call_bad_{tmp_path.name}", "source_path": str(unsupported_path), "file_name": "bad.txt", }, ) assert unsupported.status_code == 415 oversized_path = _write_audio_fixture(tmp_path / "fixtures" / "big.wav", size=4096) oversized = recording_client.post( "/recordings/register", headers=_supervisor_headers(), json={ "call_id": f"call_big_{tmp_path.name}", "source_path": str(oversized_path), "file_name": "big.wav", }, ) assert oversized.status_code == 413 denied = recording_client.get("/recordings", headers=_operator_headers()) assert denied.status_code == 403 def test_recording_import_upload_requires_trusted_service_when_admin_fallback_disabled(tmp_path, monkeypatch): monkeypatch.setenv("CC_RECORDINGS_DIR", str(tmp_path / "managed")) monkeypatch.setenv("RECORDING_IMPORT_ALLOW_ADMIN", "0") monkeypatch.setenv("RECORDING_IMPORT_TRUSTED_SERVICE_SUBJECTS", "svc:asterisk-bridge") monkeypatch.setenv("ALLOW_LEGACY_HEADER_AUTH", "0") monkeypatch.setenv("APP_TOKEN_SECRET", "track9-service-token-test") recording_client = TestClient(recording_app) fixture = _write_audio_fixture(tmp_path / "fixtures" / "svc-upload.wav") with fixture.open("rb") as handle: denied = recording_client.post( "/recordings/import-upload", headers=_admin_headers(), data={ "call_id": f"call_upload_denied_{tmp_path.name}", "interaction_id": f"int_upload_denied_{tmp_path.name}", "file_name": "svc-upload.wav", "mime_type": "audio/wav", "duration_seconds": "4", }, files={"file": ("svc-upload.wav", handle, "audio/wav")}, ) assert denied.status_code == 403 token = issue_app_token( subject="svc:asterisk-bridge", username="asterisk-bridge", role="admin", auth_source="service", provider="track9-test", ttl_seconds=300, ) with fixture.open("rb") as handle: allowed = recording_client.post( "/recordings/import-upload", headers={"Authorization": f"Bearer {token}"}, data={ "call_id": f"call_upload_allowed_{tmp_path.name}", "interaction_id": f"int_upload_allowed_{tmp_path.name}", "file_name": "svc-upload.wav", "mime_type": "audio/wav", "duration_seconds": "4", }, files={"file": ("svc-upload.wav", handle, "audio/wav")}, ) assert allowed.status_code == 200 assert allowed.json()["recording_id"].startswith("rec_")