Logan CusanoandClaude Opus 5.5
7ecfca7648
transcript_correction now also discards a correction that drops or changes a bare number the recogniser heard (the live "7" -> "ShotSpotter" case) or grows the transcript past max(4 words, 25%). place_verifier.verify() derives an anchor from the talkgroup's own name plus the receiving node's state when no area is configured; no state means no anchor, and a configured place without an anchor keeps the load-bearing skip. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
312 lines
13 KiB
Python
312 lines
13 KiB
Python
"""
|
|
Unit tests for the transcript correction pass (server-26#36).
|
|
|
|
Two properties carry real risk and are pinned hardest here:
|
|
|
|
* SCOPE RESOLUTION — talkgroup reference data must rank ABOVE system data.
|
|
A system spanning several counties can have a talkgroup covering one
|
|
municipality, and burying that municipality's streets under a county-wide
|
|
list is the failure this whole feature exists to avoid.
|
|
|
|
* SEGMENT ALIGNMENT — scene extraction maps scenes to transmissions by index
|
|
(segment_indices), so a corrected array of the wrong length would silently
|
|
attribute the wrong audio to a scene. Anything but an exact 1:1 match must
|
|
be discarded whole.
|
|
"""
|
|
import pytest
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from app.internal import transcript_correction as tc
|
|
|
|
SYSTEM = {
|
|
"vocabulary": ["Croton-Harmon", "Metro-North"],
|
|
"ten_codes": {"10-4": "acknowledged", "10-13": "officer needs assistance"},
|
|
"area_context": {
|
|
"county": "Westchester",
|
|
"state": "New York",
|
|
"local_knowledge": [
|
|
{"term": "Route 9", "meaning": "north-south state highway"},
|
|
{"term": "Saw Mill Parkway"},
|
|
],
|
|
},
|
|
"config": {
|
|
"talkgroups": [
|
|
{
|
|
"id": 9048,
|
|
"name": "Ossining - Police Dispatch",
|
|
"vocabulary": ["Snowden Avenue", "Croton-Harmon"],
|
|
"area_context": {
|
|
"municipality": "Ossining",
|
|
"local_knowledge": [{"term": "Sing Sing", "meaning": "state prison"}],
|
|
},
|
|
},
|
|
{"id": 9600, "name": "Harrison - Police/EMS Dispatch"},
|
|
{"id": 9563, "ten_codes": {"10-4": "on scene"}},
|
|
]
|
|
},
|
|
}
|
|
|
|
|
|
def _system(doc=SYSTEM):
|
|
return patch.object(tc.fstore, "doc_get_cached", AsyncMock(return_value=doc))
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _api_key():
|
|
"""
|
|
The dev venv has no GEMINI_API_KEY, and correct() returns early without one
|
|
— which would make every assertion below pass for the wrong reason.
|
|
"""
|
|
with patch.object(tc.settings, "gemini_api_key", "test-key"):
|
|
yield
|
|
|
|
|
|
# ── Scope resolution ────────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_talkgroup_vocabulary_ranks_above_system():
|
|
with _system():
|
|
ctx = await tc.resolve_context("sys-1", 9048)
|
|
assert ctx["vocabulary"][0] == "Snowden Avenue", "talkgroup terms must come first"
|
|
assert "Metro-North" in ctx["vocabulary"], "system terms are still inherited"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_duplicate_terms_are_not_repeated():
|
|
"""Croton-Harmon is on both scopes; it should appear once, at talkgroup rank."""
|
|
with _system():
|
|
ctx = await tc.resolve_context("sys-1", 9048)
|
|
assert [t.lower() for t in ctx["vocabulary"]].count("croton-harmon") == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_talkgroup_area_precedes_system_area():
|
|
with _system():
|
|
ctx = await tc.resolve_context("sys-1", 9048)
|
|
joined = "\n".join(ctx["area_lines"])
|
|
assert joined.index("Ossining") < joined.index("Westchester")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_talkgroup_without_own_data_inherits_system():
|
|
with _system():
|
|
ctx = await tc.resolve_context("sys-1", 9600)
|
|
assert ctx["vocabulary"] == ["Croton-Harmon", "Metro-North"]
|
|
assert any("Westchester" in line for line in ctx["area_lines"])
|
|
assert ctx["area"].get("municipality") is None, "inherits, invents nothing"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_talkgroup_ten_code_overrides_system_meaning():
|
|
with _system():
|
|
ctx = await tc.resolve_context("sys-1", 9563)
|
|
assert ctx["ten_codes"]["10-4"] == "on scene"
|
|
assert ctx["ten_codes"]["10-13"] == "officer needs assistance"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("system_id, tgid", [(None, 9048), ("sys-1", None)])
|
|
async def test_missing_scope_is_not_an_error(system_id, tgid):
|
|
with _system():
|
|
ctx = await tc.resolve_context(system_id, tgid)
|
|
assert isinstance(ctx["vocabulary"], list)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unconfigured_system_yields_empty_context():
|
|
with _system(doc=None):
|
|
ctx = await tc.resolve_context("sys-1", 9048)
|
|
assert ctx == {
|
|
"vocabulary": [], "ten_codes": {}, "area_lines": [],
|
|
"area": {}, "system_area": {}, "tg_area": {},
|
|
}
|
|
|
|
|
|
# ── Correction behaviour ────────────────────────────────────────────────────
|
|
|
|
def _gemini(payload):
|
|
return patch.object(tc, "_sync_gemini", lambda model, prompt: payload)
|
|
|
|
|
|
SEGS = [{"start": 0.0, "end": 1.0, "text": "Headquarters, 11-9."},
|
|
{"start": 1.0, "end": 2.0, "text": "Shout out to Optum."},
|
|
{"start": 2.0, "end": 3.0, "text": "360 north, back to Rose."}]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_short_transcript_is_never_sent():
|
|
"""9 of 29 calls in the sample window were <=3 words. Nothing to correct."""
|
|
with patch.object(tc, "_sync_gemini") as m:
|
|
out = await tc.correct("c1", "10-4.", None, system_id="sys-1")
|
|
assert out == (None, None, False)
|
|
m.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disabled_by_setting():
|
|
with patch.object(tc.settings, "transcript_correction_enabled", False), \
|
|
patch.object(tc, "_sync_gemini") as m:
|
|
assert await tc.correct("c1", "a b c d e", None) == (None, None, False)
|
|
m.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_segments_corrected_when_lengths_match():
|
|
payload = {"corrected": "Headquarters, 11-9. Show it out to Ossining. 360 north, back to Route 9.",
|
|
"segments": ["Headquarters, 11-9.", "Show it out to Ossining.", "360 north, back to Route 9."]}
|
|
with _system(), _gemini(payload):
|
|
text, segs, not_speech = await tc.correct("c1", "x y z w", SEGS, system_id="sys-1", talkgroup_id=9048)
|
|
assert not_speech is False
|
|
assert segs is not None and len(segs) == 3
|
|
assert segs[1]["text"] == "Show it out to Ossining."
|
|
assert segs[1]["start"] == 1.0, "timing must survive correction untouched"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wrong_segment_count_is_discarded_whole():
|
|
"""A short array would silently misattribute audio to the wrong scene."""
|
|
payload = {"corrected": "fine", "segments": ["only", "two"]}
|
|
with _system(), _gemini(payload):
|
|
text, segs, _ = await tc.correct("c1", "x y z w", SEGS, system_id="sys-1")
|
|
assert segs is None
|
|
assert text == "fine", "the joined correction still stands"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_non_string_segment_entries_are_discarded():
|
|
payload = {"corrected": None, "segments": ["ok", 42, "ok"]}
|
|
with _system(), _gemini(payload):
|
|
_, segs, _ = await tc.correct("c1", "x y z w", SEGS, system_id="sys-1")
|
|
assert segs is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unchanged_segments_report_no_correction():
|
|
payload = {"corrected": None, "segments": [s["text"] for s in SEGS]}
|
|
with _system(), _gemini(payload):
|
|
text, segs, _ = await tc.correct("c1", "x y z w", SEGS, system_id="sys-1")
|
|
assert (text, segs) == (None, None)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_echoed_transcript_counts_as_no_change():
|
|
with _system(), _gemini({"corrected": " x y z w "}):
|
|
text, _, _ = await tc.correct("c1", "x y z w", None, system_id="sys-1")
|
|
assert text is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_not_speech_is_surfaced():
|
|
with _system(), _gemini({"corrected": None, "not_speech": True}):
|
|
_, _, not_speech = await tc.correct("c1", "10-11. 10-12. 10-13. 10-14.", None, system_id="sys-1")
|
|
assert not_speech is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_model_failure_leaves_the_transcript_alone():
|
|
"""Correction is an improvement, never a dependency."""
|
|
def boom(model, prompt):
|
|
raise RuntimeError("gemini exploded")
|
|
with _system(), patch.object(tc, "_sync_gemini", boom):
|
|
assert await tc.correct("c1", "x y z w", SEGS, system_id="sys-1") == (None, None, False)
|
|
|
|
|
|
# ── Code-token guard (server-26#162) ────────────────────────────────────────
|
|
# Caught live: the same call came back with "10-7" rewritten to "10-13" in one
|
|
# place and "10-4" in another. A real code swapped for a different real code
|
|
# reads exactly as trustworthy as a correct one — worse than leaving the raw
|
|
# mishearing in place, since nothing downstream can tell it happened.
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_changed_ten_code_is_discarded():
|
|
payload = {"corrected": "10-13, we're back in town."}
|
|
with _system(), _gemini(payload):
|
|
text, _, _ = await tc.correct("c1", "10-7, we're back in town.", None, system_id="sys-1")
|
|
assert text is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invented_code_token_is_discarded():
|
|
"""Nothing code-shaped in the original — the model added one from nothing."""
|
|
payload = {"corrected": "ShotSpotter, 10-4, group of 3 shooting outside."}
|
|
with _system(), _gemini(payload):
|
|
text, _, _ = await tc.correct("c1", "Seven, group of 3 shooting outside.", None, system_id="sys-1")
|
|
assert text is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_legitimate_place_correction_with_unchanged_codes_still_applies():
|
|
"""The guard must not collateral-damage a correction that never touches
|
|
a code token — Home/Forest for Holmes/4th-and-Rowe is exactly the kind of
|
|
fix this pass exists to make."""
|
|
payload = {"corrected": "10-13 coming over on Home Street and Forest Ave, 4-2."}
|
|
with _system(), _gemini(payload):
|
|
text, _, _ = await tc.correct(
|
|
"c1", "10-13 coming over on Holmes Street and 4th and Rowe, 4-2.",
|
|
None, system_id="sys-1",
|
|
)
|
|
assert text == "10-13 coming over on Home Street and Forest Ave, 4-2."
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_segment_code_change_discards_segments_only():
|
|
"""A code change in one segment discards the whole segments array (same
|
|
all-or-nothing rule as a length mismatch), but the independently-checked
|
|
joined correction still stands if it kept its own codes intact. The
|
|
joined `text`/`corrected` pair here is deliberately code-free — this test
|
|
isolates the segment-level guard, not the joined-text one."""
|
|
payload = {
|
|
"corrected": "Show it out to Ossining, back to Route 9.",
|
|
"segments": ["Headquarters, 10-13.", "Show it out to Ossining.", "360 north, back to Route 9."],
|
|
}
|
|
with _system(), _gemini(payload):
|
|
# A realistic joined original (code-free, like the correction): a
|
|
# placeholder like "x y z w" now trips the server-26#162 growth guard.
|
|
text, segs, _ = await tc.correct("c1", "Shout out to Optum, back to Rose.", SEGS,
|
|
system_id="sys-1", talkgroup_id=9048)
|
|
assert segs is None
|
|
assert text == "Show it out to Ossining, back to Route 9."
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reference_data_reaches_the_prompt():
|
|
seen = {}
|
|
def capture(model, prompt):
|
|
seen["prompt"] = prompt
|
|
return {"corrected": None}
|
|
with _system(), patch.object(tc, "_sync_gemini", capture):
|
|
await tc.correct("c1", "x y z w", None, system_id="sys-1",
|
|
talkgroup_id=9048, talkgroup_name="Ossining - Police Dispatch")
|
|
p = seen["prompt"]
|
|
assert "Snowden Avenue" in p and "Ossining - Police Dispatch" in p
|
|
assert "Sing Sing — state prison" in p, "a term without its meaning is half the information"
|
|
assert "Ossining, Westchester, New York" in p, "state must reach the prompt (server-26#36)"
|
|
assert "10-13=officer needs assistance" in p
|
|
|
|
|
|
# ── server-26#162, the rest: bare numbers and invented words ────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_bare_number_turned_into_a_word_is_rejected():
|
|
"""The live case: a bare "7" came back as "ShotSpotter"."""
|
|
payload = {"corrected": "Car ShotSpotter on Home Street and Forest Ave."}
|
|
with _system(), _gemini(payload):
|
|
text, _, _ = await tc.correct("c1", "Car 7 on Holmes Street and 4th and Rowe.", None, system_id="sys-1")
|
|
assert text is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_place_fix_that_keeps_its_numbers_still_passes():
|
|
payload = {"corrected": "Car 7 on Home Street and Forest Ave."}
|
|
with _system(), _gemini(payload):
|
|
text, _, _ = await tc.correct("c1", "Car 7 on Holmes Street and 4th and Rowe.", None, system_id="sys-1")
|
|
assert text == "Car 7 on Home Street and Forest Ave."
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_correction_that_writes_new_sentences_is_rejected():
|
|
payload = {"corrected": "Car 7 on Main Street. Suspect is armed and fled northbound in a red sedan."}
|
|
with _system(), _gemini(payload):
|
|
text, _, _ = await tc.correct("c1", "Car 7 on Mane Street.", None, system_id="sys-1")
|
|
assert text is None
|