From 2217b00ff6fc7c334652f14a0e50907ca024a623 Mon Sep 17 00:00:00 2001 From: xavierk Date: Tue, 1 Sep 2026 22:33:26 +0530 Subject: [PATCH] feat: collector tracer bullet (#21)\n\nSmartctl acquisition with validation\nSysfs controller identity acquisition\nIdentity normalization (strip, no case fold, blank handling)\nSQLite store in WAL mode with six entities\nSchema versioning via PRAGMA user_version\nInvariant validation (negative bytes, etc.)\n17 passing tests across two test files\n\nCloses #21 --- pyproject.toml | 22 ++ src/fenris/__init__.py | 2 + src/fenris/collector.py | 281 +++++++++++++++++++++ src/fenris/store.py | 187 ++++++++++++++ tests/test_collector_tracer.py | 358 +++++++++++++++++++++++++++ tests/test_identity_normalization.py | 96 +++++++ 6 files changed, 946 insertions(+) create mode 100644 pyproject.toml create mode 100644 src/fenris/__init__.py create mode 100644 src/fenris/collector.py create mode 100644 src/fenris/store.py create mode 100644 tests/test_collector_tracer.py create mode 100644 tests/test_identity_normalization.py diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..998e8b9 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,22 @@ +[project] +name = "fenris" +version = "0.3.0" +description = "NVMe wear monitor with persistent TUI" +requires-python = ">=3.9" +dependencies = [ + "textual>=0.40.0", +] + +[project.optional-dependencies] +dev = [ + "pytest>=7.0.0", + "pytest-cov>=4.0.0", +] + +[tool.pytest.ini_options] +testpaths = ["tests"] +python_files = ["test_*.py"] +python_functions = ["test_*"] +markers = [ + "slow: marks tests as slow", +] diff --git a/src/fenris/__init__.py b/src/fenris/__init__.py new file mode 100644 index 0000000..3df3b7e --- /dev/null +++ b/src/fenris/__init__.py @@ -0,0 +1,2 @@ +"""Fenris: NVMe wear monitor with persistent TUI.""" +__version__ = "0.3.0" diff --git a/src/fenris/collector.py b/src/fenris/collector.py new file mode 100644 index 0000000..bc16f7a --- /dev/null +++ b/src/fenris/collector.py @@ -0,0 +1,281 @@ +"""Collector: acquires counters and identity, writes to observation store. + +This module implements the thinnest complete write path: +- Acquire counters and thermal evidence from smartctl -a -j +- Acquire controller identity from sysfs +- Normalize identity exactly once at write time +- Validate every row against store invariants +- Commit one well-formed sample + +No code path outside the collector interrogates the device. +""" +import json +import sqlite3 +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, Optional, Tuple + +from .store import init_store, get_store_path + + +class AcquisitionError(Exception): + """Raised when acquisition fails - whole run is refused.""" + pass + + +class InvariantViolationError(Exception): + """Raised when a row would violate store invariants - writes nothing.""" + pass + + +def acquire_from_smartctl(smartctl_data: Dict[str, Any]) -> Dict[str, Any]: + """Acquire counters and thermal evidence from smartctl -a -j data. + + Validates that all required fields are present. + Raises AcquisitionError on any failure. + """ + required_fields = [ + "nvme_smart_health_information_log", + "user_capacity", + "model_name", + "serial_number", + "firmware_version", + ] + + for field in required_fields: + if field not in smartctl_data: + raise AcquisitionError(f"Missing required field in smartctl data: {field}") + + log = smartctl_data["nvme_smart_health_information_log"] + required_log_fields = [ + "data_units_written", + "data_units_read", + "percentage_used", + "power_on_hours", + "temperature", + ] + + for field in required_log_fields: + if field not in log: + raise AcquisitionError(f"Missing required field in SMART log: {field}") + + return { + "model": smartctl_data["model_name"], + "serial": smartctl_data["serial_number"], + "firmware_rev": smartctl_data["firmware_version"], + "capacity_bytes": smartctl_data["user_capacity"]["bytes"], + "percentage_used": log["percentage_used"], + "available_spare": log.get("available_spare"), + "media_errors": log.get("media_errors", 0), + "power_on_hours": log["power_on_hours"], + "power_cycles": log.get("power_cycles"), + "unsafe_shutdowns": log.get("unsafe_shutdowns"), + "temperature_c": log["temperature"], + "data_units_written": log["data_units_written"], + "data_units_read": log["data_units_read"], + "bytes_written": log["data_units_written"] * 512000, + "bytes_read": log["data_units_read"] * 512000, + "critical_warning": log.get("critical_warning", 0), + } + + +def acquire_from_sysfs(sysfs_path: Path) -> Dict[str, Any]: + """Acquire controller identity from sysfs. + + Reads identity from: + - /sys/class/nvme//subsysnqn (primary) + - /sys/class/nvme//model + - /sys/class/nvme//serial + - /sys/class/nvme//firmware_rev + - /sys/class/nvme//transport/ (optional) + + Raises AcquisitionError on any failure. + """ + identity_files = { + "subnqn": "subsysnqn", + "mn": "model", + "sn": "serial", + "fr": "firmware_rev", + } + + identity = {} + for key, filename in identity_files.items(): + filepath = sysfs_path / filename + if not filepath.exists(): + raise AcquisitionError(f"Missing sysfs file: {filepath}") + + try: + value = filepath.read_text().strip() + identity[key] = value if value else "" + except Exception as e: + raise AcquisitionError(f"Failed to read {filepath}: {e}") + + # Transport info (optional) + transport_dir = sysfs_path / "transport" + if transport_dir.exists(): + try: + transport_file = transport_dir / "trstring" + if transport_file.exists(): + identity["transport"] = transport_file.read_text().strip() + else: + identity["transport"] = None + except Exception: + identity["transport"] = None + else: + identity["transport"] = None + + return identity + + +def normalize_identity(identity: Dict[str, Any]) -> str: + """Normalize identity exactly once at write time. + + Rules: + - Strip trailing spaces and newlines + - No case folding + - Empty-after-strip stored blank + + Returns normalized identity key. + """ + # Primary key: normalized kernel-exposed subsystem NQN + key = identity.get("subnqn", "") + if key: + key = key.rstrip() + return key + + # Fallback 1: kernel composite (not implemented yet) + # Fallback 2: model|serial + mn = identity.get("mn", "").rstrip() + sn = identity.get("sn", "").rstrip() + if mn or sn: + return f"{mn}|{sn}" + + # All keys blank - degraded identity + return "" + + +def compute_identity_degraded(identity: Dict[str, Any]) -> bool: + """Check if identity is degraded (all key rungs empty).""" + key = normalize_identity(identity) + return key == "" + + +def validate_sample_invariants(sample: Dict[str, Any], conn: sqlite3.Connection) -> None: + """Validate sample against store invariants. + + Raises InvariantViolationError if any invariant is violated. + """ + # TODO: Implement more complex invariants as needed + # For now, just check basic constraints + + if sample.get("bytes_written", 0) < 0: + raise InvariantViolationError("Negative bytes_written") + + if sample.get("bytes_read", 0) < 0: + raise InvariantViolationError("Negative bytes_read") + + +def write_sample( + sample: Dict[str, Any], + identity: Dict[str, Any], + conn: sqlite3.Connection, + clock, +) -> None: + """Write one sample to the observation store. + + Identity normalization happens exactly once here. + """ + # Normalize identity exactly once at write time + identity_key = normalize_identity(identity) + identity_degraded = compute_identity_degraded(identity) + + # TODO: Implement full sample writing with controller segment handling + # For now, just insert a basic sample with identity fields + + cursor = conn.execute( + """ + INSERT INTO samples ( + ts, device, subnqn, sn, mn, 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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + sample["ts"], + sample["device"], + identity.get("subnqn", ""), + identity.get("sn", ""), + identity.get("mn", ""), + identity.get("fr", ""), + sample["capacity_bytes"], + sample["percentage_used"], + sample["available_spare"], + sample["media_errors"], + sample["power_on_hours"], + sample["power_cycles"], + sample["unsafe_shutdowns"], + sample["temperature_c"], + sample["data_units_written"], + sample["data_units_read"], + sample["bytes_written"], + sample["bytes_read"], + sample["critical_warning"], + ), + ) + + conn.commit() + + +def run_collection( + smartctl_data: Dict[str, Any], + sysfs_path: Path, + config: Dict[str, Any], + clock, +) -> Dict[str, Any]: + """Run one collection run. + + This is the main entry point for the collector. + Returns the run outcome. + """ + try: + # Acquire counters and thermal evidence + counters = acquire_from_smartctl(smartctl_data) + + # Acquire controller identity + identity = acquire_from_sysfs(sysfs_path) + + # Build sample with injected clock + sample = { + "ts": clock.utcnow().isoformat(), + "device": config["device"], + **counters, + **identity, + } + + # Initialize store if needed + store_path = get_store_path(config) + conn = init_store(store_path) + + try: + # Validate invariants + validate_sample_invariants(sample, conn) + + # Write sample + write_sample(sample, identity, conn, clock) + + return { + "ok": True, + "sample_count": 1, + "store_path": str(store_path), + } + finally: + conn.close() + + except (AcquisitionError, InvariantViolationError) as e: + return { + "ok": False, + "error": str(e), + "error_type": type(e).__name__, + } diff --git a/src/fenris/store.py b/src/fenris/store.py new file mode 100644 index 0000000..3c3b4a0 --- /dev/null +++ b/src/fenris/store.py @@ -0,0 +1,187 @@ +"""Observation store: SQLite database for persisting observation history. + +This module handles: +- Store initialization with WAL mode +- Schema versioning with PRAGMA user_version +- The six entities: samples, hour_observations, day_aggregates, + monitoring_periods, controller_segments, endurance_baseline +""" +import sqlite3 +from pathlib import Path +from typing import Optional + + +# Schema version - increment on each migration +SCHEMA_VERSION = 1 + + +def get_store_path(config: dict) -> Path: + """Get the store path from config.""" + return Path(config["store_path"]) + + +def init_store(store_path: Path) -> sqlite3.Connection: + """Initialize the observation store if not present. + + Creates the database with WAL mode and all six entities. + Returns a connection to the store. + """ + conn = sqlite3.connect(str(store_path)) + + # Enable WAL mode for concurrent reads during writes + conn.execute("PRAGMA journal_mode=WAL") + + # Check if this is a new database + cursor = conn.execute("PRAGMA user_version") + current_version = cursor.fetchone()[0] + + if current_version == 0: + # New database - create schema + _create_schema(conn) + conn.execute(f"PRAGMA user_version={SCHEMA_VERSION}") + conn.commit() + elif current_version > SCHEMA_VERSION: + # Unknown newer version - refuse + conn.close() + raise ValueError( + f"Observation store written by a newer Fenris (version {current_version}) " + f"— upgrade Fenris" + ) + elif current_version < SCHEMA_VERSION: + # Older version - apply migrations + _apply_migrations(conn, current_version) + conn.execute(f"PRAGMA user_version={SCHEMA_VERSION}") + conn.commit() + + return conn + + +def _create_schema(conn: sqlite3.Connection): + """Create the initial schema with all six entities.""" + + # Samples: raw collection runs (14-day retention) + conn.execute(""" + CREATE TABLE IF NOT EXISTS samples ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + ts TEXT NOT NULL, -- ISO 8601 UTC timestamp + device TEXT NOT NULL, + -- Normalized controller-identity fields captured at acquisition + 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 + ) + """) + + # Hour observations: UTC-hour usage-habit split + conn.execute(""" + CREATE TABLE IF NOT EXISTS hour_observations ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + hour TEXT NOT NULL UNIQUE, -- ISO 8601 UTC hour (e.g., "2026-09-01T12:00:00Z") + 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 + ) + """) + + # Day aggregates: derived from hour observations + conn.execute(""" + CREATE TABLE IF NOT EXISTS day_aggregates ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + day TEXT NOT NULL UNIQUE, -- ISO 8601 UTC day (e.g., "2026-09-01") + 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 + ) + """) + + # Monitoring periods: tracking when monitoring was enabled/disabled + conn.execute(""" + CREATE TABLE IF NOT EXISTS monitoring_periods ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + started_at TEXT NOT NULL, -- ISO 8601 UTC timestamp + ended_at TEXT, -- NULL if currently active + end_cause TEXT CHECK(end_cause IN ('user_disabled', 'migrated', 'unknown_gap')) + ) + """) + + # Controller segments: identity key plus metadata snapshot + conn.execute(""" + CREATE TABLE IF NOT EXISTS controller_segments ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + opened_at TEXT NOT NULL, -- ISO 8601 UTC timestamp + identity_key TEXT, -- Normalized identity key (NULL if degraded) + identity_degraded BOOLEAN DEFAULT 0, + subnqn TEXT, + sn TEXT, + mn TEXT, + fr TEXT, + vid TEXT, + ssvid TEXT, + transport TEXT + ) + """) + + # Endurance baseline: one active row, replaced on edit + conn.execute(""" + CREATE TABLE IF NOT EXISTS 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, -- 'user' or 'machine_match' + verified BOOLEAN DEFAULT 0, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ) + """) + + +def _apply_migrations(conn: sqlite3.Connection, current_version: int): + """Apply forward-only migrations from current_version to SCHEMA_VERSION.""" + # Future migrations will go here + # For now, just upgrade to current version + pass + + +def is_store_faulty(store_path: Path) -> bool: + """Check if the store is present but cannot be read or trusted.""" + if not store_path.exists(): + return False + + try: + conn = sqlite3.connect(f"file:{store_path}?mode=ro", uri=True) + conn.execute("PRAGMA user_version") + conn.close() + return False + except sqlite3.Error: + return True diff --git a/tests/test_collector_tracer.py b/tests/test_collector_tracer.py new file mode 100644 index 0000000..f0c758d --- /dev/null +++ b/tests/test_collector_tracer.py @@ -0,0 +1,358 @@ +"""Collector tracer bullet test. + +Tests the thinnest complete write path through the system: +- Input: smartctl-JSON fixture, sysfs fixture tree, config fixture, injected clock +- Output: resulting store contents, run outcome + +Seam: write side of the observation store database file. +""" +import json +import os +import sqlite3 +import tempfile +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, Generator + +import pytest + +# Add src to path for imports +import sys +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +from fenris.collector import run_collection, AcquisitionError, InvariantViolationError +from fenris.store import init_store, get_store_path + + +# 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) + + # Controller identity files + (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 info (optional, but we'll include it) + 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"), + } + + +@pytest.fixture +def clock_fixture(): + """Injected clock returning fixed time.""" + class FakeClock: + def __init__(self): + self.now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc) + + def utcnow(self): + return self.now + + return FakeClock() + + +# Test: Collector writes one well-formed sample + +def test_collector_writes_one_sample( + smartctl_fixture: Dict[str, Any], + sysfs_fixture_tree: Path, + config_fixture: Dict[str, Any], + clock_fixture, +): + """Given a smartctl fixture and sysfs fixture tree, + when the collector runs, + then one well-formed sample is written to the observation store.""" + result = run_collection( + smartctl_data=smartctl_fixture, + sysfs_path=sysfs_fixture_tree / "sys" / "class" / "nvme" / "nvme0", + config=config_fixture, + clock=clock_fixture, + ) + + # Verify run succeeded + assert result["ok"] is True, f"Collection failed: {result.get('error')}" + + # Verify store contents + conn = sqlite3.connect(config_fixture["store_path"]) + cursor = conn.execute("SELECT COUNT(*) FROM samples") + count = cursor.fetchone()[0] + assert count == 1 + + cursor = conn.execute("SELECT * FROM samples") + row = cursor.fetchone() + assert row is not None + + # Verify row contents match fixtures + # Row structure: id, ts, device, subnqn, sn, mn, 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 + assert row[2] == "/dev/nvme0" # device + assert row[3] == "nqn.2014-08.org.nvmexpress:uuid:12345678-1234-1234-1234-123456789abc" # subnqn + assert row[4] == "S4EWNX0N123456" # sn + assert row[5] == "Samsung SSD 970 EVO Plus 1TB" # mn + assert row[6] == "2B2QEXM7" # fr + assert row[7] == 1024000000000 # capacity_bytes + assert row[8] == 5 # percentage_used + assert row[15] == 12345678 # data_units_written + assert row[17] == 12345678 * 512000 # bytes_written + + conn.close() + + +# Test: Identity normalization applied exactly once at write time + +def test_identity_normalization( + smartctl_fixture: Dict[str, Any], + sysfs_fixture_tree: Path, + config_fixture: Dict[str, Any], + clock_fixture, +): + """Given sysfs identity with trailing spaces/newlines, + when the collector writes, + then identity is normalized exactly once at write time.""" + # Create identity with trailing whitespace + identity = { + "subnqn": "nqn.2014-08.org.nvmexpress:uuid:12345678-1234-1234-1234-123456789abc \n", + "mn": "Samsung SSD 970 EVO Plus 1TB\n", + "sn": "S4EWNX0N123456\n", + "fr": "2B2QEXM7", + "transport": "pcie", + } + + from fenris.collector import normalize_identity + + # Normalize once + key1 = normalize_identity(identity) + # Normalize again - should be identical + key2 = normalize_identity(identity) + + assert key1 == key2 + assert key1 == "nqn.2014-08.org.nvmexpress:uuid:12345678-1234-1234-1234-123456789abc" + assert "\n" not in key1 + assert key1 == key1.rstrip() # No trailing whitespace + + +# Test: Any acquisition failure fails the whole run + +def test_acquisition_failure_fails_run( + sysfs_fixture_tree: Path, + config_fixture: Dict[str, Any], + clock_fixture, +): + """Given a smartctl fixture with missing fields, + when the collector runs, + then the whole run fails and writes nothing.""" + # Missing required field + bad_smartctl = { + "json_format_version": [1, 0], + "smartctl": {"version": [7, 3]}, + # Missing nvme_smart_health_information_log + } + + result = run_collection( + smartctl_data=bad_smartctl, + sysfs_path=sysfs_fixture_tree / "sys" / "class" / "nvme" / "nvme0", + config=config_fixture, + clock=clock_fixture, + ) + + # Verify run failed + assert result["ok"] is False + assert "Missing required field" in result["error"] + + # Verify nothing was written + if os.path.exists(config_fixture["store_path"]): + conn = sqlite3.connect(config_fixture["store_path"]) + cursor = conn.execute("SELECT COUNT(*) FROM samples") + count = cursor.fetchone()[0] + assert count == 0 + conn.close() + else: + # Store wasn't even created - also valid + pass + + +# Test: Store initializes with six entities + +def test_store_initialization(config_fixture: Dict[str, Any]): + """Given no existing store, + when the collector runs, + then the store is initialized with six entities.""" + store_path = Path(config_fixture["store_path"]) + + # Store shouldn't exist yet + assert not store_path.exists() + + # Initialize store + conn = init_store(store_path) + + # Verify all six entities exist + cursor = conn.execute("SELECT name FROM sqlite_master WHERE type='table'") + tables = {row[0] for row in cursor.fetchall()} + + expected_tables = { + "samples", + "hour_observations", + "day_aggregates", + "monitoring_periods", + "controller_segments", + "endurance_baseline", + } + + # sqlite_sequence is a system table created by AUTOINCREMENT + expected_tables.add("sqlite_sequence") + assert expected_tables == tables + + conn.close() + + +# Test: Schema versioning with PRAGMA user_version + +def test_schema_versioning(config_fixture: Dict[str, Any]): + """Given a store with unknown newer version, + when the collector runs, + then it refuses to proceed.""" + store_path = Path(config_fixture["store_path"]) + + # Create a store with newer version + conn = sqlite3.connect(str(store_path)) + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA user_version=999") # Unknown newer version + conn.close() + + # Try to initialize - should fail + with pytest.raises(ValueError, match="newer Fenris"): + init_store(store_path) + + +def test_schema_version_current(config_fixture: Dict[str, Any]): + """Given a store with current version, + when the collector runs, + then it proceeds without migration.""" + from fenris.store import SCHEMA_VERSION + + store_path = Path(config_fixture["store_path"]) + + # Initialize store + conn1 = init_store(store_path) + conn1.close() + + # Open again - should succeed + conn2 = init_store(store_path) + + # Verify version is current + cursor = conn2.execute("PRAGMA user_version") + version = cursor.fetchone()[0] + assert version == SCHEMA_VERSION + + conn2.close() + + +def test_schema_version_older(config_fixture: Dict[str, Any]): + """Given a store with older version, + when the collector runs, + then it applies migrations and proceeds.""" + from fenris.store import SCHEMA_VERSION + + store_path = Path(config_fixture["store_path"]) + + # Create a store with older version + conn = sqlite3.connect(str(store_path)) + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA user_version=0") # Older version + conn.close() + + # Initialize store - should apply migrations + conn = init_store(store_path) + + # Verify version is current + cursor = conn.execute("PRAGMA user_version") + version = cursor.fetchone()[0] + assert version == SCHEMA_VERSION + + conn.close() + + +# Test: Invariant-violating run writes nothing + +def test_invariant_violation_writes_nothing( + smartctl_fixture: Dict[str, Any], + sysfs_fixture_tree: Path, + config_fixture: Dict[str, Any], + clock_fixture, +): + """Given a smartctl fixture that would violate store invariants, + when the collector runs, + then it writes nothing and fails visibly.""" + # Create a fixture that would cause invariant violation + # (negative bytes_written - we'll mock this) + bad_smartctl = smartctl_fixture.copy() + bad_smartctl["nvme_smart_health_information_log"] = { + **smartctl_fixture["nvme_smart_health_information_log"], + "data_units_written": -1, # This will cause negative bytes_written + } + + result = run_collection( + smartctl_data=bad_smartctl, + sysfs_path=sysfs_fixture_tree / "sys" / "class" / "nvme" / "nvme0", + config=config_fixture, + clock=clock_fixture, + ) + + # Verify run failed + assert result["ok"] is False + assert "InvariantViolation" in result["error_type"] + + # Verify nothing was written + if os.path.exists(config_fixture["store_path"]): + conn = sqlite3.connect(config_fixture["store_path"]) + cursor = conn.execute("SELECT COUNT(*) FROM samples") + count = cursor.fetchone()[0] + assert count == 0 + conn.close() diff --git a/tests/test_identity_normalization.py b/tests/test_identity_normalization.py new file mode 100644 index 0000000..2fd8e52 --- /dev/null +++ b/tests/test_identity_normalization.py @@ -0,0 +1,96 @@ +"""Test identity normalization per specification. + +From spec §2.3: +- strip trailing spaces and newlines +- no case folding +- empty-after-strip stored blank + +Padded and unpadded renderings of the same field yield byte-identical stored values. +""" +import pytest +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +from fenris.collector import normalize_identity + + +def test_strip_trailing_spaces(): + """Trailing spaces are stripped.""" + identity = {"subnqn": "nqn.2014-08.org.nvmexpress:uuid:12345678 "} + result = normalize_identity(identity) + assert result == "nqn.2014-08.org.nvmexpress:uuid:12345678" + + +def test_strip_trailing_newlines(): + """Trailing newlines are stripped.""" + identity = {"subnqn": "nqn.2014-08.org.nvmexpress:uuid:12345678\n"} + result = normalize_identity(identity) + assert result == "nqn.2014-08.org.nvmexpress:uuid:12345678" + + +def test_strip_trailing_spaces_and_newlines(): + """Trailing spaces and newlines are stripped.""" + identity = {"subnqn": "nqn.2014-08.org.nvmexpress:uuid:12345678 \n\n"} + result = normalize_identity(identity) + assert result == "nqn.2014-08.org.nvmexpress:uuid:12345678" + + +def test_no_case_folding(): + """Case is preserved - no case folding.""" + identity = {"subnqn": "NQN.2014-08.ORG.NVMEXPRESS:UUID:12345678"} + result = normalize_identity(identity) + assert result == "NQN.2014-08.ORG.NVMEXPRESS:UUID:12345678" + + +def test_empty_after_strip_stored_blank(): + """Empty after strip is stored as blank string.""" + identity = {"subnqn": " \n\n "} + result = normalize_identity(identity) + assert result == "" + + +def test_fallback_to_model_serial(): + """When subnqn is empty, falls back to model|serial.""" + identity = { + "subnqn": "", + "mn": "Samsung SSD 970 EVO Plus 1TB", + "sn": "S4EWNX0N123456", + } + result = normalize_identity(identity) + assert result == "Samsung SSD 970 EVO Plus 1TB|S4EWNX0N123456" + + +def test_fallback_model_serial_normalized(): + """Model and serial are also normalized.""" + identity = { + "subnqn": "", + "mn": "Samsung SSD 970 EVO Plus 1TB\n", + "sn": "S4EWNX0N123456 ", + } + result = normalize_identity(identity) + assert result == "Samsung SSD 970 EVO Plus 1TB|S4EWNX0N123456" + + +def test_all_keys_blank_returns_blank(): + """When all keys are blank, returns blank (degraded identity).""" + identity = { + "subnqn": "", + "mn": "", + "sn": "", + } + result = normalize_identity(identity) + assert result == "" + + +def test_byte_identical_for_padded_unpadded(): + """Padded and unpadded renderings yield byte-identical values.""" + padded = {"subnqn": "nqn.2014-08.org.nvmexpress:uuid:12345678 \n"} + unpadded = {"subnqn": "nqn.2014-08.org.nvmexpress:uuid:12345678"} + + result_padded = normalize_identity(padded) + result_unpadded = normalize_identity(unpadded) + + assert result_padded == result_unpadded + assert result_padded == "nqn.2014-08.org.nvmexpress:uuid:12345678"