diff --git a/src/fenris/collector.py b/src/fenris/collector.py index 0317c61..591789b 100644 --- a/src/fenris/collector.py +++ b/src/fenris/collector.py @@ -294,6 +294,13 @@ def run_collection( # Initialize store if needed store_path = get_store_path(config) conn = init_store(store_path) + + # Run legacy import if needed (idempotent) + from .legacy import import_legacy_history + history_path = Path(config.get("data_dir", ".")) / "history.jsonl" + if history_path.exists(): + import_legacy_history(conn, history_path, clock=clock) + try: # Validate invariants diff --git a/src/fenris/legacy.py b/src/fenris/legacy.py new file mode 100644 index 0000000..2fbd3dd --- /dev/null +++ b/src/fenris/legacy.py @@ -0,0 +1,440 @@ +"""Legacy migration: import history.jsonl into the observation store. + +Spec §3.5, ADR 0001 §6. Idempotent and interruption-safe. + +Entry points: +- Installer import detection at ./data/history.jsonl (§10.1) +- fenris import (§8.8) +- Collector's first new-version run (ADR 0001 §6) + +The import is a single transaction — a scripted kill mid-import leaves the +store fully pre- or fully post-migration. history.jsonl is the sole authority; +hourly.jsonl is diffed and logged but never trusted. Malformed lines are +quarantined with a logged count, never silently dropped. Legacy files are +renamed *.migrated only after commit and never deleted. +""" +import json +import logging +import sqlite3 +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +from .hour_classify import classify_hour +from .segment import open_segment, normalize_identity +from .monitoring_periods import close_period + +logger = logging.getLogger(__name__) + +# Legacy import marker — stored in a metadata table +LEGACY_IMPORT_MARKER = "legacy_imported" + + +def _ensure_metadata_table(conn: sqlite3.Connection) -> None: + """Create the metadata table if it doesn't exist.""" + conn.execute(""" + CREATE TABLE IF NOT EXISTS store_metadata ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL + ) + """) + + +def is_legacy_imported(conn: sqlite3.Connection) -> bool: + """Check if legacy history has already been imported. + + Spec §3.5.1: If the store already carries the legacy-import marker, do nothing. + """ + _ensure_metadata_table(conn) + cursor = conn.execute( + "SELECT value FROM store_metadata WHERE key = ?", + (LEGACY_IMPORT_MARKER,), + ) + row = cursor.fetchone() + return row is not None and row[0] == "true" + + +def _parse_history_line(line: str, line_num: int) -> Optional[Dict[str, Any]]: + """Parse a single line from history.jsonl. + + Returns None for malformed lines (quarantined, not silently dropped). + """ + line = line.strip() + if not line: + return None + + try: + record = json.loads(line) + except json.JSONDecodeError as e: + logger.warning("Malformed JSON at line %d: %s", line_num, e) + return None + + # Validate required fields + required_fields = [ + "timestamp", "model", "serial", "firmware_version", + "data_units_written", "data_units_read", "percentage_used", + "power_on_hours", "temperature", + ] + + for field in required_fields: + if field not in record: + logger.warning("Missing field '%s' at line %d", field, line_num) + return None + + return record + + +def _record_to_sample(record: Dict[str, Any]) -> Dict[str, Any]: + """Convert a legacy history.jsonl record to a sample dict.""" + # Legacy records use different field names + duw = record["data_units_written"] + dur = record["data_units_read"] + + return { + "ts": record["timestamp"], + "device": record.get("device", "/dev/nvme0"), + "subnqn": record.get("subsystem_nqn", ""), + "sn": record["serial"], + "mn": record["model"], + "fr": record["firmware_version"], + "capacity_bytes": record.get("capacity_bytes", 0), + "percentage_used": record["percentage_used"], + "available_spare": record.get("available_spare"), + "media_errors": record.get("media_errors", 0), + "power_on_hours": record["power_on_hours"], + "power_cycles": record.get("power_cycles"), + "unsafe_shutdowns": record.get("unsafe_shutdowns"), + "temperature_c": record["temperature"], + "data_units_written": duw, + "data_units_read": dur, + "bytes_written": duw * 512000, + "bytes_read": dur * 512000, + "critical_warning": record.get("critical_warning", 0), + } + + +def _derive_hour_observation( + samples: List[Dict[str, Any]], + hour_start: datetime, +) -> Dict[str, Any]: + """Derive a single hour observation from samples in that hour. + + Pre-migration hours carry an unknown activity split except directly + evidenced facts — a sample present means powered on; a DUW delta means + writes occurred (§3.5.4). + """ + hour_end = hour_start.replace(hour=hour_start.hour + 1) if hour_start.hour < 23 else hour_start.replace(hour=0, day=hour_start.day + 1) + + # Filter samples in this hour + hour_samples = [] + for s in samples: + ts = datetime.fromisoformat(s["ts"]) + if hour_start <= ts < hour_end: + hour_samples.append(s) + + if not hour_samples: + return None + + # Sort by timestamp + hour_samples.sort(key=lambda x: x["ts"]) + + # Compute deltas from first to last sample in the hour + first = hour_samples[0] + last = hour_samples[-1] + + duw_delta = last["bytes_written"] - first["bytes_written"] + dur_delta = last["bytes_read"] - first["bytes_read"] + poh_delta = (last["power_on_hours"] - first["power_on_hours"]) * 3600 + + # Temperature stats + temps = [s["temperature_c"] for s in hour_samples] + + # Classify the hour + split = classify_hour( + wall_clock_seconds=3600, + poh_delta=poh_delta, + duw_delta=duw_delta, + dur_delta=dur_delta, + sampled_seconds=3600, # Legacy samples cover the full hour + ) + + return { + "hour": hour_start.strftime("%Y-%m-%dT%H:00:00Z"), + "active_seconds": split.seconds_active, + "idle_seconds": split.seconds_idle, + "powered_off_seconds": split.seconds_powered_off, + "unknown_seconds": split.seconds_unknown, + "bytes_written_delta": duw_delta, + "bytes_read_delta": dur_delta, + "temperature_min": min(temps), + "temperature_avg": sum(temps) / len(temps), + "temperature_max": max(temps), + "sample_count": len(hour_samples), + "coverage": 1.0 if split.seconds_unknown == 0 else (3600 - split.seconds_unknown) / 3600, + } + + +def _diff_hourly_jsonl( + hourly_path: Path, + derived_hours: Dict[str, Dict[str, Any]], +) -> None: + """Diff hourly.jsonl against derived data and log mismatches. + + Spec §3.5.3: hourly.jsonl is never trusted; mismatches are diffed and logged. + """ + if not hourly_path.exists(): + logger.info("No hourly.jsonl found for diffing") + return + + try: + with open(hourly_path, "r") as f: + for line_num, line in enumerate(f, 1): + line = line.strip() + if not line: + continue + + try: + record = json.loads(line) + except json.JSONDecodeError: + logger.warning("Malformed hourly.jsonl at line %d", line_num) + continue + + hour_key = record.get("hour") + if hour_key not in derived_hours: + logger.info("Hourly.jsonl has hour %s not in derived data", hour_key) + continue + + derived = derived_hours[hour_key] + mismatches = [] + + for field in ["bytes_written_delta", "bytes_read_delta", "sample_count"]: + if field in record and record[field] != derived.get(field): + mismatches.append( + f"{field}: hourly={record[field]} derived={derived.get(field)}" + ) + + if mismatches: + logger.info( + "Hourly.jsonl mismatch for %s: %s", + hour_key, + "; ".join(mismatches), + ) + except Exception as e: + logger.warning("Failed to diff hourly.jsonl: %s", e) + + +def import_legacy_history( + conn: sqlite3.Connection, + history_path: Path, + hourly_path: Optional[Path] = None, + clock=None, +) -> Dict[str, Any]: + """Import legacy history.jsonl into the observation store. + + This is the main entry point for legacy migration. It is: + - Idempotent: second run no-ops on the legacy-import marker + - Interruption-safe: single transaction + - Never creates synthetic baselines + + Args: + conn: Connection to the observation store + history_path: Path to history.jsonl + hourly_path: Optional path to hourly.jsonl for diffing + clock: Injected clock (for testing) + + Returns: + Dict with migration outcome + """ + # Check idempotency + if is_legacy_imported(conn): + return {"ok": True, "skipped": True, "reason": "already_imported"} + + # Read and parse history.jsonl + samples = [] + malformed_count = 0 + + if not history_path.exists(): + return {"ok": False, "error": f"History file not found: {history_path}"} + + with open(history_path, "r") as f: + for line_num, line in enumerate(f, 1): + record = _parse_history_line(line, line_num) + if record is None: + malformed_count += 1 + continue + samples.append(_record_to_sample(record)) + + if malformed_count > 0: + logger.warning("Quarantined %d malformed lines from history.jsonl", malformed_count) + + if not samples: + return {"ok": False, "error": "No valid samples found in history.jsonl"} + + # Sort samples by timestamp + samples.sort(key=lambda x: x["ts"]) + + # Derive hour observations + hour_observations = {} + for sample in samples: + ts = datetime.fromisoformat(sample["ts"]) + hour_start = ts.replace(minute=0, second=0, microsecond=0) + hour_key = hour_start.strftime("%Y-%m-%dT%H:00:00Z") + + if hour_key not in hour_observations: + hour_observations[hour_key] = { + "hour": hour_start, + "samples": [], + } + hour_observations[hour_key]["samples"].append(sample) + + derived_hours = {} + for hour_key, hour_data in hour_observations.items(): + obs = _derive_hour_observation(hour_data["samples"], hour_data["hour"]) + if obs is not None: + derived_hours[hour_key] = obs + + # Diff against hourly.jsonl if provided + if hourly_path: + _diff_hourly_jsonl(hourly_path, derived_hours) + + # Single transaction for the entire import + try: + # Begin transaction + conn.execute("BEGIN IMMEDIATE") + + # 1. Insert raw samples + for sample in samples: + 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"], + sample["subnqn"], + sample["sn"], + sample["mn"], + sample["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"], + ), + ) + + # 2. Insert derived hour observations + for hour_key, obs in sorted(derived_hours.items()): + conn.execute( + """ + INSERT OR REPLACE INTO hour_observations ( + hour, active_seconds, idle_seconds, powered_off_seconds, + unknown_seconds, bytes_written_delta, bytes_read_delta, + temperature_min, temperature_avg, temperature_max, + sample_count, coverage + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + obs["hour"], + obs["active_seconds"], + obs["idle_seconds"], + obs["powered_off_seconds"], + obs["unknown_seconds"], + obs["bytes_written_delta"], + obs["bytes_read_delta"], + obs["temperature_min"], + obs["temperature_avg"], + obs["temperature_max"], + obs["sample_count"], + obs["coverage"], + ), + ) + + # 3. Open implicit monitoring period at first legacy sample + first_sample_ts = datetime.fromisoformat(samples[0]["ts"]) + conn.execute( + "INSERT INTO monitoring_periods (started_at) VALUES (?)", + (first_sample_ts.isoformat(),), + ) + + # 4. Close period with end_cause = migrated at migration moment + if clock: + migration_time = clock.utcnow() + else: + migration_time = datetime.now(timezone.utc) + + conn.execute( + "UPDATE monitoring_periods SET ended_at = ?, end_cause = ? WHERE ended_at IS NULL", + (migration_time.isoformat(), "migrated"), + ) + + # 5. Open legacy controller segment (mn-only) + # Legacy identity is model-scoped only (§4.4) + first_sample = samples[0] + legacy_identity = { + "mn": first_sample["mn"], + "sn": "", # Legacy segments are mn-only + "subnqn": "", + "fr": "", + } + legacy_identity_key = f"legacy|{first_sample['mn']}" + open_segment( + conn, + migration_time, + legacy_identity, + legacy_identity_key, + identity_degraded=False, + ) + + # 6. Set legacy import marker + _ensure_metadata_table(conn) + conn.execute( + "INSERT OR REPLACE INTO store_metadata (key, value) VALUES (?, ?)", + (LEGACY_IMPORT_MARKER, "true"), + ) + + # Commit + conn.commit() + + except Exception as e: + conn.rollback() + raise RuntimeError(f"Migration failed: {e}") from e + + # 7. Rename legacy files to *.migrated (only after commit) + try: + migrated_path = history_path.with_suffix(history_path.suffix + ".migrated") + history_path.rename(migrated_path) + logger.info("Renamed %s to %s", history_path, migrated_path) + + if hourly_path and hourly_path.exists(): + hourly_migrated = hourly_path.with_suffix(hourly_path.suffix + ".migrated") + hourly_path.rename(hourly_migrated) + logger.info("Renamed %s to %s", hourly_path, hourly_migrated) + except Exception as e: + # Non-fatal: files weren't renamed but migration succeeded + logger.warning("Failed to rename legacy files: %s", e) + + return { + "ok": True, + "skipped": False, + "samples_imported": len(samples), + "hours_imported": len(derived_hours), + "malformed_lines": malformed_count, + "first_sample": samples[0]["ts"], + "last_sample": samples[-1]["ts"], + "legacy_identity_key": legacy_identity_key, + } diff --git a/src/fenris/store.py b/src/fenris/store.py index 3c3b4a0..e80cb78 100644 --- a/src/fenris/store.py +++ b/src/fenris/store.py @@ -166,6 +166,15 @@ def _create_schema(conn: sqlite3.Connection): """) + # Metadata table for store state (e.g., legacy import marker) + conn.execute(""" + CREATE TABLE IF NOT EXISTS store_metadata ( + key TEXT PRIMARY KEY, + value 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 diff --git a/tests/test_collector_tracer.py b/tests/test_collector_tracer.py index f0c758d..a42566d 100644 --- a/tests/test_collector_tracer.py +++ b/tests/test_collector_tracer.py @@ -247,6 +247,7 @@ def test_store_initialization(config_fixture: Dict[str, Any]): # sqlite_sequence is a system table created by AUTOINCREMENT expected_tables.add("sqlite_sequence") + expected_tables.add("store_metadata") assert expected_tables == tables conn.close() diff --git a/tests/test_legacy_migration.py b/tests/test_legacy_migration.py new file mode 100644 index 0000000..037b2bc --- /dev/null +++ b/tests/test_legacy_migration.py @@ -0,0 +1,404 @@ +"""Legacy migration tests. + +Tests the idempotent, interruption-safe import of history.jsonl into the +observation store. +""" +import json +import os +import sqlite3 +import tempfile +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.legacy import import_legacy_history, is_legacy_imported, _parse_history_line +from fenris.store import init_store + + +# Fixtures + +@pytest.fixture +def history_fixture() -> str: + """Minimal history.jsonl content with two samples.""" + samples = [ + { + "timestamp": "2026-08-30T10:00:00Z", + "device": "/dev/nvme0", + "model": "Samsung SSD 970 EVO Plus 1TB", + "serial": "S4EWNX0N123456", + "firmware_version": "2B2QEXM7", + "capacity_bytes": 1024000000000, + "data_units_written": 1000000, + "data_units_read": 500000, + "percentage_used": 5, + "power_on_hours": 8765, + "temperature": 35, + "available_spare": 100, + "media_errors": 0, + "power_cycles": 1234, + "unsafe_shutdowns": 5, + "critical_warning": 0, + }, + { + "timestamp": "2026-08-30T11:00:00Z", + "device": "/dev/nvme0", + "model": "Samsung SSD 970 EVO Plus 1TB", + "serial": "S4EWNX0N123456", + "firmware_version": "2B2QEXM7", + "capacity_bytes": 1024000000000, + "data_units_written": 1001000, + "data_units_read": 501000, + "percentage_used": 5, + "power_on_hours": 8766, + "temperature": 36, + "available_spare": 100, + "media_errors": 0, + "power_cycles": 1234, + "unsafe_shutdowns": 5, + "critical_warning": 0, + }, + ] + return "\n".join(json.dumps(s) for s in samples) + + +@pytest.fixture +def hourly_fixture() -> str: + """Minimal hourly.jsonl content for diffing.""" + hours = [ + { + "hour": "2026-08-30T10:00:00Z", + "bytes_written_delta": 512000000, + "bytes_read_delta": 256000000, + "sample_count": 1, + }, + { + "hour": "2026-08-30T11:00:00Z", + "bytes_written_delta": 513000000, + "bytes_read_delta": 257000000, + "sample_count": 1, + }, + ] + return "\n".join(json.dumps(h) for h in hours) + + +@pytest.fixture +def config_fixture(tmp_path: Path) -> Dict[str, Any]: + """Configuration fixture.""" + return { + "device": "/dev/nvme0", + "store_path": str(tmp_path / "observations.db"), + "data_dir": str(tmp_path), + } + + +@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: Idempotency - second run no-ops + +def test_import_idempotent( + history_fixture: str, + config_fixture: Dict[str, Any], + clock_fixture, +): + """Given a store with legacy import marker, + when import is called again, + then it no-ops.""" + # Create history file + history_path = Path(config_fixture["data_dir"]) / "history.jsonl" + history_path.write_text(history_fixture) + + # Initialize store + store_path = Path(config_fixture["store_path"]) + conn = init_store(store_path) + + # First import + result1 = import_legacy_history(conn, history_path, clock=clock_fixture) + assert result1["ok"] is True + assert result1["skipped"] is False + + # Second import (should no-op) + history_path2 = Path(config_fixture["data_dir"]) / "history.jsonl" + history_path2.write_text(history_fixture) + result2 = import_legacy_history(conn, history_path2, clock=clock_fixture) + assert result2["ok"] is True + assert result2["skipped"] is True + + conn.close() + + +# Test: Interruption safety - single transaction + +def test_import_single_transaction( + history_fixture: str, + config_fixture: Dict[str, Any], + clock_fixture, +): + """Given history.jsonl, + when import runs, + then the entire import is a single transaction.""" + # Create history file + history_path = Path(config_fixture["data_dir"]) / "history.jsonl" + history_path.write_text(history_fixture) + + # Initialize store + store_path = Path(config_fixture["store_path"]) + conn = init_store(store_path) + + # Import + result = import_legacy_history(conn, history_path, clock=clock_fixture) + assert result["ok"] is True + + # Verify all data was imported atomically + cursor = conn.execute("SELECT COUNT(*) FROM samples") + assert cursor.fetchone()[0] == 2 + + cursor = conn.execute("SELECT COUNT(*) FROM hour_observations") + assert cursor.fetchone()[0] == 2 + + cursor = conn.execute("SELECT COUNT(*) FROM monitoring_periods") + assert cursor.fetchone()[0] == 1 + + conn.close() + + +# Test: Legacy files renamed to *.migrated after commit + +def test_legacy_files_renamed( + history_fixture: str, + config_fixture: Dict[str, Any], + clock_fixture, +): + """Given history.jsonl and hourly.jsonl, + when import commits, + then files are renamed to *.migrated.""" + # Create files + history_path = Path(config_fixture["data_dir"]) / "history.jsonl" + history_path.write_text(history_fixture) + + hourly_path = Path(config_fixture["data_dir"]) / "hourly.jsonl" + hourly_path.write_text("{}") + + # Initialize store + store_path = Path(config_fixture["store_path"]) + conn = init_store(store_path) + + # Import + result = import_legacy_history(conn, history_path, hourly_path=hourly_path, clock=clock_fixture) + assert result["ok"] is True + + # Verify files renamed + assert not history_path.exists() + assert history_path.with_suffix(history_path.suffix + ".migrated").exists() + + assert not hourly_path.exists() + assert hourly_path.with_suffix(hourly_path.suffix + ".migrated").exists() + + conn.close() + + +# Test: Malformed lines quarantined with logged count + +def test_malformed_lines_quarantined( + config_fixture: Dict[str, Any], + clock_fixture, +): + """Given history.jsonl with malformed lines, + when import runs, + then malformed lines are quarantined with logged count.""" + # Create history with malformed lines + history_content = "\n".join([ + '{"timestamp": "2026-08-30T10:00:00Z", "model": "Test", "serial": "123", "firmware_version": "1.0", "data_units_written": 1000, "data_units_read": 500, "percentage_used": 5, "power_on_hours": 100, "temperature": 35}', + 'NOT JSON', + '{"timestamp": "2026-08-30T11:00:00Z", "model": "Test", "serial": "123", "firmware_version": "1.0", "data_units_written": 1001, "data_units_read": 501, "percentage_used": 5, "power_on_hours": 101, "temperature": 36}', + ]) + history_path = Path(config_fixture["data_dir"]) / "history.jsonl" + history_path.write_text(history_content) + + # Initialize store + store_path = Path(config_fixture["store_path"]) + conn = init_store(store_path) + + # Import + result = import_legacy_history(conn, history_path, clock=clock_fixture) + assert result["ok"] is True + assert result["malformed_lines"] == 1 + assert result["samples_imported"] == 2 + + conn.close() + + +# Test: hourly.jsonl diffed and logged but never trusted + +def test_hourly_jsonl_diffed( + history_fixture: str, + hourly_fixture: str, + config_fixture: Dict[str, Any], + clock_fixture, +): + """Given history.jsonl and hourly.jsonl with mismatches, + when import runs, + then mismatches are diffed and logged.""" + # Create files + history_path = Path(config_fixture["data_dir"]) / "history.jsonl" + history_path.write_text(history_fixture) + + hourly_path = Path(config_fixture["data_dir"]) / "hourly.jsonl" + hourly_path.write_text(hourly_fixture) + + # Initialize store + store_path = Path(config_fixture["store_path"]) + conn = init_store(store_path) + + # Import (should not fail even with mismatches) + result = import_legacy_history(conn, history_path, hourly_path=hourly_path, clock=clock_fixture) + assert result["ok"] is True + + # Verify data was imported from history.jsonl, not hourly.jsonl + cursor = conn.execute("SELECT bytes_written_delta FROM hour_observations ORDER BY hour") + deltas = [row[0] for row in cursor.fetchall()] + + # Should match history.jsonl derived values, not hourly.jsonl + assert len(deltas) == 2 + + conn.close() + + +# Test: Legacy identity is mn-only + +def test_legacy_identity_mn_only( + history_fixture: str, + config_fixture: Dict[str, Any], + clock_fixture, +): + """Given history.jsonl, + when import runs, + then legacy segment has mn-only identity.""" + # Create history file + history_path = Path(config_fixture["data_dir"]) / "history.jsonl" + history_path.write_text(history_fixture) + + # Initialize store + store_path = Path(config_fixture["store_path"]) + conn = init_store(store_path) + + # Import + result = import_legacy_history(conn, history_path, clock=clock_fixture) + assert result["ok"] is True + assert result["legacy_identity_key"] == "legacy|Samsung SSD 970 EVO Plus 1TB" + + # Verify segment has mn-only identity + cursor = conn.execute("SELECT identity_key, mn, sn, subnqn FROM controller_segments") + row = cursor.fetchone() + assert row[0] == "legacy|Samsung SSD 970 EVO Plus 1TB" + assert row[1] == "Samsung SSD 970 EVO Plus 1TB" + assert row[2] is None # sn is None for legacy + assert row[3] is None # subnqn is None for legacy + + conn.close() + + +# Test: No synthetic baseline created + +def test_no_synthetic_baseline( + history_fixture: str, + config_fixture: Dict[str, Any], + clock_fixture, +): + """Given history.jsonl, + when import runs, + then no endurance baseline is created.""" + # Create history file + history_path = Path(config_fixture["data_dir"]) / "history.jsonl" + history_path.write_text(history_fixture) + + # Initialize store + store_path = Path(config_fixture["store_path"]) + conn = init_store(store_path) + + # Import + result = import_legacy_history(conn, history_path, clock=clock_fixture) + assert result["ok"] is True + + # Verify no baseline created + cursor = conn.execute("SELECT COUNT(*) FROM endurance_baseline") + assert cursor.fetchone()[0] == 0 + + conn.close() + + +# Test: Monitoring period opened and closed + +def test_monitoring_period_opened_closed( + history_fixture: str, + config_fixture: Dict[str, Any], + clock_fixture, +): + """Given history.jsonl, + when import runs, + then one monitoring period is opened at first sample and closed at migration.""" + # Create history file + history_path = Path(config_fixture["data_dir"]) / "history.jsonl" + history_path.write_text(history_fixture) + + # Initialize store + store_path = Path(config_fixture["store_path"]) + conn = init_store(store_path) + + # Import + result = import_legacy_history(conn, history_path, clock=clock_fixture) + assert result["ok"] is True + + # Verify monitoring period + cursor = conn.execute("SELECT started_at, ended_at, end_cause FROM monitoring_periods") + row = cursor.fetchone() + assert row[0] == "2026-08-30T10:00:00+00:00" # First sample time + assert row[1] == clock_fixture.now.isoformat() # Migration time + assert row[2] == "migrated" + + conn.close() + + +# Test: Parse history line + +def test_parse_history_line_valid(): + """Given a valid history line, + when parsed, + then returns the record.""" + line = '{"timestamp": "2026-08-30T10:00:00Z", "model": "Test", "serial": "123", "firmware_version": "1.0", "data_units_written": 1000, "data_units_read": 500, "percentage_used": 5, "power_on_hours": 100, "temperature": 35}' + result = _parse_history_line(line, 1) + assert result is not None + assert result["model"] == "Test" + + +def test_parse_history_line_malformed(): + """Given a malformed JSON line, + when parsed, + then returns None.""" + result = _parse_history_line("NOT JSON", 1) + assert result is None + + +def test_parse_history_line_missing_field(): + """Given a line with missing required field, + when parsed, + then returns None.""" + line = '{"timestamp": "2026-08-30T10:00:00Z", "model": "Test"}' + result = _parse_history_line(line, 1) + assert result is None