362 lines
11 KiB
Python
362 lines
11 KiB
Python
"""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 os
|
|
import sqlite3
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Dict
|
|
|
|
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
|
|
from fenris.store import init_store
|
|
|
|
|
|
# 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 the observation, derived-history, and metadata tables 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")
|
|
expected_tables.add("store_metadata")
|
|
expected_tables.add("local_days")
|
|
expected_tables.add("pending_publications")
|
|
expected_tables.add("local_day_unallocated_evidence")
|
|
expected_tables.add("local_day_segment_totals")
|
|
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()
|