vikingowl
65c54416f3
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.
41 lines
1.2 KiB
Python
41 lines
1.2 KiB
Python
import os
|
|
|
|
from claude_matrix_bot.reset_watcher.diff import State, WindowState
|
|
from claude_matrix_bot.reset_watcher.state import load_state, save_state
|
|
|
|
|
|
def test_save_then_load_roundtrips(tmp_path) -> None:
|
|
path = tmp_path / "state.json"
|
|
s = State(
|
|
version=1,
|
|
last_run_ts=1_700_000_000,
|
|
windows={
|
|
"seven_day": WindowState(
|
|
resets_at=1_700_500_000, utilization=0.55, alerted_thresholds=[50]
|
|
),
|
|
},
|
|
)
|
|
|
|
save_state(str(path), s)
|
|
loaded = load_state(str(path))
|
|
|
|
assert loaded is not None
|
|
assert loaded.last_run_ts == s.last_run_ts
|
|
assert loaded.windows["seven_day"].resets_at == 1_700_500_000
|
|
assert loaded.windows["seven_day"].utilization == 0.55
|
|
assert loaded.windows["seven_day"].alerted_thresholds == [50]
|
|
|
|
|
|
def test_load_missing_returns_none(tmp_path) -> None:
|
|
assert load_state(str(tmp_path / "nope.json")) is None
|
|
|
|
|
|
def test_save_is_atomic(tmp_path) -> None:
|
|
path = tmp_path / "state.json"
|
|
s = State(version=1, last_run_ts=1, windows={})
|
|
save_state(str(path), s)
|
|
|
|
# No stray temp file left behind.
|
|
leftovers = [p for p in os.listdir(tmp_path) if p != "state.json"]
|
|
assert leftovers == []
|