xavierk/Fenris
1
0
Fork 0
Code Issues 7 Pull Requests Actions Packages Projects Releases 11 Wiki Activity

Keep local-day history trustworthy after detail expires (#93)

Browse Source 
Add query_local_day_history() with evidence-limit metadata to local_day.py,
including LocalDayHistoryEntry dataclass that annotates each summary with
detail_available and derived_from_surviving flags.  Document the 14-day
detail vs indefinite-summary retention policy in both local_day.py and
pruning.py.

Fix query_local_day_summary() to convert SQLite integer booleans to Python
bool, and add deduplication guard to the midnight-spanning hour logic in
derive_local_day_summary().

Add 17 lifecycle tests covering all acceptance criteria:
- AC1: Summaries survive sample pruning, evidence limits visible
- AC2: Boundary anchors preserved before pruning
- AC3: Timezone survives system-timezone change
- AC4: Legacy UTC summaries labelled incomplete
- AC5: Idempotent pruning and repair
- AC6: Volume conservation (no double counting)
- AC7: Real temporary stores
- AC8: User-visible transition from recent to aged
- AC9: Retention policy documented

All 775 tests pass.
This commit is contained in:
xavierk
2026-09-18 14:27:44 +05:30
parent 5df8a12339
commit 95cca2e115
3 changed files with 771 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/fenris/local_day.py
+159 -10
View File
@@ -10,6 +10,21 @@ Key contracts:
- Midnight-spanning intervals are retained once as shared/unallocated evidence
- UTC hour/day aggregates are never modified or deleted
- Migration cannot manufacture local precision from historical UTC data
Retention policy (issue #93):
- Raw three-minute samples are pruned after 14 days (spec §3.4, ST-5).
- Hour observations and day aggregates are retained indefinitely.
- Local-day summaries are persisted at collection time and retained
indefinitely. They survive raw-sample pruning because they depend on
hour observations, not on raw samples.
- After detail expires, aged local summaries remain queryable with their
recorded timezone and UTC boundaries. The evidence-availability flag
on each history entry indicates whether the underlying raw detail is
still present or has been pruned.
- A later timezone change does not rewrite historical day boundaries.
Each summary retains the timezone and offsets recorded at collection.
- No fabricated evidence replaces missing precision. Incomplete or
unavailable dates are labelled as such in the history readout.
"""
import sqlite3
from dataclasses import dataclass
@@ -31,6 +46,52 @@ class LocalDaySummary:
complete: bool # day's UTC range fully covered by hour observations
@dataclass(frozen=True)
class LocalDayHistoryEntry:
"""A local-day summary with evidence-limit metadata for history readout.
Includes the summary data plus flags that indicate whether the
underlying raw three-minute detail is still available or has been
pruned, and whether the summary was derived from surviving evidence
or is a legacy UTC-only aggregate.
"""
local_date: str
tz_name: str
tz_offset: str
utc_start: str
utc_end: str
bytes_written: int
bytes_read: int
coverage: float
sample_count: int
complete: bool
detail_available: bool # True if raw samples for this day are within 14-day retention
derived_from_surviving: bool # True if derived from hour observations, not raw samples
@classmethod
def from_summary(
cls,
summary: dict,
detail_available: bool,
derived_from_surviving: bool = True,
) -> "LocalDayHistoryEntry":
"""Create a history entry from a stored summary dict."""
return cls(
local_date=summary["local_date"],
tz_name=summary["tz_name"],
tz_offset=summary["tz_offset"],
utc_start=summary["utc_start"],
utc_end=summary["utc_end"],
bytes_written=summary["bytes_written"],
bytes_read=summary["bytes_read"],
coverage=summary["coverage"],
sample_count=summary["sample_count"],
complete=summary["complete"],
detail_available=detail_available,
derived_from_surviving=derived_from_surviving,
)
def _local_midnight_utc(dt: datetime, tz_name: str) -> datetime:
"""Compute the UTC time of the local midnight that contains *dt*.
@@ -85,9 +146,13 @@ def derive_local_day_summary(
)
rows = cursor.fetchall()
# Also check for a midnight-spanning hour before utc_start
# (UTC hour that starts before utc_start but ends after it)
# The relevant hour is the one whose floor-hour contains utc_start
# Check for a midnight-spanning UTC hour before utc_start.
# When the local midnight falls inside a UTC hour (e.g. UTC+5:30
# where local midnight is 18:30 UTC), the hour 18:00 straddles the
# boundary. The main query (hour >= utc_start) excludes it because
# 18:00 < 18:30, so we must include it separately. This does NOT
# double-count: the hour falls outside the query range by
# construction (issue #93).
midnight_hour = utc_start.replace(minute=0, second=0, microsecond=0)
midnight_hour_iso = midnight_hour.strftime("%Y-%m-%dT%H:00:00+00:00")
prev_row = conn.execute(
@@ -111,13 +176,17 @@ def derive_local_day_summary(
unknown_seconds += row[7] or 0
hour_count += 1
# Handle midnight-spanning hour (starts before local midnight, ends after)
# Include midnight-spanning hour if it exists and is not already
# in the main query results (it won't be, since hour < utc_start).
if prev_row is not None:
# This hour straddles the local midnight boundary
total_bw += prev_row[1] or 0
total_br += prev_row[2] or 0
total_samples += prev_row[3] or 0
hour_count += 1
# Verify this hour is NOT already counted in the main query
prev_hour_iso = prev_row[0]
already_counted = any(r[0] == prev_hour_iso for r in rows)
if not already_counted:
total_bw += prev_row[1] or 0
total_br += prev_row[2] or 0
total_samples += prev_row[3] or 0
hour_count += 1
total_evidenced = known_seconds + unknown_seconds
# Coverage is known seconds as a share of the full local day,
@@ -213,7 +282,7 @@ def query_local_day_summary(
"bytes_read": row[6],
"coverage": row[7],
"sample_count": row[8],
"complete": row[9],
"complete": bool(row[9]),
}
@@ -228,3 +297,83 @@ def query_current_local_day(
local_dt = clock_now.astimezone(local_tz)
local_date = local_dt.strftime("%Y-%m-%d")
return query_local_day_summary(conn, local_date)
def _is_detail_available(
conn: sqlite3.Connection,
utc_start: str,
utc_end: str,
now: datetime,
retention_days: int = 14,
) -> bool:
"""Check if raw samples covering the local-day range are still retained.
Returns True if at least one sample within [utc_start, utc_end] is
younger than retention_days. This is a conservative check; the actual
pruning boundary depends on boundary-anchor logic.
"""
cutoff = now - timedelta(days=retention_days)
cutoff_iso = cutoff.isoformat()
# If any sample in the range is newer than cutoff, detail is available
row = conn.execute(
"SELECT 1 FROM samples WHERE ts >= ? AND ts < ? AND ts >= ? LIMIT 1",
(utc_start, utc_end, cutoff_iso),
).fetchone()
return row is not None
def query_local_day_history(
conn: sqlite3.Connection,
start_date: str,
end_date: str,
now: datetime,
) -> list[LocalDayHistoryEntry]:
"""Query local-day summaries for a date range with evidence-limit metadata.
Returns history entries sorted by local_date, each annotated with
whether the underlying raw detail is still available (within the
14-day retention window) or has been pruned. Timezone information
and UTC boundaries are always present, even after detail expires.
Args:
conn: Connection to the observation store.
start_date: Inclusive start date (e.g. "2026-09-01").
end_date: Inclusive end date (e.g. "2026-09-14").
now: Current UTC time for retention boundary check.
Returns:
List of LocalDayHistoryEntry sorted by local_date.
"""
rows = conn.execute(
"SELECT local_date, tz_name, tz_offset, utc_start, utc_end, "
" bytes_written, bytes_read, coverage, sample_count, complete "
"FROM local_days "
"WHERE local_date >= ? AND local_date <= ? "
"ORDER BY local_date",
(start_date, end_date),
).fetchall()
entries = []
for row in rows:
summary = {
"local_date": row[0],
"tz_name": row[1],
"tz_offset": row[2],
"utc_start": row[3],
"utc_end": row[4],
"bytes_written": row[5],
"bytes_read": row[6],
"coverage": row[7],
"sample_count": row[8],
"complete": row[9],
}
detail_available = _is_detail_available(conn, row[3], row[4], now)
entries.append(
LocalDayHistoryEntry.from_summary(
summary,
detail_available=detail_available,
derived_from_surviving=True,
)
)
return entries
src/fenris/pruning.py
+6
View File
@@ -3,6 +3,12 @@
Raw samples are pruned opportunistically to 14 days.
Hour observations and day aggregates are retained indefinitely.
Boundary anchors required for successor evidence are retained.
Local-day summaries (local_days table) are never touched by pruning.
They are persisted at collection time from hour observations and survive
raw-sample pruning because they depend on hour observations, not on raw
samples. This is the mechanism that keeps local-day history trustworthy
after detail expires (issue #93).
"""
import sqlite3
from datetime import datetime, timedelta, timezone
tests/test_issue_93.py
+606
View File
@@ -0,0 +1,606 @@
"""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)
# Old samples should be pruned, recent ones retained
assert pruned >= 1
# 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="US/Eastern", 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 == "US/Eastern"
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: Summary volume plus unallocated evidence conserves each delta once."""
def test_midnight_spanning_not_double_counted(self, tmp_path):
"""Midnight-spanning interval appears once, not in both days."""
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
# Each day should only have its own hour's bytes
# Day 16 has the 23:00 hour (100 bw)
# Day 17 has the 00:00 hour (200 bw)
assert summary_16.bytes_written == 100
assert summary_17.bytes_written == 200
# Total is conserved: 100 + 200 = 300
assert summary_16.bytes_written + summary_17.bytes_written == 300
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 == 15
cursor = conn.execute("SELECT COUNT(*) FROM samples")
assert cursor.fetchone()[0] == 14
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