xavierk
7d219c4697
Hour classification (PR-4): - Powered-off: POH delta < 90% of wall-clock span - Active: DUW delta >= 256 MiB - Idle: powered on + sampled + below active threshold - Unknown: unsampled without POH evidence - Four splits sum to exactly wall_clock_seconds - Disabled time is never an hour state Monitoring periods (FL-8): - ensure_period_open: opens period at run moment if none exists - close_period: closes with end cause - is_inside_period: checks timestamp against period bounds - Never backdated; wall-clock outside periods excluded from denominator Day aggregates (ST-4, PR-5): - Derived monotonically from hour rows - UTC-bounded; no 23/25-hour days - Coverage: known seconds / period wall-clock - Gap hours inside periods contribute unknown seconds - Hours outside periods excluded entirely - No absent hour interpolated/estimated/fabricated (FL-3) Raw sample pruning (ST-5): - Prunes samples older than 14 days - Hour observations and day aggregates retained indefinitely Closes #22
219 lines
7.2 KiB
Python
219 lines
7.2 KiB
Python
"""Hour classification tests.
|
|
|
|
From spec §5.1 — each UTC hour is classified by named constants:
|
|
- Powered-off: power-on-hours delta < 90% of wall-clock span
|
|
- Active: DUW delta >= 256 MiB
|
|
- Idle: powered on + sampled + below active threshold
|
|
- Unknown: everything else
|
|
|
|
Four splits sum to exactly 3600s. Disabled time is never an hour state.
|
|
"""
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
from fenris.hour_classify import classify_hour, HourSplit, ACTIVE_THRESHOLD_BYTES
|
|
|
|
HOUR_SECONDS = 3600
|
|
|
|
|
|
class TestPoweredOff:
|
|
"""Spec §5.1: Powered-off when power-on-hours delta < 90% of wall-clock span."""
|
|
|
|
def test_below_90_percent_poh_is_powered_off(self):
|
|
split = classify_hour(
|
|
wall_clock_seconds=HOUR_SECONDS,
|
|
poh_delta=0,
|
|
duw_delta=0,
|
|
dur_delta=0,
|
|
)
|
|
assert split.seconds_powered_off == HOUR_SECONDS
|
|
assert split.seconds_active == 0
|
|
assert split.seconds_idle == 0
|
|
assert split.seconds_unknown == 0
|
|
|
|
def test_exactly_90_percent_poh_is_not_powered_off(self):
|
|
split = classify_hour(
|
|
wall_clock_seconds=HOUR_SECONDS,
|
|
poh_delta=3240,
|
|
duw_delta=0,
|
|
dur_delta=0,
|
|
)
|
|
assert split.seconds_powered_off == 0
|
|
assert split.seconds_unknown == HOUR_SECONDS
|
|
|
|
def test_just_below_90_percent_poh_is_powered_off(self):
|
|
split = classify_hour(
|
|
wall_clock_seconds=HOUR_SECONDS,
|
|
poh_delta=3239,
|
|
duw_delta=0,
|
|
dur_delta=0,
|
|
)
|
|
assert split.seconds_powered_off == HOUR_SECONDS
|
|
|
|
def test_100_percent_poh_is_not_powered_off(self):
|
|
split = classify_hour(
|
|
wall_clock_seconds=HOUR_SECONDS,
|
|
poh_delta=HOUR_SECONDS,
|
|
duw_delta=0,
|
|
dur_delta=0,
|
|
)
|
|
assert split.seconds_powered_off == 0
|
|
|
|
|
|
class TestActive:
|
|
"""Spec §5.1: Active when DUW delta >= 256 MiB."""
|
|
|
|
def test_above_256_mib_is_active(self):
|
|
split = classify_hour(
|
|
wall_clock_seconds=HOUR_SECONDS,
|
|
poh_delta=HOUR_SECONDS,
|
|
duw_delta=ACTIVE_THRESHOLD_BYTES,
|
|
dur_delta=1000,
|
|
sampled_seconds=HOUR_SECONDS,
|
|
)
|
|
assert split.seconds_active == HOUR_SECONDS
|
|
|
|
def test_just_below_256_mib_is_idle(self):
|
|
split = classify_hour(
|
|
wall_clock_seconds=HOUR_SECONDS,
|
|
poh_delta=HOUR_SECONDS,
|
|
duw_delta=ACTIVE_THRESHOLD_BYTES - 1,
|
|
dur_delta=1000,
|
|
sampled_seconds=HOUR_SECONDS,
|
|
)
|
|
assert split.seconds_idle == HOUR_SECONDS
|
|
assert split.seconds_active == 0
|
|
|
|
def test_powered_off_takes_priority_over_active_writes(self):
|
|
"""Spec §5.1 order: powered-off checked first."""
|
|
split = classify_hour(
|
|
wall_clock_seconds=HOUR_SECONDS,
|
|
poh_delta=0,
|
|
duw_delta=ACTIVE_THRESHOLD_BYTES,
|
|
dur_delta=1000,
|
|
)
|
|
assert split.seconds_powered_off == HOUR_SECONDS
|
|
assert split.seconds_active == 0
|
|
|
|
|
|
class TestIdle:
|
|
"""Spec §5.1: Idle when powered on, sampled, below active threshold."""
|
|
|
|
def test_idle_with_writes_below_threshold(self):
|
|
split = classify_hour(
|
|
wall_clock_seconds=HOUR_SECONDS,
|
|
poh_delta=HOUR_SECONDS,
|
|
duw_delta=1024 * 1024, # 1 MiB
|
|
dur_delta=500,
|
|
sampled_seconds=HOUR_SECONDS,
|
|
)
|
|
assert split.seconds_idle == HOUR_SECONDS
|
|
assert split.seconds_active == 0
|
|
|
|
def test_idle_no_writes(self):
|
|
"""Powered on, sampled, zero writes -> idle."""
|
|
split = classify_hour(
|
|
wall_clock_seconds=HOUR_SECONDS,
|
|
poh_delta=HOUR_SECONDS,
|
|
duw_delta=0,
|
|
dur_delta=0,
|
|
sampled_seconds=HOUR_SECONDS,
|
|
)
|
|
assert split.seconds_idle == HOUR_SECONDS
|
|
|
|
|
|
class TestUnknown:
|
|
"""Spec §5.1: Unknown when unsampled without POH evidence."""
|
|
|
|
def test_no_sample_no_writes_is_unknown(self):
|
|
split = classify_hour(
|
|
wall_clock_seconds=HOUR_SECONDS,
|
|
poh_delta=HOUR_SECONDS,
|
|
duw_delta=0,
|
|
dur_delta=0,
|
|
)
|
|
assert split.seconds_unknown == HOUR_SECONDS
|
|
|
|
def test_partial_unknown(self):
|
|
"""Partial hour -> split unknown for unsampled portion."""
|
|
split = classify_hour(
|
|
wall_clock_seconds=HOUR_SECONDS,
|
|
poh_delta=HOUR_SECONDS,
|
|
duw_delta=0,
|
|
dur_delta=0,
|
|
sampled_seconds=1800,
|
|
)
|
|
assert split.seconds_unknown == 1800
|
|
assert split.seconds_idle == 1800
|
|
|
|
def test_powered_off_with_no_sample_still_powered_off(self):
|
|
"""POH < 90% is powered off regardless of sample status."""
|
|
split = classify_hour(
|
|
wall_clock_seconds=HOUR_SECONDS,
|
|
poh_delta=0,
|
|
duw_delta=0,
|
|
dur_delta=0,
|
|
sampled_seconds=0,
|
|
)
|
|
assert split.seconds_powered_off == HOUR_SECONDS
|
|
assert split.seconds_unknown == 0
|
|
|
|
|
|
class TestSumTo3600:
|
|
"""Spec §5.1: Four splits sum to exactly wall_clock_seconds."""
|
|
|
|
def test_all_scenarios_sum_to_wall_clock(self):
|
|
scenarios = [
|
|
{"wall_clock_seconds": 3600, "poh_delta": 0, "duw_delta": 0, "dur_delta": 0},
|
|
{"wall_clock_seconds": 3600, "poh_delta": 3600, "duw_delta": 0, "dur_delta": 0},
|
|
{"wall_clock_seconds": 3600, "poh_delta": 3600, "duw_delta": ACTIVE_THRESHOLD_BYTES, "dur_delta": 1000, "sampled_seconds": 3600},
|
|
{"wall_clock_seconds": 3600, "poh_delta": 3600, "duw_delta": 1000, "dur_delta": 500, "sampled_seconds": 3600},
|
|
{"wall_clock_seconds": 3600, "poh_delta": 3239, "duw_delta": 0, "dur_delta": 0},
|
|
{"wall_clock_seconds": 1800, "poh_delta": 0, "duw_delta": 0, "dur_delta": 0},
|
|
{"wall_clock_seconds": 3600, "poh_delta": 3600, "duw_delta": 0, "dur_delta": 0, "sampled_seconds": 1800},
|
|
]
|
|
for s in scenarios:
|
|
split = classify_hour(**s)
|
|
total = (
|
|
split.seconds_active
|
|
+ split.seconds_idle
|
|
+ split.seconds_powered_off
|
|
+ split.seconds_unknown
|
|
)
|
|
assert total == s["wall_clock_seconds"], (
|
|
f"Scenario {s}: splits sum to {total}, expected {s['wall_clock_seconds']}"
|
|
)
|
|
|
|
def test_partial_wall_clock(self):
|
|
split = classify_hour(
|
|
wall_clock_seconds=1800,
|
|
poh_delta=0,
|
|
duw_delta=0,
|
|
dur_delta=0,
|
|
)
|
|
total = (
|
|
split.seconds_active
|
|
+ split.seconds_idle
|
|
+ split.seconds_powered_off
|
|
+ split.seconds_unknown
|
|
)
|
|
assert total == 1800
|
|
assert split.seconds_powered_off == 1800
|
|
|
|
|
|
class TestDisabledTimeNotAnHourState:
|
|
"""Spec §5.2: Disabled time is not an hour state — excluded from numerator/denominator."""
|
|
|
|
def test_classify_hour_has_no_disabled_state(self):
|
|
split = classify_hour(
|
|
wall_clock_seconds=HOUR_SECONDS,
|
|
poh_delta=0,
|
|
duw_delta=0,
|
|
dur_delta=0,
|
|
)
|
|
assert not hasattr(split, "seconds_disabled")
|