#!/usr/bin/env python3 """ Test suite for Speaker API v1 endpoints. Covers: - PUT /recordings//speakers/assign (17 tests) - POST /recordings//speakers/identify (10 tests) - PUT /settings/auto-summarization (5 tests) - Regression for GET /speakers and GET /recordings//speakers (2 tests) Pattern follows tests/test_api_v1_upload.py — standalone, no pytest fixtures. """ import json import secrets import sys import os from unittest.mock import patch, MagicMock # Add parent directory so we can import the app sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from src.app import app, db from src.models import User, APIToken, Recording, Speaker from src.utils.token_auth import hash_token # --------------------------------------------------------------------------- # Test data constants # --------------------------------------------------------------------------- SAMPLE_TRANSCRIPTION_JSON = json.dumps([ {"speaker": "SPEAKER_00", "sentence": "Hi, I'm Alice."}, {"speaker": "SPEAKER_01", "sentence": "Hello Alice, I'm Bob."}, {"speaker": "SPEAKER_00", "sentence": "Nice to meet you, Bob."}, ]) SAMPLE_TRANSCRIPTION_TEXT = ( "[SPEAKER_00]: Hi, I'm Alice.\n" "[SPEAKER_01]: Hello Alice, I'm Bob.\n" "[SPEAKER_00]: Nice to meet you, Bob." ) SAMPLE_EMBEDDINGS = json.dumps({ "SPEAKER_00": [0.1] * 256, "SPEAKER_01": [0.2] * 256, }) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _get_or_create_test_user(suffix=""): """Get or create a test user. Returns (user, created_bool).""" username = f"speaker_test_user{suffix}" user = User.query.filter_by(username=username).first() created = False if not user: user = User( username=username, email=f"{username}@local.test", name="Test User" if not suffix else None, ) db.session.add(user) db.session.commit() created = True return user, created def _create_api_token(user): """Create a fresh API token. Returns (token_record, plaintext).""" plaintext = f"test-token-{secrets.token_urlsafe(16)}" token = APIToken( user_id=user.id, token_hash=hash_token(plaintext), name="test-api-token", ) db.session.add(token) db.session.commit() return token, plaintext def _create_test_recording(user, transcription=None, speaker_embeddings=None, status="COMPLETED"): """Create a Recording owned by *user*.""" rec = Recording( user_id=user.id, title="Test Recording", status=status, transcription=transcription, speaker_embeddings=speaker_embeddings, ) db.session.add(rec) db.session.commit() return rec def _create_test_speaker(user, name="Alice"): """Create a Speaker owned by *user*.""" speaker = Speaker(name=name, user_id=user.id) db.session.add(speaker) db.session.commit() return speaker def _cleanup(*objects): """Delete DB objects in reverse order, committing once.""" for obj in reversed(objects): try: db.session.delete(obj) except Exception: db.session.rollback() try: merged = db.session.merge(obj) db.session.delete(merged) except Exception: pass db.session.commit() # ========================================================================= # Group 1: PUT /recordings//speakers/assign (17 tests) # ========================================================================= def test_assign_no_auth(): """No token -> 302 redirect (Flask-Login).""" with app.app_context(): user, cu = _get_or_create_test_user() rec = _create_test_recording(user, transcription=SAMPLE_TRANSCRIPTION_JSON) client = app.test_client() try: resp = client.put(f"/api/v1/recordings/{rec.id}/speakers/assign", json={"speaker_map": {}}) assert resp.status_code in (302, 401), f"Expected 302/401, got {resp.status_code}" return True finally: _cleanup(rec) if cu: _cleanup(user) def test_assign_recording_not_found(): """Nonexistent recording ID -> 404.""" with app.app_context(): user, cu = _get_or_create_test_user() token_rec, token = _create_api_token(user) client = app.test_client() try: resp = client.put("/api/v1/recordings/999999/speakers/assign", headers={"X-API-Token": token}, json={"speaker_map": {}}) assert resp.status_code == 404, f"Expected 404, got {resp.status_code}" return True finally: _cleanup(token_rec) if cu: _cleanup(user) def test_assign_wrong_user_recording(): """Other user's recording -> 403.""" with app.app_context(): owner, co = _get_or_create_test_user("_owner") other, cu = _get_or_create_test_user("_other") token_rec, token = _create_api_token(other) rec = _create_test_recording(owner, transcription=SAMPLE_TRANSCRIPTION_JSON) client = app.test_client() try: resp = client.put(f"/api/v1/recordings/{rec.id}/speakers/assign", headers={"X-API-Token": token}, json={"speaker_map": {"SPEAKER_00": "Alice"}}) assert resp.status_code == 403, f"Expected 403, got {resp.status_code}" return True finally: _cleanup(rec, token_rec) if cu: _cleanup(other) if co: _cleanup(owner) def test_assign_missing_speaker_map(): """Body {} -> 400 'speaker_map is required'.""" with app.app_context(): user, cu = _get_or_create_test_user() token_rec, token = _create_api_token(user) rec = _create_test_recording(user, transcription=SAMPLE_TRANSCRIPTION_JSON) client = app.test_client() try: resp = client.put(f"/api/v1/recordings/{rec.id}/speakers/assign", headers={"X-API-Token": token}, json={}) assert resp.status_code == 400, f"Expected 400, got {resp.status_code}" body = resp.get_json() assert "speaker_map" in body.get("error", "").lower(), f"Unexpected error: {body}" return True finally: _cleanup(rec, token_rec) if cu: _cleanup(user) def test_assign_invalid_speaker_map_type(): """speaker_map: 'string' -> 400.""" with app.app_context(): user, cu = _get_or_create_test_user() token_rec, token = _create_api_token(user) rec = _create_test_recording(user, transcription=SAMPLE_TRANSCRIPTION_JSON) client = app.test_client() try: resp = client.put(f"/api/v1/recordings/{rec.id}/speakers/assign", headers={"X-API-Token": token}, json={"speaker_map": "not a dict"}) assert resp.status_code == 400, f"Expected 400, got {resp.status_code}" return True finally: _cleanup(rec, token_rec) if cu: _cleanup(user) def test_assign_string_value_json_transcript(): """Happy path: string names update JSON segments + participants.""" with app.app_context(): user, cu = _get_or_create_test_user() token_rec, token = _create_api_token(user) rec = _create_test_recording(user, transcription=SAMPLE_TRANSCRIPTION_JSON) client = app.test_client() try: resp = client.put(f"/api/v1/recordings/{rec.id}/speakers/assign", headers={"X-API-Token": token}, json={"speaker_map": {"SPEAKER_00": "Alice", "SPEAKER_01": "Bob"}}) assert resp.status_code == 200, f"Expected 200, got {resp.status_code}" body = resp.get_json() assert body.get("success") is True # Verify participants participants = body["recording"]["participants"] assert "Alice" in participants and "Bob" in participants # Verify transcription was updated db.session.refresh(rec) segments = json.loads(rec.transcription) assert segments[0]["speaker"] == "Alice" assert segments[1]["speaker"] == "Bob" return True finally: _cleanup(rec, token_rec) if cu: _cleanup(user) def test_assign_object_value_with_name(): """Happy path: {name, isMe} object format.""" with app.app_context(): user, cu = _get_or_create_test_user() token_rec, token = _create_api_token(user) rec = _create_test_recording(user, transcription=SAMPLE_TRANSCRIPTION_JSON) client = app.test_client() try: resp = client.put(f"/api/v1/recordings/{rec.id}/speakers/assign", headers={"X-API-Token": token}, json={"speaker_map": { "SPEAKER_00": {"name": "Alice", "isMe": False}, }}) assert resp.status_code == 200, f"Expected 200, got {resp.status_code}" db.session.refresh(rec) segments = json.loads(rec.transcription) assert segments[0]["speaker"] == "Alice" return True finally: _cleanup(rec, token_rec) if cu: _cleanup(user) def test_assign_is_me_flag_with_user_name(): """isMe: true resolves to user.name.""" with app.app_context(): user, cu = _get_or_create_test_user() # user.name == "Test User" token_rec, token = _create_api_token(user) rec = _create_test_recording(user, transcription=SAMPLE_TRANSCRIPTION_JSON) client = app.test_client() try: resp = client.put(f"/api/v1/recordings/{rec.id}/speakers/assign", headers={"X-API-Token": token}, json={"speaker_map": { "SPEAKER_00": {"name": "", "isMe": True}, }}) assert resp.status_code == 200, f"Expected 200, got {resp.status_code}" db.session.refresh(rec) segments = json.loads(rec.transcription) assert segments[0]["speaker"] == "Test User", f"Got {segments[0]['speaker']}" return True finally: _cleanup(rec, token_rec) if cu: _cleanup(user) def test_assign_is_me_flag_without_user_name(): """isMe: true falls back to 'Me' when user.name is None.""" with app.app_context(): user, cu = _get_or_create_test_user("_noname") # Ensure user.name is None user.name = None db.session.commit() token_rec, token = _create_api_token(user) rec = _create_test_recording(user, transcription=SAMPLE_TRANSCRIPTION_JSON) client = app.test_client() try: resp = client.put(f"/api/v1/recordings/{rec.id}/speakers/assign", headers={"X-API-Token": token}, json={"speaker_map": { "SPEAKER_00": {"name": "", "isMe": True}, }}) assert resp.status_code == 200, f"Expected 200, got {resp.status_code}" db.session.refresh(rec) segments = json.loads(rec.transcription) assert segments[0]["speaker"] == "Me", f"Got {segments[0]['speaker']}" return True finally: _cleanup(rec, token_rec) if cu: _cleanup(user) def test_assign_plain_text_transcript(): """Replaces [SPEAKER_XX] in plain text format.""" with app.app_context(): user, cu = _get_or_create_test_user() token_rec, token = _create_api_token(user) rec = _create_test_recording(user, transcription=SAMPLE_TRANSCRIPTION_TEXT) client = app.test_client() try: resp = client.put(f"/api/v1/recordings/{rec.id}/speakers/assign", headers={"X-API-Token": token}, json={"speaker_map": {"SPEAKER_00": "Alice", "SPEAKER_01": "Bob"}}) assert resp.status_code == 200, f"Expected 200, got {resp.status_code}" db.session.refresh(rec) assert "[Alice]" in rec.transcription assert "[Bob]" in rec.transcription assert "[SPEAKER_00]" not in rec.transcription return True finally: _cleanup(rec, token_rec) if cu: _cleanup(user) def test_assign_speaker_xx_filtered_from_participants(): """Unresolved SPEAKER_XX labels excluded from participants.""" with app.app_context(): user, cu = _get_or_create_test_user() token_rec, token = _create_api_token(user) rec = _create_test_recording(user, transcription=SAMPLE_TRANSCRIPTION_JSON) client = app.test_client() try: # Only assign one speaker - SPEAKER_01 stays unresolved resp = client.put(f"/api/v1/recordings/{rec.id}/speakers/assign", headers={"X-API-Token": token}, json={"speaker_map": {"SPEAKER_00": "Alice"}}) assert resp.status_code == 200, f"Expected 200, got {resp.status_code}" body = resp.get_json() participants = body["recording"]["participants"] assert "SPEAKER_01" not in participants, f"SPEAKER_01 should be filtered: {participants}" assert "Alice" in participants return True finally: _cleanup(rec, token_rec) if cu: _cleanup(user) def test_assign_invalid_value_type(): """Array value -> 400 'Invalid value type'.""" with app.app_context(): user, cu = _get_or_create_test_user() token_rec, token = _create_api_token(user) rec = _create_test_recording(user, transcription=SAMPLE_TRANSCRIPTION_JSON) client = app.test_client() try: resp = client.put(f"/api/v1/recordings/{rec.id}/speakers/assign", headers={"X-API-Token": token}, json={"speaker_map": {"SPEAKER_00": [1, 2, 3]}}) assert resp.status_code == 400, f"Expected 400, got {resp.status_code}" body = resp.get_json() assert "invalid value type" in body.get("error", "").lower(), f"Unexpected: {body}" return True finally: _cleanup(rec, token_rec) if cu: _cleanup(user) def test_assign_empty_speaker_map(): """Empty speaker_map {} -> 200 with no changes.""" with app.app_context(): user, cu = _get_or_create_test_user() token_rec, token = _create_api_token(user) rec = _create_test_recording(user, transcription=SAMPLE_TRANSCRIPTION_JSON) client = app.test_client() try: resp = client.put(f"/api/v1/recordings/{rec.id}/speakers/assign", headers={"X-API-Token": token}, json={"speaker_map": {}}) assert resp.status_code == 200, f"Expected 200, got {resp.status_code}" body = resp.get_json() assert body.get("success") is True # Transcription should be unchanged db.session.refresh(rec) segments = json.loads(rec.transcription) assert segments[0]["speaker"] == "SPEAKER_00" return True finally: _cleanup(rec, token_rec) if cu: _cleanup(user) def test_assign_regenerate_summary(): """regenerate_summary: true -> job_queue.enqueue called, summary_queued: true.""" with app.app_context(): user, cu = _get_or_create_test_user() token_rec, token = _create_api_token(user) rec = _create_test_recording(user, transcription=SAMPLE_TRANSCRIPTION_JSON) client = app.test_client() try: mock_jq = MagicMock() mock_jq.enqueue = MagicMock(return_value="job-123") with patch("src.services.job_queue.job_queue", mock_jq): resp = client.put(f"/api/v1/recordings/{rec.id}/speakers/assign", headers={"X-API-Token": token}, json={ "speaker_map": {"SPEAKER_00": "Alice"}, "regenerate_summary": True, }) assert resp.status_code == 200, f"Expected 200, got {resp.status_code}" body = resp.get_json() assert body.get("summary_queued") is True mock_jq.enqueue.assert_called_once() return True finally: _cleanup(rec, token_rec) if cu: _cleanup(user) def test_assign_embeddings_updated(): """With speaker_embeddings -> update_speaker_embedding called, counts returned.""" with app.app_context(): user, cu = _get_or_create_test_user() token_rec, token = _create_api_token(user) rec = _create_test_recording( user, transcription=SAMPLE_TRANSCRIPTION_JSON, speaker_embeddings=SAMPLE_EMBEDDINGS, ) speaker = _create_test_speaker(user, "Alice") client = app.test_client() try: mock_update = MagicMock() mock_snippets = MagicMock(return_value=2) with patch("src.services.speaker_embedding_matcher.update_speaker_embedding", mock_update), \ patch("src.services.speaker_snippets.create_speaker_snippets", mock_snippets): resp = client.put(f"/api/v1/recordings/{rec.id}/speakers/assign", headers={"X-API-Token": token}, json={"speaker_map": {"SPEAKER_00": "Alice"}}) assert resp.status_code == 200, f"Expected 200, got {resp.status_code}" body = resp.get_json() assert body.get("embeddings_updated") >= 1, f"embeddings_updated: {body}" mock_update.assert_called() return True finally: _cleanup(rec, speaker, token_rec) if cu: _cleanup(user) def test_assign_no_transcription(): """Recording without transcription -> speakers applied to empty content gracefully.""" with app.app_context(): user, cu = _get_or_create_test_user() token_rec, token = _create_api_token(user) rec = _create_test_recording(user, transcription=None) client = app.test_client() try: resp = client.put(f"/api/v1/recordings/{rec.id}/speakers/assign", headers={"X-API-Token": token}, json={"speaker_map": {"SPEAKER_00": "Alice"}}) # Should succeed (or at least not 500) assert resp.status_code in (200, 400), f"Expected 200/400, got {resp.status_code}" return True finally: _cleanup(rec, token_rec) if cu: _cleanup(user) def test_assign_whitespace_name_trimmed(): """Names with leading/trailing whitespace get trimmed.""" with app.app_context(): user, cu = _get_or_create_test_user() token_rec, token = _create_api_token(user) rec = _create_test_recording(user, transcription=SAMPLE_TRANSCRIPTION_JSON) client = app.test_client() try: resp = client.put(f"/api/v1/recordings/{rec.id}/speakers/assign", headers={"X-API-Token": token}, json={"speaker_map": {"SPEAKER_00": " Alice "}}) assert resp.status_code == 200, f"Expected 200, got {resp.status_code}" db.session.refresh(rec) segments = json.loads(rec.transcription) assert segments[0]["speaker"] == "Alice", f"Name not trimmed: '{segments[0]['speaker']}'" return True finally: _cleanup(rec, token_rec) if cu: _cleanup(user) # ========================================================================= # Group 2: POST /recordings//speakers/identify (10 tests) # ========================================================================= def test_identify_no_auth(): """No token -> 302.""" with app.app_context(): user, cu = _get_or_create_test_user() rec = _create_test_recording(user, transcription=SAMPLE_TRANSCRIPTION_JSON) client = app.test_client() try: resp = client.post(f"/api/v1/recordings/{rec.id}/speakers/identify") assert resp.status_code in (302, 401), f"Expected 302/401, got {resp.status_code}" return True finally: _cleanup(rec) if cu: _cleanup(user) def test_identify_recording_not_found(): """Nonexistent ID -> 404.""" with app.app_context(): user, cu = _get_or_create_test_user() token_rec, token = _create_api_token(user) client = app.test_client() try: resp = client.post("/api/v1/recordings/999999/speakers/identify", headers={"X-API-Token": token}) assert resp.status_code == 404, f"Expected 404, got {resp.status_code}" return True finally: _cleanup(token_rec) if cu: _cleanup(user) def test_identify_wrong_user_recording(): """Other user's recording -> 403.""" with app.app_context(): owner, co = _get_or_create_test_user("_id_owner") other, cu = _get_or_create_test_user("_id_other") token_rec, token = _create_api_token(other) rec = _create_test_recording(owner, transcription=SAMPLE_TRANSCRIPTION_JSON) client = app.test_client() try: resp = client.post(f"/api/v1/recordings/{rec.id}/speakers/identify", headers={"X-API-Token": token}) assert resp.status_code == 403, f"Expected 403, got {resp.status_code}" return True finally: _cleanup(rec, token_rec) if cu: _cleanup(other) if co: _cleanup(owner) def test_identify_no_transcription(): """No transcription -> 400.""" with app.app_context(): user, cu = _get_or_create_test_user() token_rec, token = _create_api_token(user) rec = _create_test_recording(user, transcription=None) client = app.test_client() try: resp = client.post(f"/api/v1/recordings/{rec.id}/speakers/identify", headers={"X-API-Token": token}) assert resp.status_code == 400, f"Expected 400, got {resp.status_code}" return True finally: _cleanup(rec, token_rec) if cu: _cleanup(user) def test_identify_non_json_transcription(): """Plain text -> 400.""" with app.app_context(): user, cu = _get_or_create_test_user() token_rec, token = _create_api_token(user) rec = _create_test_recording(user, transcription=SAMPLE_TRANSCRIPTION_TEXT) client = app.test_client() try: resp = client.post(f"/api/v1/recordings/{rec.id}/speakers/identify", headers={"X-API-Token": token}) assert resp.status_code == 400, f"Expected 400, got {resp.status_code}" return True finally: _cleanup(rec, token_rec) if cu: _cleanup(user) def test_identify_json_but_not_list(): """Dict JSON -> 400.""" with app.app_context(): user, cu = _get_or_create_test_user() token_rec, token = _create_api_token(user) rec = _create_test_recording(user, transcription=json.dumps({"key": "value"})) client = app.test_client() try: resp = client.post(f"/api/v1/recordings/{rec.id}/speakers/identify", headers={"X-API-Token": token}) assert resp.status_code == 400, f"Expected 400, got {resp.status_code}" return True finally: _cleanup(rec, token_rec) if cu: _cleanup(user) def test_identify_happy_path(): """Mock LLM returns names -> 200 with speaker_map.""" with app.app_context(): user, cu = _get_or_create_test_user() token_rec, token = _create_api_token(user) rec = _create_test_recording(user, transcription=SAMPLE_TRANSCRIPTION_JSON) client = app.test_client() try: # Build a mock LLM completion response mock_completion = MagicMock() mock_completion.choices = [MagicMock()] mock_completion.choices[0].message.content = json.dumps({ "SPEAKER_00": "Alice", "SPEAKER_01": "Bob", }) with patch("src.services.llm.call_llm_completion", return_value=mock_completion), \ patch("src.models.system.SystemSetting") as mock_ss: mock_ss.get_setting.return_value = 30000 resp = client.post(f"/api/v1/recordings/{rec.id}/speakers/identify", headers={"X-API-Token": token}) assert resp.status_code == 200, f"Expected 200, got {resp.status_code}" body = resp.get_json() assert body.get("success") is True sm = body.get("speaker_map", {}) assert sm.get("SPEAKER_00") == "Alice" assert sm.get("SPEAKER_01") == "Bob" return True finally: _cleanup(rec, token_rec) if cu: _cleanup(user) def test_identify_post_processing_unknown_values(): """'Unknown'/'N/A' cleared to ''.""" with app.app_context(): user, cu = _get_or_create_test_user() token_rec, token = _create_api_token(user) rec = _create_test_recording(user, transcription=SAMPLE_TRANSCRIPTION_JSON) client = app.test_client() try: mock_completion = MagicMock() mock_completion.choices = [MagicMock()] mock_completion.choices[0].message.content = json.dumps({ "SPEAKER_00": "Unknown", "SPEAKER_01": "N/A", }) with patch("src.services.llm.call_llm_completion", return_value=mock_completion), \ patch("src.models.system.SystemSetting") as mock_ss: mock_ss.get_setting.return_value = 30000 resp = client.post(f"/api/v1/recordings/{rec.id}/speakers/identify", headers={"X-API-Token": token}) assert resp.status_code == 200, f"Expected 200, got {resp.status_code}" body = resp.get_json() sm = body.get("speaker_map", {}) assert sm.get("SPEAKER_00") == "", f"Expected empty, got {sm.get('SPEAKER_00')}" assert sm.get("SPEAKER_01") == "", f"Expected empty, got {sm.get('SPEAKER_01')}" return True finally: _cleanup(rec, token_rec) if cu: _cleanup(user) def test_identify_no_speakers_in_transcript(): """Segments without speaker field -> 400.""" with app.app_context(): user, cu = _get_or_create_test_user() token_rec, token = _create_api_token(user) no_speakers = json.dumps([{"sentence": "Hello"}, {"sentence": "World"}]) rec = _create_test_recording(user, transcription=no_speakers) client = app.test_client() try: resp = client.post(f"/api/v1/recordings/{rec.id}/speakers/identify", headers={"X-API-Token": token}) assert resp.status_code == 400, f"Expected 400, got {resp.status_code}" return True finally: _cleanup(rec, token_rec) if cu: _cleanup(user) def test_identify_llm_error(): """LLM raises exception -> 500.""" with app.app_context(): user, cu = _get_or_create_test_user() token_rec, token = _create_api_token(user) rec = _create_test_recording(user, transcription=SAMPLE_TRANSCRIPTION_JSON) client = app.test_client() try: with patch("src.services.llm.call_llm_completion", side_effect=RuntimeError("LLM down")), \ patch("src.models.system.SystemSetting") as mock_ss: mock_ss.get_setting.return_value = 30000 resp = client.post(f"/api/v1/recordings/{rec.id}/speakers/identify", headers={"X-API-Token": token}) assert resp.status_code == 500, f"Expected 500, got {resp.status_code}" return True finally: _cleanup(rec, token_rec) if cu: _cleanup(user) # ========================================================================= # Group 3: PUT /settings/auto-summarization (5 tests) # ========================================================================= def test_auto_summarization_no_auth(): """No token -> 302.""" with app.app_context(): client = app.test_client() resp = client.put("/api/v1/settings/auto-summarization", json={"enabled": True}) assert resp.status_code in (302, 401), f"Expected 302/401, got {resp.status_code}" return True def test_auto_summarization_missing_enabled(): """Body {} -> 400.""" with app.app_context(): user, cu = _get_or_create_test_user() token_rec, token = _create_api_token(user) client = app.test_client() try: resp = client.put("/api/v1/settings/auto-summarization", headers={"X-API-Token": token}, json={}) assert resp.status_code == 400, f"Expected 400, got {resp.status_code}" body = resp.get_json() assert "enabled" in body.get("error", "").lower(), f"Unexpected: {body}" return True finally: _cleanup(token_rec) if cu: _cleanup(user) def test_auto_summarization_invalid_json(): """Non-JSON body -> 400.""" with app.app_context(): user, cu = _get_or_create_test_user() token_rec, token = _create_api_token(user) client = app.test_client() try: resp = client.put("/api/v1/settings/auto-summarization", headers={"X-API-Token": token, "Content-Type": "application/json"}, data="not valid json") assert resp.status_code == 400, f"Expected 400, got {resp.status_code}" return True finally: _cleanup(token_rec) if cu: _cleanup(user) def test_auto_summarization_enable(): """enabled: true -> updates user, returns true.""" with app.app_context(): user, cu = _get_or_create_test_user() user.auto_summarization = False db.session.commit() token_rec, token = _create_api_token(user) client = app.test_client() try: resp = client.put("/api/v1/settings/auto-summarization", headers={"X-API-Token": token}, json={"enabled": True}) assert resp.status_code == 200, f"Expected 200, got {resp.status_code}" body = resp.get_json() assert body.get("auto_summarization") is True db.session.refresh(user) assert user.auto_summarization is True return True finally: _cleanup(token_rec) if cu: _cleanup(user) def test_auto_summarization_disable(): """enabled: false -> updates user, returns false.""" with app.app_context(): user, cu = _get_or_create_test_user() user.auto_summarization = True db.session.commit() token_rec, token = _create_api_token(user) client = app.test_client() try: resp = client.put("/api/v1/settings/auto-summarization", headers={"X-API-Token": token}, json={"enabled": False}) assert resp.status_code == 200, f"Expected 200, got {resp.status_code}" body = resp.get_json() assert body.get("auto_summarization") is False db.session.refresh(user) assert user.auto_summarization is False return True finally: _cleanup(token_rec) if cu: _cleanup(user) # ========================================================================= # Group 4: Regression tests (2 tests) # ========================================================================= def test_regression_get_speakers_list(): """GET /speakers still returns user's speakers.""" with app.app_context(): user, cu = _get_or_create_test_user() token_rec, token = _create_api_token(user) speaker = _create_test_speaker(user, "Regression Speaker") client = app.test_client() try: resp = client.get("/api/v1/speakers", headers={"X-API-Token": token}) assert resp.status_code == 200, f"Expected 200, got {resp.status_code}" body = resp.get_json() names = [s["name"] for s in body.get("speakers", [])] assert "Regression Speaker" in names, f"Speaker not found: {names}" return True finally: _cleanup(speaker, token_rec) if cu: _cleanup(user) def test_regression_get_recording_speakers(): """GET /recordings//speakers still returns transcript speakers.""" with app.app_context(): user, cu = _get_or_create_test_user() token_rec, token = _create_api_token(user) rec = _create_test_recording(user, transcription=SAMPLE_TRANSCRIPTION_JSON) client = app.test_client() try: with patch("src.services.speaker_embedding_matcher.find_matching_speakers", return_value={}): resp = client.get(f"/api/v1/recordings/{rec.id}/speakers", headers={"X-API-Token": token}) assert resp.status_code == 200, f"Expected 200, got {resp.status_code}" body = resp.get_json() labels = [s["label"] for s in body.get("speakers", [])] assert "SPEAKER_00" in labels and "SPEAKER_01" in labels, f"Labels: {labels}" return True finally: _cleanup(rec, token_rec) if cu: _cleanup(user) # ========================================================================= # Runner # ========================================================================= ALL_TESTS = [ # Group 1: assign test_assign_no_auth, test_assign_recording_not_found, test_assign_wrong_user_recording, test_assign_missing_speaker_map, test_assign_invalid_speaker_map_type, test_assign_string_value_json_transcript, test_assign_object_value_with_name, test_assign_is_me_flag_with_user_name, test_assign_is_me_flag_without_user_name, test_assign_plain_text_transcript, test_assign_speaker_xx_filtered_from_participants, test_assign_invalid_value_type, test_assign_empty_speaker_map, test_assign_regenerate_summary, test_assign_embeddings_updated, test_assign_no_transcription, test_assign_whitespace_name_trimmed, # Group 2: identify test_identify_no_auth, test_identify_recording_not_found, test_identify_wrong_user_recording, test_identify_no_transcription, test_identify_non_json_transcription, test_identify_json_but_not_list, test_identify_happy_path, test_identify_post_processing_unknown_values, test_identify_no_speakers_in_transcript, test_identify_llm_error, # Group 3: auto-summarization test_auto_summarization_no_auth, test_auto_summarization_missing_enabled, test_auto_summarization_invalid_json, test_auto_summarization_enable, test_auto_summarization_disable, # Group 4: regression test_regression_get_speakers_list, test_regression_get_recording_speakers, ] def main(): print(f"Running {len(ALL_TESTS)} Speaker API tests...\n") passed = 0 failed = 0 errors = [] for test_fn in ALL_TESTS: name = test_fn.__name__ try: result = test_fn() if result: print(f" PASS {name}") passed += 1 else: print(f" FAIL {name} (returned False)") failed += 1 errors.append(name) except Exception as e: print(f" ERROR {name}: {e}") failed += 1 errors.append(name) print(f"\n{'=' * 60}") print(f"Results: {passed} passed, {failed} failed out of {len(ALL_TESTS)}") if errors: print("Failed tests:") for e in errors: print(f" - {e}") print('=' * 60) sys.exit(0 if failed == 0 else 1) if __name__ == "__main__": main()