Spaces:
Sleeping
Sleeping
File size: 4,189 Bytes
37ee895 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 | from unittest.mock import patch
from bot.coaching_runtime import CoachingRuntime
def _high_risk_snapshot() -> dict:
return {
"chat_analytics": {
"messages_per_minute_10m": 0.05,
"messages_per_minute_60m": 0.90,
},
"chatters": {
"active_10m": 2,
"active_60m": 14,
},
"agent_outcomes": {
"ignored_rate_60m": 44.0,
"llm_interactions_60m": 11,
"quality_fallback_60m": 5,
},
"sentiment": {
"avg": -0.9,
"count": 16,
},
"stream_health": {
"score": 41,
},
}
def _low_risk_snapshot() -> dict:
return {
"chat_analytics": {
"messages_per_minute_10m": 0.72,
"messages_per_minute_60m": 0.74,
},
"chatters": {
"active_10m": 12,
"active_60m": 13,
},
"agent_outcomes": {
"ignored_rate_60m": 10.0,
"llm_interactions_60m": 13,
"quality_fallback_60m": 1,
},
"sentiment": {
"avg": 0.8,
"count": 18,
},
"stream_health": {
"score": 90,
},
}
@patch("bot.coaching_runtime.hud_runtime")
def test_evaluate_and_emit_applies_cooldown_antinoise(mock_hud_runtime):
runtime = CoachingRuntime(cooldown_seconds=120)
first = runtime.evaluate_and_emit(
_high_risk_snapshot(),
channel_id="Canal_A",
now=1_000.0,
)
assert first["channel_id"] == "canal_a"
assert first["risk_band"] in {"high", "critical"}
assert first["has_alerts"] is True
assert first["hud"]["emitted"] is True
assert first["hud"]["suppressed"] is False
assert first["hud"]["suppressed_total"] == 0
assert first["hud"]["last_emitted_at"] == "1970-01-01T00:16:40Z"
assert first["hud"]["signature"]
assert mock_hud_runtime.push_message.call_count == 1
assert mock_hud_runtime.push_message.call_args.kwargs["source"] == "coaching"
second = runtime.evaluate_and_emit(
_high_risk_snapshot(),
channel_id="canal_a",
now=1_060.0,
)
assert second["hud"]["emitted"] is False
assert second["hud"]["suppressed"] is True
assert second["hud"]["suppressed_total"] == 1
assert second["hud"]["last_emitted_at"] == "1970-01-01T00:16:40Z"
assert mock_hud_runtime.push_message.call_count == 1
third = runtime.evaluate_and_emit(
_high_risk_snapshot(),
channel_id="canal_a",
now=1_130.0,
)
assert third["hud"]["emitted"] is True
assert third["hud"]["suppressed"] is False
assert third["hud"]["suppressed_total"] == 1
assert third["hud"]["last_emitted_at"] == "1970-01-01T00:18:50Z"
assert mock_hud_runtime.push_message.call_count == 2
@patch("bot.coaching_runtime.hud_runtime")
def test_evaluate_and_emit_isolated_per_channel(mock_hud_runtime):
runtime = CoachingRuntime(cooldown_seconds=120)
first = runtime.evaluate_and_emit(
_high_risk_snapshot(),
channel_id="Canal_A",
now=2_000.0,
)
second = runtime.evaluate_and_emit(
_high_risk_snapshot(),
channel_id="Canal_B",
now=2_010.0,
)
assert first["channel_id"] == "canal_a"
assert second["channel_id"] == "canal_b"
assert first["hud"]["emitted"] is True
assert second["hud"]["emitted"] is True
assert mock_hud_runtime.push_message.call_count == 2
@patch("bot.coaching_runtime.hud_runtime")
def test_evaluate_and_emit_skips_hud_when_risk_is_low(mock_hud_runtime):
runtime = CoachingRuntime(cooldown_seconds=120)
payload = runtime.evaluate_and_emit(
_low_risk_snapshot(),
channel_id="Canal_Low",
now=3_000.0,
)
assert payload["channel_id"] == "canal_low"
assert payload["risk_band"] == "low"
assert payload["has_alerts"] is False
assert payload["hud"]["emitted"] is False
assert payload["hud"]["suppressed"] is False
assert payload["hud"]["last_emitted_at"] == ""
assert payload["hud"]["suppressed_total"] == 0
mock_hud_runtime.push_message.assert_not_called()
|