Logan Cusano
d6dfe5a293
The control channel was wrong in both directions. Grants fire 0.84-1.62s
before anyone speaks, and srcaddr can drop to 0 while someone is still
talking - one recording came back "-1.61s lead, -0.00s tail", the trim
finding nothing to remove because the window had closed on live speech.
Confirmed by ear: the cut lands at a word boundary on an unfinished word.
Audio is ground truth for WHEN. The console remains the only source of
WHO, so it still supplies talkgroup, alias and rid.
START voice onset in the captured audio, with a 0.25s pre-roll that
now covers only chunk quantisation and threshold ramp-up rather
than a variable control-channel offset.
STOP call_silence_timeout seconds of silence heard in the audio.
LABEL resolved AT CLOSE from a bounded rolling history of console
observations overlapping the window, +4s/-2s, because there is
no guaranteed ordering between a grant and its audio.
SPLIT a console talkgroup change still forces a cut, since two calls
with no silence between them would otherwise merge into one.
Capture now emits raw PCM instead of MP3. Silence detection becomes
integer arithmetic per chunk with no decode, trimming becomes a byte
offset slice rather than a second ffmpeg pass, and MP3 encoding happens
exactly once at save - uploads are no longer double-encoded.
Audio with no talkgroup anywhere in its window is discarded rather than
uploaded: an untagged call silently poisons incident correlation, which
is worse than losing the audio. Logged at ERROR and counted on
/api/status.
When capture produces no audio at all the old console state machine
still runs, so a node with a broken audio path keeps reporting radio
activity. That is now the only consumer of call_idle_timeout.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1150 lines
41 KiB
Python
1150 lines
41 KiB
Python
"""
|
|
Unit tests for the MetadataWatcher segmentation state machine.
|
|
|
|
TWO MODES, both covered here:
|
|
|
|
AUDIO MODE (production) — a recording STARTS at voice onset heard in the
|
|
captured audio and STOPS after settings.call_silence_timeout seconds of
|
|
silence heard in the same audio. The OP25 console supplies only the LABEL
|
|
(talkgroup/alias/rid), resolved at CLOSE time from a rolling history, and the
|
|
forced SPLIT when the talkgroup changes with no silence between calls.
|
|
|
|
CONSOLE FALLBACK (capture down) — the older state machine: `call_log` grants
|
|
start segments, the srcaddr != 0 -> 0 edge plus call_idle_timeout ends them.
|
|
Tests that wire no audio provider exercise this path, which is exactly the
|
|
behaviour a node falls back to when PulseAudio is not producing audio.
|
|
|
|
All OP25 HTTP calls are mocked — no running services required.
|
|
"""
|
|
import pytest
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from app.config import settings
|
|
from app.internal.call_recorder import AudioActivity
|
|
from app.internal.metadata_watcher import (
|
|
ATTRIBUTION_LOOKAHEAD_SECONDS,
|
|
ATTRIBUTION_LOOKBACK_SECONDS,
|
|
MAX_SEGMENT_SECONDS,
|
|
MetadataWatcher,
|
|
OP25_OFFLINE_GRACE,
|
|
)
|
|
from app.internal.op25_client import TerminalUpdate, parse_terminal_messages
|
|
|
|
|
|
class FakeClock:
|
|
"""Manually advanced clock so idle timeouts are testable without sleeping."""
|
|
|
|
def __init__(self, start: float = 1_700_000_000.0):
|
|
self.now = start
|
|
|
|
def __call__(self) -> float:
|
|
return self.now
|
|
|
|
def advance(self, seconds: float) -> float:
|
|
self.now += seconds
|
|
return self.now
|
|
|
|
|
|
@pytest.fixture
|
|
def clock():
|
|
return FakeClock()
|
|
|
|
|
|
@pytest.fixture
|
|
def watcher(clock):
|
|
"""Console fallback mode: no audio provider wired, capture assumed down."""
|
|
w = MetadataWatcher()
|
|
w._clock = clock
|
|
w.on_call_start = AsyncMock()
|
|
w.on_call_end = AsyncMock()
|
|
return w
|
|
|
|
|
|
class FakeAudio:
|
|
"""
|
|
Stands in for CallRecorder.audio_activity.
|
|
|
|
Mirrors the recorder's own rule for what starts a new voice RUN: a
|
|
non-silent chunk more than settings.call_silence_timeout after the previous
|
|
one. Tests drive it with speak()/quiet() instead of synthesising PCM, so the
|
|
segmentation logic is tested independently of the detector.
|
|
"""
|
|
|
|
def __init__(self, clock):
|
|
self.clock = clock
|
|
self.capturing = True
|
|
self.recording = False
|
|
self.last_voice = None
|
|
self.onset = None
|
|
|
|
def speak(self, at=None):
|
|
"""Mark voice heard now (or at `at`)."""
|
|
moment = self.clock.now if at is None else at
|
|
if self.last_voice is None or (moment - self.last_voice) >= settings.call_silence_timeout:
|
|
self.onset = moment
|
|
self.last_voice = moment
|
|
return moment
|
|
|
|
def __call__(self) -> AudioActivity:
|
|
silence = 0.0 if self.last_voice is None else max(0.0, self.clock.now - self.last_voice)
|
|
return AudioActivity(
|
|
capturing=self.capturing,
|
|
recording=self.recording,
|
|
last_voice_epoch=self.last_voice,
|
|
voice_onset_epoch=self.onset,
|
|
silence_seconds=silence,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def audio(clock):
|
|
return FakeAudio(clock)
|
|
|
|
|
|
@pytest.fixture
|
|
def hearing(clock, audio):
|
|
"""Audio mode: the production wiring, with a controllable audio stream."""
|
|
w = MetadataWatcher()
|
|
w._clock = clock
|
|
w.on_call_start = AsyncMock()
|
|
w.on_call_end = AsyncMock()
|
|
w.audio_activity = audio
|
|
return w
|
|
|
|
|
|
def grant(tgid: int, time_: float, tgtag: str = "", rid: int = 101, freq: int = 851_000_000):
|
|
"""One OP25 call_log entry (see tk_p25.log_call)."""
|
|
return {
|
|
"time": time_,
|
|
"sysid": 1,
|
|
"rcvr": 0,
|
|
"freq": freq,
|
|
"slot": None,
|
|
"prio": 0,
|
|
"tgid": tgid,
|
|
"tgtag": tgtag,
|
|
"rid": rid,
|
|
"rtag": "",
|
|
}
|
|
|
|
|
|
def channel(tgid: int = 0, srcaddr: int = 0, tag: str = "", hold_tgid: int = 0):
|
|
"""One OP25 channel_update channel dict (see tk_p25.get_chan_status)."""
|
|
return {
|
|
"freq": 851_000_000,
|
|
"tgid": tgid or None,
|
|
"tag": tag,
|
|
"srcaddr": srcaddr,
|
|
"svcopts": bool(srcaddr),
|
|
"hold_tgid": hold_tgid or None,
|
|
"encrypted": 0,
|
|
"emergency": 0,
|
|
}
|
|
|
|
|
|
def update(call_log=None, channels=None) -> TerminalUpdate:
|
|
return TerminalUpdate(channels=list(channels or []), call_log=list(call_log or []))
|
|
|
|
|
|
def patched(result):
|
|
return patch(
|
|
"app.internal.metadata_watcher.op25_client.poll_terminal",
|
|
new=AsyncMock(return_value=result),
|
|
)
|
|
|
|
|
|
async def tick(watcher, result):
|
|
with patched(result):
|
|
await watcher._tick()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# op25_client message parsing
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_parser_extracts_call_log_and_channels():
|
|
messages = [
|
|
{"json_type": "trunk_update", "0": {"whatever": 1}},
|
|
{"json_type": "channel_update", "channels": [0], "0": channel(tgid=1234, srcaddr=555)},
|
|
{"json_type": "call_log", "log": [grant(1234, 100.0)]},
|
|
]
|
|
result = parse_terminal_messages(messages)
|
|
|
|
assert len(result.channels) == 1
|
|
assert result.channels[0]["tgid"] == 1234
|
|
assert len(result.call_log) == 1
|
|
assert result.call_log[0]["time"] == 100.0
|
|
|
|
|
|
def test_parser_tolerates_garbage():
|
|
"""Unknown json_types, bare dicts and malformed entries must not raise."""
|
|
assert parse_terminal_messages(None).call_log == []
|
|
assert parse_terminal_messages("nope").channels == []
|
|
assert parse_terminal_messages([None, 5, {"json_type": "mystery"}]).channels == []
|
|
|
|
# A bare dict instead of a list.
|
|
single = parse_terminal_messages({"json_type": "call_log", "log": [grant(1, 1.0), "junk"]})
|
|
assert len(single.call_log) == 1
|
|
|
|
# channel_update naming a channel that isn't present.
|
|
missing = parse_terminal_messages([{"json_type": "channel_update", "channels": [0, 1], "0": channel(9)}])
|
|
assert len(missing.channels) == 1
|
|
|
|
|
|
def test_parser_handles_multiple_channels():
|
|
messages = [{
|
|
"json_type": "channel_update",
|
|
"channels": [0, 1],
|
|
"0": channel(tgid=1111, srcaddr=1),
|
|
"1": channel(tgid=2222, srcaddr=2),
|
|
}]
|
|
result = parse_terminal_messages(messages)
|
|
assert [c["tgid"] for c in result.channels] == [1111, 2222]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Call start — driven by call_log
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_call_log_entry_starts_call(watcher, clock):
|
|
grant_time = clock.now - 0.4 # OP25 logged it before our poll noticed
|
|
await tick(watcher, update(
|
|
call_log=[grant(1234, grant_time, tgtag="Police Dispatch")],
|
|
channels=[channel(tgid=1234, srcaddr=555)],
|
|
))
|
|
|
|
assert watcher.is_active
|
|
assert watcher.current_tgid == 1234
|
|
watcher.on_call_start.assert_called_once()
|
|
|
|
payload = watcher.on_call_start.call_args[0][0]
|
|
assert payload["tgid"] == 1234
|
|
assert payload["tgid_name"] == "Police Dispatch"
|
|
assert payload["call_id"]
|
|
# The whole point: the start is OP25's timestamp, not our detection time.
|
|
assert payload["started_at_epoch"] == grant_time
|
|
assert payload["started_at"].startswith("20")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_channel_activity_alone_does_not_start_a_call(watcher):
|
|
"""No call_log entry => no call, even with a live tgid on the channel."""
|
|
await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=555)]))
|
|
|
|
assert not watcher.is_active
|
|
watcher.on_call_start.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_call_log_entry_without_tgid_is_ignored(watcher):
|
|
await tick(watcher, update(call_log=[grant(0, 100.0)]))
|
|
|
|
assert not watcher.is_active
|
|
watcher.on_call_start.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_call_log_without_time_falls_back_to_local_clock(watcher, clock):
|
|
entry = grant(1234, 0.0)
|
|
entry.pop("time")
|
|
await tick(watcher, update(call_log=[entry]))
|
|
|
|
assert watcher.is_active
|
|
assert watcher.on_call_start.call_args[0][0]["started_at_epoch"] == clock.now
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_op25_unreachable_does_not_start_call(watcher):
|
|
await tick(watcher, None)
|
|
|
|
assert not watcher.is_active
|
|
watcher.on_call_start.assert_not_called()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Console fallback: call end — srcaddr edge + idle timeout
|
|
#
|
|
# This is the path a node uses ONLY when PulseAudio capture is not producing
|
|
# audio. It is measurably wrong (the srcaddr edge fires mid-word) but it is all
|
|
# there is when there is no audio to segment on, and it keeps the node
|
|
# reporting radio activity to C2 while the audio path is broken.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_srcaddr_edge_then_idle_timeout_ends_call(watcher, clock):
|
|
await tick(watcher, update(
|
|
call_log=[grant(1234, clock.now)],
|
|
channels=[channel(tgid=1234, srcaddr=555)],
|
|
))
|
|
assert watcher.is_active
|
|
|
|
# srcaddr resets to 0 while tgid is still held — OP25's end-of-call signal.
|
|
clock.advance(0.5)
|
|
edge_time = clock.now
|
|
await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)]))
|
|
assert watcher.is_active, "the srcaddr edge alone must not close the segment"
|
|
watcher.on_call_end.assert_not_called()
|
|
|
|
# Still quiet, but not long enough yet.
|
|
clock.advance(settings.call_idle_timeout - 0.5)
|
|
await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)]))
|
|
assert watcher.is_active
|
|
|
|
# Past the timeout.
|
|
clock.advance(1.0)
|
|
await tick(watcher, update(channels=[channel()]))
|
|
assert not watcher.is_active
|
|
|
|
watcher.on_call_end.assert_called_once()
|
|
payload = watcher.on_call_end.call_args[0][0]
|
|
assert payload["end_reason"] == "idle_timeout"
|
|
# The audio ends at the last transmission plus a short pad, NOT at "now" —
|
|
# otherwise every recording carries call_idle_timeout seconds of silence.
|
|
assert payload["ended_at_epoch"] == pytest.approx(edge_time + settings.call_tail_pad_seconds)
|
|
assert payload["tgid"] == 1234
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tail_pad_is_configurable_and_defaults_to_three_seconds(watcher, clock, monkeypatch):
|
|
"""
|
|
Field measurement showed the grant->speech offset runs ~0.84-1.62s, so a
|
|
1.0s pad let short calls' windows close before voice audio even started
|
|
(clipping mid-word). The default moved to 3.0 — and it has to be a
|
|
setting, not a magic number, so it can be tuned per node without a code
|
|
change (and so tests can prove it isn't hardcoded anywhere downstream).
|
|
"""
|
|
assert settings.call_tail_pad_seconds == 3.0
|
|
|
|
monkeypatch.setattr(settings, "call_tail_pad_seconds", 5.0)
|
|
|
|
await tick(watcher, update(
|
|
call_log=[grant(1234, clock.now)],
|
|
channels=[channel(tgid=1234, srcaddr=555)],
|
|
))
|
|
clock.advance(0.5)
|
|
edge_time = clock.now
|
|
await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)]))
|
|
|
|
# Advance well past both the idle timeout AND the monkeypatched 5.0s pad so
|
|
# the "now" cap in _handle_channels never masks the pad value under test.
|
|
clock.advance(settings.call_idle_timeout + 6.0)
|
|
await tick(watcher, update(channels=[channel()]))
|
|
|
|
payload = watcher.on_call_end.call_args[0][0]
|
|
assert payload["ended_at_epoch"] == pytest.approx(edge_time + 5.0)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_short_call_window_now_covers_delayed_voice_arrival(watcher, clock):
|
|
"""
|
|
Regression test for the truncation bug: a ~0.97s control-channel call
|
|
(grant to srcaddr-drop) previously closed its window at
|
|
last_tx_end + 1.0s pad, i.e. ~1.97s after the grant — but field
|
|
measurement shows voice audio doesn't start until ~1.5s after the grant
|
|
(0.84-1.62s measured), so the old window left as little as ~0.4s of
|
|
captured speech and clipped it mid-word.
|
|
|
|
With the 3.0s default pad, the same short call's window must extend well
|
|
past the ~1.5s point where voice actually starts.
|
|
"""
|
|
call_start = clock.now
|
|
await tick(watcher, update(
|
|
call_log=[grant(1234, call_start)],
|
|
channels=[channel(tgid=1234, srcaddr=555)],
|
|
))
|
|
|
|
# The control-channel call itself is short — under 1 second.
|
|
clock.advance(0.97)
|
|
edge_time = clock.now
|
|
await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)]))
|
|
|
|
clock.advance(settings.call_idle_timeout + 0.5)
|
|
await tick(watcher, update(channels=[channel()]))
|
|
|
|
payload = watcher.on_call_end.call_args[0][0]
|
|
assert payload["end_reason"] == "idle_timeout"
|
|
|
|
voice_arrival = call_start + 1.5 # measured grant->speech offset, typical case
|
|
assert payload["ended_at_epoch"] == pytest.approx(edge_time + settings.call_tail_pad_seconds)
|
|
assert payload["ended_at_epoch"] > voice_arrival, (
|
|
"recording window must extend past the point voice audio actually arrives, "
|
|
"not just past the control-channel call_log timestamps"
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_idle_close_logs_the_measured_control_channel_idle(watcher, clock, caplog):
|
|
"""
|
|
The correct idle timeout can only be tuned from the CONTROL-CHANNEL idle, not
|
|
from silence measured in the audio (which also contains the ~1.9s P25
|
|
grant→speech delay). So the real measured value has to reach the log.
|
|
"""
|
|
await tick(watcher, update(
|
|
call_log=[grant(1234, clock.now)],
|
|
channels=[channel(tgid=1234, srcaddr=555)],
|
|
))
|
|
clock.advance(0.5)
|
|
await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)]))
|
|
|
|
with caplog.at_level("INFO", logger="drb-edge-node"):
|
|
clock.advance(settings.call_idle_timeout + 0.25)
|
|
await tick(watcher, update(channels=[channel()]))
|
|
|
|
idle_lines = [r.message for r in caplog.records if "measured control-channel idle" in r.message]
|
|
assert idle_lines, "idle-timeout closes must log the measured idle for later tuning"
|
|
assert "3.25s" in idle_lines[0]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ongoing_transmission_never_times_out(watcher, clock):
|
|
await tick(watcher, update(
|
|
call_log=[grant(1234, clock.now)],
|
|
channels=[channel(tgid=1234, srcaddr=555)],
|
|
))
|
|
|
|
for _ in range(20):
|
|
clock.advance(settings.call_idle_timeout)
|
|
await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=555)]))
|
|
assert watcher.is_active
|
|
|
|
watcher.on_call_end.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_op25_unreachable_ends_active_call_after_grace(watcher, clock):
|
|
await tick(watcher, update(
|
|
call_log=[grant(1234, clock.now)],
|
|
channels=[channel(tgid=1234, srcaddr=555)],
|
|
))
|
|
|
|
# A single failed poll must not kill the call.
|
|
clock.advance(OP25_OFFLINE_GRACE / 2)
|
|
await tick(watcher, None)
|
|
assert watcher.is_active
|
|
|
|
clock.advance(OP25_OFFLINE_GRACE)
|
|
await tick(watcher, None)
|
|
assert not watcher.is_active
|
|
assert watcher.on_call_end.call_args[0][0]["end_reason"] == "op25_unreachable"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_grant_whose_call_already_ended(watcher, clock):
|
|
"""call_log delivers a start whose transmission is already over."""
|
|
await tick(watcher, update(
|
|
call_log=[grant(1234, clock.now - 1.2)],
|
|
channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)], # already idle
|
|
))
|
|
assert watcher.is_active
|
|
|
|
clock.advance(settings.call_idle_timeout + 0.5)
|
|
await tick(watcher, update(channels=[channel()]))
|
|
|
|
assert not watcher.is_active
|
|
watcher.on_call_end.assert_called_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_closes_open_segment(watcher, clock):
|
|
await tick(watcher, update(
|
|
call_log=[grant(1234, clock.now)],
|
|
channels=[channel(tgid=1234, srcaddr=555)],
|
|
))
|
|
await watcher.stop()
|
|
|
|
assert not watcher.is_active
|
|
assert watcher.on_call_end.call_args[0][0]["end_reason"] == "shutdown"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Segmentation: same-TGID continuation, different-TGID split
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_same_tgid_grant_continues_one_recording(watcher, clock):
|
|
"""Back-and-forth on one talkgroup must stay a single call/recording."""
|
|
start = clock.now
|
|
await tick(watcher, update(
|
|
call_log=[grant(1234, start)],
|
|
channels=[channel(tgid=1234, srcaddr=555)],
|
|
))
|
|
call_id = watcher.active_call_id
|
|
|
|
# Transmission ends...
|
|
clock.advance(1.0)
|
|
await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)]))
|
|
|
|
# ...and the other party keys up on the SAME tgid inside the idle window.
|
|
clock.advance(1.0)
|
|
await tick(watcher, update(
|
|
call_log=[grant(1234, clock.now)],
|
|
channels=[channel(tgid=1234, srcaddr=777)],
|
|
))
|
|
|
|
assert watcher.is_active
|
|
assert watcher.active_call_id == call_id, "same tgid must not open a new call"
|
|
watcher.on_call_start.assert_called_once()
|
|
watcher.on_call_end.assert_not_called()
|
|
|
|
# And the whole exchange closes as one segment.
|
|
clock.advance(1.0)
|
|
await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)]))
|
|
clock.advance(settings.call_idle_timeout + 0.5)
|
|
await tick(watcher, update(channels=[channel()]))
|
|
|
|
watcher.on_call_end.assert_called_once()
|
|
payload = watcher.on_call_end.call_args[0][0]
|
|
assert payload["call_id"] == call_id
|
|
assert payload["transmissions"] == 2
|
|
assert payload["started_at_epoch"] == start
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_different_tgid_grant_splits_recording(watcher, clock):
|
|
await tick(watcher, update(
|
|
call_log=[grant(1111, clock.now, tgtag="Fire")],
|
|
channels=[channel(tgid=1111, srcaddr=1)],
|
|
))
|
|
first_id = watcher.active_call_id
|
|
|
|
clock.advance(2.0)
|
|
split_time = clock.now
|
|
await tick(watcher, update(
|
|
call_log=[grant(2222, split_time, tgtag="EMS")],
|
|
channels=[channel(tgid=2222, srcaddr=2)],
|
|
))
|
|
|
|
assert watcher.is_active
|
|
assert watcher.current_tgid == 2222
|
|
assert watcher.active_call_id != first_id
|
|
assert watcher.on_call_start.call_count == 2
|
|
watcher.on_call_end.assert_called_once()
|
|
|
|
ended = watcher.on_call_end.call_args[0][0]
|
|
assert ended["call_id"] == first_id
|
|
assert ended["tgid"] == 1111
|
|
assert ended["end_reason"] == "tgid_change"
|
|
# The outgoing segment is padded PAST where the new one begins. The buffered
|
|
# audio lags control-channel timestamps by ~1.5s, so ending exactly at the
|
|
# split cut the outgoing call's last words. The overlap is correct — the
|
|
# audio stream really does hold one call's tail then the next call's start.
|
|
assert ended["ended_at_epoch"] == split_time + settings.call_tail_pad_seconds
|
|
assert watcher.on_call_start.call_args[0][0]["started_at_epoch"] == split_time
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_multiple_call_log_entries_in_one_poll(watcher, clock):
|
|
"""
|
|
The call_log deque is drained per poll and capped at 10, so one poll can
|
|
deliver several grants. Same-tgid ones merge, different-tgid ones split.
|
|
"""
|
|
t0 = clock.now - 1.5
|
|
await tick(watcher, update(
|
|
call_log=[
|
|
grant(1111, t0),
|
|
grant(1111, t0 + 0.4), # same tgid -> continuation
|
|
grant(2222, t0 + 0.9), # different tgid -> split
|
|
],
|
|
channels=[channel(tgid=2222, srcaddr=9)],
|
|
))
|
|
|
|
assert watcher.current_tgid == 2222
|
|
assert watcher.on_call_start.call_count == 2
|
|
watcher.on_call_end.assert_called_once()
|
|
|
|
ended = watcher.on_call_end.call_args[0][0]
|
|
assert ended["tgid"] == 1111
|
|
assert ended["transmissions"] == 2
|
|
assert ended["started_at_epoch"] == t0
|
|
# Split close is padded past the new grant — see the tgid_change test above.
|
|
assert ended["ended_at_epoch"] == t0 + 0.9 + settings.call_tail_pad_seconds
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_out_of_order_call_log_entries_are_sorted(watcher, clock):
|
|
t0 = clock.now - 2.0
|
|
await tick(watcher, update(
|
|
call_log=[grant(2222, t0 + 1.0), grant(1111, t0)],
|
|
channels=[channel(tgid=2222, srcaddr=9)],
|
|
))
|
|
|
|
ended = watcher.on_call_end.call_args[0][0]
|
|
assert ended["tgid"] == 1111
|
|
assert watcher.current_tgid == 2222
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dropped_call_log_event_still_closes_segment(watcher, clock):
|
|
"""
|
|
CALL_LOG_MAX_LEN is 10, so a slow consumer loses grants. If another talkgroup
|
|
is plainly transmitting we must close immediately rather than record it under
|
|
the wrong tgid for call_idle_timeout seconds.
|
|
"""
|
|
await tick(watcher, update(
|
|
call_log=[grant(1111, clock.now)],
|
|
channels=[channel(tgid=1111, srcaddr=1)],
|
|
))
|
|
first_id = watcher.active_call_id
|
|
|
|
clock.advance(1.0)
|
|
await tick(watcher, update(channels=[channel(tgid=3333, srcaddr=7)])) # no call_log
|
|
|
|
assert not watcher.is_active
|
|
ended = watcher.on_call_end.call_args[0][0]
|
|
assert ended["call_id"] == first_id
|
|
assert ended["end_reason"] == "tgid_change_unlogged"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_other_receiver_on_another_tgid_does_not_split(watcher, clock):
|
|
"""
|
|
The dropped-call_log fallback must not fire on multi-receiver setups: another
|
|
receiver being busy says nothing about ours, and closing on it would truncate
|
|
every call.
|
|
"""
|
|
await tick(watcher, update(
|
|
call_log=[grant(1111, clock.now)],
|
|
channels=[channel(tgid=1111, srcaddr=1)],
|
|
))
|
|
first_id = watcher.active_call_id
|
|
|
|
clock.advance(0.5)
|
|
await tick(watcher, update(channels=[
|
|
channel(tgid=1111, srcaddr=0, hold_tgid=1111),
|
|
channel(tgid=3333, srcaddr=7),
|
|
]))
|
|
|
|
assert watcher.is_active
|
|
assert watcher.active_call_id == first_id
|
|
watcher.on_call_end.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_second_receiver_activity_does_not_hold_segment_open(watcher, clock):
|
|
"""Only channels on OUR talkgroup count as our transmission."""
|
|
await tick(watcher, update(
|
|
call_log=[grant(1111, clock.now)],
|
|
channels=[channel(tgid=1111, srcaddr=1)],
|
|
))
|
|
|
|
clock.advance(0.5)
|
|
await tick(watcher, update(channels=[
|
|
channel(tgid=1111, srcaddr=0, hold_tgid=1111),
|
|
channel(tgid=1111, srcaddr=0),
|
|
]))
|
|
assert watcher.is_active
|
|
|
|
clock.advance(settings.call_idle_timeout + 0.5)
|
|
await tick(watcher, update(channels=[channel(tgid=1111, srcaddr=0, hold_tgid=1111)]))
|
|
assert not watcher.is_active
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_end_never_precedes_start(watcher, clock):
|
|
"""A clock skew or odd grant order must never produce a negative duration."""
|
|
await tick(watcher, update(
|
|
call_log=[grant(1234, clock.now + 5.0)], # OP25 stamp in the future
|
|
channels=[channel(tgid=1234, srcaddr=5)],
|
|
))
|
|
await watcher.stop()
|
|
|
|
payload = watcher.on_call_end.call_args[0][0]
|
|
assert payload["ended_at_epoch"] >= payload["started_at_epoch"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# AUDIO MODE — boundaries from the audio, label from the console
|
|
#
|
|
# This is the production path. Everything above this line is the fallback that
|
|
# only runs when PulseAudio capture is down.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_audio_onset_starts_the_recording(hearing, clock, audio):
|
|
"""Voice onset opens the segment, and the recorder is told to slice to it."""
|
|
onset = audio.speak()
|
|
await tick(hearing, update(
|
|
call_log=[grant(1234, clock.now, tgtag="Police Dispatch")],
|
|
channels=[channel(tgid=1234, srcaddr=555)],
|
|
))
|
|
|
|
assert hearing.is_active
|
|
hearing.on_call_start.assert_called_once()
|
|
payload = hearing.on_call_start.call_args[0][0]
|
|
assert payload["driver"] == "audio"
|
|
# The recorder slices the ring buffer back to THIS epoch, so it has to be
|
|
# the audio onset, not the grant and not our detection time.
|
|
assert payload["started_at_epoch"] == onset
|
|
assert payload["tgid"] == 1234
|
|
assert payload["tgid_name"] == "Police Dispatch"
|
|
assert payload["attributed"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_grant_with_no_audio_does_not_start_a_recording(hearing, clock):
|
|
"""
|
|
The grant fires 0.84-1.62s before anyone speaks. Opening on it is what put
|
|
seconds of dead air at the head of every recording.
|
|
"""
|
|
await tick(hearing, update(
|
|
call_log=[grant(1234, clock.now, tgtag="Fire")],
|
|
channels=[channel(tgid=1234, srcaddr=555)],
|
|
))
|
|
|
|
assert not hearing.is_active
|
|
hearing.on_call_start.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_recording_closes_after_the_configured_silence(hearing, clock, audio):
|
|
audio.speak()
|
|
await tick(hearing, update(
|
|
call_log=[grant(1234, clock.now)],
|
|
channels=[channel(tgid=1234, srcaddr=555)],
|
|
))
|
|
|
|
clock.advance(0.5)
|
|
last_voice = audio.speak()
|
|
await tick(hearing, update(channels=[channel(tgid=1234, srcaddr=555)]))
|
|
|
|
# Quiet, but not for long enough yet.
|
|
clock.advance(settings.call_silence_timeout - 0.5)
|
|
await tick(hearing, update(channels=[channel()]))
|
|
assert hearing.is_active
|
|
hearing.on_call_end.assert_not_called()
|
|
|
|
clock.advance(0.6)
|
|
await tick(hearing, update(channels=[channel()]))
|
|
|
|
assert not hearing.is_active
|
|
payload = hearing.on_call_end.call_args[0][0]
|
|
assert payload["end_reason"] == "audio_silence"
|
|
assert payload["tgid"] == 1234
|
|
# The audio ends where the silence run began plus the threshold, so the
|
|
# trim can measure and strip exactly that run.
|
|
assert payload["ended_at_epoch"] == pytest.approx(last_voice + settings.call_silence_timeout)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_false_srcaddr_drop_mid_speech_does_not_end_the_recording(hearing, clock, audio):
|
|
"""
|
|
THE BUG THIS REARCHITECTURE EXISTS FOR. srcaddr can reset to 0 while someone
|
|
is still talking; under the old design that started the idle timer and the
|
|
window closed on top of live speech (recording 0ff35b20: "-1.61s lead,
|
|
-0.00s tail" — nothing left to trim because the cut landed mid-word).
|
|
"""
|
|
audio.speak()
|
|
await tick(hearing, update(
|
|
call_log=[grant(1234, clock.now)],
|
|
channels=[channel(tgid=1234, srcaddr=555)],
|
|
))
|
|
call_id = hearing.active_call_id
|
|
|
|
# The control channel now says the call is over. It is wrong; the audio
|
|
# keeps arriving. This runs well past call_idle_timeout, which is what
|
|
# would have closed the segment before.
|
|
for _ in range(10):
|
|
clock.advance(0.5)
|
|
audio.speak()
|
|
await tick(hearing, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)]))
|
|
assert hearing.is_active, "a false srcaddr drop must never end a recording"
|
|
|
|
assert (clock.now - hearing._started_at) > settings.call_idle_timeout
|
|
assert hearing.active_call_id == call_id
|
|
hearing.on_call_end.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_silence_close_logs_the_measured_trailing_silence(hearing, clock, audio, caplog):
|
|
"""
|
|
call_silence_timeout can only be tuned from the real trailing silence, so
|
|
the measured number has to reach the log — the audio-mode counterpart of
|
|
the "measured control-channel idle" line.
|
|
"""
|
|
audio.speak()
|
|
await tick(hearing, update(call_log=[grant(1234, clock.now)]))
|
|
|
|
with caplog.at_level("INFO", logger="drb-edge-node"):
|
|
clock.advance(settings.call_silence_timeout + 0.25)
|
|
await tick(hearing, update(channels=[channel()]))
|
|
|
|
lines = [r.message for r in caplog.records if "measured trailing silence" in r.message]
|
|
assert lines, "audio closes must log the measured silence for later tuning"
|
|
assert "3.25s" in lines[0]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_same_tgid_grant_continues_one_audio_recording(hearing, clock, audio):
|
|
"""Back-and-forth on one talkgroup must stay a single call/recording."""
|
|
onset = audio.speak()
|
|
await tick(hearing, update(
|
|
call_log=[grant(1234, clock.now)],
|
|
channels=[channel(tgid=1234, srcaddr=555)],
|
|
))
|
|
call_id = hearing.active_call_id
|
|
|
|
# The other party keys up on the SAME tgid while audio is still flowing.
|
|
clock.advance(1.0)
|
|
audio.speak()
|
|
await tick(hearing, update(
|
|
call_log=[grant(1234, clock.now)],
|
|
channels=[channel(tgid=1234, srcaddr=777)],
|
|
))
|
|
|
|
assert hearing.active_call_id == call_id, "same tgid must not open a new call"
|
|
hearing.on_call_start.assert_called_once()
|
|
hearing.on_call_end.assert_not_called()
|
|
|
|
clock.advance(settings.call_silence_timeout + 0.5)
|
|
await tick(hearing, update(channels=[channel()]))
|
|
|
|
hearing.on_call_end.assert_called_once()
|
|
payload = hearing.on_call_end.call_args[0][0]
|
|
assert payload["call_id"] == call_id
|
|
assert payload["started_at_epoch"] == onset
|
|
assert payload["transmissions"] == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_different_tgid_splits_even_with_no_silence_between(hearing, clock, audio):
|
|
"""
|
|
Back-to-back calls on two talkgroups with no gap. Pure audio segmentation
|
|
would merge them into ONE file under ONE label, which corrupts correlation.
|
|
The console talkgroup change has to force the cut.
|
|
"""
|
|
audio.speak()
|
|
await tick(hearing, update(
|
|
call_log=[grant(1111, clock.now, tgtag="Fire")],
|
|
channels=[channel(tgid=1111, srcaddr=1)],
|
|
))
|
|
first_id = hearing.active_call_id
|
|
|
|
clock.advance(2.0)
|
|
audio.speak() # still talking — no silence anywhere in this test
|
|
split = clock.now
|
|
await tick(hearing, update(
|
|
call_log=[grant(2222, split, tgtag="EMS")],
|
|
channels=[channel(tgid=2222, srcaddr=2)],
|
|
))
|
|
|
|
hearing.on_call_end.assert_called_once()
|
|
ended = hearing.on_call_end.call_args[0][0]
|
|
assert ended["call_id"] == first_id
|
|
assert ended["tgid"] == 1111
|
|
assert ended["end_reason"] == "tgid_change"
|
|
# Padded past the split: buffered audio lags the control channel, so cutting
|
|
# at the exact grant timestamp clipped the outgoing call's last words. The
|
|
# overlap between the two slices is correct.
|
|
assert ended["ended_at_epoch"] == split + settings.call_tail_pad_seconds
|
|
|
|
assert hearing.is_active and hearing.current_tgid == 2222
|
|
assert hearing.active_call_id != first_id
|
|
assert hearing.on_call_start.call_count == 2
|
|
assert hearing.on_call_start.call_args[0][0]["started_at_epoch"] == split
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_an_unlogged_tgid_change_also_splits_and_reopens(hearing, clock, audio):
|
|
"""
|
|
OP25's call_log deque is capped at 10, so grants get dropped. If our only
|
|
receiver is plainly on another talkgroup the segment is over — and in audio
|
|
mode a new one must open immediately or the audio would be dropped on the
|
|
floor until the next voice run.
|
|
"""
|
|
audio.speak()
|
|
await tick(hearing, update(
|
|
call_log=[grant(1111, clock.now)],
|
|
channels=[channel(tgid=1111, srcaddr=1)],
|
|
))
|
|
first_id = hearing.active_call_id
|
|
|
|
clock.advance(1.0)
|
|
audio.speak()
|
|
await tick(hearing, update(channels=[channel(tgid=3333, srcaddr=7)])) # no call_log
|
|
|
|
ended = hearing.on_call_end.call_args[0][0]
|
|
assert ended["call_id"] == first_id
|
|
assert ended["end_reason"] == "tgid_change_unlogged"
|
|
assert hearing.is_active and hearing.current_tgid == 3333
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_max_length_backstop_closes_and_reopens(hearing, clock, audio):
|
|
"""
|
|
A talkgroup that never goes quiet must not produce an unbounded recording —
|
|
but the audio must not be dropped either, so the segment is immediately
|
|
reopened rather than simply abandoned.
|
|
"""
|
|
audio.speak()
|
|
await tick(hearing, update(call_log=[grant(1234, clock.now)]))
|
|
first_id = hearing.active_call_id
|
|
|
|
clock.advance(MAX_SEGMENT_SECONDS / 2)
|
|
audio.speak()
|
|
await tick(hearing, update(channels=[channel(tgid=1234, srcaddr=5)]))
|
|
assert hearing.active_call_id == first_id
|
|
|
|
clock.advance(MAX_SEGMENT_SECONDS / 2 + 1)
|
|
audio.speak()
|
|
await tick(hearing, update(channels=[channel(tgid=1234, srcaddr=5)]))
|
|
|
|
ended = hearing.on_call_end.call_args[0][0]
|
|
assert ended["call_id"] == first_id
|
|
assert ended["end_reason"] == "max_length"
|
|
assert hearing.is_active, "a still-live transmission must not be dropped at the cap"
|
|
assert hearing.active_call_id != first_id
|
|
assert hearing.on_call_start.call_count == 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Attribution: resolved at close, from a bounded rolling console history
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_orphan_audio_is_discarded_flagged_and_counted(hearing, clock, audio, caplog):
|
|
"""
|
|
Audio with no console talkgroup anywhere near it — Liquidsoap fallback, a
|
|
test tone, stray noise, a dropped call_log. It must be impossible to miss
|
|
and must never be uploaded: an untagged call poisons correlation.
|
|
"""
|
|
audio.speak()
|
|
await tick(hearing, update()) # console says nothing at all
|
|
|
|
assert hearing.is_active
|
|
started = hearing.on_call_start.call_args[0][0]
|
|
assert started["tgid"] is None
|
|
assert started["attributed"] is False
|
|
|
|
with caplog.at_level("ERROR", logger="drb-edge-node"):
|
|
clock.advance(settings.call_silence_timeout + 0.5)
|
|
await tick(hearing, update())
|
|
|
|
ended = hearing.on_call_end.call_args[0][0]
|
|
assert ended["attributed"] is False
|
|
assert ended["tgid"] is None
|
|
assert hearing.unattributed_segments == 1
|
|
assert any("ORPHAN AUDIO" in r.message for r in caplog.records), \
|
|
"unattributed audio must be loud in the logs, not silent"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_grant_before_audio_onset_still_attributes_the_recording(hearing, clock, audio):
|
|
"""
|
|
The common ordering: the console grants the channel, then 0.84-1.62s later
|
|
(plus pipeline lag) the audio shows up. The lookback has to cover it.
|
|
"""
|
|
grant_time = clock.now
|
|
await tick(hearing, update(call_log=[grant(1234, grant_time, tgtag="Fire")]))
|
|
assert not hearing.is_active
|
|
|
|
clock.advance(2.0)
|
|
audio.speak()
|
|
await tick(hearing, update()) # console says nothing NOW
|
|
|
|
assert hearing.is_active
|
|
payload = hearing.on_call_start.call_args[0][0]
|
|
assert payload["tgid"] == 1234
|
|
assert payload["tgid_name"] == "Fire"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_grant_after_audio_onset_attributes_the_recording_late(hearing, clock, audio):
|
|
"""
|
|
There is no guaranteed ordering: the console is polled every 500ms, so the
|
|
grant can land after voice onset. The segment starts unattributed and picks
|
|
the talkgroup up part-way through — expected, and fine.
|
|
"""
|
|
audio.speak()
|
|
await tick(hearing, update())
|
|
|
|
assert hearing.is_active
|
|
assert hearing.on_call_start.call_args[0][0]["attributed"] is False
|
|
|
|
clock.advance(0.5)
|
|
audio.speak()
|
|
await tick(hearing, update(call_log=[grant(1234, clock.now, tgtag="EMS")]))
|
|
assert hearing.current_tgid == 1234
|
|
|
|
clock.advance(settings.call_silence_timeout + 0.5)
|
|
await tick(hearing, update())
|
|
|
|
ended = hearing.on_call_end.call_args[0][0]
|
|
assert ended["attributed"] is True
|
|
assert ended["tgid"] == 1234
|
|
assert ended["tgid_name"] == "EMS"
|
|
assert hearing.unattributed_segments == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_attribution_resolves_from_a_channel_row_when_the_grant_was_dropped(hearing, clock, audio):
|
|
"""A dropped grant is survivable: an active channel row names the talkgroup."""
|
|
audio.speak()
|
|
await tick(hearing, update())
|
|
assert hearing.on_call_start.call_args[0][0]["tgid"] is None
|
|
|
|
clock.advance(0.5)
|
|
audio.speak()
|
|
await tick(hearing, update(channels=[channel(tgid=4321, srcaddr=99, tag="Sheriff")]))
|
|
|
|
clock.advance(settings.call_silence_timeout + 0.5)
|
|
await tick(hearing, update())
|
|
|
|
ended = hearing.on_call_end.call_args[0][0]
|
|
assert ended["tgid"] == 4321
|
|
assert ended["attributed"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_console_activity_outside_the_tolerance_does_not_attribute(hearing, clock, audio):
|
|
"""
|
|
The tolerance is deliberately bounded. A grant from long before the audio is
|
|
not evidence about this audio, and borrowing it would be worse than
|
|
admitting the audio is unattributed.
|
|
"""
|
|
await tick(hearing, update(call_log=[grant(1234, clock.now)]))
|
|
|
|
clock.advance(ATTRIBUTION_LOOKBACK_SECONDS + 5.0)
|
|
audio.speak()
|
|
await tick(hearing, update())
|
|
|
|
assert hearing.is_active
|
|
assert hearing.on_call_start.call_args[0][0]["tgid"] is None
|
|
assert ATTRIBUTION_LOOKAHEAD_SECONDS > 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_directly_stated_talkgroup_is_not_overruled_at_close(hearing, clock, audio, caplog):
|
|
"""
|
|
The attribution window extends past the audio on purpose, so a neighbouring
|
|
call's console activity can fall inside it. An inference over that window
|
|
must never overrule a talkgroup the console stated outright — but the
|
|
overlap does mean the split logic missed something, so it is logged.
|
|
"""
|
|
audio.speak()
|
|
await tick(hearing, update(
|
|
call_log=[grant(1111, clock.now, tgtag="Fire")],
|
|
channels=[channel(tgid=1111, srcaddr=1)],
|
|
))
|
|
|
|
# A second talkgroup is busy on ANOTHER receiver, so no split fires, and it
|
|
# produces more console observations than ours did.
|
|
for _ in range(4):
|
|
clock.advance(0.5)
|
|
audio.speak()
|
|
await tick(hearing, update(channels=[
|
|
channel(tgid=1111, srcaddr=0, hold_tgid=1111),
|
|
channel(tgid=2222, srcaddr=7),
|
|
]))
|
|
|
|
with caplog.at_level("WARNING", logger="drb-edge-node"):
|
|
clock.advance(settings.call_silence_timeout + 0.5)
|
|
await tick(hearing, update())
|
|
|
|
ended = hearing.on_call_end.call_args[0][0]
|
|
assert ended["tgid"] == 1111, "the console said 1111 directly; inference must not overrule it"
|
|
assert any("split logic should have fired" in r.message for r in caplog.records)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_the_same_voice_run_does_not_reopen_a_second_recording(hearing, clock, audio):
|
|
"""A closed run must stay closed; only NEW audio opens the next segment."""
|
|
audio.speak()
|
|
await tick(hearing, update(call_log=[grant(1234, clock.now)]))
|
|
|
|
clock.advance(settings.call_silence_timeout + 0.5)
|
|
await tick(hearing, update())
|
|
assert not hearing.is_active
|
|
|
|
# Several more quiet polls must not resurrect it.
|
|
for _ in range(3):
|
|
clock.advance(0.5)
|
|
await tick(hearing, update())
|
|
assert not hearing.is_active
|
|
assert hearing.on_call_start.call_count == 1
|
|
|
|
# ...but the next voice run does open a new segment.
|
|
clock.advance(1.0)
|
|
audio.speak()
|
|
await tick(hearing, update(call_log=[grant(1234, clock.now)]))
|
|
assert hearing.is_active
|
|
assert hearing.on_call_start.call_count == 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Mode changes: capture loss, console loss
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_losing_capture_closes_an_audio_segment(hearing, clock, audio):
|
|
"""
|
|
An audio-driven segment must never hang open when the audio stops arriving:
|
|
with no chunks there is no silence to detect, so the mode change is the
|
|
thing that has to close it.
|
|
"""
|
|
audio.speak()
|
|
await tick(hearing, update(call_log=[grant(1234, clock.now)]))
|
|
assert hearing.is_active
|
|
|
|
audio.capturing = False
|
|
clock.advance(0.5)
|
|
await tick(hearing, update(channels=[channel(tgid=1234, srcaddr=555)]))
|
|
|
|
assert not hearing.is_active
|
|
assert hearing.on_call_end.call_args[0][0]["end_reason"] == "capture_lost"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_console_segmentation_takes_over_while_capture_is_down(hearing, clock, audio):
|
|
"""
|
|
No audio means no recordings, but the node must still report real radio
|
|
activity to C2 rather than going silent about it.
|
|
"""
|
|
audio.capturing = False
|
|
|
|
await tick(hearing, update(
|
|
call_log=[grant(1234, clock.now, tgtag="Police")],
|
|
channels=[channel(tgid=1234, srcaddr=555)],
|
|
))
|
|
assert hearing.is_active
|
|
assert hearing.on_call_start.call_args[0][0]["driver"] == "console"
|
|
|
|
clock.advance(0.5)
|
|
await tick(hearing, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)]))
|
|
clock.advance(settings.call_idle_timeout + 0.5)
|
|
await tick(hearing, update(channels=[channel()]))
|
|
|
|
assert not hearing.is_active
|
|
assert hearing.on_call_end.call_args[0][0]["end_reason"] == "idle_timeout"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_op25_unreachable_still_closes_an_audio_segment(hearing, clock, audio):
|
|
"""Without the console there is no attribution, so there is nothing to keep open."""
|
|
audio.speak()
|
|
await tick(hearing, update(call_log=[grant(1234, clock.now)]))
|
|
|
|
clock.advance(OP25_OFFLINE_GRACE + 0.5)
|
|
audio.speak()
|
|
await tick(hearing, None)
|
|
|
|
assert not hearing.is_active
|
|
assert hearing.on_call_end.call_args[0][0]["end_reason"] == "op25_unreachable"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_console_history_is_bounded(hearing, clock, audio):
|
|
"""A rolling history that grows without limit would be a slow memory leak."""
|
|
from app.internal.metadata_watcher import CONSOLE_HISTORY_MAX, CONSOLE_HISTORY_SECONDS
|
|
|
|
for _ in range(CONSOLE_HISTORY_MAX + 200):
|
|
clock.advance(0.05)
|
|
await tick(hearing, update(channels=[channel(tgid=1234, srcaddr=5)]))
|
|
|
|
assert len(hearing._console) <= CONSOLE_HISTORY_MAX
|
|
|
|
# ...and old entries age out even when the count is low.
|
|
clock.advance(CONSOLE_HISTORY_SECONDS + 1)
|
|
await tick(hearing, update())
|
|
assert all(e.epoch >= clock.now - CONSOLE_HISTORY_SECONDS for e in hearing._console)
|