"""Unit tests for the docker plugin's retention cutoff/bucketing helpers. _hour_floor and _rollup_cutoff are pure (no DB); retention.py imports its models lazily inside run_docker_retention, so importing them here doesn't register ORM tables — safe for the no-DB unit lane. The DB-backed rollup/prune itself is covered in tests/integration/test_docker.py. """ from datetime import datetime, timedelta, timezone from plugins.docker.retention import _hour_floor, _rollup_cutoff def test_hour_floor_drops_sub_hour_components(): dt = datetime(2026, 6, 19, 14, 37, 52, 123456, tzinfo=timezone.utc) assert _hour_floor(dt) == datetime(2026, 6, 19, 14, 0, 0, 0, tzinfo=timezone.utc) def test_hour_floor_on_exact_hour_is_identity(): dt = datetime(2026, 6, 19, 14, 0, 0, tzinfo=timezone.utc) assert _hour_floor(dt) == dt def test_rollup_cutoff_is_hour_aligned(): now = datetime(2026, 6, 19, 14, 37, 52, tzinfo=timezone.utc) cutoff = _rollup_cutoff(now, raw_days=7) # 7 days back from 14:37 is 14:37 on the 12th, floored to 14:00. assert cutoff == datetime(2026, 6, 12, 14, 0, 0, tzinfo=timezone.utc) # No sub-hour remainder — only whole elapsed hours get rolled up. assert cutoff.minute == 0 and cutoff.second == 0 and cutoff.microsecond == 0 def test_rollup_cutoff_scales_with_raw_days(): now = datetime(2026, 6, 19, 0, 0, 0, tzinfo=timezone.utc) assert _rollup_cutoff(now, 1) == now - timedelta(days=1) assert _rollup_cutoff(now, 30) == now - timedelta(days=30) def test_rollup_cutoff_sample_classification(): """A sample inside the raw window is kept; one before the aligned cutoff rolls up.""" now = datetime(2026, 6, 19, 14, 37, 0, tzinfo=timezone.utc) cutoff = _rollup_cutoff(now, raw_days=7) recent = datetime(2026, 6, 19, 12, 0, 0, tzinfo=timezone.utc) # within window old = datetime(2026, 6, 10, 9, 0, 0, tzinfo=timezone.utc) # older than cutoff assert not (recent < cutoff) # kept raw assert old < cutoff # rolled up