Mark-VIII/tests/test_shutdown.py

import os
import shutil
import sys
import tempfile
import unittest

sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'payload', 'user', 'remote_access', 'pager-webui'))
import server


class PagerTruthTest(unittest.TestCase):
    def setUp(self):
        self.uci = {
            'pineapd.wlan1mon.hop': '1',
            'pineapd.wlan2mon.disable': '0',
            'wireless.dummy_radio0.disabled': '0',
            'wireless.radio1.channel': 'auto',
            'wireless.radio1.band': '5g',
            'wireless.wlan1open': None,
            'wireless.wlan1wpa': None,
        }
        self.ifaces = {}
        fd, self.snap = tempfile.mkstemp(suffix='.json')
        os.close(fd)
        os.unlink(self.snap)
        self.old_file = server.PAGER_SNAPSHOT_FILE
        self.old_pineap = server.PINEAP_STATE_FILE
        server.PAGER_SNAPSHOT_FILE = self.snap
        server.PINEAP_STATE_FILE = self.snap + '.pineap'
        self.old_run = server.device_run
        server.device_run = self.fake_run
        self.old_ent = server._disable_enterprise_ap
        server._disable_enterprise_ap = lambda resume_hop=True: None

    def tearDown(self):
        server.device_run = self.old_run
        server._disable_enterprise_ap = self.old_ent
        server.PAGER_SNAPSHOT_FILE = self.old_file
        server.PINEAP_STATE_FILE = self.old_pineap
        for path in (self.snap, self.snap + '.tmp', self.snap + '.pineap'):
            try:
                os.unlink(path)
            except OSError:
                pass

    def fake_run(self, args, timeout=20, input_data=None):
        a = list(args)
        if a[:2] == ['uci', '-q'] and a[2] == 'get':
            key = a[3]
            if key.startswith('wireless.') and key.count('.') == 1:
                name = key.split('.', 1)[1]
                if name in self.ifaces:
                    return 0, 'wifi-iface\n', ''
                return 1, '', ''
            val = self.uci.get(key)
            if val is None:
                return 1, '', ''
            return 0, val + '\n', ''
        if a[:2] == ['uci', 'show']:
            sec = a[2]
            name = sec.split('.', 1)[-1]
            cfg = self.ifaces.get(name) or {}
            body = ''.join("%s.%s='%s'\n" % (sec, k, v) for k, v in cfg.items())
            return (0, body, '') if cfg or name in self.ifaces else (1, '', '')
        if a[:2] == ['uci', 'set']:
            expr = a[2]
            if '=' not in expr:
                return 0, '', ''
            key, _, val = expr.partition('=')
            parts = key.split('.')
            if len(parts) == 2 and parts[0] == 'wireless' and val == 'wifi-iface':
                self.ifaces.setdefault(parts[1], {})
                return 0, '', ''
            if len(parts) == 3 and parts[0] == 'wireless' and (
                    parts[1] in self.ifaces or parts[1] in ('wlan1open', 'wlan1wpa', 'wlan1ent')):
                self.ifaces.setdefault(parts[1], {})[parts[2]] = val
                return 0, '', ''
            self.uci[key] = val
            return 0, '', ''
        if a[:2] == ['uci', 'delete']:
            key = a[2]
            parts = key.split('.')
            if len(parts) == 2 and parts[0] == 'wireless':
                self.ifaces.pop(parts[1], None)
            self.uci.pop(key, None)
            return 0, '', ''
        if a[:2] == ['uci', 'commit']:
            return 0, '', ''
        if a[0] in ('wifi', '/etc/init.d/pineapd', 'kill', 'iw'):
            return 0, '', ''
        return 0, '', ''

    def test_restore_reverts_radio1_ap_and_hop(self):
        server.capture_pager_snapshot()
        self.assertTrue(os.path.isfile(self.snap))
        self.uci['pineapd.wlan1mon.hop'] = '0'
        self.ifaces['wlan1open'] = {'ssid': 'EvilTwin', 'disabled': '0', 'device': 'radio1'}
        self.uci['wireless.dummy_radio0.disabled'] = '1'
        result = server.restore_pager_truth()
        self.assertTrue(result['ok'])
        self.assertTrue(result['restored'])
        self.assertEqual(self.uci['pineapd.wlan1mon.hop'], '1')
        self.assertNotIn('wlan1open', self.ifaces)
        self.assertEqual(self.uci['wireless.dummy_radio0.disabled'], '0')
        self.assertFalse(os.path.isfile(self.snap))

    def test_restore_without_snapshot_is_safe(self):
        result = server.restore_pager_truth()
        self.assertTrue(result['ok'])
        self.assertFalse(result['restored'])
        self.assertEqual(result['reason'], 'no snapshot')

    def test_mode_get_reports_snapshot(self):
        server.capture_pager_snapshot()
        status, payload = server.h_mode_get(None)
        self.assertEqual(status, 200)
        self.assertTrue(payload['snapshot'])
        self.assertTrue(payload['markviii'])
        self.assertEqual(payload['pager_port'], 1471)

    def test_payload_refresh_falls_back_to_disk(self):
        tmp = tempfile.mkdtemp()
        self.addCleanup(lambda: shutil.rmtree(tmp, ignore_errors=True))
        payload_dir = tmp
        for part in ('user', 'games', 'snake'):
            payload_dir = os.path.join(payload_dir, part)
            if not os.path.isdir(payload_dir):
                os.mkdir(payload_dir)
        with open(os.path.join(payload_dir, 'payload.sh'), 'w') as handle:
            handle.write('#!/bin/sh\n')
        old_roots = server.PAYLOAD_ROOTS
        old_daemon = server._payload_daemon
        server.PAYLOAD_ROOTS = (tmp,)
        server._payload_daemon = lambda *a, **k: (500, {'error': 'portal down'})
        try:
            status, data = server.h_payloads_refresh(None)
        finally:
            server.PAYLOAD_ROOTS = old_roots
            server._payload_daemon = old_daemon
        self.assertEqual(status, 200)
        self.assertEqual(len(data['payloads']), 1)
        self.assertEqual(data['payloads'][0]['key'], 'user~games~snake')
        self.assertIn('warning', data)

    def test_respawn_keeps_original_snapshot(self):
        server.capture_pager_snapshot()
        self.uci['pineapd.wlan1mon.hop'] = '0'
        server.capture_pager_snapshot()
        result = server.restore_pager_truth()
        self.assertTrue(result['restored'])
        self.assertEqual(self.uci['pineapd.wlan1mon.hop'], '1')

    def test_uci_get_keeps_settings_default(self):
        self.assertEqual(server._uci_get('missing.key', 'UTC'), 'UTC')
        self.assertIsNone(server._uci_get('missing.key'))


if __name__ == '__main__':
    unittest.main()