""" The AI feature flags have to be an enforceable statement about the system, not just about the ingest path (server-26#75, server-26#76). Three defects motivate these tests: #75 Correlation read the raw global config/ai_features flag instead of the per-system resolution, so a system that had opted out via its own ai_flags still correlated -- with empty tags, down the thin/recency path, blindly attaching to whatever incident was most recent. #76 Transcript correction and the transcript-PATCH extraction path checked no Firestore flag at all, so "AI is off" still spent money. Plus the destructive half of PATCH /calls/{id}/transcript, which wipes a call's intelligence fields on the promise that re-extraction rebuilds them. Firestore and the lazily-imported pipeline modules are fully mocked; the functions are called directly rather than through FastAPI. """ import pytest from unittest.mock import AsyncMock, MagicMock, patch from fastapi import HTTPException from app.routers import upload, calls from app.internal import summarizer, transcription ALL_ON = { "stt_enabled": True, "correlation_enabled": True, "summaries_enabled": True, "vocabulary_learning_enabled": True, "transcript_correction_enabled": True, } def _flags(**overrides): return {**ALL_ON, **overrides} def _system(ai_flags): return {"system_id": "sys-1", "ai_flags": ai_flags or {}} def _patch_flags(global_flags, system_ai_flags): """Patch the two reads resolve_flags() makes: the global doc and the system doc.""" return ( patch("app.internal.feature_flags.get_flags", AsyncMock(return_value=global_flags)), patch("app.internal.firestore.doc_get_cached", AsyncMock(return_value=_system(system_ai_flags))), ) # -------------------------------------------------------------------------- # resolve_flags: global master off beats everything, system false beats # global true, absent system key inherits global. # -------------------------------------------------------------------------- @pytest.mark.parametrize( "global_on, system_ai_flags, expected", [ (True, {"correlation_enabled": False}, False), # #75: system opt-out holds (True, {}, True), # absent -> inherit global (True, {"correlation_enabled": True}, True), (False, {"correlation_enabled": True}, False), # global is the master switch (False, {}, False), ], ) @pytest.mark.asyncio async def test_resolve_flags_precedence(global_on, system_ai_flags, expected): g, sysdoc = _patch_flags(_flags(correlation_enabled=global_on), system_ai_flags) with g, sysdoc: _, flag = await upload._resolve_flags("sys-1") assert flag("correlation_enabled") is expected @pytest.mark.asyncio async def test_resolve_flags_without_a_system_id_does_not_read_the_system_doc(): with patch("app.internal.feature_flags.get_flags", AsyncMock(return_value=_flags())), \ patch("app.internal.firestore.doc_get_cached", AsyncMock()) as cached: _, flag = await upload._resolve_flags(None) assert flag("correlation_enabled") is True cached.assert_not_awaited() # -------------------------------------------------------------------------- # The ingest path. This is the exact shape of #75: with the global on and the # system opted out, extraction was skipped but the empty-scenes fallback still # ran, correlating the call with no tags and attaching it to whatever incident # was most recent on that system. # -------------------------------------------------------------------------- async def _run_ingest(global_correlation, system_ai_flags): g, sysdoc = _patch_flags( _flags(correlation_enabled=global_correlation), system_ai_flags ) with g, sysdoc, \ patch.object(upload, "fstore") as fs, \ patch.object(upload, "_correlate_with_consensus", AsyncMock(return_value=None)) as corr, \ patch("app.internal.transcription.transcribe_call", AsyncMock(return_value=("units respond to main street", []))), \ patch("app.internal.intelligence.extract_scenes", AsyncMock(return_value=[])) as scenes, \ patch("app.internal.alerter.check_and_dispatch", AsyncMock()): fs.doc_get = AsyncMock(return_value={}) fs.doc_set = AsyncMock() await upload._run_intelligence_pipeline( call_id="call-1", node_id="node-1", system_id="sys-1", talkgroup_id=101, talkgroup_name="PD Dispatch", gcs_uri="gs://bucket/call-1.mp3", ) return scenes, corr @pytest.mark.asyncio async def test_per_system_opt_out_blocks_the_blind_recency_fallback_too(): scenes, corr = await _run_ingest(True, {"correlation_enabled": False}) scenes.assert_not_awaited() # The regression that mattered: the no-scenes fallback correlating on empty tags. corr.assert_not_awaited() @pytest.mark.asyncio async def test_ingest_correlates_when_the_system_has_not_opted_out(): scenes, corr = await _run_ingest(True, {}) scenes.assert_awaited_once() corr.assert_awaited_once() # -------------------------------------------------------------------------- # _run_extraction_pipeline -- the transcript-PATCH path (#76). # -------------------------------------------------------------------------- async def _run_extraction(global_correlation, system_ai_flags=None): g, sysdoc = _patch_flags( _flags(correlation_enabled=global_correlation), system_ai_flags ) with g, sysdoc, \ patch.object(upload, "fstore") as fs, \ patch("app.internal.intelligence.extract_scenes", AsyncMock(return_value=[])) as scenes, \ patch("app.internal.alerter.check_and_dispatch", AsyncMock()) as alert: fs.doc_set = AsyncMock() await upload._run_extraction_pipeline( call_id="call-1", node_id="node-1", system_id="sys-1", talkgroup_id=101, talkgroup_name="PD Dispatch", transcript="units respond to main street", ) return scenes, alert, fs @pytest.mark.asyncio async def test_extraction_does_not_spend_when_correlation_is_off(): scenes, alert, fs = await _run_extraction(False) scenes.assert_not_awaited() # No incidents produced, so nothing may be stamped onto the call doc. fs.doc_set.assert_not_awaited() # Alerting is rule-based and free -- it still runs. alert.assert_awaited_once() @pytest.mark.asyncio async def test_extraction_respects_a_per_system_opt_out(): scenes, _alert, _fs = await _run_extraction(True, {"correlation_enabled": False}) scenes.assert_not_awaited() @pytest.mark.asyncio async def test_extraction_runs_when_the_flag_is_on(): scenes, _alert, _fs = await _run_extraction(True) scenes.assert_awaited_once() # -------------------------------------------------------------------------- # PATCH /calls/{id}/transcript is destructive before it is constructive. # With correlation off it must refuse rather than blank the call out. # -------------------------------------------------------------------------- @pytest.mark.asyncio async def test_transcript_patch_refuses_when_correlation_is_off(): g, sysdoc = _patch_flags(_flags(correlation_enabled=False), {}) with g, sysdoc, patch.object(calls, "fstore") as fs: fs.doc_get = AsyncMock(return_value={"call_id": "call-1", "system_id": "sys-1"}) fs.doc_set = AsyncMock() with pytest.raises(HTTPException) as exc: await calls.patch_transcript( call_id="call-1", body=MagicMock(transcript="corrected text"), background_tasks=MagicMock(), _={}, ) assert exc.value.status_code == 409 # The refusal has to land before the first write, or the call is already ruined. fs.doc_set.assert_not_awaited() # -------------------------------------------------------------------------- # Transcript correction is a second model call plus a Places lookup (#76). # -------------------------------------------------------------------------- @pytest.mark.asyncio async def test_transcript_correction_is_skipped_when_its_flag_is_off(): g, sysdoc = _patch_flags(_flags(transcript_correction_enabled=False), {}) with g, sysdoc, \ patch.object(transcription, "fstore") as fs, \ patch.object(transcription, "ai_health") as health, \ patch.object(transcription, "transcript_correction") as tc, \ patch("asyncio.to_thread", AsyncMock(return_value=("units respond", [], False))): fs.doc_set = AsyncMock() health.report_healthy = AsyncMock() health.report_failure = AsyncMock() tc.correct = AsyncMock() await transcription.transcribe_call( "call-1", "gs://bucket/call-1.mp3", "PD Dispatch", system_id="sys-1" ) tc.correct.assert_not_awaited() # -------------------------------------------------------------------------- # Summarizer: the flag guards model spend, not the free Firestore sweep. # -------------------------------------------------------------------------- @pytest.mark.asyncio async def test_summarize_incident_is_a_no_op_when_summaries_are_off(): with patch("app.internal.feature_flags.get_flags", AsyncMock(return_value=_flags(summaries_enabled=False))), \ patch.object(summarizer, "fstore") as fs, \ patch.object(summarizer, "_sync_summarize") as sync: fs.doc_get = AsyncMock() fs.doc_set = AsyncMock() await summarizer._summarize_incident( {"incident_id": "inc-1", "call_ids": ["call-1"]} ) sync.assert_not_called() fs.doc_get.assert_not_awaited() fs.doc_set.assert_not_awaited()