bzuccaro
6e3968c19a
- STA uplink disable now uses uci commit + ip link set wlan0 down instead of wifi reload, so the env check never bounces radios/APs/drops monitors - Removed the wlan0mon 'hop is off' warning: field-verified hop=0 is normal (wlan1mon also reports 0 while scanning both bands); the STA/AP pinning checks are the real signals - recon/status drops wlan0_hopping; UI keeps wlan0_sta + wlan0_pinned pills
262 lines
10 KiB
Python
262 lines
10 KiB
Python
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.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] == '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[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.assertIn(['_pineap', 'SSIDPOOL', 'DISABLE'], self.runs)
|
|
|
|
def test_applies_sane_defaults_when_missing(self):
|
|
report = server.env_check()
|
|
self.assertEqual(self.steps(report, 'sane-off UCI defaults applied')[0]['ok'], 'fixed')
|
|
for key, value in server.PINEAPD_SAFE_UCI.items():
|
|
self.assertEqual(self.uci_state[key], value)
|
|
|
|
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_clears_refilled_pool_list(self):
|
|
self.uci_state['pineapd.@ssidpool[0].ssid'] = 'QmVlcg=='
|
|
report = server.env_check()
|
|
actions = ' | '.join((r.get('action') or '') for r in report)
|
|
self.assertIn('pool-list cleared', actions)
|
|
self.assertNotIn('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_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_disabled_when_enabled(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 disabled')[0]['ok'], 'fixed')
|
|
self.assertEqual(self.uci_state['wireless.dummy_radio0.disabled'], '1')
|
|
self.assertIn(['ip', 'link', 'set', 'wlan0', 'down'], self.runs)
|
|
self.assertNotIn(['wifi', 'reload'], self.runs,
|
|
'STA fix must not bounce the radios')
|
|
|
|
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_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()
|