feat: startup environment check + self-heal (v1.3)

- env_check(): verifies daemon/pineapd/UCI/monitors/recon DB, auto-fixes
  fixable issues and re-verifies; forces runtime SSID-pool broadcast off to
  match the UI (kills the 'pool on but UI shows off' gap)
- server.py --env-check CLI; payload.sh runs it verbosely before starting,
  aborts on core failure
- serve() runs the check at every startup (boot + procd respawn)
- /api/health exposes env report + pool_runtime; /api/recon/status exposes
  wlan0_pinned (2.4GHz under-sampling warning); recon page warns when a
  radio0 AP pins wlan0mon
- recon start failures now include the daemon reason in the UI error
- shared stabilization refactor (_stabilize_uci/PINEAPD_SAFE_UCI/_raise_monitors)
- tests: test_env_check.py (18) + health/recon updates
This commit is contained in:
2026-08-19 10:25:02 -05:00
parent 37b5e821dd
commit 904843307e
8 changed files with 468 additions and 28 deletions
+7
View File
@@ -51,6 +51,13 @@ Then on the Pager menu, run **Mark VIII**:
- Re-run the payload while running to **Stop** the service.
- `PAYLOAD_GET_CONFIG pager_webui auto_mode/run_mode` skip the prompt.
Every payload run (and every service startup) first runs an **environment
check** that prints on the payload screen / `/tmp/pagerwebui.log`: daemon
reachable, pineapd alive, sane-off PineAP UCI defaults applied (SSID pool
broadcast off, 6GHz hopper disabled, monitor bands), runtime pool sync, monitor
interfaces up, and recon DB readable. Anything fixable is fixed and re-verified;
startup aborts only if a core dependency fails.
Browse `http://172.16.52.1:8080/` and log in with the device password.
## Uninstall / recovery
@@ -8,7 +8,7 @@
"title": "Mark VIII",
"author": "c4ch3c4d3",
"description": "Mark VII-style web management UI for the WiFi Pineapple Pager",
"version": "1.2",
"version": "1.3",
"category": "remote_access",
"tags": ["remote-access", "web-interface", "device-management", "pineap"],
"firmware": "Pineapple Pager 24.10.1"
@@ -2,7 +2,7 @@
# Title: Mark VIII
# Description: Mark VII-style web management UI for the WiFi Pineapple Pager
# Author: c4ch3c4d3
# Version: 1.1
# Version: 1.3
# Category: Remote-Access
# Tags: remote-access, web-interface, device-management, pineap
# Firmware: Pineapple Pager 24.10.1
@@ -28,7 +28,7 @@ get_pager_ip() {
}
LOG "cyan" "+---------------------------+"
LOG "cyan" "| Mark VIII v1.1 |"
LOG "cyan" "| Mark VIII v1.3 |"
LOG "cyan" "+---------------------------+"
if ! command -v python3 >/dev/null 2>&1; then
@@ -36,6 +36,16 @@ if ! command -v python3 >/dev/null 2>&1; then
exit 1
fi
run_env_check() {
LOG "cyan" "Running environment check..."
if ! python3 "$SCRIPT_DIR/server.py" --env-check; then
LOG "red" "Environment check FAILED. Fix the issues above and re-run the payload."
sleep 3
exit 1
fi
LOG "green" "Environment check passed."
}
if [ -f "$INIT_SCRIPT" ] && "$INIT_SCRIPT" running 2>/dev/null; then
PAGER_IP=$(get_pager_ip)
LOG "green" "Mark VIII service is running"
@@ -67,6 +77,7 @@ else
fi
if user_confirmed "$resp"; then
run_env_check
LOG "cyan" "Starting as background service..."
[ ! -f "$SCRIPT_DIR/server.py" ] && { LOG "red" "server.py not found!"; exit 1; }
cp "$SCRIPT_DIR/pagerwebui.init" "$INIT_SCRIPT"
@@ -82,6 +93,7 @@ if user_confirmed "$resp"; then
exit 0
fi
run_env_check
LOG "cyan" "Starting foreground mode..."
cleanup() {
LOG "yellow" "Stopping Mark VIII..."
+204 -21
View File
@@ -1255,7 +1255,13 @@ def h_recon_start(ctx):
# appends a scan without discarding history.
status, data = daemon_sock_call('POST', '/api/pineap/recon/new', body=body)
if status != 200 or not (data or {}).get('success'):
return 502, {'error': 'native recon scan failed', 'detail': data}
detail = data if isinstance(data, dict) else {}
reason = detail.get('error') or detail.get('detail')
if not reason and status == 0:
reason = 'daemon unreachable'
return 502, {'error': 'native recon scan failed',
'detail': ('recon/new: %s' % reason) if reason else None,
'daemon': detail or None}
_recon_scan_state['active'] = True
_recon_scan_state['started'] = time.time()
_recon_scan_state['duration'] = scan_time
@@ -1352,6 +1358,7 @@ def h_recon_status(ctx):
'active': last_activity is not None and int(time.time()) - last_activity < 300,
'scanning': scanning, 'scan_remaining': remaining, 'stale': stale,
'hopper_online': _hopper_online(),
'wlan0_pinned': _wlan0_pinned(),
'history_reset': _recon_history_reset()}
@@ -3502,6 +3509,41 @@ def _sigsegv_count():
return out.count('SIGSEGV')
PINEAPD_SAFE_UCI = {
'pineapd.@ssidpool[0].disable': '1',
'pineapd.wlan2mon.disable': '1',
'pineapd.wlan2mon.hop': '0',
'pineapd.wlan1mon.bands': '5',
'pineapd.wlan0mon.bands': '2',
'pineapd.wlan1mon.hop': '0',
}
def _apply_uci_wanted(wanted):
"""Idempotently apply a wanted UCI key/value set. Returns changed keys."""
actions = []
for key, value in wanted.items():
rc, out, err = device_run(['uci', 'get', key])
if rc != 0 or out.strip() != value:
device_run(['uci', 'set', '%s=%s' % (key, value)])
actions.append(key.split('.')[-1])
return actions
def _stabilize_uci():
"""Sane-off UCI pass shared by the health monitor and the startup env
check. Clears the refilled pool list too. Commits when changed and returns
the action labels."""
actions = _apply_uci_wanted(PINEAPD_SAFE_UCI)
rc, out, err = device_run(['uci', 'get', 'pineapd.@ssidpool[0].ssid'])
if rc == 0 and out.strip():
device_run(['uci', 'delete', 'pineapd.@ssidpool[0].ssid'])
actions.append('pool-list cleared')
if actions:
device_run(['uci', 'commit', 'pineapd'])
return actions
def _stabilize_pineapd():
"""Idempotent crash-source pass. Field-verified SIGSEGV/terminate sources
on this firmware:
@@ -3513,26 +3555,8 @@ def _stabilize_pineapd():
4. a large refilled pool (collect refills it; crashes observed even
with broadcast disabled)
The pool list itself is cleared. Returns what changed."""
actions = []
wanted = {
'pineapd.@ssidpool[0].disable': '1',
'pineapd.wlan2mon.disable': '1',
'pineapd.wlan2mon.hop': '0',
'pineapd.wlan1mon.bands': '5',
'pineapd.wlan0mon.bands': '2',
'pineapd.wlan1mon.hop': '0',
}
for key, value in wanted.items():
rc, out, err = device_run(['uci', 'get', key])
if rc != 0 or out.strip() != value:
device_run(['uci', 'set', '%s=%s' % (key, value)])
actions.append(key.split('.')[-1])
rc, out, err = device_run(['uci', 'get', 'pineapd.@ssidpool[0].ssid'])
if rc == 0 and out.strip():
device_run(['uci', 'delete', 'pineapd.@ssidpool[0].ssid'])
actions.append('pool-list cleared')
actions = _stabilize_uci()
if actions:
device_run(['uci', 'commit', 'pineapd'])
return 'stabilized: ' + ', '.join(actions)
return 'pineapd restart'
@@ -3589,10 +3613,18 @@ def health_check():
return dict(h)
def _bring_monitors_up(h):
def _raise_monitors():
"""Bring any down monitor interfaces up. Returns the interfaces raised."""
raised = []
for name in ('wlan1mon', 'wlan0mon'):
if _monitor_down(name):
device_run(['ip', 'link', 'set', name, 'up'], timeout=10)
raised.append(name)
return raised
def _bring_monitors_up(h):
_raise_monitors()
h['last_action'] = 'monitor interfaces brought up'
h['monitor_fixes'] = h.get('monitor_fixes', 0) + 1
@@ -3603,6 +3635,15 @@ def h_health(ctx):
h['pool_disabled'] = _uci_section('pineapd.@ssidpool[0]').get('disable') == '1'
h['wlan1mon_up'] = not _monitor_down('wlan1mon')
h['wlan0mon_up'] = not _monitor_down('wlan0mon')
h['pool_runtime'] = ENV_CHECK_STATE.get('pool_runtime')
if ENV_CHECK_STATE.get('report'):
h['env'] = {
'overall': ENV_CHECK_STATE['overall'],
'updated': ENV_CHECK_STATE['updated'],
'counts': {k: _env_count(ENV_CHECK_STATE['report'], k)
for k in ('pass', 'fixed', 'warn', 'fail')},
'steps': ENV_CHECK_STATE['report'],
}
return 200, h
@@ -3619,6 +3660,126 @@ def start_health_monitor():
threading.Thread(target=_health_loop, daemon=True).start()
# --------------------------------------------------------------------------
# Startup environment check: run once at service startup (and via
# ``server.py --env-check`` on the payload screen) to make the device match
# the UI before the user interacts with it.
# --------------------------------------------------------------------------
ENV_CHECK_STATE = {'report': None, 'overall': None, 'updated': 0, 'pool_runtime': None}
def _env_step(report, ok, detail, action=None):
step = {'ok': ok, 'detail': detail}
if action:
step['action'] = action
report.append(step)
def _env_overall(report):
if any(r['ok'] == 'fail' for r in report):
return 'fail'
if any(r['ok'] == 'fixed' for r in report):
return 'fixed'
if any(r['ok'] == 'warn' for r in report):
return 'warn'
return 'pass'
def _env_count(report, kind):
return sum(1 for r in report if r['ok'] == kind)
def _pineapd_alive():
rc, out, err = device_run(['pidof', 'pineapd'], timeout=10)
return rc == 0 and bool((out or '').strip())
def _sync_pool_runtime():
"""Force pineapd's runtime SSID-pool broadcast off so it matches the
sane-off UCI default the UI derives from. Returns (state, detail)."""
rc, out, err = _pineap('SSIDPOOL', 'DISABLE', timeout=15)
if rc != 0:
detail = (err or out or '').strip() or 'no output'
return 'unknown', 'SSIDPOOL DISABLE failed: %s' % detail[-200:]
return 'disabled', 'SSIDPOOL DISABLE sent'
def _wlan0_pinned():
"""True when a radio0 AP (OpenAP/Evil WPA) is enabled. A phy's channel is
held by its AP interface, so an enabled radio0 AP pins wlan0mon to one
2.4GHz channel and 2.4GHz recon results are under-sampled."""
for name in ('wlan0open', 'wlan0wpa'):
cfg = _uci_wifi_iface(name) or {}
if cfg and cfg.get('disabled') != '1':
return True
return False
def env_check():
"""Full environment pass, run at service startup and via
``server.py --env-check``. Auto-fixes everything fixable, re-verifies, and
returns a list of step reports (ok: pass|fixed|warn|fail). Only core
dependencies (daemon, pineapd, recon DB) can fail."""
report = []
status, data = daemon_sock_call('GET', '/api/pineap/get_config')
if status != 200:
_env_step(report, 'fail', 'daemon unreachable (status %s)' % status)
else:
_env_step(report, 'pass', 'daemon reachable')
if _pineapd_alive():
_env_step(report, 'pass', 'pineapd running')
else:
actions = _stabilize_uci()
device_run(['/etc/init.d/pineapd', 'restart'], timeout=30)
if _pineapd_alive():
_env_step(report, 'fixed', 'pineapd was down; stabilized and restarted',
', '.join(actions) or 'restart')
else:
_env_step(report, 'fail', 'pineapd did not come back after restart',
', '.join(actions) or 'restart')
actions = _stabilize_uci()
if actions:
_env_step(report, 'fixed', 'pineapd sane-off UCI defaults applied',
', '.join(actions))
else:
_env_step(report, 'pass', 'pineapd sane-off UCI defaults already set')
pool_runtime, pool_detail = _sync_pool_runtime()
ENV_CHECK_STATE['pool_runtime'] = pool_runtime
if pool_runtime == 'disabled':
_env_step(report, 'pass', 'SSID-pool broadcast off (runtime)')
else:
_env_step(report, 'warn', pool_detail)
raised = _raise_monitors()
if raised:
_env_step(report, 'fixed', 'monitor interfaces brought up', ', '.join(raised))
else:
_env_step(report, 'pass', 'monitors up (wlan0mon, wlan1mon)')
try:
rows = _db_rows(RECON_DB, 'SELECT count(*) AS c FROM scan', timeout=15)
count = rows[0]['c'] if rows else 0
_env_step(report, 'pass', 'recon DB readable (%d scans)' % count)
except RuntimeError as exc:
_env_step(report, 'fail', 'recon DB unreadable: %s' % exc)
if _wlan0_pinned():
_env_step(report, 'warn', '2.4GHz under-sampled: a radio0 AP is up and pins '
'wlan0mon to one channel during scans')
else:
_env_step(report, 'pass', 'no radio0 AP pins wlan0mon')
ENV_CHECK_STATE['report'] = report
ENV_CHECK_STATE['overall'] = _env_overall(report)
ENV_CHECK_STATE['updated'] = time.time()
return report
def h_attacks_clients(ctx):
"""Clients (recon devices) plus the APs matching an SSID, for targeting."""
ssid = ((ctx.query or {}).get('ssid') or '').strip()
@@ -5164,6 +5325,10 @@ def _recon_watchdog_loop():
def serve():
try:
env_check()
except Exception:
pass
threading.Thread(target=live_loop, daemon=True).start()
threading.Thread(target=_recon_watchdog_loop, daemon=True).start()
start_health_monitor()
@@ -5177,5 +5342,23 @@ def serve():
threading.Thread(target=_handle_conn, args=(conn, addr), daemon=True).start()
def env_check_cli():
"""Run the environment check and print a verbose report to stdout.
Returns the process exit code (0 = pass/warn, 1 = core failure)."""
report = env_check()
for step in report:
line = '[%s] %s' % (step['ok'].upper(), step['detail'])
if step.get('action'):
line += ' (%s)' % step['action']
print(line)
counts = {k: _env_count(report, k) for k in ('pass', 'fixed', 'warn', 'fail')}
print('ENVIRONMENT CHECK: %s (%d pass, %d fixed, %d warn, %d fail)' % (
ENV_CHECK_STATE['overall'].upper(), counts['pass'], counts['fixed'],
counts['warn'], counts['fail']))
return 0 if ENV_CHECK_STATE['overall'] != 'fail' else 1
if __name__ == '__main__':
if '--env-check' in sys.argv:
sys.exit(env_check_cli())
serve()
@@ -516,6 +516,9 @@ const Live = (() => {
if (h.pineap_up === false) {
el.textContent = 'PINEAPD DOWN';
el.className = 'health-chip bad';
} else if (h.env && h.env.overall === 'fail') {
el.textContent = 'ENV CHECK FAIL';
el.className = 'health-chip bad';
} else if (h.pool_disabled) {
el.textContent = 'POOL OFF';
el.className = 'health-chip warn';
@@ -177,7 +177,8 @@ views.dashboard = (root) => {
liveCards.health.textContent = (h2.pineap_up ? 'pineapd up' : 'pineapd DOWN') +
' · wlan0mon ' + (h2.wlan0mon_up ? 'up' : 'down') +
' · wlan1mon ' + (h2.wlan1mon_up ? 'up' : 'down') +
(h2.pool_disabled ? ' · pool off' : '');
(h2.pool_disabled ? ' · pool off' : '') +
((h2.env || {}).overall ? ' · env ' + h2.env.overall : '');
liveCards.health.style.color = h2.pineap_up ? '' : '#b71c1c';
}).catch(() => {});
PagerAPI.get('/api/recon/status').then((r) => {
@@ -1460,7 +1461,8 @@ views.recon = (root) => {
apBand: 'all', apEnc: 'all', gps: null, wigle: null,
compare: [], history: {}, mapBand: null,
archive: null, archives: [], scanRemaining: null,
hopperOnline: null, historyReset: false, scanErr: null };
hopperOnline: null, historyReset: false, scanErr: null,
wlan0Pinned: false };
const cols = reconLoadCols();
// ---- title cards (stat cards with optional mini charts) ----
@@ -1658,9 +1660,10 @@ views.recon = (root) => {
}
if (state.hopperOnline === false) bits.push('Hopper radio offline — fewer networks seen');
if (state.historyReset) bits.push('History reset — previous scans archived (see Previous Scans)');
if (state.wlan0Pinned) bits.push('2.4GHz under-sampled — OpenAP/Evil WPA holds wlan0mon');
if (state.scanErr) bits.push(state.scanErr);
scanStatus.textContent = bits.join(' · ');
scanStatus.classList.toggle('warn', state.hopperOnline === false || state.historyReset || !!state.scanErr);
scanStatus.classList.toggle('warn', state.hopperOnline === false || state.historyReset || state.wlan0Pinned || !!state.scanErr);
}
scanToggle.addEventListener('change', () => {
if (pendingScan) { scanToggle.checked = !scanToggle.checked; return; }
@@ -2522,6 +2525,7 @@ views.recon = (root) => {
state.scanRemaining = r.data.scan_remaining != null ? r.data.scan_remaining : null;
state.hopperOnline = r.data.hopper_online;
state.historyReset = !!r.data.history_reset;
state.wlan0Pinned = !!r.data.wlan0_pinned;
if (!pendingScan) scanToggle.checked = scanning;
renderScanBar();
if (wasScanning !== scanning) restartPoll();
+230
View File
@@ -0,0 +1,230 @@
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_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()
+2 -1
View File
@@ -224,7 +224,8 @@ class DaemonSockTest(unittest.TestCase):
type('C', (), {'args': (), 'body': {'scan_time': 30}})())
self.assertEqual(status, 502)
self.assertEqual(data['error'], 'native recon scan failed')
self.assertEqual(data['detail'], {'error': 'no radio'})
self.assertEqual(data['detail'], 'recon/new: no radio')
self.assertEqual(data['daemon'], {'error': 'no radio'})
self.assertFalse(server._recon_scan_state['active'])