MeshTerm/tests/test_dashboard.py

"""Dashboard tests: the live screen's state, its chart plumbing, and its data feeds.

The screen is driven headless against fake sessions and canned observations, the same
approach as the live trace/TX screen tests. (The braille chart renderer itself is
covered in ``test_braillechart``.)
"""

from __future__ import annotations

import re
from datetime import datetime, timedelta
from pathlib import Path

from meshterm.core.events import MeshEvent
from meshterm.core.models import Observation, utcnow
from meshterm.persistence.repository import Repository
from meshterm.services.monitor_service import ACTIVITY_BUCKETS
from meshterm.ui.dashboard_screen import DashboardScreen


class _FakeSession:
    def __init__(self) -> None:
        self.repaints = 0

    def invalidate(self) -> None:
        self.repaints += 1


def _screen(window=None, histogram=None, kinds=None) -> DashboardScreen:
    screen = DashboardScreen(
        session=_FakeSession(),
        resolve=lambda h: {"a1b2": "Alice", "3d63": "YUL"}.get(h, ""),
        window=list(window or []),
        activity=lambda: tuple(histogram or (0,) * ACTIVITY_BUCKETS),
        activity_flags=lambda: (True,) * ACTIVITY_BUCKETS,
        kind_counts=lambda: dict(kinds or {}),
    )
    screen.note_viewport(48)  # the frame records this before every real paint
    return screen


def _obs(node="a1b2", kind="advert", snr=5.0, rssi=-90.0, age_s=0, **extra) -> Observation:
    return Observation(
        node=node, name=node, kind=kind, snr=snr, rssi=rssi,
        observed_at=utcnow() - timedelta(seconds=age_s), **extra,
    )


def _plain(lines: list[str]) -> str:
    return "\n".join(lines)


def _stripped(lines: list[str]) -> list[str]:
    """The rendered lines with ANSI escapes removed, for structural assertions."""
    return [re.sub(r"\x1b\[[0-9;]*m", "", line) for line in lines]


# --- the screen ----------------------------------------------------------------------


def test_footer_advertises_scroll_only_when_the_overview_overflows() -> None:
    """The scroll keys join the footer only when the body is taller than the viewport."""
    screen = _screen()
    # Fits (the frame recorded a body no taller than the viewport): Esc alone acts.
    screen.note_metrics(total=10, viewport=20)
    assert screen.footer_hint == "Esc back"
    # Overflows: the scroll atom appears, ahead of Esc.
    screen.note_metrics(total=40, viewport=20)
    assert screen.footer_hint == "↑↓ PgUp/PgDn scroll · Esc back"


def test_dashboard_renders_all_three_sections() -> None:
    """Activity, Traffic, and RF health all render from a seeded window.

    (The feed panel moved out to the Live feed tool — see ``test_livefeed``.)
    """
    screen = _screen(
        window=[_obs(), _obs(node="3d63", node_type=2, snr=-2.0)],
        histogram=[3] + [0] * (ACTIVITY_BUCKETS - 1),
        kinds={"advert": 5, "ack": 1},
    )
    body = _plain(screen.render_body(100))
    assert "Activity" in body and "Traffic" in body and "RF health" in body
    assert "Feed" not in body  # the packet stream lives in the Live feed tool now
    assert "nodes heard" in body and "(1 repeater)" in body
    assert "advert" in body and "Alice" in body  # traffic class + the busiest node


def test_dashboard_activity_chart_reads_newest_right_with_mirrored_scale() -> None:
    """'now' anchors the right edge and the scale marks mirror on both gutters.

    Peak 9 over 3 rows (12 dots): the top gutter's ┤ tick crosses its row's third
    dot (an 11-dot bar → 9·11/12 ≈ 8), so the mark reads 8, not the peak itself.
    """
    screen = _screen(histogram=[9] + [0] * (ACTIVITY_BUCKETS - 1))
    lines = _stripped(screen.render_body(80))
    top = next(line for line in lines if "┤" in line)
    assert top.strip().startswith("8 ┤")
    assert top.rstrip().endswith("├ 8")
    caption = next(line for line in lines if "now" in line)
    assert caption.index("−") < caption.index("now")  # oldest left, newest right
    # The lone newest-minute burst draws against the chart's right gutter, and the
    # left half of the chart is bare flatline.
    chart = [line for line in lines if "┤" in line or "│" in line]
    bottom = chart[-1]
    left_half = bottom[3 : 3 + (len(bottom) - 6) // 2]
    assert all(ch in (chr(0x2800), chr(0x2800 | 0x40 | 0x80), " ") for ch in left_half)
    assert any(0x2800 <= ord(ch) <= 0x28FF and ord(ch) & 0x3F for ch in chart[0])


def test_dashboard_activity_chart_fills_the_width() -> None:
    """The chart stretches to the render width: wider terminal, more minutes shown."""
    screen = _screen(histogram=[5] * ACTIVITY_BUCKETS)
    for width in (60, 110):
        lines = _stripped(screen.render_body(width))
        top = next(line for line in lines if "┤" in line)
        assert len(top) == width  # gutters + chart consume every cell
    narrow = next(line for line in _stripped(screen.render_body(60)) if "└" in line)
    wide = next(line for line in _stripped(screen.render_body(110)) if "└" in line)
    assert wide.count("─") > narrow.count("─")


def test_dashboard_pulse_drops_heard_only_when_it_wont_fit() -> None:
    """The pulse line keeps ' heard' at full width and sheds it instead of wrapping."""
    from meshterm.ui.tui.render import render_lines

    def pulse_row(width: int) -> str:
        grid = screen._pulse_grid(shown, width)
        return _stripped(render_lines(grid, width))[0]

    screen = _screen(window=[_obs(), _obs(node="3d63", node_type=2)])
    shown = [1] * 120
    roomy = pulse_row(100)
    assert "nodes heard" in roomy
    tight = pulse_row(len(roomy.rstrip()) - 1)
    assert "heard" not in tight and "nodes" in tight


def test_dashboard_traffic_breaks_packets_out_by_payload_class() -> None:
    """The raw ``packet`` bucket meters per payload class, glossed and iconed."""
    screen = _screen(
        window=[_obs()],
        kinds={"advert": 4, "packet:GRP_TXT": 3, "packet:TRACE": 2, "packet": 1},
    )
    body = _plain(_stripped(screen.render_body(100)))
    assert "channel text" in body and "trace" in body  # glossed, not raw typenames
    assert "GRP_TXT" not in body
    # The class-less remainder keeps a plain "packet" row alongside the classed ones.
    assert "packet" in body
    # Decoded families before the raw classes, class-less packet closing the block.
    assert body.index("advert") < body.index("channel text") < body.index("trace")


def test_dashboard_traffic_disambiguates_raw_advert_and_ack_rows() -> None:
    """An overheard ADVERT/ACK frame's row reads "raw …" so no two meters match."""
    screen = _screen(
        window=[_obs()],
        kinds={"advert": 4, "packet:ADVERT": 2, "ack": 3, "packet:ACK": 1},
    )
    body = _plain(_stripped(screen.render_body(100)))
    assert "raw advert" in body and "raw ack" in body


def test_dashboard_live_observations_land_in_the_window() -> None:
    """A hub observation folds into the trailing window and repaints the charts."""
    screen = _screen()
    screen.on_event(MeshEvent.observation_event(_obs(snr=7.5)))
    body = _plain(screen.render_body(100))
    assert "+7.5 dB median" in body  # the RF section now has a reception to describe
    assert screen._session.repaints >= 1


def test_dashboard_prunes_the_window() -> None:
    """Observations older than the window drop out of the statistics."""
    stale = _obs(age_s=3 * 3600, snr=-12.0)
    screen = _screen(window=[stale])
    body = _plain(screen.render_body(100))
    assert "no receptions in the window yet" in body  # stats pruned the stale row


def test_dashboard_radio_rows_render_device_stats() -> None:
    """The Device-info dynamic numbers (noise floor, airtime, battery) fold in."""
    screen = _screen(window=[_obs()])
    screen.stats = {
        "noise_floor": -104, "last_rssi": -95, "last_snr": 5.5,
        "tx_air_secs": 12, "rx_air_secs": 340,
    }
    screen.battery = {"level": 4010}
    body = _plain(screen.render_body(100))
    assert "-104 dBm noise floor" in body
    assert "TX 12 s · RX 340 s" in body
    assert "4.01 V" in body


def test_repository_kind_counts_bucket_packets_by_payload_class(tmp_path: Path) -> None:
    """Stored packet frames with a payload class seed as ``packet:<TYPENAME>`` buckets."""
    repo = Repository(tmp_path / "kinds.db")
    run = repo.start_run("monitor", {}, None)
    repo.record_observation(run, _obs(node="a", kind="advert"))
    repo.record_observation(
        run, _obs(node="", kind="packet", raw={"payload_typename": "GRP_TXT"})
    )
    repo.record_observation(
        run, _obs(node="", kind="packet", raw={"payload_typename": "GRP_TXT"})
    )
    repo.record_observation(run, _obs(node="", kind="packet"))  # class-less frame

    counts = repo.kind_counts()
    assert counts["advert"] == 1
    assert counts["packet:GRP_TXT"] == 2
    assert counts["packet"] == 1  # only the class-less frame stays a bare packet
    repo.close()


def test_monitor_tallies_live_packets_by_payload_class(tmp_path: Path) -> None:
    """A live classed packet observation lands in its ``packet:<TYPENAME>`` bucket."""
    from meshterm.services.monitor_service import MonitorService

    monitor = MonitorService.__new__(MonitorService)
    monitor._activity = {}
    monitor._kind_counts = {}
    monitor._count_packet(
        MeshEvent.observation_event(
            _obs(node="", kind="packet", raw={"payload_typename": "TRACE"})
        )
    )
    monitor._count_packet(MeshEvent.observation_event(_obs(kind="advert")))
    assert monitor._kind_counts == {"packet:TRACE": 1, "advert": 1}


# --- the repository window -----------------------------------------------------------


def test_recent_observations_windows_and_orders(tmp_path: Path) -> None:
    """The seed query returns only the window, oldest first, hydrated with kinds."""
    repo = Repository(tmp_path / "dash.db")
    run = repo.start_run("monitor", {}, None)
    repo.record_observation(run, _obs(node="old", age_s=3 * 3600))
    repo.record_observation(run, _obs(node="mid", kind="packet", age_s=600, path="3d63"))
    repo.record_observation(run, _obs(node="new", age_s=5))

    window = repo.recent_observations(since=utcnow() - timedelta(hours=2))
    assert [o.node for o in window] == ["mid", "new"]
    assert window[0].kind == "packet" and window[0].path == "3d63"
    repo.close()


def test_recent_observations_rehydrate_channel_text_for_decryption(tmp_path: Path) -> None:
    """A stored GRP_TXT frame comes back with the fields the packet viewer decrypts from."""
    from Crypto.Cipher import AES
    from Crypto.Hash import HMAC, SHA256

    from meshterm.core.channels import channel_hash, derive_secret
    from meshterm.ui.packet_viewer import PacketEntry, PacketViewer

    name, secret = "#general", derive_secret("#general")
    plain = (0).to_bytes(4, "little") + bytes([0]) + b"hi from history"
    plain += b"\x00" * (-len(plain) % 16)
    crypted = AES.new(secret, AES.MODE_ECB).encrypt(plain)
    mac = HMAC.new(secret, digestmod=SHA256)
    mac.update(crypted)
    raw = {
        "payload_typename": "GRP_TXT",
        "chan_hash": channel_hash(secret),
        "cipher_mac": mac.digest()[:2].hex(),
        "crypted": crypted.hex(),
    }

    repo = Repository(tmp_path / "chan.db")
    run = repo.start_run("monitor", {}, None)
    repo.record_observation(run, _obs(node="src", kind="packet", path="3d63", raw=raw))
    repo.record_observation(run, _obs(node="adv", kind="advert"))  # a plain advert nearby

    window = {o.node: o for o in repo.recent_observations(since=utcnow() - timedelta(hours=2))}
    stored = window["src"]
    assert stored.raw is not None and stored.raw["chan_hash"] == channel_hash(secret)
    assert window["adv"].raw is None  # nothing to rebuild from → no raw, as before

    # The stored frame decrypts straight out of history — the whole point of persisting it.
    viewer = PacketViewer(
        [PacketEntry.from_observation(stored)], 0,
        resolve=lambda h: "", channels=[(name, secret)],
    )
    body = "\n".join(re.sub(r"\x1b\[[0-9;]*m", "", ln) for ln in viewer.render_body(80))
    assert "#general" in body and "hi from history" in body
    repo.close()


def test_recent_observations_rehydrate_the_packet_payload_class(tmp_path: Path) -> None:
    """A stored ``packet`` comes back carrying its payload class, so a list can name it."""
    repo = Repository(tmp_path / "cls.db")
    run = repo.start_run("monitor", {}, None)
    repo.record_observation(
        run,
        _obs(node="", kind="packet", path="3d63", raw={"payload_typename": "TRACE"}),
    )
    repo.record_observation(run, _obs(node="a1b2", kind="advert"))  # a plain advert nearby

    window = {o.node or "": o for o in repo.recent_observations(since=utcnow() - timedelta(hours=2))}
    assert window[""].raw is not None and window[""].raw["payload_typename"] == "TRACE"
    assert window["a1b2"].raw is None  # an advert keeps no packet raw, as before
    repo.close()


def test_quarter_hour_activity_buckets_by_15_minute_slice(tmp_path: Path) -> None:
    """Observations fall into 96 quarter-hour-of-day slots (``HH * 4 + MM // 15``).

    The slots are local time-of-day, so the observations are stamped at local
    wall-clock instants (naive → ``.astimezone()``) — the slice index then matches
    each ``HH:MM`` directly, in any runner zone.
    """
    repo = Repository(tmp_path / "q.db")
    run = repo.start_run("monitor", {}, None)

    def at(hh: int, mm: int) -> Observation:
        return Observation(
            node="n", name="n", kind="advert",
            observed_at=datetime(2026, 7, 8, hh, mm).astimezone(),
        )

    for hh, mm in [(0, 0), (0, 44), (17, 55), (17, 59)]:
        repo.record_observation(run, at(hh, mm))
    slots = repo.quarter_hour_activity()
    assert len(slots) == 96
    assert slots[0] == 1        # 00:00 → slot 0
    assert slots[2] == 1        # 00:44 → slot 2 (44 // 15)
    assert slots[71] == 2       # 17:55 and 17:59 → slot 17*4 + 3
    assert sum(slots) == 4
    repo.close()