simple/tests/test_fetch_routes.py

"""PWA <-> server HTTP contract tests.

`pysimFetch(path, body)` sends POST whenever a body is passed and GET
otherwise; the server's routes live in `PysimHandler._do_GET` / `_do_POST`.
The test-script endpoints were registered in `_do_GET` while the PWA POSTs
to them, so every Start answered 404 "Not found" (live 2026-09-29; present
since the phase-1 commit and never exercised over HTTP).  The matrix below
is the guard for that contract; the smoke test boots the real handler and
checks the three endpoints.
"""

import json
import pathlib
import re
import threading
import unittest
import urllib.error
import urllib.request

from pysim_simple_server.__main__ import _build_http_server
from pysim_simple_server.server import PysimHandler, _CARD_FREE_GET, _CARD_LOCK

ROOT = pathlib.Path(__file__).resolve().parents[1]


def _pwa_calls():
    """[(path, has_body)] for every literal pysimFetch() call in the PWA."""
    html = (ROOT / 'frontend' / 'index.html').read_text(encoding='utf-8')
    return [(m.group(1).split('?')[0], m.group(2) == ',')
            for m in re.finditer(r"pysimFetch\(\s*'([^']+)'\s*(,|\))", html)]


def _server_routes():
    """((GET exact, GET prefixes), (POST exact, POST prefixes)) of /api paths."""
    src = (ROOT / 'pysim_simple_server' / 'server.py').read_text(encoding='utf-8')
    get_seg = src[src.index('def _do_GET('):src.index('def do_POST(')]
    post_seg = src[src.index('def _do_POST('):src.index('def log_message(')]

    def routes(seg):
        exact = set(re.findall(r"self\.path == '(/api/[^']+)'", seg))
        for group in re.findall(r"self\.path in \(([^)]*)\)", seg):
            exact |= set(re.findall(r"'(/api/[^']+)'", group))
        prefixes = set(re.findall(r"self\.path\.startswith\('(/api/[^']+)'\)", seg))
        return exact, prefixes - {'/api/'}

    return routes(get_seg), routes(post_seg)


class FetchRouteMatrixTests(unittest.TestCase):
    def test_every_fetch_matches_a_route_with_the_right_method(self):
        (get, get_pref), (post, post_pref) = _server_routes()
        self.assertTrue(get and post, 'route extraction found nothing')

        def in_routes(path, exact, prefixes):
            return path in exact or any(path.startswith(p) for p in prefixes)

        for path, has_body in _pwa_calls():
            with self.subTest(path=path, post=has_body):
                if has_body:
                    self.assertTrue(in_routes(path, post, post_pref),
                                    '%s is POSTed (body) but has no POST route' % path)
                else:
                    self.assertTrue(
                        in_routes(path, get, get_pref) or in_routes(path, post, post_pref),
                        '%s is fetched without a body but has no GET/POST route' % path)


class TestScriptHttpTests(unittest.TestCase):
    """The test-script endpoints must answer on POST (the PWA's method)."""

    def setUp(self):
        self.server = _build_http_server('127.0.0.1', 0, PysimHandler)
        self.server.log_requests = False
        self.server.app = None
        threading.Thread(target=self.server.serve_forever, daemon=True).start()
        self.port = self.server.server_address[1]

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()

    def _post(self, path, body):
        req = urllib.request.Request(
            'http://127.0.0.1:%d%s' % (self.port, path),
            data=json.dumps(body).encode(),
            headers={'Content-Type': 'application/json'}, method='POST')
        try:
            with urllib.request.urlopen(req, timeout=5) as res:
                return res.status, json.loads(res.read() or b'{}')
        except urllib.error.HTTPError as e:
            return e.code, json.loads(e.read() or b'{}')

    def test_start_stop_clear_answer_on_post(self):
        # an empty script is invalid, but it proves the route exists (400)
        status, resp = self._post('/api/test/run', {})
        self.assertEqual(status, 400, resp)
        self.assertIn('error', resp)
        status, resp = self._post('/api/test/stop', {})
        self.assertEqual(status, 200, resp)
        status, resp = self._post('/api/test/clear', {})
        self.assertEqual(status, 200, resp)

    def test_the_test_endpoints_are_not_get_routes(self):
        with self.assertRaises(urllib.error.HTTPError) as cm:
            urllib.request.urlopen('http://127.0.0.1:%d/api/test/run' % self.port, timeout=5)
        self.assertEqual(cm.exception.code, 404)
        cm.exception.close()


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


class CardFreeGetTests(unittest.TestCase):
    """Cached GETs must answer while the card lock is held.

    During a long RAM install the UI polls several cached endpoints every few
    seconds; while they queued behind the install they ate the browser's
    per-host connection budget and starved /api/status - the modal's progress
    bar froze around 70% and jumped to 100% at the end (live 2026-09-29).
    """

    def setUp(self):
        self.server = _build_http_server('127.0.0.1', 0, PysimHandler)
        self.server.log_requests = False
        self.server.app = None
        self.server.sl = None
        self.server.scc = None
        self.server.stk_pending = None
        self.server.menu_active = False
        self.server.sim_menu = None
        self.server.event_list = None
        self.server.terminal_profile = None
        self.server.cli_terminal_profile = None
        threading.Thread(target=self.server.serve_forever, daemon=True).start()
        self.port = self.server.server_address[1]

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()

    def _get(self, path, timeout=2):
        return urllib.request.urlopen(
            'http://127.0.0.1:%d%s' % (self.port, path), timeout=timeout)

    def test_cached_endpoints_answer_while_the_card_lock_is_held(self):
        with _CARD_LOCK:
            for path in ('/api/status', '/api/stk-status', '/api/poll-status',
                         '/api/proactive-log', '/api/menu', '/api/events'):
                with self.subTest(path=path):
                    with self._get(path) as res:
                        self.assertEqual(res.status, 200)

    def test_card_touching_gets_stay_locked(self):
        # cardinfo runs a card command; the ES10 endpoints select/STORE DATA
        for path in ('/api/cardinfo', '/api/esim/chip', '/api/esim/profiles'):
            self.assertNotIn(path, _CARD_FREE_GET)


def _mini_cap_hex():
    """A minimal parseable CAP archive (Header + Applet + one Import)."""
    import io as _io
    import struct as _struct
    import zipfile as _zipfile

    def comp(tag, payload):
        return bytes([tag]) + _struct.pack('>H', len(payload)) + payload

    aid = bytes.fromhex('0102030405')
    header = comp(0x01, _struct.pack('>I', 0xDECAFFED) + bytes([1, 2, 0, 1, 2]) +
                  bytes([len(aid)]) + aid)
    applet = comp(0x03, bytes([1, len(aid)]) + aid + b'\x00\x01')
    imports = comp(0x04, bytes([1]) + bytes([0, 1, 7]) + bytes.fromhex('A0000000620101'))
    buf = _io.BytesIO()
    with _zipfile.ZipFile(buf, 'w') as z:
        z.writestr('Header.cap', header)
        z.writestr('Applet.cap', applet)
        z.writestr('Import.cap', imports)
    return buf.getvalue().hex().upper()


class CapCompatHttpTests(unittest.TestCase):
    """POST /api/cap-compat exists (the CAP compatibility test, v3.6.48)."""

    def setUp(self):
        self.server = _build_http_server('127.0.0.1', 0, PysimHandler)
        self.server.log_requests = False
        self.server.app = None
        self.server.sl = None
        self.server.scc = None
        threading.Thread(target=self.server.serve_forever, daemon=True).start()
        self.port = self.server.server_address[1]

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()

    def _post(self, path, body):
        req = urllib.request.Request(
            'http://127.0.0.1:%d%s' % (self.port, path),
            data=json.dumps(body).encode(),
            headers={'Content-Type': 'application/json'}, method='POST')
        try:
            with urllib.request.urlopen(req, timeout=5) as res:
                return res.status, json.loads(res.read() or b'{}')
        except urllib.error.HTTPError as e:
            return e.code, json.loads(e.read() or b'{}')

    def test_cap_compat_route_exists(self):
        # without a card session it answers 503 (route present), not 404
        status, resp = self._post('/api/cap-compat', {'cap_hex': '00'})
        self.assertEqual(status, 503, resp)
        self.assertIn('error', resp)

    def test_cap_compat_rejects_a_bad_probe(self):
        # a parseable CAP + a bad override -> 400 from the probe validation,
        # before any card I/O (a dummy scc object is enough)
        self.server.scc = object()
        status, resp = self._post('/api/cap-compat',
                                  {'cap_hex': _mini_cap_hex(), 'probe_imports': {'all': 'x'}})
        self.assertEqual(status, 400, resp)
        self.assertIn('import probe', resp['error'])


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