Files
FabledSteward/tests/core/test_norms.py
T
bvandeusenandClaude Opus 5.5 4c84bc8560
CI / unit (push) Failing after 0s
CI / lint (push) Failing after 0s
CI / integration (push) Failing after 1s
CI / publish (push) Skipped
feat(norms): Steward learns each metric's normal per hour of day, then week (#5739)
A metric registered with learn_normal() gets, for every resource reporting
it, a usual hold (the 90th percentile across past days of each hour's p95)
and a usual peak for each slot:
- an hour of the day once 5 days of history exist;
- an hour of the week once history reaches back 3 weeks, so weekly jobs are
  learned too.
Each slot takes in the hours either side, so a job that runs a little late
stays normal. One SQL statement relearns everything hourly into
metric_norms, from the rollup plus the raw window not rolled up yet;
norms_at() reads the slot a moment falls in. Nothing is registered yet:
Traefik (#5741) and hosts (#5742) opt in.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-14 12:48:44 -04:00

28 lines
1.1 KiB
Python

"""Each metric's normal, by time slot (#5739): the slot arithmetic and the
registry. Learning itself runs in SQL; tests/integration/test_norms.py
covers it against Postgres."""
from __future__ import annotations
from datetime import datetime, timedelta, timezone
from steward.core import norms
def test_a_day_slot_is_the_utc_hour():
at = datetime(2026, 10, 14, 2, 30, tzinfo=timezone(timedelta(hours=-5))) # 07:30 UTC
assert norms.slot_of(at, "day") == 7
def test_a_week_slot_counts_hours_from_monday_utc():
monday = datetime(2026, 10, 12, 0, 10, tzinfo=timezone.utc)
assert norms.slot_of(monday, "week") == 0
assert norms.slot_of(monday + timedelta(days=2, hours=2), "week") == 50 # Wed 02:10
assert norms.slot_of(monday + timedelta(days=6, hours=23), "week") == 167
def test_a_learned_metric_is_registered_once_with_its_floor():
norms.learn_normal("test", "x_pct", 2)
norms.learn_normal("test", "x_pct", 3)
assert norms.LEARNED[("test", "x_pct")] == norms.Learned("test", "x_pct", 3.0)
del norms.LEARNED[("test", "x_pct")]