604 lines
24 KiB
Python
604 lines
24 KiB
Python
"""Lifecycle tests for local-day history after detail expires (issue #93).
|
|
|
|
Verifies the full lifecycle: raw samples collected, local-day summaries
|
|
persisted, raw samples pruned after 14 days, and aged local-day summaries
|
|
remain queryable with correct timezone, boundaries, and evidence-limit
|
|
metadata.
|
|
|
|
Seams:
|
|
- Write side: collector.run_collection() → local_day derivation → persist
|
|
- Pruning: prune_old_samples() removes raw samples but preserves local_days
|
|
- Read side: query_local_day_history() returns entries with evidence flags
|
|
- Repair: repair_derivation() does not touch existing local_days
|
|
"""
|
|
import sqlite3
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Dict
|
|
|
|
import pytest
|
|
import sys
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
from fenris.store import init_store, SCHEMA_VERSION
|
|
from fenris.local_day import (
|
|
derive_local_day_summary,
|
|
persist_local_day,
|
|
query_local_day_summary,
|
|
query_local_day_history,
|
|
LocalDaySummary,
|
|
LocalDayHistoryEntry,
|
|
_is_detail_available,
|
|
)
|
|
from fenris.pruning import prune_old_samples, RAW_SAMPLE_RETENTION_DAYS
|
|
from fenris.repair import repair_derivation
|
|
from fenris.monitoring_periods import ensure_period_open, close_period
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_smartctl(duw: int, dur: int) -> Dict[str, Any]:
|
|
return {
|
|
"json_format_version": [1, 0],
|
|
"smartctl": {"version": [7, 3], "svn_revision": "5155",
|
|
"build_info": "(local build)"},
|
|
"nvme_smart_health_information_log": {
|
|
"critical_warning": 0, "temperature": 35,
|
|
"available_spare": 100, "available_spare_threshold": 10,
|
|
"percentage_used": 5, "data_units_written": duw,
|
|
"data_units_read": dur, "power_on_hours": 8765,
|
|
"power_cycles": 1234, "unsafe_shutdowns": 5,
|
|
"media_errors": 0, "num_err_log_entries": 0,
|
|
},
|
|
"user_capacity": {"bytes": 1024000000000, "units": "bytes"},
|
|
"model_name": "Samsung SSD 970 EVO Plus 1TB",
|
|
"serial_number": "S4EWNX0N123456",
|
|
"firmware_version": "2B2QEXM7",
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def sysfs_tree(tmp_path: Path) -> Path:
|
|
ctrl_dir = tmp_path / "sys" / "class" / "nvme" / "nvme0"
|
|
ctrl_dir.mkdir(parents=True)
|
|
(ctrl_dir / "subsysnqn").write_text(
|
|
"nqn.2014-08.org.nvmexpress:uuid:12345678-1234-1234-1234-123456789abc\n"
|
|
)
|
|
(ctrl_dir / "model").write_text("Samsung SSD 970 EVO Plus 1TB\n")
|
|
(ctrl_dir / "serial").write_text("S4EWNX0N123456\n")
|
|
(ctrl_dir / "firmware_rev").write_text("2B2QEXM7\n")
|
|
transport_dir = ctrl_dir / "transport"
|
|
transport_dir.mkdir()
|
|
(transport_dir / "address").write_text("0000:03:00.0")
|
|
(transport_dir / "trstring").write_text("pcie")
|
|
return tmp_path
|
|
|
|
|
|
class _Clock:
|
|
def __init__(self, initial: datetime):
|
|
self.now = initial
|
|
def utcnow(self):
|
|
return self.now
|
|
|
|
|
|
def _insert_hour(conn, hour_iso, bw=0, br=0, sample_count=1,
|
|
active=3600, idle=0, powered_off=0, unknown=0):
|
|
"""Insert a UTC hour observation."""
|
|
known = active + idle + powered_off
|
|
coverage = known / 3600.0
|
|
conn.execute(
|
|
"INSERT INTO hour_observations "
|
|
"(hour, active_seconds, idle_seconds, powered_off_seconds, unknown_seconds, "
|
|
" bytes_written_delta, bytes_read_delta, sample_count, coverage) "
|
|
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
|
(hour_iso, active, idle, powered_off, unknown, bw, br, sample_count, coverage),
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def _insert_sample(conn, ts_iso, bw=0, br=0, device="/dev/nvme0"):
|
|
"""Insert a raw sample."""
|
|
conn.execute(
|
|
"INSERT INTO samples (ts, device, data_units_written, data_units_read, "
|
|
" bytes_written, bytes_read, percentage_used) VALUES (?, ?, 0, 0, ?, ?, 0)",
|
|
(ts_iso, device, bw, br),
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# AC1: Advance through 14-day boundary, detail expires, summaries remain
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestDetailExpiresSummariesRemain:
|
|
"""AC1: After 14-day pruning, local-day summaries survive."""
|
|
|
|
def test_local_days_survive_sample_pruning(self, tmp_path):
|
|
"""Local-day summaries persist after raw samples are pruned."""
|
|
conn = init_store(tmp_path / "obs.db")
|
|
now = datetime(2026, 10, 1, 12, 0, 0, tzinfo=timezone.utc)
|
|
|
|
# Create local-day summaries for dates spanning the 14-day boundary
|
|
# Day 1: 15 days ago (detail should be pruned)
|
|
old_summary = LocalDaySummary(
|
|
local_date="2026-09-16", tz_name="UTC", tz_offset="+00:00",
|
|
utc_start="2026-09-16T00:00:00+00:00",
|
|
utc_end="2026-09-17T00:00:00+00:00",
|
|
bytes_written=1000, bytes_read=500,
|
|
coverage=0.8, sample_count=10, complete=True,
|
|
)
|
|
persist_local_day(conn, old_summary)
|
|
|
|
# Day 2: 10 days ago (detail should be retained)
|
|
recent_summary = LocalDaySummary(
|
|
local_date="2026-09-21", tz_name="UTC", tz_offset="+00:00",
|
|
utc_start="2026-09-21T00:00:00+00:00",
|
|
utc_end="2026-09-22T00:00:00+00:00",
|
|
bytes_written=2000, bytes_read=800,
|
|
coverage=0.9, sample_count=15, complete=True,
|
|
)
|
|
persist_local_day(conn, recent_summary)
|
|
|
|
# Insert raw samples (old and recent)
|
|
_insert_sample(conn, "2026-09-16T10:00:00+00:00", bw=500, br=200)
|
|
_insert_sample(conn, "2026-09-16T10:05:00+00:00", bw=1000, br=500)
|
|
_insert_sample(conn, "2026-09-21T10:00:00+00:00", bw=1000, br=400)
|
|
_insert_sample(conn, "2026-09-21T10:05:00+00:00", bw=2000, br=800)
|
|
|
|
# Run pruning
|
|
pruned = prune_old_samples(conn, now, retention_days=14)
|
|
|
|
# The handcrafted totals lack replacement provenance, so retain samples.
|
|
assert pruned == 0
|
|
|
|
# Local-day summaries must still be queryable
|
|
old_result = query_local_day_summary(conn, "2026-09-16")
|
|
assert old_result is not None
|
|
assert old_result["bytes_written"] == 1000
|
|
assert old_result["bytes_read"] == 500
|
|
|
|
recent_result = query_local_day_summary(conn, "2026-09-21")
|
|
assert recent_result is not None
|
|
assert recent_result["bytes_written"] == 2000
|
|
|
|
conn.close()
|
|
|
|
def test_history_query_shows_evidence_limits(self, tmp_path):
|
|
"""History entries annotate detail-availability correctly."""
|
|
conn = init_store(tmp_path / "obs.db")
|
|
now = datetime(2026, 10, 1, 12, 0, 0, tzinfo=timezone.utc)
|
|
|
|
# Old summary (detail pruned)
|
|
old = LocalDaySummary(
|
|
local_date="2026-09-16", tz_name="UTC", tz_offset="+00:00",
|
|
utc_start="2026-09-16T00:00:00+00:00",
|
|
utc_end="2026-09-17T00:00:00+00:00",
|
|
bytes_written=1000, bytes_read=500,
|
|
coverage=0.8, sample_count=10, complete=True,
|
|
)
|
|
persist_local_day(conn, old)
|
|
|
|
# Recent summary (detail available)
|
|
recent = LocalDaySummary(
|
|
local_date="2026-09-21", tz_name="UTC", tz_offset="+00:00",
|
|
utc_start="2026-09-21T00:00:00+00:00",
|
|
utc_end="2026-09-22T00:00:00+00:00",
|
|
bytes_written=2000, bytes_read=800,
|
|
coverage=0.9, sample_count=15, complete=True,
|
|
)
|
|
persist_local_day(conn, recent)
|
|
|
|
# Insert a recent sample (no old samples → old detail not available)
|
|
_insert_sample(conn, "2026-09-21T10:00:00+00:00", bw=1000, br=400)
|
|
|
|
# Query history
|
|
entries = query_local_day_history(conn, "2026-09-16", "2026-09-21", now)
|
|
assert len(entries) == 2
|
|
|
|
old_entry = next(e for e in entries if e.local_date == "2026-09-16")
|
|
assert old_entry.detail_available is False
|
|
assert old_entry.bytes_written == 1000
|
|
|
|
recent_entry = next(e for e in entries if e.local_date == "2026-09-21")
|
|
assert recent_entry.detail_available is True
|
|
assert recent_entry.bytes_written == 2000
|
|
|
|
conn.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# AC2: Preserve summaries and boundary evidence before pruning
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestBoundaryPreservation:
|
|
"""AC2: Summaries and boundary evidence preserved before pruning."""
|
|
|
|
def test_boundary_anchor_not_pruned_when_needed(self, tmp_path):
|
|
"""Raw samples needed as boundary anchors are retained."""
|
|
conn = init_store(tmp_path / "obs.db")
|
|
now = datetime(2026, 10, 1, 12, 0, 0, tzinfo=timezone.utc)
|
|
|
|
# Old sample before cutoff, recent sample after cutoff
|
|
# The old sample is a boundary anchor
|
|
_insert_sample(conn, "2026-09-16T23:55:00+00:00", bw=100)
|
|
_insert_sample(conn, "2026-09-17T12:30:00+00:00", bw=200)
|
|
|
|
pruned = prune_old_samples(conn, now, retention_days=14)
|
|
|
|
# Boundary anchor should be retained
|
|
cursor = conn.execute(
|
|
"SELECT COUNT(*) FROM samples WHERE ts = '2026-09-16T23:55:00+00:00'"
|
|
)
|
|
assert cursor.fetchone()[0] == 1
|
|
conn.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# AC3: Aged summaries retain timezone after system-timezone change
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestTimezonePreservation:
|
|
"""AC3: Historical summaries retain recorded timezone after TZ change."""
|
|
|
|
def test_timezone_survives_pruning(self, tmp_path):
|
|
"""Old summary in Asia/Kolkata retains its timezone after pruning."""
|
|
conn = init_store(tmp_path / "obs.db")
|
|
now = datetime(2026, 10, 1, 12, 0, 0, tzinfo=timezone.utc)
|
|
|
|
# Old summary in Asia/Kolkata
|
|
old = LocalDaySummary(
|
|
local_date="2026-09-16", tz_name="Asia/Kolkata", tz_offset="+05:30",
|
|
utc_start="2026-09-15T18:30:00+00:00",
|
|
utc_end="2026-09-16T18:30:00+00:00",
|
|
bytes_written=3000, bytes_read=1200,
|
|
coverage=0.7, sample_count=8, complete=False,
|
|
)
|
|
persist_local_day(conn, old)
|
|
|
|
# Prune
|
|
prune_old_samples(conn, now, retention_days=14)
|
|
|
|
# Summary must retain its timezone
|
|
result = query_local_day_summary(conn, "2026-09-16")
|
|
assert result is not None
|
|
assert result["tz_name"] == "Asia/Kolkata"
|
|
assert result["tz_offset"] == "+05:30"
|
|
# UTC boundaries are preserved (not rewritten)
|
|
assert result["utc_start"] == "2026-09-15T18:30:00+00:00"
|
|
assert result["utc_end"] == "2026-09-16T18:30:00+00:00"
|
|
conn.close()
|
|
|
|
def test_history_entry_shows_tz_after_pruning(self, tmp_path):
|
|
"""History entry for aged summary shows correct timezone info."""
|
|
conn = init_store(tmp_path / "obs.db")
|
|
now = datetime(2026, 10, 1, 12, 0, 0, tzinfo=timezone.utc)
|
|
|
|
old = LocalDaySummary(
|
|
local_date="2026-09-16", tz_name="America/New_York", tz_offset="-04:00",
|
|
utc_start="2026-09-16T04:00:00+00:00",
|
|
utc_end="2026-09-17T04:00:00+00:00",
|
|
bytes_written=5000, bytes_read=2000,
|
|
coverage=0.6, sample_count=5, complete=False,
|
|
)
|
|
persist_local_day(conn, old)
|
|
|
|
entries = query_local_day_history(conn, "2026-09-16", "2026-09-16", now)
|
|
assert len(entries) == 1
|
|
entry = entries[0]
|
|
assert entry.tz_name == "America/New_York"
|
|
assert entry.tz_offset == "-04:00"
|
|
assert entry.utc_start == "2026-09-16T04:00:00+00:00"
|
|
assert entry.detail_available is False
|
|
conn.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# AC4: Legacy UTC summaries without local-day total remain incomplete
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestLegacyDataHandling:
|
|
"""AC4: Legacy UTC summaries that can't establish local-day total."""
|
|
|
|
def test_incomplete_legacy_labelled_correctly(self, tmp_path):
|
|
"""Legacy summary without full evidence is labelled incomplete."""
|
|
conn = init_store(tmp_path / "obs.db")
|
|
now = datetime(2026, 10, 1, 12, 0, 0, tzinfo=timezone.utc)
|
|
|
|
# Simulate a legacy summary that can't establish a full local-day total
|
|
legacy = LocalDaySummary(
|
|
local_date="2026-09-16", tz_name="UTC", tz_offset="+00:00",
|
|
utc_start="2026-09-16T00:00:00+00:00",
|
|
utc_end="2026-09-17T00:00:00+00:00",
|
|
bytes_written=500, bytes_read=200,
|
|
coverage=0.3, sample_count=3, complete=False,
|
|
)
|
|
persist_local_day(conn, legacy)
|
|
|
|
result = query_local_day_summary(conn, "2026-09-16")
|
|
assert result is not None
|
|
assert result["complete"] is False
|
|
assert result["coverage"] == pytest.approx(0.3, abs=0.01)
|
|
conn.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# AC5: Migration, pruning, and repair are interruption-safe and idempotent
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestIdempotencySafety:
|
|
"""AC5: Operations are idempotent and interruption-safe."""
|
|
|
|
def test_repeated_pruning_no_side_effects(self, tmp_path):
|
|
"""Running pruning multiple times produces the same result."""
|
|
conn = init_store(tmp_path / "obs.db")
|
|
now = datetime(2026, 10, 1, 12, 0, 0, tzinfo=timezone.utc)
|
|
|
|
# Create a local-day summary
|
|
summary = LocalDaySummary(
|
|
local_date="2026-09-16", tz_name="UTC", tz_offset="+00:00",
|
|
utc_start="2026-09-16T00:00:00+00:00",
|
|
utc_end="2026-09-17T00:00:00+00:00",
|
|
bytes_written=1000, bytes_read=500,
|
|
coverage=0.8, sample_count=10, complete=True,
|
|
)
|
|
persist_local_day(conn, summary)
|
|
_insert_sample(conn, "2026-09-16T10:00:00+00:00", bw=500)
|
|
_insert_sample(conn, "2026-09-16T10:05:00+00:00", bw=1000)
|
|
|
|
# Prune twice
|
|
prune_old_samples(conn, now, retention_days=14)
|
|
prune_old_samples(conn, now, retention_days=14)
|
|
|
|
# Local-day summary unchanged
|
|
result = query_local_day_summary(conn, "2026-09-16")
|
|
assert result is not None
|
|
assert result["bytes_written"] == 1000
|
|
conn.close()
|
|
|
|
def test_repair_does_not_touch_local_days(self, tmp_path):
|
|
"""Repair derivation does not modify existing local_days entries."""
|
|
conn = init_store(tmp_path / "obs.db")
|
|
|
|
# Create a local-day summary
|
|
summary = LocalDaySummary(
|
|
local_date="2026-09-16", tz_name="UTC", tz_offset="+00:00",
|
|
utc_start="2026-09-16T00:00:00+00:00",
|
|
utc_end="2026-09-17T00:00:00+00:00",
|
|
bytes_written=1000, bytes_read=500,
|
|
coverage=0.8, sample_count=10, complete=True,
|
|
)
|
|
persist_local_day(conn, summary)
|
|
|
|
# Run repair (should not touch local_days)
|
|
result = repair_derivation(conn)
|
|
assert result.ok is True
|
|
|
|
# Local-day summary unchanged
|
|
stored = query_local_day_summary(conn, "2026-09-16")
|
|
assert stored is not None
|
|
assert stored["bytes_written"] == 1000
|
|
assert stored["bytes_read"] == 500
|
|
conn.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# AC6: Volume conservation — no double counting of midnight-spanning bytes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestVolumeConservation:
|
|
"""AC6: UTC-hour totals never become fabricated local-day totals."""
|
|
|
|
def test_hourly_bytes_are_unavailable_at_local_precision(self, tmp_path):
|
|
"""UTC-day hour bytes cannot be split across local-day boundaries."""
|
|
conn = init_store(tmp_path / "obs.db")
|
|
|
|
# Insert hour observations that span midnight
|
|
# UTC hour 2026-09-16T23:00 contains midnight boundary
|
|
_insert_hour(conn, "2026-09-16T23:00:00+00:00", bw=100, br=50)
|
|
_insert_hour(conn, "2026-09-17T00:00:00+00:00", bw=200, br=80)
|
|
|
|
# Derive summaries for both days
|
|
clock_16 = datetime(2026, 9, 16, 12, 0, 0, tzinfo=timezone.utc)
|
|
summary_16 = derive_local_day_summary(conn, "UTC", clock_16)
|
|
assert summary_16 is not None
|
|
|
|
clock_17 = datetime(2026, 9, 17, 12, 0, 0, tzinfo=timezone.utc)
|
|
summary_17 = derive_local_day_summary(conn, "UTC", clock_17)
|
|
assert summary_17 is not None
|
|
|
|
assert summary_16.bytes_written == 0
|
|
assert summary_17.bytes_written == 0
|
|
assert summary_16.activity_precision == "unavailable"
|
|
assert summary_17.activity_precision == "unavailable"
|
|
conn.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# AC7: Use existing retention/repair entry points with real stores
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestExistingEntryPoints:
|
|
"""AC7: Use existing entry points with real temporary stores."""
|
|
|
|
def test_prune_with_real_temp_store(self, tmp_path):
|
|
"""Pruning works with a real temporary store."""
|
|
conn = init_store(tmp_path / "obs.db")
|
|
now = datetime(2026, 10, 1, 12, 0, 0, tzinfo=timezone.utc)
|
|
|
|
# Insert old and recent samples
|
|
for days_ago in range(1, 30):
|
|
ts = (now - timedelta(days=days_ago)).isoformat()
|
|
_insert_sample(conn, ts, bw=days_ago * 100)
|
|
|
|
pruned = prune_old_samples(conn, now, retention_days=14)
|
|
assert pruned == 0
|
|
|
|
cursor = conn.execute("SELECT COUNT(*) FROM samples")
|
|
assert cursor.fetchone()[0] == 29
|
|
conn.close()
|
|
|
|
def test_repair_with_real_temp_store(self, tmp_path):
|
|
"""Repair works with a real temporary store."""
|
|
conn = init_store(tmp_path / "obs.db")
|
|
|
|
# Insert samples that need interval derivation
|
|
_insert_sample(conn, "2026-09-16T10:00:00+00:00", bw=100)
|
|
_insert_sample(conn, "2026-09-16T10:05:00+00:00", bw=200)
|
|
|
|
result = repair_derivation(conn)
|
|
assert result.ok is True
|
|
conn.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# AC8: User-visible transition from recent detail to older summaries
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestUserVisibleTransition:
|
|
"""AC8: Transition from recent detail to aged summaries."""
|
|
|
|
def test_recent_vs_aged_history_entries(self, tmp_path):
|
|
"""Recent entries show detail_available=True, aged show False."""
|
|
conn = init_store(tmp_path / "obs.db")
|
|
now = datetime(2026, 10, 1, 12, 0, 0, tzinfo=timezone.utc)
|
|
|
|
# Create summaries for recent and aged dates
|
|
for days_ago, bw in [(5, 1000), (10, 2000), (15, 3000)]:
|
|
date = (now - timedelta(days=days_ago)).strftime("%Y-%m-%d")
|
|
utc_start = (now - timedelta(days=days_ago)).replace(
|
|
hour=0, minute=0, second=0, microsecond=0
|
|
).isoformat()
|
|
utc_end = (now - timedelta(days=days_ago - 1)).replace(
|
|
hour=0, minute=0, second=0, microsecond=0
|
|
).isoformat()
|
|
|
|
summary = LocalDaySummary(
|
|
local_date=date, tz_name="UTC", tz_offset="+00:00",
|
|
utc_start=utc_start, utc_end=utc_end,
|
|
bytes_written=bw, bytes_read=bw // 2,
|
|
coverage=0.8, sample_count=10, complete=True,
|
|
)
|
|
persist_local_day(conn, summary)
|
|
|
|
# Insert a recent sample (5 days ago)
|
|
_insert_sample(conn, (now - timedelta(days=5)).isoformat(), bw=500)
|
|
|
|
# Query full range
|
|
start = (now - timedelta(days=15)).strftime("%Y-%m-%d")
|
|
end = (now - timedelta(days=5)).strftime("%Y-%m-%d")
|
|
entries = query_local_day_history(conn, start, end, now)
|
|
|
|
assert len(entries) == 3
|
|
|
|
# Recent entry has detail
|
|
recent = next(e for e in entries if e.local_date == (now - timedelta(days=5)).strftime("%Y-%m-%d"))
|
|
assert recent.detail_available is True
|
|
|
|
# Aged entries do not have detail
|
|
aged_10 = next(e for e in entries if e.local_date == (now - timedelta(days=10)).strftime("%Y-%m-%d"))
|
|
assert aged_10.detail_available is False
|
|
|
|
aged_15 = next(e for e in entries if e.local_date == (now - timedelta(days=15)).strftime("%Y-%m-%d"))
|
|
assert aged_15.detail_available is False
|
|
conn.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# AC9: Document the 14-day detail vs indefinite-summary policy
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRetentionPolicyDocumentation:
|
|
"""AC9: Module docstring documents the retention policy."""
|
|
|
|
def test_module_has_retention_policy_doc(self):
|
|
"""The local_day module documents the 14-day vs indefinite policy."""
|
|
import fenris.local_day as ld
|
|
docstring = ld.__doc__
|
|
assert "14 day" in docstring.lower() or "fourteen" in docstring.lower()
|
|
assert "indefinite" in docstring.lower()
|
|
assert "retention" in docstring.lower()
|
|
|
|
def test_pruning_module_documents_local_days_preservation(self):
|
|
"""The pruning module documents that local_days are preserved."""
|
|
import fenris.pruning as pr
|
|
docstring = pr.__doc__
|
|
assert "local_days" in docstring.lower() or "local day" in docstring.lower()
|
|
assert "preserved" in docstring.lower() or "never" in docstring.lower()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Evidence-limit visibility
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestEvidenceLimitVisibility:
|
|
"""History entries expose evidence-availability metadata."""
|
|
|
|
def test_detail_available_flag_reflects_samples(self, tmp_path):
|
|
"""detail_available is True only when samples exist within retention."""
|
|
conn = init_store(tmp_path / "obs.db")
|
|
now = datetime(2026, 10, 1, 12, 0, 0, tzinfo=timezone.utc)
|
|
|
|
# Summary for a date with no samples
|
|
no_samples = LocalDaySummary(
|
|
local_date="2026-09-16", tz_name="UTC", tz_offset="+00:00",
|
|
utc_start="2026-09-16T00:00:00+00:00",
|
|
utc_end="2026-09-17T00:00:00+00:00",
|
|
bytes_written=500, bytes_read=200,
|
|
coverage=0.4, sample_count=3, complete=False,
|
|
)
|
|
persist_local_day(conn, no_samples)
|
|
|
|
# Summary for a date with a recent sample
|
|
with_sample = LocalDaySummary(
|
|
local_date="2026-09-21", tz_name="UTC", tz_offset="+00:00",
|
|
utc_start="2026-09-21T00:00:00+00:00",
|
|
utc_end="2026-09-22T00:00:00+00:00",
|
|
bytes_written=1000, bytes_read=400,
|
|
coverage=0.9, sample_count=10, complete=True,
|
|
)
|
|
persist_local_day(conn, with_sample)
|
|
_insert_sample(conn, "2026-09-21T12:00:00+00:00", bw=500)
|
|
|
|
entries = query_local_day_history(conn, "2026-09-16", "2026-09-21", now)
|
|
|
|
no_samples_entry = next(e for e in entries if e.local_date == "2026-09-16")
|
|
assert no_samples_entry.detail_available is False
|
|
|
|
with_sample_entry = next(e for e in entries if e.local_date == "2026-09-21")
|
|
assert with_sample_entry.detail_available is True
|
|
conn.close()
|
|
|
|
def test_empty_history_range_returns_empty_list(self, tmp_path):
|
|
"""Querying an empty date range returns an empty list."""
|
|
conn = init_store(tmp_path / "obs.db")
|
|
now = datetime(2026, 10, 1, 12, 0, 0, tzinfo=timezone.utc)
|
|
|
|
entries = query_local_day_history(conn, "2026-01-01", "2026-01-31", now)
|
|
assert entries == []
|
|
conn.close()
|
|
|
|
def test_history_entry_from_summary_factory(self, tmp_path):
|
|
"""LocalDayHistoryEntry.from_summary creates entry from dict."""
|
|
summary = {
|
|
"local_date": "2026-09-16",
|
|
"tz_name": "UTC",
|
|
"tz_offset": "+00:00",
|
|
"utc_start": "2026-09-16T00:00:00+00:00",
|
|
"utc_end": "2026-09-17T00:00:00+00:00",
|
|
"bytes_written": 1000,
|
|
"bytes_read": 500,
|
|
"coverage": 0.8,
|
|
"sample_count": 10,
|
|
"complete": True,
|
|
}
|
|
entry = LocalDayHistoryEntry.from_summary(
|
|
summary, detail_available=True, derived_from_surviving=True
|
|
)
|
|
assert entry.local_date == "2026-09-16"
|
|
assert entry.detail_available is True
|
|
assert entry.derived_from_surviving is True
|
|
assert entry.bytes_written == 1000
|