Fenris/tests/test_collector_history_tracer.py

"""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 == 6  # v1→v2→v3→v4→v5→v6→v7

        # 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


class TestCollectionAtomicity:
    """Collection exposes sample and derived evidence as one publication."""

    @pytest.mark.asyncio
    async def test_collection_is_visible_through_cli_and_tui_readers(
        self, config_fixture, smartctl_fixture, sysfs_fixture_tree, monkeypatch
    ):
        """Ordinary readers observe matching published sample and activity evidence."""
        from fenris.status import get_status, read_status
        from fenris.tui import FenrisTuiApp

        monkeypatch.setenv("TZ", "UTC")
        service = {
            "boot_enabled": True,
            "timer_active": True,
            "last_collect_ok": True,
        }
        monkeypatch.setattr("fenris.status.query_service_state", lambda: service)

        now = datetime.now(timezone.utc).replace(
            minute=35, second=0, microsecond=0,
        )
        first = {
            **smartctl_fixture,
            "nvme_smart_health_information_log": {
                **smartctl_fixture["nvme_smart_health_information_log"],
                "data_units_written": 12345678,
                "data_units_read": 9876543,
            },
        }
        second = {
            **smartctl_fixture,
            "nvme_smart_health_information_log": {
                **smartctl_fixture["nvme_smart_health_information_log"],
                "data_units_written": 12345698,
                "data_units_read": 9876553,
            },
        }
        sysfs_path = sysfs_fixture_tree / "sys" / "class" / "nvme" / "nvme0"

        assert run_collection(
            first, sysfs_path, config_fixture, FakeClock(now - timedelta(minutes=5))
        )["ok"] is True
        assert run_collection(
            second, sysfs_path, config_fixture, FakeClock(now)
        )["ok"] is True

        store_path = Path(config_fixture["store_path"])
        with read_status(store_path, now, query_services=False) as (reader, composition):
            assert reader is not None
            assert composition.sample_count == 2
            assert composition.day_count == 1
            utc_day = reader.execute(
                "SELECT bytes_written_delta, bytes_read_delta FROM day_aggregates"
            ).fetchone()
            local_day = reader.execute(
                "SELECT bytes_written, bytes_read FROM local_days"
            ).fetchone()
            assert tuple(utc_day) == (10_240_000, 5_120_000)
            assert tuple(local_day) == (10_240_000, 5_120_000)

        cli_output = get_status(
            store_path, now, query_services=True, query_journal=False
        )
        assert "Monitoring" in cli_output

        app = FenrisTuiApp(store_path=store_path)
        async with app.run_test(size=(100, 30)) as pilot:
            await pilot.pause()
            live_readout = str(app.query_one("#live-readout").render())
            assert "W 0.010 GB" in live_readout
            assert "R 0.005 GB" in live_readout

        repeated_result = run_collection(
            second, sysfs_path, config_fixture, FakeClock(now + timedelta(minutes=5))
        )
        assert repeated_result["ok"] is True
        with read_status(
            store_path, now + timedelta(minutes=5), query_services=False
        ) as (reader, composition):
            assert reader is not None
            assert composition.sample_count == 3
            utc_day = reader.execute(
                "SELECT bytes_written_delta, bytes_read_delta FROM day_aggregates"
            ).fetchone()
            local_day = reader.execute(
                "SELECT bytes_written, bytes_read FROM local_days"
            ).fetchone()
            assert tuple(utc_day) == (10_240_000, 5_120_000)
            assert tuple(local_day) == (10_240_000, 5_120_000)

    def test_failed_local_day_publication_keeps_previous_publication(
        self, config_fixture, smartctl_fixture, sysfs_fixture_tree, monkeypatch
    ):
        """A failed final derivation step leaves all prior reader state intact."""
        from fenris.status import read_status

        monkeypatch.setenv("TZ", "UTC")
        now = datetime.now(timezone.utc).replace(second=0, microsecond=0)
        first_clock = FakeClock(now)
        first = {
            **smartctl_fixture,
            "nvme_smart_health_information_log": {
                **smartctl_fixture["nvme_smart_health_information_log"],
                "data_units_written": 12345678,
                "data_units_read": 9876543,
            },
        }
        first_result = run_collection(
            first,
            sysfs_fixture_tree / "sys" / "class" / "nvme" / "nvme0",
            config_fixture,
            first_clock,
        )
        assert first_result["ok"] is True, first_result

        writer = sqlite3.connect(config_fixture["store_path"])
        writer.execute(
            "CREATE TRIGGER fail_local_day_publication "
            "BEFORE INSERT ON local_days "
            "BEGIN SELECT RAISE(ABORT, 'injected local-day publication failure'); END"
        )
        writer.commit()
        writer.close()

        next_time = now + timedelta(minutes=5)
        second = {
            **smartctl_fixture,
            "nvme_smart_health_information_log": {
                **smartctl_fixture["nvme_smart_health_information_log"],
                "data_units_written": 12345698,
                "data_units_read": 9876548,
            },
        }
        failed_result = run_collection(
            second,
            sysfs_fixture_tree / "sys" / "class" / "nvme" / "nvme0",
            config_fixture,
            FakeClock(next_time),
        )

        assert failed_result["ok"] is False
        assert "injected local-day publication failure" in failed_result["error"]

        with read_status(
            Path(config_fixture["store_path"]), next_time, query_services=False
        ) as (reader, composition):
            assert reader is not None
            assert composition.sample_count == 1
            public_counts = reader.execute(
                "SELECT (SELECT COUNT(*) FROM samples), "
                "(SELECT COUNT(*) FROM controller_segments), "
                "(SELECT COUNT(*) FROM monitoring_periods), "
                "(SELECT COUNT(*) FROM hour_observations), "
                "(SELECT COUNT(*) FROM day_aggregates), "
                "(SELECT COUNT(*) FROM local_days)"
            ).fetchone()

        assert tuple(public_counts) == (1, 1, 1, 0, 0, 0)


# ---------------------------------------------------------------------------
# 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",
        }