vikingowl/claude-matrix-bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
claude-matrix-bot/tests/test_diff.py
T
vikingowl 65c54416f3 feat(reset-watcher): add diff, state, and format logic 
Pure functions for the watcher's core: compute_alerts decides what to emit
given prior state vs. fresh usage, with configurable threshold list and a
reset-shift tolerance to filter minute-boundary jitter. State is atomically
persisted as JSON. Alerts render to (plain, html) tuples for Matrix, with
severity-colored percentages, blockquote framing, and a Unicode progress
bar for the cold-start armed message.
2026-05-18 17:22:31 +02:00

436 lines
16 KiB
Python
Raw Permalink Blame History

from claude_matrix_bot.reset_watcher.diff import (
Alert,
State,
WindowSnapshot,
WindowState,
compute_alerts,
)
WEEKLY = ("seven_day", "seven_day_opus", "seven_day_sonnet")
THRESHOLDS = (50, 80)
def _fresh(util_by_window: dict[str, tuple[float, int]]) -> dict[str, WindowSnapshot]:
return {
name: WindowSnapshot(utilization=util, resets_at=reset)
for name, (util, reset) in util_by_window.items()
}
def test_cold_start_emits_no_alerts_and_records_initial_state() -> None:
fresh = _fresh(
{
"five_hour": (0.42, 1_000_000),
"seven_day": (0.55, 2_000_000),
"seven_day_opus": (0.10, 2_000_000),
"seven_day_sonnet": (0.62, 2_000_000),
}
)
alerts, new_state = compute_alerts(
old_state=None,
fresh=fresh,
now=999_000,
thresholds=THRESHOLDS,
weekly_buckets=WEEKLY,
)
assert alerts == []
assert new_state.windows["seven_day"].resets_at == 2_000_000
assert new_state.windows["seven_day"].utilization == 0.55
assert new_state.windows["seven_day"].alerted_thresholds == []
def test_no_changes_no_alerts() -> None:
old = State(
version=1,
last_run_ts=998_000,
windows={
"five_hour": WindowState(resets_at=1_000_000, utilization=0.42, alerted_thresholds=[]),
"seven_day": WindowState(resets_at=2_000_000, utilization=0.40, alerted_thresholds=[]),
"seven_day_opus": WindowState(resets_at=2_000_000, utilization=0.10, alerted_thresholds=[]),
"seven_day_sonnet": WindowState(resets_at=2_000_000, utilization=0.20, alerted_thresholds=[]),
},
)
fresh = _fresh(
{
"five_hour": (0.43, 1_000_000),
"seven_day": (0.41, 2_000_000),
"seven_day_opus": (0.11, 2_000_000),
"seven_day_sonnet": (0.21, 2_000_000),
}
)
alerts, _ = compute_alerts(
old_state=old, fresh=fresh, now=999_000, thresholds=THRESHOLDS, weekly_buckets=WEEKLY,
)
assert alerts == []
def test_natural_rollover_does_not_alert() -> None:
old = State(
version=1,
last_run_ts=999_999,
windows={
"five_hour": WindowState(resets_at=1_000_000, utilization=0.90, alerted_thresholds=[]),
"seven_day": WindowState(resets_at=2_000_000, utilization=0.30, alerted_thresholds=[50]),
"seven_day_opus": WindowState(resets_at=2_000_000, utilization=0.10, alerted_thresholds=[]),
"seven_day_sonnet": WindowState(resets_at=2_000_000, utilization=0.20, alerted_thresholds=[]),
},
)
fresh = _fresh(
{
"five_hour": (0.05, 1_018_000), # new window after old expired
"seven_day": (0.31, 2_000_000),
"seven_day_opus": (0.11, 2_000_000),
"seven_day_sonnet": (0.21, 2_000_000),
}
)
alerts, new_state = compute_alerts(
old_state=old, fresh=fresh, now=1_010_000, thresholds=THRESHOLDS, weekly_buckets=WEEKLY,
)
assert alerts == []
assert new_state.windows["five_hour"].resets_at == 1_018_000
def test_mid_window_reset_shift_alerts() -> None:
old = State(
version=1,
last_run_ts=998_000,
windows={
"five_hour": WindowState(resets_at=1_000_000, utilization=0.42, alerted_thresholds=[]),
"seven_day": WindowState(resets_at=2_000_000, utilization=0.40, alerted_thresholds=[]),
"seven_day_opus": WindowState(resets_at=2_000_000, utilization=0.10, alerted_thresholds=[]),
"seven_day_sonnet": WindowState(resets_at=2_000_000, utilization=0.20, alerted_thresholds=[]),
},
)
fresh = _fresh(
{
"five_hour": (0.42, 1_005_000), # shifted while still in old window
"seven_day": (0.40, 2_000_000),
"seven_day_opus": (0.10, 2_000_000),
"seven_day_sonnet": (0.20, 2_000_000),
}
)
alerts, _ = compute_alerts(
old_state=old, fresh=fresh, now=999_000, thresholds=THRESHOLDS, weekly_buckets=WEEKLY,
)
assert len(alerts) == 1
assert alerts[0].kind == "reset_shift"
assert alerts[0].window == "five_hour"
assert alerts[0].detail["old_resets_at"] == 1_000_000
assert alerts[0].detail["new_resets_at"] == 1_005_000
def test_weekly_crosses_50_percent_alerts_once() -> None:
old = State(
version=1,
last_run_ts=998_000,
windows={
"five_hour": WindowState(resets_at=1_000_000, utilization=0.42, alerted_thresholds=[]),
"seven_day": WindowState(resets_at=2_000_000, utilization=0.49, alerted_thresholds=[]),
"seven_day_opus": WindowState(resets_at=2_000_000, utilization=0.10, alerted_thresholds=[]),
"seven_day_sonnet": WindowState(resets_at=2_000_000, utilization=0.20, alerted_thresholds=[]),
},
)
fresh = _fresh(
{
"five_hour": (0.43, 1_000_000),
"seven_day": (0.51, 2_000_000),
"seven_day_opus": (0.11, 2_000_000),
"seven_day_sonnet": (0.21, 2_000_000),
}
)
alerts, new_state = compute_alerts(
old_state=old, fresh=fresh, now=999_000, thresholds=THRESHOLDS, weekly_buckets=WEEKLY,
)
assert [a.kind for a in alerts] == ["threshold_crossed"]
assert alerts[0].window == "seven_day"
assert alerts[0].detail["threshold"] == 50
assert alerts[0].detail["utilization_pct"] == 51.0
assert 50 in new_state.windows["seven_day"].alerted_thresholds
def test_threshold_already_alerted_does_not_refire() -> None:
old = State(
version=1,
last_run_ts=998_000,
windows={
"five_hour": WindowState(resets_at=1_000_000, utilization=0.42, alerted_thresholds=[]),
"seven_day": WindowState(resets_at=2_000_000, utilization=0.55, alerted_thresholds=[50]),
"seven_day_opus": WindowState(resets_at=2_000_000, utilization=0.10, alerted_thresholds=[]),
"seven_day_sonnet": WindowState(resets_at=2_000_000, utilization=0.20, alerted_thresholds=[]),
},
)
fresh = _fresh(
{
"five_hour": (0.43, 1_000_000),
"seven_day": (0.60, 2_000_000),
"seven_day_opus": (0.11, 2_000_000),
"seven_day_sonnet": (0.21, 2_000_000),
}
)
alerts, _ = compute_alerts(
old_state=old, fresh=fresh, now=999_000, thresholds=THRESHOLDS, weekly_buckets=WEEKLY,
)
assert alerts == []
def test_jumps_past_both_thresholds_in_one_step_fires_both() -> None:
old = State(
version=1,
last_run_ts=998_000,
windows={
"five_hour": WindowState(resets_at=1_000_000, utilization=0.42, alerted_thresholds=[]),
"seven_day": WindowState(resets_at=2_000_000, utilization=0.10, alerted_thresholds=[]),
"seven_day_opus": WindowState(resets_at=2_000_000, utilization=0.10, alerted_thresholds=[]),
"seven_day_sonnet": WindowState(resets_at=2_000_000, utilization=0.20, alerted_thresholds=[]),
},
)
fresh = _fresh(
{
"five_hour": (0.43, 1_000_000),
"seven_day": (0.85, 2_000_000),
"seven_day_opus": (0.11, 2_000_000),
"seven_day_sonnet": (0.21, 2_000_000),
}
)
alerts, new_state = compute_alerts(
old_state=old, fresh=fresh, now=999_000, thresholds=THRESHOLDS, weekly_buckets=WEEKLY,
)
thresholds_fired = sorted(a.detail["threshold"] for a in alerts if a.window == "seven_day")
assert thresholds_fired == [50, 80]
assert sorted(new_state.windows["seven_day"].alerted_thresholds) == [50, 80]
def test_window_rollover_resets_alerted_thresholds() -> None:
old = State(
version=1,
last_run_ts=998_000,
windows={
"five_hour": WindowState(resets_at=1_000_000, utilization=0.42, alerted_thresholds=[]),
"seven_day": WindowState(resets_at=2_000_000, utilization=0.95, alerted_thresholds=[50, 80]),
"seven_day_opus": WindowState(resets_at=2_000_000, utilization=0.10, alerted_thresholds=[]),
"seven_day_sonnet": WindowState(resets_at=2_000_000, utilization=0.20, alerted_thresholds=[]),
},
)
fresh = _fresh(
{
"five_hour": (0.43, 1_000_000),
"seven_day": (0.05, 2_604_800), # new weekly window, fresh
"seven_day_opus": (0.01, 2_604_800),
"seven_day_sonnet": (0.02, 2_604_800),
}
)
alerts, new_state = compute_alerts(
old_state=old, fresh=fresh, now=2_100_000, thresholds=THRESHOLDS, weekly_buckets=WEEKLY,
)
assert alerts == []
assert new_state.windows["seven_day"].alerted_thresholds == []
assert new_state.windows["seven_day"].resets_at == 2_604_800
def test_five_hour_utilization_does_not_trigger_threshold_alert() -> None:
old = State(
version=1,
last_run_ts=998_000,
windows={
"five_hour": WindowState(resets_at=1_000_000, utilization=0.10, alerted_thresholds=[]),
"seven_day": WindowState(resets_at=2_000_000, utilization=0.10, alerted_thresholds=[]),
"seven_day_opus": WindowState(resets_at=2_000_000, utilization=0.10, alerted_thresholds=[]),
"seven_day_sonnet": WindowState(resets_at=2_000_000, utilization=0.20, alerted_thresholds=[]),
},
)
fresh = _fresh(
{
"five_hour": (0.95, 1_000_000), # 5h at 95% — should NOT alert
"seven_day": (0.11, 2_000_000),
"seven_day_opus": (0.11, 2_000_000),
"seven_day_sonnet": (0.21, 2_000_000),
}
)
alerts, _ = compute_alerts(
old_state=old, fresh=fresh, now=999_000, thresholds=THRESHOLDS, weekly_buckets=WEEKLY,
)
assert alerts == []
def test_all_three_weekly_buckets_evaluated_independently() -> None:
old = State(
version=1,
last_run_ts=998_000,
windows={
"five_hour": WindowState(resets_at=1_000_000, utilization=0.42, alerted_thresholds=[]),
"seven_day": WindowState(resets_at=2_000_000, utilization=0.49, alerted_thresholds=[]),
"seven_day_opus": WindowState(resets_at=2_000_000, utilization=0.79, alerted_thresholds=[50]),
"seven_day_sonnet": WindowState(resets_at=2_000_000, utilization=0.45, alerted_thresholds=[]),
},
)
fresh = _fresh(
{
"five_hour": (0.43, 1_000_000),
"seven_day": (0.51, 2_000_000), # crosses 50
"seven_day_opus": (0.81, 2_000_000), # crosses 80 (50 already alerted)
"seven_day_sonnet": (0.46, 2_000_000), # still under 50
}
)
alerts, _ = compute_alerts(
old_state=old, fresh=fresh, now=999_000, thresholds=THRESHOLDS, weekly_buckets=WEEKLY,
)
fired = {(a.window, a.detail["threshold"]) for a in alerts}
assert fired == {("seven_day", 50), ("seven_day_opus", 80)}
def test_new_window_starting_above_threshold_alerts_for_that_threshold() -> None:
"""If a fresh weekly window begins (e.g. plan switch) and current util is already > 50%,
we should still emit the 50% alert — otherwise we'd silently miss it."""
old = State(
version=1,
last_run_ts=998_000,
windows={
"five_hour": WindowState(resets_at=1_000_000, utilization=0.42, alerted_thresholds=[]),
"seven_day": WindowState(resets_at=2_000_000, utilization=0.95, alerted_thresholds=[50, 80]),
"seven_day_opus": WindowState(resets_at=2_000_000, utilization=0.10, alerted_thresholds=[]),
"seven_day_sonnet": WindowState(resets_at=2_000_000, utilization=0.20, alerted_thresholds=[]),
},
)
fresh = _fresh(
{
"five_hour": (0.43, 1_000_000),
"seven_day": (0.60, 2_604_800), # fresh window, but already at 60%
"seven_day_opus": (0.11, 2_604_800),
"seven_day_sonnet": (0.21, 2_604_800),
}
)
alerts, new_state = compute_alerts(
old_state=old, fresh=fresh, now=2_100_000, thresholds=THRESHOLDS, weekly_buckets=WEEKLY,
)
assert any(a.window == "seven_day" and a.detail["threshold"] == 50 for a in alerts)
assert 50 in new_state.windows["seven_day"].alerted_thresholds
def _state_with_five_hour(reset: int) -> State:
return State(
version=1,
last_run_ts=998_000,
windows={
"five_hour": WindowState(resets_at=reset, utilization=0.42, alerted_thresholds=[]),
"seven_day": WindowState(resets_at=2_000_000, utilization=0.40, alerted_thresholds=[]),
"seven_day_opus": WindowState(resets_at=2_000_000, utilization=0.10, alerted_thresholds=[]),
"seven_day_sonnet": WindowState(resets_at=2_000_000, utilization=0.20, alerted_thresholds=[]),
},
)
def test_sub_tolerance_reset_shift_is_suppressed() -> None:
old = _state_with_five_hour(1_000_000)
fresh = _fresh(
{
"five_hour": (0.42, 1_000_060), # +60s, default tolerance 120s
"seven_day": (0.40, 2_000_000),
"seven_day_opus": (0.10, 2_000_000),
"seven_day_sonnet": (0.20, 2_000_000),
}
)
alerts, _ = compute_alerts(
old_state=old, fresh=fresh, now=999_000,
thresholds=THRESHOLDS, weekly_buckets=WEEKLY,
)
assert alerts == []
def test_above_tolerance_reset_shift_alerts() -> None:
old = _state_with_five_hour(1_000_000)
fresh = _fresh(
{
"five_hour": (0.42, 1_000_300), # +300s, way above default 120s
"seven_day": (0.40, 2_000_000),
"seven_day_opus": (0.10, 2_000_000),
"seven_day_sonnet": (0.20, 2_000_000),
}
)
alerts, _ = compute_alerts(
old_state=old, fresh=fresh, now=999_000,
thresholds=THRESHOLDS, weekly_buckets=WEEKLY,
)
assert [a.kind for a in alerts] == ["reset_shift"]
assert alerts[0].detail["delta_seconds"] == 300
def test_explicit_tolerance_override() -> None:
"""Custom tolerance lets a 60s shift fire if the operator wants stricter detection."""
old = _state_with_five_hour(1_000_000)
fresh = _fresh(
{
"five_hour": (0.42, 1_000_060),
"seven_day": (0.40, 2_000_000),
"seven_day_opus": (0.10, 2_000_000),
"seven_day_sonnet": (0.20, 2_000_000),
}
)
alerts, _ = compute_alerts(
old_state=old, fresh=fresh, now=999_000,
thresholds=THRESHOLDS, weekly_buckets=WEEKLY,
reset_shift_tolerance_sec=30,
)
assert [a.kind for a in alerts] == ["reset_shift"]
def test_sub_tolerance_jitter_preserves_threshold_dedup() -> None:
"""A 60s minute-boundary jitter must NOT reset alerted_thresholds, otherwise 50%/80%
would re-fire on every poll that crosses a minute boundary."""
old = State(
version=1,
last_run_ts=998_000,
windows={
"five_hour": WindowState(resets_at=1_000_000, utilization=0.42, alerted_thresholds=[]),
"seven_day": WindowState(resets_at=2_000_000, utilization=0.55, alerted_thresholds=[50]),
"seven_day_opus": WindowState(resets_at=2_000_000, utilization=0.10, alerted_thresholds=[]),
"seven_day_sonnet": WindowState(resets_at=2_000_000, utilization=0.20, alerted_thresholds=[]),
},
)
fresh = _fresh(
{
"five_hour": (0.43, 1_000_000),
"seven_day": (0.60, 2_000_060), # +60s sub-tolerance jitter on weekly
"seven_day_opus": (0.11, 2_000_000),
"seven_day_sonnet": (0.21, 2_000_000),
}
)
alerts, new_state = compute_alerts(
old_state=old, fresh=fresh, now=999_000,
thresholds=THRESHOLDS, weekly_buckets=WEEKLY,
)
# No re-fire of 50% threshold.
assert alerts == []
# Dedup state carried across the jitter.
assert 50 in new_state.windows["seven_day"].alerted_thresholds
Reference in New Issue View Git Blame Copy Permalink