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, timezone
from pathlib import Path

from meshterm.core.events import MeshEvent
from meshterm.core.models import Ack, Message, 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, active=True) -> 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 {}),
        hub_active=lambda: active,
    )
    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_dashboard_renders_all_four_sections() -> None:
    """Activity, Traffic, RF health, and Feed all render from a seeded window."""
    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
    assert "RF health" in body and "Feed" in body
    assert "● live" in body
    assert "nodes heard" in body and "(1 repeater)" in body
    assert "advert" in body and "Alice" in body and "YUL" in body


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_live_events_land_in_the_feed_and_window() -> None:
    """Observations, messages, and acks each fold into the feed as they arrive."""
    screen = _screen()
    screen.on_event(MeshEvent.observation_event(_obs(snr=7.5)))
    screen.on_event(MeshEvent.message_event(Message(text="hi", sender="a1b2")))
    screen.on_event(MeshEvent.ack_event(Ack(code="01c3")))
    body = _plain(screen.render_body(100))
    assert "+7.5 dB" in body
    assert "message" in body and "ack" in body
    # Newest first: the ack row sits above the message row, which sits above the advert.
    assert body.index("ack") < body.index("message") < body.index("advert")


def test_dashboard_packet_rows_show_their_relay_path() -> None:
    """An RX-log packet reads as its relay path, resolved to names where known.

    The path renders through the shared compact path widget, so each name carries its
    own hue — assert on the stripped text, not the raw ANSI.
    """
    screen = _screen()
    screen.on_event(
        MeshEvent.observation_event(_obs(kind="packet", path="3d63,a1b2", snr=1.0))
    )
    assert "via YUL → Alice" in "\n".join(_stripped(screen.render_body(100)))


def test_dashboard_feed_names_a_relayed_packet_by_its_payload_class() -> None:
    """A relayed packet naming no origin reads as its payload class, never a bare '?'."""
    screen = _screen()
    raw = {"payload_typename": "TRACE", "route_typename": "FLOOD"}
    screen.on_event(
        MeshEvent.observation_event(
            _obs(node="", kind="packet", snr=1.0, path="3d63,a1b2", raw=raw)
        )
    )
    feed = _plain(screen.render_body(100)).split("Feed")[1]
    assert "trace" in feed  # the payload-class gloss stands in for the missing identity
    assert "?" not in feed  # …instead of the useless placeholder


def test_dashboard_feed_names_a_channel_message_by_its_sender() -> None:
    """A channel message's ``Name:`` prefix names the node lane, not the bare channel."""
    from meshterm.core.models import Message

    screen = _screen()
    screen.on_event(
        MeshEvent.message_event(Message(text="Alice: hi all", channel=3, is_channel=True))
    )
    feed = _plain(screen.render_body(100)).split("Feed")[1]
    assert "Alice" in feed          # the parsed sender leads the row
    assert "ch 3" in feed           # …with the channel kept as the trailing context


def test_dashboard_prunes_the_window_but_keeps_the_feed() -> 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_page_keys_move_the_feed_selection() -> None:
    """With a feed row highlighted, PgUp/PgDn walk the selection a windowful at a time."""
    window = [_obs(node=f"n{i}", age_s=i) for i in range(20)]
    screen = _screen(window=window)
    screen._feed_window.page = 5  # as if the last paint settled a five-row window
    assert screen._selected == 0  # the newest packet is highlighted from the start
    screen.handle("pagedown")
    assert screen._selected == 5  # the selection travelled down a page, not just the view
    screen.handle("pageup")
    assert screen._selected == 0

    # With nothing highlighted, the page keys slide the feed window instead.
    screen._selected = None
    screen.handle("pagedown")
    assert screen._feed_window.top > 0 and screen._selected is None


def test_dashboard_feed_windows_inside_the_fixed_screen() -> None:
    """The charts stay pinned: the body fits the viewport and the feed rows window."""
    window = [_obs(node=f"n{i}", age_s=i) for i in range(40)]
    screen = _screen(window=window)
    screen.note_viewport(24)
    lines = screen.render_body(100)
    assert len(lines) <= 24  # charts + feed window == the viewport, never more
    body = _plain(_stripped(lines))
    assert "Activity" in body and "Feed" in body  # the chrome is all still there
    assert "↓" in body and "more" in body  # hidden feed rows are counted below


def test_dashboard_without_a_device_reads_as_waiting() -> None:
    """With the hub idle the feed heading says so instead of pretending to be live."""
    screen = _screen(active=False)
    assert "○ waiting for a device" in _plain(screen.render_body(100))


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


# --- 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``)."""
    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, tzinfo=timezone.utc),
        )

    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()