import importlib import io import os import sqlite3 import sys import tempfile import unittest from contextlib import redirect_stdout sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'payload', 'user', 'remote_access', 'pager-webui')) import server def setUpModule(): importlib.reload(server) class EnvCheckTest(unittest.TestCase): def setUp(self): fd, self.db = tempfile.mkstemp(suffix='.db') os.close(fd) conn = sqlite3.connect(self.db) conn.execute('CREATE TABLE scan(id INTEGER PRIMARY KEY, time INT, name TEXT)') conn.execute("INSERT INTO scan (time, name) VALUES (1786466531, 'pager')") conn.commit() conn.close() server.RECON_DB = self.db self.runs = [] self.ping_ok = True self.daemon_ok = True self.ip_link_ok = True self.iface_up = {'wlan0mon': True, 'wlan1mon': True} self.uci_state = {} server.ENV_CHECK_STATE.update({'report': None, 'overall': None, 'updated': 0, 'pool_runtime': None}) self.old_iface_up = server._iface_up server._iface_up = lambda name: self.iface_up.get(name, True) self.old_daemon = server.daemon_sock_call self.old_run = server.device_run server.daemon_sock_call = self.fake_daemon server.device_run = self.fake_run def tearDown(self): server._iface_up = self.old_iface_up server.daemon_sock_call = self.old_daemon server.device_run = self.old_run try: os.unlink(self.db) except OSError: pass def fake_daemon(self, method, path, body=None, timeout=10): if self.daemon_ok: return 200, {'autossidpool': False} return 0, None def fake_run(self, args, timeout=20, input_data=None): self.runs.append(list(args)) a = list(args) if a[0] == 'pidof' and a[1] == 'pineapple': return (0, '23456\n', '') if self.daemon_ok else (1, '', '') if a[0] == 'pidof' and a[1] == 'pineapd': return (0, '12345\n', '') if self.ping_ok else (1, '', '') if a[0] == 'uci': if a[1] == 'set': k, _, v = a[2].partition('=') self.uci_state[k] = v return (0, '', '') if a[1] == 'delete': for k in list(self.uci_state): if k == a[2] or k.startswith(a[2] + '.'): del self.uci_state[k] return (0, '', '') if a[1] == 'get': return (0, self.uci_state.get(a[2], '') + '\n', '') if a[1] == 'commit': return (0, '', '') if a[1] == 'show': sec = a[2] return (0, ''.join("%s=%s\n" % (k, v) for k, v in self.uci_state.items() if k.startswith(sec + '.')), '') if a[0] == '_pineap': return (0, '', '') if a[:3] == ['ip', 'link', 'set']: if self.ip_link_ok: self.iface_up[a[3]] = True return (0, '', '') return (1, '', 'interface unavailable') if a[0] in ('ip', '/etc/init.d/pineapd'): return (0, '', '') return (0, '', '') def safe_set(self): for key, value in server.PINEAPD_SAFE_UCI.items(): self.uci_state[key] = value def steps(self, report, needle): return [r for r in report if needle in r['detail']] def test_pass_when_state_sane(self): self.safe_set() report = server.env_check() self.assertEqual(server.ENV_CHECK_STATE['overall'], 'pass') self.assertEqual([r['ok'] for r in report], ['pass'] * len(report)) self.assertEqual(server.ENV_CHECK_STATE['pool_runtime'], 'disabled') self.assertNotIn(['_pineap', 'SSIDPOOL', 'DISABLE'], self.runs) def test_warns_sane_defaults_when_missing_without_mutating(self): self.uci_state['pineapd.@ssidpool[0].ssid'] = 'QmVlcg==' report = server.env_check() self.assertEqual( self.steps(report, 'live PineAP UCI left unchanged')[0]['ok'], 'warn') self.assertNotIn(['/etc/init.d/pineapd', 'restart'], self.runs) self.assertNotIn(['/etc/init.d/pineapd', 'stop'], self.runs) self.assertFalse(any(a[:2] == ['uci', 'set'] for a in self.runs)) self.assertFalse(any(a[:2] == ['uci', 'commit'] for a in self.runs)) self.assertFalse(any(a[:2] == ['uci', 'delete'] for a in self.runs)) self.assertEqual(self.uci_state['pineapd.@ssidpool[0].ssid'], 'QmVlcg==') for key in server.PINEAPD_SAFE_UCI: self.assertNotIn(key, self.uci_state) def test_uci_pass_when_already_set(self): self.safe_set() report = server.env_check() self.assertEqual(self.steps(report, 'sane-off UCI defaults already set')[0]['ok'], 'pass') def test_does_not_commit_refilled_pool_while_live(self): self.safe_set() self.uci_state['pineapd.@ssidpool[0].ssid'] = 'QmVlcg==' report = server.env_check() actions = ' | '.join((r.get('action') or '') for r in report) self.assertNotIn('pool-list cleared', actions) self.assertIn('pineapd.@ssidpool[0].ssid', self.uci_state) def test_restarts_pineapd_when_down(self): self.safe_set() self.ping_ok = False self.pidof_calls = 0 real_ping = self.fake_run def ping_then_up(args, timeout=20, input_data=None): if args[0] == 'pidof' and args[1] == 'pineapd': self.pidof_calls += 1 if self.pidof_calls > 1: return (0, '12345\n', '') return real_ping(args, timeout=timeout, input_data=input_data) server.device_run = ping_then_up report = server.env_check() self.assertEqual(self.steps(report, 'pineapd was down')[0]['ok'], 'fixed') self.assertIn(['/etc/init.d/pineapd', 'restart'], self.runs) def test_fail_when_pineapd_stays_down(self): self.safe_set() self.ping_ok = False report = server.env_check() self.assertEqual(self.steps(report, 'did not come back')[0]['ok'], 'fail') self.assertEqual(server.ENV_CHECK_STATE['overall'], 'fail') def test_fail_when_daemon_unreachable(self): self.daemon_ok = False report = server.env_check() self.assertEqual(self.steps(report, 'daemon unreachable')[0]['ok'], 'fail') self.assertEqual(server.ENV_CHECK_STATE['overall'], 'fail') def test_raises_down_monitors(self): self.safe_set() self.iface_up = {'wlan0mon': False, 'wlan1mon': True} report = server.env_check() self.assertEqual(self.steps(report, 'monitor interfaces brought up')[0]['ok'], 'fixed') self.assertIn(['ip', 'link', 'set', 'wlan0mon', 'up'], self.runs) def test_unavailable_monitor_fails_startup_contract(self): self.safe_set() self.iface_up = {'wlan0mon': False, 'wlan1mon': True} self.ip_link_ok = False report = server.env_check() step = self.steps(report, 'monitor interfaces unavailable')[0] self.assertEqual(step['ok'], 'fail') self.assertEqual(server.ENV_CHECK_STATE['overall'], 'fail') def test_monitors_up_pass(self): self.safe_set() report = server.env_check() self.assertEqual(self.steps(report, 'monitors up')[0]['ok'], 'pass') def test_recon_db_unreadable_fails(self): self.safe_set() os.unlink(self.db) report = server.env_check() self.assertEqual(self.steps(report, 'recon DB unreadable')[0]['ok'], 'fail') self.assertEqual(server.ENV_CHECK_STATE['overall'], 'fail') def test_recon_db_readable_reports_count(self): self.safe_set() report = server.env_check() step = self.steps(report, 'recon DB readable')[0] self.assertEqual(step['ok'], 'pass') self.assertIn('(1 scans)', step['detail']) def test_wlan0_pinned_warns(self): self.safe_set() self.uci_state['wireless.wlan0wpa.disabled'] = '0' report = server.env_check() self.assertEqual(self.steps(report, '2.4GHz under-sampled')[0]['ok'], 'warn') def test_wlan0_not_pinned_when_absent(self): self.safe_set() report = server.env_check() self.assertEqual(self.steps(report, 'no radio0 AP pins wlan0mon')[0]['ok'], 'pass') def test_wlan0_not_pinned_when_disabled(self): self.safe_set() self.uci_state['wireless.wlan0open.disabled'] = '1' self.uci_state['wireless.wlan0wpa.disabled'] = '1' report = server.env_check() self.assertEqual(self.steps(report, 'no radio0 AP pins wlan0mon')[0]['ok'], 'pass') def test_sta_uplink_warns_when_enabled_without_mutating(self): self.safe_set() self.uci_state['wireless.dummy_radio0.mode'] = 'sta' self.uci_state['wireless.dummy_radio0.ifname'] = 'wlan0' self.uci_state['wireless.dummy_radio0.disabled'] = '0' report = server.env_check() self.assertEqual(self.steps(report, 'dummy_radio0 STA uplink is enabled')[0]['ok'], 'warn') self.assertEqual(self.uci_state['wireless.dummy_radio0.disabled'], '0') self.assertNotIn(['ip', 'link', 'set', 'wlan0', 'down'], self.runs) self.assertNotIn(['wifi', 'reload'], self.runs) self.assertFalse(any( a[:2] == ['uci', 'set'] and 'dummy_radio0' in a[2] for a in self.runs if len(a) > 2)) def test_sta_uplink_pass_when_absent(self): self.safe_set() report = server.env_check() self.assertEqual(self.steps(report, 'no STA uplink pinning phy0')[0]['ok'], 'pass') def test_sta_uplink_pass_when_disabled(self): self.safe_set() self.uci_state['wireless.dummy_radio0.mode'] = 'sta' self.uci_state['wireless.dummy_radio0.disabled'] = '1' report = server.env_check() self.assertEqual(self.steps(report, 'no STA uplink pinning phy0')[0]['ok'], 'pass') def test_recon_status_exposes_sta(self): self.safe_set() self.uci_state['wireless.dummy_radio0.mode'] = 'sta' status, payload = server.h_recon_status(type('C', (), {'query': {}})()) self.assertEqual(status, 200) self.assertTrue(payload['wlan0_sta']) def test_cli_exits_zero_on_pass(self): self.safe_set() buf = io.StringIO() with redirect_stdout(buf): code = server.env_check_cli() self.assertEqual(code, 0) self.assertIn('[PASS]', buf.getvalue()) self.assertIn('ENVIRONMENT CHECK: PASS', buf.getvalue()) def test_cli_exits_one_on_fail(self): self.safe_set() self.daemon_ok = False buf = io.StringIO() with redirect_stdout(buf): code = server.env_check_cli() self.assertEqual(code, 1) self.assertIn('[FAIL]', buf.getvalue()) def test_startup_check_retries_core_failure(self): reports = [ [{'ok': 'fail', 'detail': 'daemon unreachable'}], [{'ok': 'pass', 'detail': 'daemon reachable'}], ] old_check = server.env_check old_sleep = server.time.sleep def check(): report = reports.pop(0) server.ENV_CHECK_STATE['overall'] = report[0]['ok'] return report server.env_check = check server.time.sleep = lambda seconds: None try: result = server.startup_env_check(attempts=2, delay=0) finally: server.env_check = old_check server.time.sleep = old_sleep self.assertEqual(result[0]['ok'], 'pass') def test_startup_check_raises_after_retries(self): old_check = server.env_check old_sleep = server.time.sleep def check(): server.ENV_CHECK_STATE['overall'] = 'fail' return [{'ok': 'fail', 'detail': 'daemon unreachable'}] server.env_check = check server.time.sleep = lambda seconds: None try: with self.assertRaises(RuntimeError): server.startup_env_check(attempts=2, delay=0) finally: server.env_check = old_check server.time.sleep = old_sleep def test_health_exposes_env_and_pool_runtime(self): self.safe_set() server.env_check() status, payload = server.h_health(type('C', (), {'query': {}})()) self.assertEqual(status, 200) self.assertEqual(payload['pool_runtime'], 'disabled') self.assertEqual(payload['env']['overall'], 'pass') self.assertEqual(payload['env']['counts']['pass'], len(payload['env']['steps'])) def test_recon_status_exposes_wlan0_pinned(self): self.safe_set() self.uci_state['wireless.wlan0open.disabled'] = '0' status, payload = server.h_recon_status(type('C', (), {'query': {}})()) self.assertEqual(status, 200) self.assertTrue(payload['wlan0_pinned']) if __name__ == '__main__': unittest.main()