bzuccaro/Mark-VIII
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(reliability): rfplan review fixes — cli commit, assoc poll, ensure_attack wiring, idle reload

Browse Source
This commit is contained in:
bzuccaro
2026-08-22 14:35:12 -06:00
parent 15cd3c5eb8
commit bace45d6e4
3 changed files with 206 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_mk8_rfplan.py
+139 -16
View File
@@ -21,14 +21,29 @@ class RfPlanTest(unittest.TestCase):
self.seq = []
self.paused = 0
self.resumed = 0
self.sleeps = []
self.uci_show = {}
self.gets = {}
self.iw_fail_left = 0
self._old = (server.device_run, server._pause_hop,
server._resume_hop, server._read_hop)
server._resume_hop, server._read_hop,
mk8_rfplan.time.sleep)
def fake_get(key):
if key == 'pineapd.wlan1mon.hop':
return 0, '0\n', ''
return self.gets.get(key, (1, '', 'entry not found'))
def fake_run(args, timeout=20, input_data=None):
args = list(args)
self.seq.append('cmd:' + ' '.join(str(a) for a in args))
self.runs.append((args, timeout))
if args[0] == 'iw':
if self.iw_fail_left > 0:
self.iw_fail_left -= 1
return 0, 'Not connected.\n', ''
return 0, ('Connected to aa:bb:cc:dd:ee:ff (on wlan1up)\n'
'\tSSID: Net\n'), ''
if args[:2] == ['uci', 'show']:
cfg = self.uci_show.get(args[2])
if cfg is None:
@@ -36,13 +51,10 @@ class RfPlanTest(unittest.TestCase):
out = ''.join("%s.%s='%s'\n" % (args[2], k, v)
for k, v in sorted(cfg.items()))
return 0, out, ''
if args[0] == 'uci' and args[1] == '-q':
return fake_get(args[3])
if args[:2] == ['uci', 'get']:
if args[2] == 'pineapd.wlan1mon.hop':
return 0, '0\n', ''
return 1, '', 'entry not found'
if args[0] == 'iw':
return 0, ('Connected to aa:bb:cc:dd:ee:ff (on wlan1up)\n'
'\tSSID: Net\n'), ''
return fake_get(args[2])
return 0, '', ''
def fake_pause():
@@ -57,15 +69,21 @@ class RfPlanTest(unittest.TestCase):
server._pause_hop = fake_pause
server._resume_hop = fake_resume
server._read_hop = lambda: '0'
mk8_rfplan.time.sleep = lambda s: self.sleeps.append(s)
def tearDown(self):
(server.device_run, server._pause_hop,
server._resume_hop, server._read_hop) = self._old
server._resume_hop, server._read_hop,
mk8_rfplan.time.sleep) = self._old
@property
def cmds(self):
return [s[len('cmd:'):] for s in self.seq if s.startswith('cmd:')]
def mutations(self):
return [c for c in self.cmds
if c.startswith(('uci set', 'uci commit'))]
def test_current_role_reads_uci(self):
cases = [
({'mode': 'sta', 'disabled': '0'}, 'uplink'),
@@ -94,33 +112,71 @@ class RfPlanTest(unittest.TestCase):
'uci set wireless.wlan1up.key=key',
'uci set wireless.wlan1up.disabled=0',
'uci set network.cli=interface',
'uci set network.cli.proto=dhcp'):
'uci set network.cli.proto=dhcp',
'uci commit network'):
self.assertIn(expected, cmds)
commit = cmds.index('uci commit wireless')
commit_net = cmds.index('uci commit network')
commit_wireless = cmds.index('uci commit wireless')
for c in ('uci set wireless.wlan1up.mode=sta',
'uci set wireless.wlan1up.disabled=0',
'uci set network.cli=interface',
'uci set network.cli.proto=dhcp'):
self.assertLess(cmds.index(c), commit)
reload_at = cmds.index('wifi reload')
self.assertLess(commit, reload_at)
self.assertLess(cmds.index(c), commit_net)
self.assertLess(commit_net, cmds.index('uci commit wireless'))
self.assertLess(commit_wireless, cmds.index('wifi reload'))
self.assertEqual(self.paused, 1)
self.assertLess(self.seq.index('pause'),
self.seq.index('cmd:wifi reload'))
self.assertEqual(result['assoc'], 'aa:bb:cc:dd:ee:ff')
def test_uplink_enables_present_but_disabled_cli_network(self):
self.gets['network.cli'] = (0, 'interface\n', '')
self.gets['network.cli.disabled'] = (0, '1\n', '')
result = mk8_rfplan.set_role('uplink', ssid='Net')
self.assertTrue(result.get('ok'), result)
cmds = self.cmds
self.assertIn('uci set network.cli.disabled=0', cmds)
self.assertIn('uci commit network', cmds)
self.assertNotIn('uci set network.cli=interface', cmds)
def test_assoc_poll_succeeds_on_third_attempt(self):
self.iw_fail_left = 2
result = mk8_rfplan.set_role('uplink', ssid='Net')
self.assertTrue(result.get('ok'), result)
self.assertEqual(result['assoc'], 'aa:bb:cc:dd:ee:ff')
self.assertEqual(len(self.sleeps), 3)
iw_calls = [c for c in self.cmds
if c.startswith('iw dev wlan1up link')]
self.assertEqual(len(iw_calls), 3)
def test_assoc_poll_exhaustion_reverts_uplink(self):
self.iw_fail_left = 99
result = mk8_rfplan.set_role('uplink', ssid='Net')
self.assertFalse(result['ok'])
self.assertIn('association failed', result['error'])
self.assertEqual(len(self.sleeps), 5)
self.assertIn('uci set wireless.wlan1up.disabled=1', self.cmds)
self.assertEqual(self.resumed, 1)
def test_set_role_uplink_requires_ssid(self):
result = mk8_rfplan.set_role('uplink')
self.assertFalse(result['ok'])
self.assertIn('ssid', result['error'])
self.assertEqual(self.runs, [])
self.assertEqual(self.mutations(), [])
self.assertEqual(self.paused, 0)
def test_idle_reloads_so_sta_disassociates_now(self):
self.uci_show['wireless.wlan1up'] = {'mode': 'sta', 'disabled': '0'}
result = mk8_rfplan.set_role('idle')
self.assertTrue(result.get('ok'), result)
self.assertIn('uci set wireless.wlan1up.disabled=1', self.cmds)
self.assertIn('wifi reload', self.cmds)
self.assertEqual(self.resumed, 1)
def test_exclusivity_switch(self):
self.uci_show['wireless.wlan1up'] = {'mode': 'ap'}
mk8_rfplan.ensure_attack()
self.assertEqual([c for c in self.cmds
if c.startswith(('uci set', 'uci commit'))], [])
self.assertEqual(self.mutations(), [])
self.assertEqual(self.resumed, 0)
self.assertEqual(self.paused, 0)
self.uci_show['wireless.wlan1up'] = {'mode': 'sta', 'disabled': '0'}
@@ -150,6 +206,73 @@ class RfPlanTest(unittest.TestCase):
self.assertEqual(status, 200)
self.assertEqual(data['role'], 'idle')
def test_ensure_attack_wired_into_attacks_deploy(self):
calls = []
def fake_set_role(role, ssid=None, psk=None):
calls.append(role)
return {'ok': True, 'role': role}
self._patch_rfplan(lambda: 'uplink', fake_set_role)
old_dep = server._deploy_wpa_open
old_state = server.update_pineap_state
server._deploy_wpa_open = lambda kind, body: {}
server.update_pineap_state = lambda *a, **kw: {}
try:
status, payload = server.h_attacks_deploy(
CtxStub({'kind': 'wpa'}))
finally:
self._unpatch_rfplan()
server._deploy_wpa_open = old_dep
server.update_pineap_state = old_state
self.assertEqual(status, 200)
self.assertEqual(calls, ['attack'])
def test_attacks_deploy_skips_switch_when_not_uplink(self):
calls = []
def fake_set_role(role, ssid=None, psk=None):
calls.append(role)
return {'ok': True}
self._patch_rfplan(lambda: 'idle', fake_set_role)
old_dep = server._deploy_wpa_open
old_state = server.update_pineap_state
server._deploy_wpa_open = lambda kind, body: {}
server.update_pineap_state = lambda *a, **kw: {}
try:
status, _ = server.h_attacks_deploy(CtxStub({'kind': 'wpa'}))
finally:
self._unpatch_rfplan()
server._deploy_wpa_open = old_dep
server.update_pineap_state = old_state
self.assertEqual(status, 200)
self.assertEqual(calls, [])
def test_ensure_attack_wired_into_radio1_ap_request(self):
calls = []
def fake_set_role(role, ssid=None, psk=None):
calls.append(role)
return {'ok': True, 'role': role}
self._patch_rfplan(lambda: 'uplink', fake_set_role)
try:
status, payload = server.h_pineap_wifi_set_ap(
CtxStub({'open': {'enabled': False, 'channel': 36}}))
finally:
self._unpatch_rfplan()
self.assertEqual(status, 200)
self.assertEqual(calls, ['attack'])
def _patch_rfplan(self, current_role, set_role):
self._rf_old = (mk8_rfplan.current_role, mk8_rfplan.set_role)
mk8_rfplan.current_role = current_role
mk8_rfplan.set_role = set_role
def _unpatch_rfplan(self):
mk8_rfplan.current_role, mk8_rfplan.set_role = self._rf_old
if __name__ == '__main__':
unittest.main()