1
"""The monitor stays open; explicit health checks return meaningful status."""
2
import contextlib
3
import importlib.util
4
import io
5
from pathlib import Path
6
from types import SimpleNamespace
7
import tempfile
8
import unittest
9
from unittest.mock import patch
10
 
11
spec = importlib.util.spec_from_file_location("monitor_cli", Path(__file__).with_name("monitor_device.py"))
12
monitor = importlib.util.module_from_spec(spec)
13
spec.loader.exec_module(monitor)
14
 
15
 
16
class Port:
17
    def __init__(self, chunks):
18
        self.chunks = iter(chunks)
19
        self.reads = 0
20
        self.closed = False
21
        self.writes = []
22
 
23
    def read(self, size):
24
        self.reads += 1
25
        item = next(self.chunks, KeyboardInterrupt())
26
        if isinstance(item, BaseException):
27
            raise item
28
        return item
29
 
30
    def close(self):
31
        self.closed = True
32
 
33
    def write(self, data):
34
        self.writes.append(data)
35
 
36
    def flush(self):
37
        pass
38
 
39
 
40
class MonitorTests(unittest.TestCase):
41
    def test_monitor_redacts_canonical_credentials_even_with_a_stale_export(self):
42
        with tempfile.TemporaryDirectory() as directory:
43
            root = Path(directory)
44
            (root / ".credential.env").write_text('VIEWER_PASSWORD="canonical-secret"\n')
45
            (root / "worker").mkdir()
46
            (root / "worker/.dev.vars").write_text('VIEWER_PASSWORD="stale-secret"\n')
47
            port = Port([b"HTTPS canonical-secret\n", KeyboardInterrupt()])
48
            output = io.StringIO()
49
            with (
50
                patch.object(monitor, "ROOT", root),
51
                patch.object(monitor, "open_port", return_value=port),
52
                patch.dict("sys.modules", {"serial": SimpleNamespace(SerialException=OSError)}),
53
                contextlib.redirect_stdout(output),
54
            ):
55
                self.assertEqual(monitor.main([]), 0)
56
            self.assertNotIn("canonical-secret", output.getvalue())
57
            self.assertEqual((root / "artifacts/radio/autonomous-monitor.log").read_text(), "HTTPS [redacted]\n")
58
 
59
    def run_monitor(self, port, args, clock=None):
60
        with tempfile.TemporaryDirectory() as directory, contextlib.ExitStack() as stack:
61
            stack.enter_context(patch.object(monitor, "ROOT", Path(directory)))
62
            stack.enter_context(patch.object(monitor, "open_port", return_value=port))
63
            stack.enter_context(patch.dict("sys.modules", {"serial": SimpleNamespace(SerialException=OSError)}))
64
            stack.enter_context(contextlib.redirect_stdout(io.StringIO()))
65
            if clock is not None:
66
                stack.enter_context(patch.object(monitor.time, "monotonic", side_effect=clock))
67
            return monitor.main(args)
68
 
69
    def test_monitor_continues_after_heartbeat_until_interrupted(self):
70
        port = Port([b"HTTPS heartbeat: status=200\n", KeyboardInterrupt()])
71
        self.assertEqual(self.run_monitor(port, []), 0)
72
        self.assertEqual(port.reads, 2)
73
        self.assertTrue(port.closed)
74
 
75
    def test_check_timeout_is_a_failure(self):
76
        port = Port([])
77
        self.assertEqual(self.run_monitor(port, ["--check", "--seconds", "1"], [0, 2]), 1)
78
        self.assertTrue(port.closed)
79
 
80
    def test_reboot_check_requires_both_on_air_and_a_good_heartbeat(self):
81
        port = Port([b"HTTPS heartbeat: status=200\n", b"ON AIR: autonomous Wi-Fi signaling\n"])
82
        self.assertEqual(self.run_monitor(port, ["--check", "--reset"]), 0)
83
        self.assertEqual(port.reads, 2)
84
        self.assertEqual(port.writes, [b'{"cmd":"restart_device"}\n'])
85
        self.assertTrue(port.closed)