bandsaunter/tests/test_ui.py

"""The live display, and keeping the driver from writing over it."""
import os
import subprocess
import sys
import tempfile
import time

import numpy as np
import pytest
from rich.console import Console

from bandsaunter.config import ScanConfig
from bandsaunter.ranges import parse_range_list
from bandsaunter.recorder import HitRecord
from bandsaunter.scanner import Scanner
from bandsaunter.simulator import SimulatedDevice
from bandsaunter.ui import ScanDisplay, _sparkline


def _display(height: int, hits: int, recording: bool) -> ScanDisplay:
    console = Console(width=100, height=height,
                      file=open(os.devnull, "w"))
    cfg = ScanConfig(ranges=parse_range_list("144M-148M"),
                     output_dir=tempfile.mkdtemp())
    scanner = Scanner(cfg, device=SimulatedDevice().open())
    scanner.prepare()
    display = ScanDisplay(scanner, console=console)
    display.attach()
    step = scanner.plan[0]
    display.on_step(0, 6, step, np.full(1024, -70.0),
                    np.linspace(step.low, step.high, 1024))
    for i in range(hits):
        display.hits.append(HitRecord(
            frequency=146e6 + i * 1e5, started_at=time.time(), duration=5.0,
            snr_db=20.0, classification="FM broadcast station (stereo)"))
    display._rec.active = recording
    display._rec.frequency = 146.52e6
    display._rec.mode = "nfm"
    return display


def _rendered_height(display: ScanDisplay) -> int:
    probe = Console(width=100, height=200, file=open(os.devnull, "w"),
                    record=True)
    probe.print(display.render())
    return len(probe.export_text().rstrip("\n").split("\n"))


@pytest.mark.parametrize("height", [16, 20, 24, 30, 40, 60])
@pytest.mark.parametrize("hits,recording", [(0, False), (3, False),
                                            (12, False), (12, True)])
def test_the_display_never_outgrows_the_terminal(height, hits, recording):
    """A frame taller than the terminal cannot be redrawn in place.

    Every refresh then scrolls another copy into the scrollback, which is why
    the header ends up on screen several times over.
    """
    display = _display(height, hits, recording)
    assert _rendered_height(display) <= height, \
        f"{hits} hits, rec={recording}: overflowed a {height}-line terminal"


def test_older_hits_are_dropped_not_the_layout():
    """When space runs short the list shortens; the panels stay."""
    display = _display(20, 12, True)
    probe = Console(width=100, height=200, file=open(os.devnull, "w"),
                    record=True)
    probe.print(display.render())
    text = probe.export_text()
    assert "receiver" in text and "sweep" in text
    assert "more above" in text, "no sign that hits were trimmed"


def test_a_tall_terminal_shows_every_hit():
    display = _display(60, 12, False)
    probe = Console(width=100, height=200, file=open(os.devnull, "w"),
                    record=True)
    probe.print(display.render())
    assert "more above" not in probe.export_text()


def test_sparkline_keeps_narrow_carriers_visible():
    values = np.full(200, -70.0)
    values[100] = -20.0
    assert "█" in _sparkline(values, 60)


# ---------------------------------------------------------------------------
# The driver writes its own messages straight to file descriptor 2
# ---------------------------------------------------------------------------

def test_quiet_driver_suppresses_writes_to_fd_2():
    """librtlsdr prints from C, so Python-level redirection cannot catch it."""
    code = (
        "import os, sys; sys.path.insert(0, %r)\n"
        "from bandsaunter.device import quiet_driver\n"
        "with quiet_driver():\n"
        "    os.write(2, b'CHATTER\\n')\n"
        "os.write(2, b'AFTERWARDS\\n')\n"
    ) % os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    out = subprocess.run([sys.executable, "-c", code], capture_output=True,
                         text=True, timeout=30)
    assert "CHATTER" not in out.stderr
    assert "AFTERWARDS" in out.stderr, "stderr was not restored"


def test_quiet_driver_restores_on_an_exception():
    code = (
        "import os, sys; sys.path.insert(0, %r)\n"
        "from bandsaunter.device import quiet_driver\n"
        "try:\n"
        "    with quiet_driver():\n"
        "        raise ValueError('boom')\n"
        "except ValueError:\n"
        "    pass\n"
        "os.write(2, b'RESTORED\\n')\n"
    ) % os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    out = subprocess.run([sys.executable, "-c", code], capture_output=True,
                         text=True, timeout=30)
    assert "RESTORED" in out.stderr


def test_driver_messages_can_be_turned_back_on_for_debugging():
    code = (
        "import os, sys; sys.path.insert(0, %r)\n"
        "from bandsaunter.device import quiet_driver\n"
        "with quiet_driver():\n"
        "    os.write(2, b'CHATTER\\n')\n"
    ) % os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    env = dict(os.environ, BANDSAUNTER_DRIVER_MESSAGES="1")
    out = subprocess.run([sys.executable, "-c", code], capture_output=True,
                         text=True, timeout=30, env=env)
    assert "CHATTER" in out.stderr


def test_python_errors_still_reach_stderr():
    """Silencing the driver must not swallow a traceback."""
    code = (
        "import sys; sys.path.insert(0, %r)\n"
        "from bandsaunter.device import quiet_driver\n"
        "with quiet_driver():\n"
        "    pass\n"
        "raise RuntimeError('visible')\n"
    ) % os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    out = subprocess.run([sys.executable, "-c", code], capture_output=True,
                         text=True, timeout=30)
    assert "visible" in out.stderr