"""Additional smartctl diagnostics (issue #111).

Field list fixed from real `smartctl -a -j` output (smartctl 7.5, NVMe 1.4):
nvme_smart_health_information_log carries warning_temp_time, critical_comp_time,
num_err_log_entries, host_reads, host_writes and controller_busy_time. The
thermal-management transition fields (thm_temp1/2_trans_count, thm_temp1/2_total_time)
appear only on drives that implement them, so they stay nullable.
"""
import sqlite3
import sys
from datetime import datetime, timezone
from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).parent.parent / "src"))

from fenris.activity_readout import diagnostics_text, readout
from fenris.activity_selection import ActivitySelection
from fenris.collector import acquire_from_smartctl, run_collection
from fenris.store import SCHEMA_VERSION, init_store, migrate_to_latest

NEW_COLUMNS = (
    "host_reads", "host_writes", "controller_busy_time", "num_err_log_entries",
    "warning_temp_time", "critical_comp_time",
    "thm_temp1_trans_count", "thm_temp1_total_time",
    "thm_temp2_trans_count", "thm_temp2_total_time",
)


def _smartctl(**extra_log):
    """Shape of real smartctl -a -j output for an NVMe drive."""
    log = {
        "critical_warning": 0, "temperature": 37, "available_spare": 100,
        "available_spare_threshold": 5, "percentage_used": 52,
        "data_units_read": 207986339, "data_units_written": 202978899,
        "host_reads": 3543447796, "host_writes": 2458441042,
        "controller_busy_time": 61626, "power_cycles": 5324,
        "power_on_hours": 11906, "unsafe_shutdowns": 159, "media_errors": 0,
        "num_err_log_entries": 0, "warning_temp_time": 0, "critical_comp_time": 0,
        "temperature_sensors": [37],
    }
    log.update(extra_log)
    return {
        "nvme_smart_health_information_log": log,
        "user_capacity": {"bytes": 512110190592},
        "model_name": "Drive", "serial_number": "SN1", "firmware_version": "FW1",
    }


class TestAcquisition:
    def test_real_fields_are_captured(self):
        got = acquire_from_smartctl(_smartctl())
        assert got["host_reads"] == 3543447796
        assert got["host_writes"] == 2458441042
        assert got["controller_busy_time"] == 61626
        assert got["num_err_log_entries"] == 0
        assert got["warning_temp_time"] == 0
        assert got["critical_comp_time"] == 0

    def test_absent_thermal_management_fields_are_none_not_zero(self):
        got = acquire_from_smartctl(_smartctl())
        for key in ("thm_temp1_trans_count", "thm_temp1_total_time",
                    "thm_temp2_trans_count", "thm_temp2_total_time"):
            assert got[key] is None

    def test_thermal_management_fields_captured_when_reported(self):
        got = acquire_from_smartctl(_smartctl(
            thm_temp1_trans_count=3, thm_temp1_total_time=120,
            thm_temp2_trans_count=1, thm_temp2_total_time=45))
        assert (got["thm_temp1_trans_count"], got["thm_temp1_total_time"]) == (3, 120)
        assert (got["thm_temp2_trans_count"], got["thm_temp2_total_time"]) == (1, 45)

    def test_missing_optional_counters_do_not_fail_the_run(self):
        data = _smartctl()
        for key in ("host_reads", "host_writes", "controller_busy_time",
                    "num_err_log_entries", "warning_temp_time", "critical_comp_time"):
            del data["nvme_smart_health_information_log"][key]
        got = acquire_from_smartctl(data)
        assert got["host_reads"] is None and got["warning_temp_time"] is None


@pytest.fixture
def sysfs(tmp_path):
    ctrl = tmp_path / "nvme0"
    ctrl.mkdir()
    for name, value in (("subsysnqn", "nqn.test"), ("model", "Drive"),
                        ("serial", "SN1"), ("firmware_rev", "FW1")):
        (ctrl / name).write_text(value + "\n")
    return ctrl


class _Clock:
    def utcnow(self):
        return datetime(2026, 9, 1, 12, 0, tzinfo=timezone.utc)


def test_collection_persists_new_fields(tmp_path, sysfs):
    config = {"device": "/dev/nvme0", "store_path": str(tmp_path / "o.db")}
    result = run_collection(smartctl_data=_smartctl(thm_temp1_trans_count=3),
                            sysfs_path=sysfs, config=config, clock=_Clock())
    assert result["ok"], result
    conn = sqlite3.connect(config["store_path"])
    conn.row_factory = sqlite3.Row
    row = conn.execute("SELECT * FROM samples").fetchone()
    assert row["host_reads"] == 3543447796
    assert row["controller_busy_time"] == 61626
    assert row["thm_temp1_trans_count"] == 3
    assert row["thm_temp2_trans_count"] is None


def _legacy_store(path):
    """A populated schema-6 store, as shipped before this change."""
    conn = init_store(path)
    for column in reversed(NEW_COLUMNS):
        conn.execute(f"ALTER TABLE samples DROP COLUMN {column}")
    conn.execute("PRAGMA user_version=6")
    for i in range(3):
        conn.execute(
            "INSERT INTO samples (ts, device, bytes_written, bytes_read, "
            "power_on_hours, critical_warning, segment_id) VALUES (?,?,?,?,?,?,1)",
            ("2026-09-30T1%d:00:00+00:00" % i, "/dev/nvme0", 10 + i, 5, 100 + i, 0))
    conn.commit()
    conn.close()


class TestMigration:
    def test_schema_version_is_seven(self):
        assert SCHEMA_VERSION == 7

    def test_populated_legacy_store_keeps_rows_and_gains_null_columns(self, tmp_path):
        db = tmp_path / "o.db"
        _legacy_store(db)
        before = sqlite3.connect(db).execute(
            "SELECT id, ts, device, bytes_written, power_on_hours FROM samples "
            "ORDER BY id").fetchall()
        assert migrate_to_latest(db) == 1
        conn = sqlite3.connect(db)
        assert conn.execute("PRAGMA user_version").fetchone()[0] == 7
        assert conn.execute(
            "SELECT id, ts, device, bytes_written, power_on_hours FROM samples "
            "ORDER BY id").fetchall() == before
        cols = {r[1] for r in conn.execute("PRAGMA table_info(samples)")}
        assert set(NEW_COLUMNS) <= cols
        nulls = conn.execute(
            "SELECT %s FROM samples" % " OR ".join(c + " IS NOT NULL" for c in NEW_COLUMNS)
        ).fetchall()
        assert nulls == [(0,)] * 3  # legacy rows stay unknown, never zero

    def test_migration_is_idempotent_and_init_store_applies_it(self, tmp_path):
        db = tmp_path / "o.db"
        _legacy_store(db)
        conn = init_store(db)
        assert conn.execute("PRAGMA user_version").fetchone()[0] == 7
        conn.close()
        assert migrate_to_latest(db) == 0

    def test_pending_payload_without_new_fields_still_publishes(self, tmp_path):
        """Observations staged by the previous version lack the new keys."""
        db = tmp_path / "o.db"
        conn = init_store(db)
        from fenris.collector import write_sample
        sample = {k: v for k, v in acquire_from_smartctl(_smartctl()).items()
                  if k not in NEW_COLUMNS}
        sample.update(ts="2026-09-01T12:00:00+00:00", device="/dev/nvme0")
        write_sample(sample, {"subnqn": "nqn.test"}, conn,
                     datetime(2026, 9, 1, 12, tzinfo=timezone.utc))
        assert conn.execute("SELECT host_reads FROM samples").fetchone() == (None,)

    def test_rollback_notes_procedure_restores_prior_schema(self, tmp_path):
        """Documented rollback: drop the columns, set user_version back to 6."""
        db = tmp_path / "o.db"
        _legacy_store(db)
        migrate_to_latest(db)
        conn = sqlite3.connect(db)
        conn.execute(
            "UPDATE samples SET host_reads = 1 WHERE id = 1")
        for column in NEW_COLUMNS:
            conn.execute(f"ALTER TABLE samples DROP COLUMN {column}")
        conn.execute("PRAGMA user_version=6")
        conn.commit()
        assert conn.execute("SELECT COUNT(*) FROM samples").fetchone() == (3,)
        cols = {r[1] for r in conn.execute("PRAGMA table_info(samples)")}
        assert not set(NEW_COLUMNS) & cols


NOW = datetime(2026, 9, 30, 20, 0, tzinfo=timezone.utc)


def _diag_sample(conn, ts, **values):
    cols = ["ts", "device", "bytes_written", "bytes_read", "segment_id"] + list(values)
    conn.execute(
        "INSERT INTO samples (%s) VALUES (%s)" % (
            ",".join(cols), ",".join("?" * len(cols))),
        [ts, "/dev/nvme0n1", 1, 1, 1] + list(values.values()))
    conn.commit()


class TestDiagnosticsPanel:
    @pytest.fixture
    def conn(self, tmp_path):
        c = init_store(tmp_path / "o.db")
        yield c
        c.close()

    def _text(self, conn):
        selection = ActivitySelection()
        selection.toggle_diagnostics()
        return diagnostics_text(readout(conn, selection, NOW))

    def test_new_counters_show_value_and_delta(self, conn):
        _diag_sample(conn, "2026-09-30T18:00:00+00:00", host_reads=1000,
                     host_writes=500, controller_busy_time=60,
                     num_err_log_entries=0, warning_temp_time=2,
                     critical_comp_time=0)
        _diag_sample(conn, "2026-09-30T19:50:00+00:00", host_reads=1300,
                     host_writes=550, controller_busy_time=65,
                     num_err_log_entries=1, warning_temp_time=2,
                     critical_comp_time=0)
        text = self._text(conn)
        assert "Host reads 1300 Δ +300" in text
        assert "Host writes 550 Δ +50" in text
        assert "Controller busy time 65 min Δ +5" in text
        assert "Error log entries 1 Δ +1" in text
        assert "Warning temperature time 2 min Δ ±0" in text
        assert "Critical temperature time 0 min Δ ±0" in text

    def test_legacy_rows_read_n_a_and_absent_thermal_management_is_hidden(self, conn):
        _diag_sample(conn, "2026-09-30T19:50:00+00:00", power_on_hours=5)
        text = self._text(conn)
        assert "Host reads n/a" in text
        assert "Thermal management" not in text

    def test_thermal_management_shown_when_the_drive_reports_it(self, conn):
        _diag_sample(conn, "2026-09-30T19:50:00+00:00", thm_temp1_trans_count=3,
                     thm_temp1_total_time=120, thm_temp2_trans_count=1,
                     thm_temp2_total_time=45)
        text = self._text(conn)
        assert "Thermal management 1 transitions 3 Δ n/a" in text
        assert "Thermal management 1 time 120 s Δ n/a" in text
        assert "Thermal management 2 transitions 1 Δ n/a" in text

    def test_unmigrated_store_still_renders_diagnostics(self, tmp_path):
        """The dashboard may open a store before the upgrade migrates it."""
        db = tmp_path / "legacy.db"
        _legacy_store(db)
        conn = sqlite3.connect(db)
        selection = ActivitySelection()
        selection.toggle_diagnostics()
        text = diagnostics_text(readout(conn, selection, NOW))
        assert "Host reads n/a" in text and "Power-on hours" in text