"""The `disks` source: physical drives in, `disk.*@dev` samples out, nothing invented.

Phase 09 task 02. The behaviours this pins, in the order it hurts to break them:

* **One row per physical disk, never per partition.** The rows come from the agent's
  whole-`/sys/block` list; the filesystems in the same snapshot must not grow a card.
* **A value the drive does not publish stays absent.** Not 0, not a stored sample — a
  stored 0 and "this drive reports nothing" are indistinguishable on a 90-day chart, and
  one of them is a lie (`AGENTS.md` rule 2).
* **Nothing is a rate.** Power-on hours and the sector counters are stored as reported;
  a second poll produces the same number, not a delta.
* **Why a card is empty is a fact, not a mystery.** `smartctl_available` is three-way
  (yes / no, install smartmontools / the agent does not say) because those are three
  different sentences for the operator, and `smart_errors` carries the agent's own line.
* **The colour comes from the shared table.** A row carries a `severity` dict for the two
  rated numbers (`temp_c`, `percentage_used`) and the drive's own `health` word, so the page
  paints a class instead of comparing a number to a bound — and the sector counters get no
  verdict at all, because a non-zero one is a chip and zero is not an achievement.
* **Events go under the dashboard name** `disks`, once per verdict change — the phase-03
  lesson: events filed under the sample source leave the page's feed empty forever.
* **The job is its own, at a 5-minute cadence**, and no host targets means no job at all,
  so an unconfigured deployment never dials (invariant 3).
"""
from __future__ import annotations

from typing import Any

import httpx
import pytest

from dashboards.collectors import disks_jobs
from dashboards.config import Settings, Target
from dashboards.severity import THRESHOLDS
from dashboards.sources.disks import (
    DASHBOARD,
    SEVERITY_FIELDS,
    SMART_SERIES,
    SOURCE,
    UNGRADED_FIELDS,
    DisksCollector,
    disk_summary,
    fleet_disk_summary,
    normalize_health,
    parse_disks_snapshot,
    row_severity,
    smartctl_available,
)


class NeverDialClient(httpx.AsyncClient):  # pragma: no cover - reaching it fails a test
    """A client that fails the test the moment anything tries to dial it."""

    async def get(self, *args: Any, **kwargs: Any) -> httpx.Response:
        raise AssertionError("a job factory must not dial; only the poll may")


def parse(snapshot: dict[str, Any], *, name: str = "driveripper",
          collector: DisksCollector | None = None,
          now: float = 1_700_000_000.0):
    """One poll of the source, with the defaults the tests do not care about filled in."""
    return parse_disks_snapshot(name, snapshot, collector or DisksCollector(), now=now)


def metric_names(result: Any) -> set[str]:
    return {sample.metric for sample in result.samples}


def values(result: Any) -> dict[str, float]:
    return {sample.metric: sample.value for sample in result.samples}


# ---- the rows: one per physical disk -------------------------------------- #

def test_every_physical_disk_the_agent_lists_becomes_a_row(disk_snapshot: Any) -> None:
    snapshot = disk_snapshot()
    result = parse(snapshot)
    rows = result.payload["disks"]
    assert [row["device"] for row in rows] == [
        device["name"] for device in snapshot["disk_io"]["devices"]
    ], "the card set is the agent's whole-disk list, in its order"
    agent_rows = {device["name"]: device for device in snapshot["disk_io"]["devices"]}
    for row in rows:
        source_row = agent_rows[row["device"]]
        assert row["path"] == f"/dev/{row['device']}"
        assert row["size_bytes"] == source_row["size_bytes"], "capacity is the agent's, not ours"
        assert row["rotational"] == source_row["rotational"], "SSD-vs-spinning is a fact"
        assert row["smart"] is not None


def test_filesystems_never_grow_a_disk_card(disk_snapshot: Any) -> None:
    """"Not per-partition" is the phase's one hard UI constraint, so it is asserted, not hoped.

    The snapshot carries a dozen mounts (`/`, `/srv`, `/boot`); a card per mount would
    double-count the same platter and bury the drive that is actually failing.
    """
    snapshot = disk_snapshot(mutate=lambda snap: snap["filesystems"].append(
        {"device": "/dev/mapper/srv", "mountpoint": "/srv", "fstype": "xfs",
         "used_pct": 91.0}))
    rows = parse(snapshot).payload["disks"]
    assert all("/" not in row["device"] for row in rows), "a mountpoint leaked into the cards"
    assert {row["device"] for row in rows}.isdisjoint(
        {fs["mountpoint"] for fs in snapshot["filesystems"]})


def test_the_smart_block_is_carried_verbatim_and_the_verdict_is_normalised(
        disk_snapshot: Any) -> None:
    snapshot = disk_snapshot()
    rows = parse(snapshot).payload["disks"]
    ata = next(row for row in rows if row["smart"] and row["smart"]["reallocated_sector_ct"])
    assert ata["smart"]["reallocated_sector_ct"] == 8.0
    assert ata["smart"]["current_pending_sector_ct"] == 3.0
    assert ata["smart"]["offline_uncorrectable"] == 2.0
    assert ata["smart"]["power_on_hours"] == 17_520.0
    assert ata["smart"]["temp_c"] == 31.0
    # `PASSED` is smartctl's word; the app's colour vocabulary is `ok`
    assert ata["smart"]["overall_health"] == "PASSED", "the drive's own word stays verbatim"
    assert ata["health"] == "ok"


@pytest.mark.parametrize("word,expected", [
    ("PASSED", "ok"), ("OK", "ok"), (" ok ", "ok"), ("passed", "ok"),
    ("FAILED", "failed"), ("failed!", "failed"),
    ("UNKNOWN", None), ("", None), (None, None), (7, None),
])
def test_only_the_drives_own_verdict_words_count(word: Any, expected: str | None) -> None:
    """`UNKNOWN` is not a pass: no measurement, no colour (`severity.py`'s first rule)."""
    assert normalize_health(word) == expected


def test_a_host_that_reports_no_disks_answers_with_an_empty_list_not_an_error() -> None:
    """gateway is a VM on virtio with nothing behind it; `/disks` renders "no disks"."""
    result = parse({"ts": 1.0, "disk_io": {"devices": []}, "errors": []}, name="gateway")
    assert result.payload["disks"] == []
    assert result.payload["summary"]["disks"] == 0
    assert result.samples == []
    assert result.payload["smartctl_available"] is None


def test_a_malformed_device_row_is_skipped_rather_than_crashing(disk_snapshot: Any) -> None:
    """`/disks` must survive an agent that answered something odd.

    A device with no name would render a card called `/dev/`, which is a worse lie than a
    missing card; a non-object row is what a truncated JSON body looks like.
    """
    clean = parse(disk_snapshot(mode="healthy")).payload["disks"]

    def mangle(snapshot: dict[str, Any]) -> None:
        snapshot["disk_io"]["devices"].append("not-a-dict")
        snapshot["disk_io"]["devices"].append({"name": "   ", "smart": {"temp_c": 44.0}})

    mangled = parse(disk_snapshot(mode="healthy", mutate=mangle)).payload["disks"]
    assert mangled == clean, "only the two junk rows were dropped, in place and unchanged"
    assert all(row["device"].strip() for row in mangled)


# ---- samples: the history namespace, counters as reported ----------------- #

def test_the_series_are_the_pinned_disk_names_per_device(disk_snapshot: Any) -> None:
    snapshot = disk_snapshot(mode="healthy")
    result = parse(snapshot)
    devices = [device["name"] for device in snapshot["disk_io"]["devices"]]
    # every series this source may write is `<base>@<device>` from the pinned table
    allowed = {f"{base}@{device}" for device in devices for _field, base in SMART_SERIES}
    assert metric_names(result) <= allowed
    assert all(name.startswith("disk.") and "@" in name for name in metric_names(result)), \
        "an unqualified disk name would collide with /system's per-mount series"
    # all six bases show up: the ATA shape supplies the sector counters, the NVMe shape
    # the wear percentage, so between them the whole table is exercised
    assert {name.split("@")[0] for name in metric_names(result)} == {
        base for _field, base in SMART_SERIES}
    # and the family is the disks one, so /api/metrics?source=disks answers for this page
    assert {sample.source for sample in result.samples} == {SOURCE} == {"disks"}


# ---- why a card is `n/a` --------------------------------------------------- #

def test_a_denied_device_nulls_only_that_disk(disk_snapshot: Any) -> None:
    snapshot = disk_snapshot(mode="denied")
    result = parse(snapshot)
    denied = snapshot["disk_io"]["devices"][-1]["name"]
    by_device = {row["device"]: row for row in result.payload["disks"]}
    assert by_device[denied]["smart"] is None
    assert by_device[denied]["health"] is None
    assert not [name for name in metric_names(result) if name.endswith(f"@{denied}")]
    assert any(row["smart"] for row in result.payload["disks"]), "its neighbours are fine"
    assert result.payload["smart_errors"] == [
        line for line in snapshot["errors"] if line.startswith("smart:")]


def test_a_missing_binary_is_answered_as_not_installed(disk_snapshot: Any) -> None:
    """"smartctl not available" is the sentence the page needs, and it comes from here."""
    result = parse(disk_snapshot(mode="no_binary"))
    assert result.payload["smartctl_available"] is False
    assert all(row["smart"] is None for row in result.payload["disks"])
    assert len(result.payload["smart_errors"]) == 1, "one missing binary is one fact"
    assert result.samples == [], "nothing was read, so nothing is stored"


def test_a_silent_opt_out_is_unknown_rather_than_not_installed(disk_snapshot: Any) -> None:
    """`--no-smart` and an agent predating the pass say nothing; guessing is worse.

    The page still has an honest state: `None` renders `n/a` with no colour, and the cards
    are dimmed rather than red (invariant 3).
    """
    result = parse(disk_snapshot(mode="silent"))
    assert result.payload["smartctl_available"] is None
    assert result.payload["smart_errors"] == []


def test_smartctl_available_is_true_the_moment_any_disk_answers() -> None:
    rows = [{"smart": {"temp_c": 31.0}}, {"smart": None}]
    assert smartctl_available(rows, ["smart: sdb: Permission denied"]) is True


def test_smart_errors_ignore_the_lines_that_are_not_about_smart() -> None:
    """`errors` also carries /proc failures, and those belong to `/system`, not here."""
    result = parse({"ts": 1.0, "disk_io": {"devices": []},
                    "errors": ["cpu: OSError: /proc/stat", "smart: sdb: denied"]})
    assert result.payload["smart_errors"] == ["smart: sdb: denied"]


# ---- the host summary ------------------------------------------------------ #

def test_summary_counts_reads_alarm_drives_and_failures_apart(disk_snapshot: Any) -> None:
    result = parse(disk_snapshot(mode="healthy"))
    summary = result.payload["summary"]
    devices = result.payload["disks"]
    assert summary["disks"] == len(devices) == summary["with_smart"]
    assert summary["health_failed"] == 0
    # every ATA shape carries reallocated/pending/uncorrectable sectors; the NVMe shapes
    # carry zeros, so the alarm count is exactly the number of ATA-shaped disks
    ata_disks = [row for row in devices if row["smart"]["reallocated_sector_ct"]]
    assert summary["alarm_disks"] == len(ata_disks) > 0


def test_a_drive_answering_failed_is_counted_and_the_neighbours_are_not(
        disk_snapshot: Any) -> None:
    result = parse(disk_snapshot(mode="failed"))
    assert result.payload["summary"]["health_failed"] == 1
    assert result.payload["disks"][0]["health"] == "failed"


def test_an_unreadable_drive_is_not_counted_as_failed(disk_snapshot: Any) -> None:
    """"Cannot read it" and "it is failing" are different alarms (rule 2)."""
    result = parse(disk_snapshot(mode="no_binary"))
    summary = result.payload["summary"]
    assert summary["disks"] > 0
    assert summary["with_smart"] == summary["health_failed"] == summary["alarm_disks"] == 0


def test_disk_summary_ignores_a_zero_counter(disk_snapshot: Any) -> None:
    rows = parse(disk_snapshot(mode="healthy")).payload["disks"]
    clean = [{**row, "smart": {**(row["smart"] or {}),
                               "reallocated_sector_ct": 0.0,
                               "current_pending_sector_ct": 0.0,
                               "offline_uncorrectable": 0.0}} for row in rows]
    assert disk_summary(clean)["alarm_disks"] == 0


# ---- events: one per verdict change, filed under the dashboard ------------- #

def test_a_drive_crossing_into_failed_writes_one_error_event(disk_snapshot: Any) -> None:
    collector = DisksCollector()
    snapshot = disk_snapshot(mode="healthy")
    device = snapshot["disk_io"]["devices"][0]["name"]
    assert parse(snapshot, collector=collector).events == [], "first sighting is not news"
    assert parse(snapshot, collector=collector).events == [], "an unchanged verdict is quiet"

    degraded = disk_snapshot(mode="failed")
    events = parse(degraded, collector=collector).events
    assert len(events) == 1
    event = events[0]
    assert event.source == DASHBOARD == "disks", (
        "a feed filtered by dashboard must find these — phase 03's empty-feed bug")
    assert event.host == "driveripper" and event.kind == "smart-health"
    assert event.severity == "error"
    assert device in event.message and "failed" in event.message
    assert event.data == {"device": device, "from": "ok", "to": "failed"}

    assert parse(degraded, collector=collector).events == []


def test_recovering_from_failed_is_a_quiet_info_event(disk_snapshot: Any) -> None:
    collector = DisksCollector()
    parse(disk_snapshot(mode="failed"), collector=collector)
    events = parse(disk_snapshot(mode="healthy"), collector=collector).events
    assert [event.severity for event in events] == ["info"]


def test_a_disk_that_left_and_came_back_is_a_first_sighting_again(disk_snapshot: Any) -> None:
    """State is replaced per poll, so a removed drive cannot compare against its old self."""
    collector = DisksCollector()
    snapshot = disk_snapshot(mode="healthy")
    device = snapshot["disk_io"]["devices"][0]["name"]
    parse(snapshot, collector=collector)
    empty = {"ts": snapshot["ts"], "errors": [], "disk_io": {"devices": []}}
    parse(empty, collector=collector)
    assert collector.previous_health("driveripper") == {}
    assert parse(disk_snapshot(mode="failed"), collector=collector).events == []
    assert collector.previous_health("driveripper")[device] == "failed"


def test_two_hosts_do_not_share_verdict_state(disk_snapshot: Any) -> None:
    collector = DisksCollector()
    parse(disk_snapshot("driveripper"), name="driveripper", collector=collector)
    parse(disk_snapshot("deskwork"), name="deskwork", collector=collector)
    assert set(collector.hosts) == {"driveripper", "deskwork"}
    assert len(collector.hosts["deskwork"]) == 1, "a host compares only against its own disks"


# ---- the job: its own, slow, and absent rather than dialling --------------- #

def test_one_disks_job_per_host_agent_at_the_slow_cadence() -> None:
    targets = [Target("driveripper", "http://a:9121"), Target("deskwork", "http://b:9122")]
    # the client is never used by the factory, only captured by the job closures
    jobs = disks_jobs(targets, NeverDialClient(), DisksCollector(), 300.0, timeout=30.0)
    assert [job.key for job in jobs] == ["disks:driveripper", "disks:deskwork"]
    assert [job.dashboard for job in jobs] == ["disks", "disks"]
    assert [job.label for job in jobs] == ["driveripper", "deskwork"], (
        "the label is the host: it is what the event feed and the alert print")
    assert [job.interval for job in jobs] == [300.0, 300.0], "SMART is not a 10 s signal"


def test_the_default_cadence_is_five_minutes_with_a_slow_fetch_timeout() -> None:
    """The defaults are the phase's decision (300 s) and the agent's SMART budget (30 s)."""
    settings = Settings()
    assert settings.disks_interval == 300.0
    assert settings.disks_timeout == 30.0, "a wedged smartctl pass costs the agent 20 s"


def test_no_host_targets_means_no_disks_job_and_nothing_dialled() -> None:
    """`disks` has no upstream of its own, so unconfigured means *no task at all*.

    The 404 `{"configured": false}` half of invariant 3 is task 03's route; this is the
    half that matters first: with no targets the app must not open a socket, and
    `/api/health` must not carry a `disks` source that is red because nobody installed one.
    """
    assert disks_jobs([], NeverDialClient(), DisksCollector(), 300.0) == []


def test_a_host_that_has_never_polled_has_no_previous_verdicts() -> None:
    assert DisksCollector().previous_health("never-seen") == {}


# ---- the verdicts a card paints (task 03's severity rows) ------------------ #

def hot_drive(disk_snapshot: Any, **fields: Any) -> dict[str, Any]:
    """The healthy capture with one drive's rated numbers moved onto a chosen bound.

    Driven through the fixture's `mutate` so the numbers under test are the ones the agent's
    own parser would have produced for a drive running hot and half-worn out.
    """
    def tweak(snapshot: dict[str, Any]) -> None:
        (snapshot["disk_io"]["devices"][0].get("smart") or {}).update(fields)

    return disk_snapshot(mode="healthy", mutate=tweak)


def test_a_row_carries_a_verdict_for_each_rated_number_and_for_the_drives_word(
        disk_snapshot: Any) -> None:
    """"72 °C, 85 % wear, PASSED" arrives as `fail`/`warn`/`ok`, never as a bound in JS."""
    rows = parse(hot_drive(disk_snapshot, temp_c=72.0, percentage_used=85.0)).payload["disks"]
    assert rows[0]["severity"] == {"temp_c": "fail", "percentage_used": "warn",
                                   "health": "ok"}


def test_a_drive_answering_failed_is_the_row_that_reads_red(disk_snapshot: Any) -> None:
    """`health` is the drive's own verdict, so its colour is a translation, not a threshold."""
    rows = parse(disk_snapshot(mode="failed")).payload["disks"]
    assert rows[0]["health"] == "failed" and rows[0]["severity"]["health"] == "fail"
    assert rows[1]["severity"]["health"] == "ok", "its neighbours keep their own verdict"


def test_an_unreadable_drive_gets_no_verdict_at_all(disk_snapshot: Any) -> None:
    """No `smart` block, no colour: the keys are present and every one is `None`.

    The card must be able to read `severity.temp_c` without branching on a missing key, and
    must not render green for a drive the agent could not open (`AGENTS.md` rule 2).
    """
    rows = parse(disk_snapshot(mode="no_binary")).payload["disks"]
    assert rows, "the drive is still listed, dimmed"
    for row in rows:
        assert row["severity"] == {"temp_c": None, "percentage_used": None, "health": None}


def test_a_missing_number_stays_unjudged_rather_than_good(disk_snapshot: Any) -> None:
    """An ATA spinning disk publishes no `percentage_used`; that row reads `n/a`, not `ok`."""
    rows = parse(disk_snapshot(mode="healthy")).payload["disks"]
    spinning = [row for row in rows if row["smart"]["percentage_used"] is None]
    assert spinning, "the recorded ATA shape has no wear field, which is the case here"
    for row in spinning:
        assert row["severity"]["percentage_used"] is None
        assert row["severity"]["temp_c"] == "ok", "the number it does report is still judged"


def test_the_sector_counters_are_never_part_of_a_rows_verdict(disk_snapshot: Any) -> None:
    """The alarm fields appear on the row and in `summary`, never as an `ok|warn|fail`."""
    rows = parse(disk_snapshot(mode="healthy")).payload["disks"]
    alarming = next(row for row in rows if row["smart"]["reallocated_sector_ct"])
    assert set(alarming["severity"]) == {field for field, _metric in SEVERITY_FIELDS} | {"health"}
    assert not set(alarming["severity"]) & set(UNGRADED_FIELDS)
    assert alarming["smart"]["reallocated_sector_ct"] == 8.0, "the number still travels"


def test_the_rated_rows_exist_in_the_one_threshold_table() -> None:
    """`SEVERITY_FIELDS` may only name metrics `severity.THRESHOLDS` actually carries.

    A row here without a bound there would silently colour nothing; a bound for a field the
    source never reads is the same drift seen from the other side.
    """
    for field, metric in SEVERITY_FIELDS:
        assert metric in THRESHOLDS, f"{field} is rated by a metric nobody defines"
    assert UNGRADED_FIELDS and not set(UNGRADED_FIELDS) & {
        field for field, _metric in SEVERITY_FIELDS}
    assert {base for _field, base in SMART_SERIES} >= {
        metric for _field, metric in SEVERITY_FIELDS}, (
        "a rated number that is not stored cannot be charted in its own colour")


def test_row_severity_tolerates_a_row_with_nothing_in_it() -> None:
    """The API hands this a dict built from an agent that answered oddly; it must not raise."""
    assert row_severity({}) == {"temp_c": None, "percentage_used": None, "health": None}
    assert row_severity({"health": "no-such-word", "smart": {"temp_c": "hot"}}) == {
        "temp_c": None, "percentage_used": None, "health": None}


# ---- the fleet header the API sums, not the page --------------------------- #

def test_the_fleet_counts_drives_across_the_hosts_that_answered(
        disk_snapshot: Any) -> None:
    """`/disks` says how many drives the fleet has and how many it could read."""
    hosts = [parse(disk_snapshot("driveripper"), name="driveripper").payload,
             parse(disk_snapshot("deskwork"), name="deskwork").payload]
    fleet = fleet_disk_summary(hosts)
    disks = sum(host["summary"]["disks"] for host in hosts)
    assert fleet["hosts_total"] == fleet["hosts_live"] == 2
    assert fleet["disks"] == disks == len(hosts[0]["disks"]) + len(hosts[1]["disks"])
    assert fleet["with_smart"] == disks
    assert fleet["health_failed"] == 0
    # the recorded ATA shape carries reallocated sectors, so the fleet's alarm count is the
    # sum of the hosts' own chip counts, not a number this function invented
    assert fleet["alarm_disks"] == sum(host["summary"]["alarm_disks"] for host in hosts) > 0


def test_a_host_that_never_answered_is_counted_but_never_summed(
        disk_snapshot: Any) -> None:
    """A dead agent shrinks `hosts_live`, not the drive totals presented as current.

    Adding a stub's zeros to the sums would halve the fleet's alarm count on the strength of
    a host that said nothing — the `/llm` fleet block's rule, same failure shape.
    """
    live = parse(disk_snapshot(mode="failed"), name="driveripper").payload
    stub = {"name": "deskwork", "ts": None, "unreachable": True, "disks": []}
    fleet = fleet_disk_summary([live, stub])
    assert fleet["hosts_total"] == 2 and fleet["hosts_live"] == 1
    assert fleet["disks"] == len(live["disks"]), "only the answering host is in the sum"
    assert fleet["health_failed"] == 1


def test_a_payload_with_no_countable_summary_contributes_nothing() -> None:
    """"Answered" is not "counted": a summary that is missing or junk adds zero, not a guess."""
    assert fleet_disk_summary([{"name": "gw", "disks": []}])["disks"] == 0
    weird = {"name": "gw", "summary": {"disks": True, "with_smart": "many",
                                       "health_failed": None, "alarm_disks": 2}}
    fleet = fleet_disk_summary([weird])
    assert fleet["disks"] == fleet["with_smart"] == fleet["health_failed"] == 0
    assert fleet["alarm_disks"] == 2
    assert fleet["hosts_live"] == 1


def test_an_empty_fleet_answers_with_zeros_rather_than_none() -> None:
    """No hosts configured never reaches this route, but an empty list must not crash either."""
    assert fleet_disk_summary([]) == {"hosts_total": 0, "hosts_live": 0, "disks": 0,
                                      "with_smart": 0, "health_failed": 0, "alarm_disks": 0}