"""Collector history tracer tests (issue #73). Tests the end-to-end history pipeline: sample acquisition → interval derivation → hour observation → day aggregate, with concurrent-read safety, cross-hour handling, and display states. Seams: - write side: run_collection() → observation store - read side: get_status(), compute_projection() → observation store """ import os 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.collector import run_collection, normalize_identity from fenris.store import init_store, get_store_path, SCHEMA_VERSION from fenris.monitoring_periods import ensure_period_open, close_period, get_open_period from fenris.day_aggregate import derive_day, derive_all_days # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture def smartctl_fixture() -> Dict[str, Any]: """Minimal smartctl -a -j output with required fields.""" 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": 12345678, "data_units_read": 9876543, "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_fixture_tree(tmp_path: Path) -> Path: """Create a minimal sysfs fixture tree with controller identity.""" 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 @pytest.fixture def config_fixture(tmp_path: Path) -> Dict[str, Any]: """Configuration fixture naming the device.""" return { "device": "/dev/nvme0", "store_path": str(tmp_path / "observations.db"), } class FakeClock: """Injected clock returning controlled time.""" def __init__(self, initial: datetime): self.now = initial def utcnow(self): return self.now def advance(self, **kwargs): self.now = self.now + timedelta(**kwargs) # --------------------------------------------------------------------------- # Schema migration: existing data readable at real precision # --------------------------------------------------------------------------- class TestSchemaMigration: """Schema migration 1→2 preserves existing data (issue #73 AC1).""" def test_migration_bumps_version(self, tmp_path): """Migration from v1 to v2 succeeds.""" from fenris.store import migrate_to_latest, SCHEMA_VERSION # Create a v1 store directly (simulating pre-migration state) db = tmp_path / "test.db" conn = sqlite3.connect(str(db)) conn.execute("PRAGMA journal_mode=WAL") # Create v1 schema manually conn.execute(""" CREATE TABLE samples ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts TEXT NOT NULL, device TEXT NOT NULL, subnqn TEXT, sn TEXT, mn TEXT, fr TEXT, capacity_bytes INTEGER, percentage_used INTEGER, available_spare INTEGER, media_errors INTEGER, power_on_hours INTEGER, power_cycles INTEGER, unsafe_shutdowns INTEGER, temperature_c INTEGER, data_units_written INTEGER, data_units_read INTEGER, bytes_written INTEGER, bytes_read INTEGER, critical_warning INTEGER ) """) conn.execute(""" CREATE TABLE hour_observations ( id INTEGER PRIMARY KEY AUTOINCREMENT, hour TEXT NOT NULL UNIQUE, active_seconds INTEGER DEFAULT 0, idle_seconds INTEGER DEFAULT 0, powered_off_seconds INTEGER DEFAULT 0, unknown_seconds INTEGER DEFAULT 0, bytes_written_delta INTEGER DEFAULT 0, bytes_read_delta INTEGER DEFAULT 0, temperature_min INTEGER, temperature_avg REAL, temperature_max INTEGER, sample_count INTEGER DEFAULT 0, coverage REAL DEFAULT 0.0 ) """) conn.execute(""" CREATE TABLE day_aggregates ( id INTEGER PRIMARY KEY AUTOINCREMENT, day TEXT NOT NULL UNIQUE, active_seconds INTEGER DEFAULT 0, idle_seconds INTEGER DEFAULT 0, powered_off_seconds INTEGER DEFAULT 0, unknown_seconds INTEGER DEFAULT 0, bytes_written_delta INTEGER DEFAULT 0, bytes_read_delta INTEGER DEFAULT 0, sample_count INTEGER DEFAULT 0, coverage REAL DEFAULT 0.0 ) """) conn.execute("CREATE TABLE monitoring_periods (id INTEGER PRIMARY KEY AUTOINCREMENT, started_at TEXT NOT NULL, ended_at TEXT, end_cause TEXT)") conn.execute("CREATE TABLE controller_segments (id INTEGER PRIMARY KEY AUTOINCREMENT, opened_at TEXT NOT NULL, identity_key TEXT, identity_degraded BOOLEAN DEFAULT 0, subnqn TEXT, sn TEXT, mn TEXT, fr TEXT, vid TEXT, ssvid TEXT, transport TEXT)") conn.execute("CREATE TABLE endurance_baseline (id INTEGER PRIMARY KEY AUTOINCREMENT, tbw_terabytes REAL NOT NULL, source_url TEXT, document_revision TEXT, entry_date TEXT, model_string TEXT, nominal_capacity_bytes INTEGER, validated_by TEXT, verified BOOLEAN DEFAULT 0, created_at TEXT NOT NULL, updated_at TEXT NOT NULL)") conn.execute("CREATE TABLE store_metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)") conn.execute("PRAGMA user_version=1") conn.execute("INSERT INTO samples (ts, device, mn, sn, fr, capacity_bytes, percentage_used, available_spare, media_errors, power_on_hours, power_cycles, unsafe_shutdowns, temperature_c, data_units_written, data_units_read, bytes_written, bytes_read, critical_warning) VALUES ('2026-09-01T12:00:00+00:00', '/dev/nvme0', 'Test', 'SN', 'FR', 1000000000000, 5, 100, 0, 1000, 100, 0, 35, 1000000, 500000, 512000000000, 256000000000, 0)") conn.commit() conn.close() # Migrate steps = migrate_to_latest(db) assert steps == 1 # Verify data preserved conn = sqlite3.connect(str(db)) row = conn.execute("SELECT ts, mn FROM samples").fetchone() version = conn.execute("PRAGMA user_version").fetchone()[0] conn.close() assert row[0] == "2026-09-01T12:00:00+00:00" assert row[1] == "Test" assert version == SCHEMA_VERSION def test_newer_schema_refused(self, tmp_path): """Store with user_version > SCHEMA_VERSION is refused.""" db = tmp_path / "test.db" conn = sqlite3.connect(str(db)) conn.execute("PRAGMA user_version=%d" % (SCHEMA_VERSION + 1)) conn.commit() conn.close() with pytest.raises(ValueError, match="newer Fenris"): init_store(db) def test_existing_data_preserved_after_migration(self, config_fixture, smartctl_fixture, sysfs_fixture_tree): """Existing sample data is not lost or modified by migration.""" clock = FakeClock(datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)) # Write first sample run_collection(smartctl_fixture, sysfs_fixture_tree / "sys" / "class" / "nvme" / "nvme0", config_fixture, clock) conn = sqlite3.connect(config_fixture["store_path"]) row = conn.execute("SELECT ts, device, bytes_written FROM samples").fetchone() conn.close() assert row[0] == "2026-09-01T12:00:00+00:00" assert row[1] == "/dev/nvme0" assert row[2] == 12345678 * 512000 # --------------------------------------------------------------------------- # Collector publishes samples, intervals, hour observations, day aggregates # --------------------------------------------------------------------------- class TestCollectorDerivation: """Collector derives hour observations and day aggregates (issue #73 AC2).""" def _make_sample(self, duw_units: int, ts: str) -> Dict[str, Any]: """Build a smartctl fixture with specific DUW.""" 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_units, "data_units_read": 9876543, "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", } def test_same_hour_20mb_derives_hour_obs(self, config_fixture, sysfs_fixture_tree): """Two samples in same hour with 20 MB delta → hour_obs gets 20 MB.""" t1 = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc) t2 = datetime(2026, 9, 1, 12, 5, 0, tzinfo=timezone.utc) # DUW units: 12345678 * 512000 = ~6.3 TB; 20 MB = 20*1024*1024 / 512000 ≈ 40 units duw1 = 12345678 duw2 = duw1 + 40 # ~20 MB more clock1 = FakeClock(t1) s1 = self._make_sample(duw1, t1.isoformat()) r1 = run_collection(s1, sysfs_fixture_tree / "sys" / "class" / "nvme" / "nvme0", config_fixture, clock1) assert r1["ok"] clock2 = FakeClock(t2) s2 = self._make_sample(duw2, t2.isoformat()) r2 = run_collection(s2, sysfs_fixture_tree / "sys" / "class" / "nvme" / "nvme0", config_fixture, clock2) assert r2["ok"] # Check hour_observation was derived conn = sqlite3.connect(config_fixture["store_path"]) hour = conn.execute( "SELECT bytes_written_delta, sample_count FROM hour_observations WHERE hour LIKE '2026-09-01T12%'" ).fetchone() conn.close() assert hour is not None, "Hour observation should exist for 12:00" assert hour[1] >= 2 # at least 2 samples contributed # bytes_written_delta should be the 20 MB delta (40 * 512000 = 20480000) assert hour[0] == 40 * 512000 def test_zero_delta_derives_hour_obs(self, config_fixture, sysfs_fixture_tree): """Two samples in same hour with no DUW change → 0 B written.""" t1 = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc) t2 = datetime(2026, 9, 1, 12, 5, 0, tzinfo=timezone.utc) duw = 12345678 # same for both clock1 = FakeClock(t1) s1 = self._make_sample(duw, t1.isoformat()) run_collection(s1, sysfs_fixture_tree / "sys" / "class" / "nvme" / "nvme0", config_fixture, clock1) clock2 = FakeClock(t2) s2 = self._make_sample(duw, t2.isoformat()) run_collection(s2, sysfs_fixture_tree / "sys" / "class" / "nvme" / "nvme0", config_fixture, clock2) conn = sqlite3.connect(config_fixture["store_path"]) hour = conn.execute( "SELECT bytes_written_delta FROM hour_observations WHERE hour LIKE '2026-09-01T12%'" ).fetchone() conn.close() assert hour is not None assert hour[0] == 0 def test_cross_hour_100mb_unattributed(self, config_fixture, sysfs_fixture_tree): """Samples in different hours → delta is unattributed to any hour.""" t1 = datetime(2026, 9, 1, 11, 55, 0, tzinfo=timezone.utc) t2 = datetime(2026, 9, 1, 12, 5, 0, tzinfo=timezone.utc) duw1 = 12345678 duw2 = duw1 + 200 # ~100 MB clock1 = FakeClock(t1) s1 = self._make_sample(duw1, t1.isoformat()) run_collection(s1, sysfs_fixture_tree / "sys" / "class" / "nvme" / "nvme0", config_fixture, clock1) clock2 = FakeClock(t2) s2 = self._make_sample(duw2, t2.isoformat()) run_collection(s2, sysfs_fixture_tree / "sys" / "class" / "nvme" / "nvme0", config_fixture, clock2) conn = sqlite3.connect(config_fixture["store_path"]) # Hour observations should NOT contain the cross-hour delta hour11 = conn.execute( "SELECT bytes_written_delta FROM hour_observations WHERE hour LIKE '2026-09-01T11%'" ).fetchone() hour12 = conn.execute( "SELECT bytes_written_delta FROM hour_observations WHERE hour LIKE '2026-09-01T12%'" ).fetchone() conn.close() # Neither hour should have the full 100 MB delta attributed # (they may have 0 or partial, but not 200*512000) full_delta = 200 * 512000 if hour11 is not None: assert hour11[0] != full_delta, "Hour 11 should not have full cross-hour delta" if hour12 is not None: assert hour12[0] != full_delta, "Hour 12 should not have full cross-hour delta" def test_readonly_sees_consistent_snapshot(self, config_fixture, sysfs_fixture_tree): """A read-only reader sees valid pre- or post-publication snapshot.""" t1 = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc) t2 = datetime(2026, 9, 1, 12, 5, 0, tzinfo=timezone.utc) clock1 = FakeClock(t1) run_collection(self._make_sample(12345678, t1.isoformat()), sysfs_fixture_tree / "sys" / "class" / "nvme" / "nvme0", config_fixture, clock1) # Open read-only ro_conn = sqlite3.connect( "file:%s?mode=ro" % config_fixture["store_path"], uri=True ) count_before = ro_conn.execute("SELECT COUNT(*) FROM samples").fetchone()[0] ro_conn.close() assert count_before == 1 # Write second sample clock2 = FakeClock(t2) run_collection(self._make_sample(12345718, t2.isoformat()), sysfs_fixture_tree / "sys" / "class" / "nvme" / "nvme0", config_fixture, clock2) # Read-only reader sees 2 samples now ro_conn2 = sqlite3.connect( "file:%s?mode=ro" % config_fixture["store_path"], uri=True ) count_after = ro_conn2.execute("SELECT COUNT(*) FROM samples").fetchone()[0] ro_conn2.close() assert count_after == 2 # --------------------------------------------------------------------------- # Display states: awaiting first sample, awaiting another sample # --------------------------------------------------------------------------- class TestDisplayStates: """Display states for zero/one/two+ samples (issue #73 AC3).""" def test_zero_samples_awaiting_first(self, tmp_path): """Zero samples → 'awaiting first sample' state.""" from fenris.status import get_status db = tmp_path / "test.db" init_store(db) now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc) from unittest.mock import patch with patch("fenris.status.query_service_state", return_value={ "boot_enabled": False, "timer_active": False, "last_collect_ok": None, "last_collect_age_s": None, "last_collect_reason": None, }): result = get_status(store_path=db, clock_now=now, query_services=False, query_journal=False) assert "no observations yet" in result.lower() or "awaiting" in result.lower() def test_one_sample_awaiting_another(self, tmp_path): """One sample → 'awaiting another sample' state.""" from fenris.status import get_status db = tmp_path / "test.db" conn = init_store(db) conn.execute( "INSERT INTO samples (ts, device, mn, sn, fr, capacity_bytes, " "percentage_used, available_spare, media_errors, power_on_hours, " "power_cycles, unsafe_shutdowns, temperature_c, " "data_units_written, data_units_read, bytes_written, bytes_read, " "critical_warning) " "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ("2026-09-01T12:00:00+00:00", "/dev/nvme0", "Test", "SN", "FR", 1000000000000, 5, 100, 0, 1000, 100, 0, 35, 1000000, 500000, 512000000000, 256000000000, 0), ) conn.commit() conn.close() now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc) from unittest.mock import patch with patch("fenris.status.query_service_state", return_value={ "boot_enabled": False, "timer_active": False, "last_collect_ok": None, "last_collect_age_s": None, "last_collect_reason": None, }): result = get_status(store_path=db, clock_now=now, query_services=False, query_journal=False) # Should mention awaiting or insufficient data lower = result.lower() assert "awaiting" in lower or "another sample" in lower or "no projection" in lower def test_one_sample_awaiting_another_in_tui(self, tmp_path): """One sample → TUI shows awaiting state.""" from fenris.tui import FenrisTuiApp db = tmp_path / "test.db" conn = init_store(db) conn.execute( "INSERT INTO samples (ts, device, mn, sn, fr, capacity_bytes, " "percentage_used, available_spare, media_errors, power_on_hours, " "power_cycles, unsafe_shutdowns, temperature_c, " "data_units_written, data_units_read, bytes_written, bytes_read, " "critical_warning) " "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ("2026-09-01T12:00:00+00:00", "/dev/nvme0", "Test", "SN", "FR", 1000000000000, 5, 100, 0, 1000, 100, 0, 35, 1000000, 500000, 512000000000, 256000000000, 0), ) conn.commit() conn.close() now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc) # Test the projection handles single sample from fenris.projection import compute_projection, ConfidenceState conn = sqlite3.connect(db) proj = compute_projection(conn, now) conn.close() # With only one sample, projection should be unavailable assert proj.confidence_state == ConfidenceState.UNSUPPORTED def test_two_same_hour_samples_show_measured_usage(self, config_fixture, sysfs_fixture_tree): """Two compatible same-hour samples show measured usage.""" t1 = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc) t2 = datetime(2026, 9, 1, 12, 5, 0, tzinfo=timezone.utc) clock1 = FakeClock(t1) run_collection(self._make_sample_helper(12345678), sysfs_fixture_tree / "sys" / "class" / "nvme" / "nvme0", config_fixture, clock1) clock2 = FakeClock(t2) run_collection(self._make_sample_helper(12345718), sysfs_fixture_tree / "sys" / "class" / "nvme" / "nvme0", config_fixture, clock2) from fenris.status import get_status from unittest.mock import patch now = datetime(2026, 9, 1, 12, 10, 0, tzinfo=timezone.utc) with patch("fenris.status.query_service_state", return_value={ "boot_enabled": False, "timer_active": False, "last_collect_ok": None, "last_collect_age_s": None, "last_collect_reason": None, }): result = get_status(store_path=Path(config_fixture["store_path"]), clock_now=now, query_services=False, query_journal=False) # Should not say "no observations" or "awaiting" lower = result.lower() assert "no observations yet" not in lower def _make_sample_helper(self, duw_units: 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_units, "data_units_read": 9876543, "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", } # --------------------------------------------------------------------------- # Pause crossing and counter reset # --------------------------------------------------------------------------- class TestPauseCrossing: """Delta across monitoring period gap (issue #73 AC4).""" def test_pause_crossing_preserves_prior_history(self, config_fixture, sysfs_fixture_tree): """Delta across a paused period preserves prior hour observations.""" t1 = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc) t_pause = datetime(2026, 9, 1, 13, 0, 0, tzinfo=timezone.utc) t_resume = datetime(2026, 9, 1, 14, 0, 0, tzinfo=timezone.utc) t2 = datetime(2026, 9, 1, 14, 5, 0, tzinfo=timezone.utc) duw1 = 12345678 duw2 = duw1 + 100 # First sample (opens period) clock1 = FakeClock(t1) run_collection(self._make_sample_for_pause(duw1), sysfs_fixture_tree / "sys" / "class" / "nvme" / "nvme0", config_fixture, clock1) # Pause conn = sqlite3.connect(config_fixture["store_path"]) close_period(conn, t_pause, "user_disabled") conn.close() # Resume with new sample clock_resume = FakeClock(t_resume) run_collection(self._make_sample_for_pause(duw1), sysfs_fixture_tree / "sys" / "class" / "nvme" / "nvme0", config_fixture, clock_resume) # Second sample after resume clock2 = FakeClock(t2) run_collection(self._make_sample_for_pause(duw2), sysfs_fixture_tree / "sys" / "class" / "nvme" / "nvme0", config_fixture, clock2) # Verify prior hour observations are intact conn = sqlite3.connect(config_fixture["store_path"]) hour_12 = conn.execute( "SELECT bytes_written_delta FROM hour_observations WHERE hour LIKE '2026-09-01T12%'" ).fetchone() conn.close() # Hour 12 should still have data from the first collection assert hour_12 is not None, "Prior hour observation should be preserved" def _make_sample_for_pause(self, duw_units: 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_units, "data_units_read": 9876543, "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", } class TestCounterReset: """Counter reset / replacement opens new segment (issue #73 AC6).""" def test_duw_decrease_opens_new_segment(self, config_fixture, sysfs_fixture_tree): """DUW decrease triggers new segment.""" t1 = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc) t2 = datetime(2026, 9, 1, 12, 5, 0, tzinfo=timezone.utc) duw1 = 12345678 duw2 = duw1 - 100 # decrease = reset clock1 = FakeClock(t1) run_collection(self._make_sample_for_reset(duw1), sysfs_fixture_tree / "sys" / "class" / "nvme" / "nvme0", config_fixture, clock1) clock2 = FakeClock(t2) run_collection(self._make_sample_for_reset(duw2), sysfs_fixture_tree / "sys" / "class" / "nvme" / "nvme0", config_fixture, clock2) conn = sqlite3.connect(config_fixture["store_path"]) segments = conn.execute("SELECT COUNT(*) FROM controller_segments").fetchone()[0] samples = conn.execute("SELECT COUNT(*) FROM samples").fetchone()[0] conn.close() # Should have 2 segments (new one opened for DUW decrease) assert segments == 2 # Should have 2 samples assert samples == 2 def _make_sample_for_reset(self, duw_units: 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_units, "data_units_read": 9876543, "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", } # --------------------------------------------------------------------------- # Derivation failure preserves prior history # --------------------------------------------------------------------------- class TestDerivationFailure: """Injected derivation failure preserves prior history (issue #73 AC6).""" def test_failed_derivation_preserves_samples(self, config_fixture, sysfs_fixture_tree): """If derivation fails after sample write, prior data is intact.""" t1 = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc) t2 = datetime(2026, 9, 1, 12, 5, 0, tzinfo=timezone.utc) clock1 = FakeClock(t1) run_collection(self._make_sample_for_failure(12345678), sysfs_fixture_tree / "sys" / "class" / "nvme" / "nvme0", config_fixture, clock1) # Verify first sample exists conn = sqlite3.connect(config_fixture["store_path"]) count = conn.execute("SELECT COUNT(*) FROM samples").fetchone()[0] conn.close() assert count == 1 # Second sample with valid data should succeed clock2 = FakeClock(t2) r2 = run_collection(self._make_sample_for_failure(12345718), sysfs_fixture_tree / "sys" / "class" / "nvme" / "nvme0", config_fixture, clock2) assert r2["ok"] # Both samples should exist conn = sqlite3.connect(config_fixture["store_path"]) count = conn.execute("SELECT COUNT(*) FROM samples").fetchone()[0] conn.close() assert count == 2 def _make_sample_for_failure(self, duw_units: 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_units, "data_units_read": 9876543, "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", } # --------------------------------------------------------------------------- # Monitoring period is opened by collector # --------------------------------------------------------------------------- class TestMonitoringPeriod: """Collector ensures monitoring period is open (issue #73 AC2).""" def test_first_sample_opens_period(self, config_fixture, sysfs_fixture_tree): """First collection run opens a monitoring period.""" t1 = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc) clock1 = FakeClock(t1) run_collection(self._make_sample_simple(), sysfs_fixture_tree / "sys" / "class" / "nvme" / "nvme0", config_fixture, clock1) conn = sqlite3.connect(config_fixture["store_path"]) period = get_open_period(conn) conn.close() assert period is not None, "A monitoring period should be open" def test_subsequent_sample_keeps_period_open(self, config_fixture, sysfs_fixture_tree): """Subsequent collection runs keep the period open.""" t1 = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc) t2 = datetime(2026, 9, 1, 12, 5, 0, tzinfo=timezone.utc) clock1 = FakeClock(t1) run_collection(self._make_sample_simple(), sysfs_fixture_tree / "sys" / "class" / "nvme" / "nvme0", config_fixture, clock1) clock2 = FakeClock(t2) run_collection(self._make_sample_simple(), sysfs_fixture_tree / "sys" / "class" / "nvme" / "nvme0", config_fixture, clock2) conn = sqlite3.connect(config_fixture["store_path"]) period = get_open_period(conn) periods_count = conn.execute("SELECT COUNT(*) FROM monitoring_periods").fetchone()[0] conn.close() assert period is not None assert periods_count == 1 # Still only one period def _make_sample_simple(self) -> 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": 12345678, "data_units_read": 9876543, "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", }