simple/tests/test_ram_batch.py

#!/usr/bin/env python3
"""Batch CAP install tests (v3.28.0).

The batch worker applies one install plan to every card that gets equipped,
resolving the keys from each card's own preset (matched by ICCID).  A card is
attempted once per run - re-equipping it is reported, never re-installed - a
failure never stops the batch, and the stop flag finishes a running install
first.
"""

import json
import sys
import threading
import time
import unittest
import urllib.error
import urllib.request
from pathlib import Path
from types import SimpleNamespace
from unittest import mock

PROJECTS = Path(__file__).resolve().parents[2]
PY_SIM = PROJECTS / 'pysim'
if str(PY_SIM) not in sys.path:
    sys.path.insert(0, str(PY_SIM))

from pysim_simple_server.__main__ import _build_http_server
from pysim_simple_server import presets
from pysim_simple_server import server as srv
from pysim_simple_server.server import PysimHandler, _CARD_LOCK


def _plan(kind='install-cap'):
    return {'kind': kind,
            'seq': [{'name': 'INSTALL [for load]', 'apdu': '80E60200'},
                    {'name': 'LOAD (1/1)', 'apdu': '80E80000'},
                    {'name': 'INSTALL [for install]', 'apdu': '80E60000'}],
            'spi2': '01', 'include_cpi': True, 'ram_format': 'compact',
            'response': {'load_file_aid': 'F0414C46416101',
                         'module_aid': 'F0414C4641610101'}}


class _FakeStore:
    def __init__(self, entries):
        self.entries = entries

    def find_by_iccid(self, iccid):
        want = presets.normalize_iccid(iccid)
        for p in self.entries:
            if presets.normalize_iccid(p.get('iccid')) == want:
                return p
        return None


def _preset(iccid, name='Card', keyset=1):
    byte = '%02X' % ((keyset << 4) | 0x05)
    ks = {'kic': byte, 'kid': byte, 'kicKey': '00' * 16, 'kidKey': '11' * 16,
          'cntr': '0000000001'}
    tar = {'role': 'isd', 'tar': '000000', 'msl': '16', 'desc': '', 'keyset': keyset}
    return {'id': 'p-' + iccid, 'name': name, 'iccid': iccid, 'keysets': [ks],
            'tars': [tar]}


def _fake_server(iccid, store, session=1, ready=None):
    """A server-shaped object carrying only the attributes the real server
    object has.  The connection state is the module global `_CARD_CONNECTED`
    (v3.28.0 audit: a fake `connected` attribute here hid a bug - the worker
    read it and never saw a card); `card_ready_session` is the marker
    _apply_equipped_card sets when the session is fully initialized (it
    defaults to the session, i.e. a finished equip)."""
    return SimpleNamespace(iccid=iccid, card_presets=store,
                           card_session=session,
                           card_ready_session=session if ready is None else ready,
                           scc=object())


class BatchStateTests(unittest.TestCase):
    def setUp(self):
        srv._RAM_BATCH = None

    def tearDown(self):
        srv._RAM_BATCH = None

    def test_the_report_counts_and_caps_the_attempts(self):
        st = srv._ram_batch_new_state(_plan())
        srv._ram_batch_record(st, {'iccid': '8970119000004600098', 'result': 'installed'})
        srv._ram_batch_record(st, {'iccid': '8970119000004600106', 'result': 'failed',
                                   'failed_step': 'LOAD (3/7)', 'por_sw': '6A80'})
        srv._ram_batch_record(st, {'iccid': '8970119000004600114', 'result': 'skipped',
                                   'reason': 'no preset matches this ICCID'})
        self.assertEqual(st['counts'], {'installed': 1, 'failed': 1, 'skipped': 1})
        self.assertEqual(st['attempts_total'], 3)
        # the notice names the last attempt's outcome
        self.assertIn('skipped: no preset matches this ICCID', st['notice'])
        srv._ram_batch_record(st, {'iccid': '8970119000004600122', 'result': 'failed',
                                   'failed_step': 'LOAD (3/7)', 'por_sw': '6A80'})
        self.assertIn('failed at LOAD (3/7) (6A80)', st['notice'])
        # the report keeps the last _RAM_BATCH_ATTEMPTS_MAX entries
        for _ in range(srv._RAM_BATCH_ATTEMPTS_MAX + 10):
            srv._ram_batch_record(st, {'result': 'skipped', 'reason': 'x'})
        self.assertEqual(len(st['attempts']), srv._RAM_BATCH_ATTEMPTS_MAX)
        self.assertEqual(st['attempts_total'], 4 + srv._RAM_BATCH_ATTEMPTS_MAX + 10)

    def test_the_failure_fields_name_the_failing_step(self):
        result = {'failed_step': 2, 'steps': [
            {'name': 'INSTALL [for load]', 'sw': '9000', 'por_status': 'por_ok'},
            {'name': 'LOAD (1/2)', 'sw': '9000', 'por_status': 'por_error',
             'por_sw': '6A80', 'por_error': 'remote SW 6A80'},
        ]}
        fields = srv._ram_batch_failure_fields(result)
        self.assertEqual(fields['failed_step'], 'LOAD (1/2)')
        self.assertEqual(fields['por_sw'], '6A80')
        self.assertEqual(fields['por_error'], 'remote SW 6A80')
        # a result without steps stays quiet
        self.assertEqual(srv._ram_batch_failure_fields({})['failed_step'], '')


class BatchCardTests(unittest.TestCase):
    """The per-card decision and one run (the install chain itself is mocked)."""

    def setUp(self):
        srv._RAM_BATCH = None

    def tearDown(self):
        srv._RAM_BATCH = None

    def test_a_card_without_a_preset_is_skipped(self):
        st = srv._ram_batch_new_state(_plan())
        srv._ram_batch_handle_card(st, _fake_server('8970119000004600098', _FakeStore([])))
        self.assertEqual(st['attempts'][0]['result'], 'skipped')
        self.assertIn('no preset', st['attempts'][0]['reason'])
        self.assertEqual(st['seen']['8970119000004600098'], 'no preset')

    def test_a_card_without_a_readable_iccid_is_skipped(self):
        st = srv._ram_batch_new_state(_plan())
        srv._ram_batch_handle_card(st, _fake_server(None, _FakeStore([])))
        self.assertIn('no ICCID', st['attempts'][0]['reason'])

    def test_the_preset_is_matched_by_iccid_and_the_install_runs(self):
        st = srv._ram_batch_new_state(_plan())
        preset = _preset('8970119000004600098', 'Alfa')
        calls = []

        def fake_run(server, body, plan, scc):
            calls.append(body)
            return {'success': True, 'final_cntr': '0000000002',
                    'load_file_aid': 'F0414C46416101', 'module_aid': 'F0414C4641610101'}

        with mock.patch.object(srv, '_ram_plan_run', fake_run):
            srv._ram_batch_handle_card(st, _fake_server('8970119000004600098',
                                                        _FakeStore([preset])))
        self.assertEqual(len(calls), 1)
        self.assertEqual(calls[0]['preset_id'], 'p-8970119000004600098')
        self.assertEqual(calls[0]['tar'], '000000')       # the ISD role TAR
        rec = st['attempts'][0]
        self.assertEqual(rec['result'], 'installed')
        self.assertEqual(rec['preset'], 'Alfa')
        self.assertEqual(st['counts']['installed'], 1)

    def test_a_failed_install_is_recorded_with_the_step_and_sw(self):
        st = srv._ram_batch_new_state(_plan())
        preset = _preset('8970119000004600098')

        def fake_run(server, body, plan, scc):
            return {'success': False, 'error': 'remote SW 6A80', 'failed_step': 2,
                    'steps': [{'name': 'INSTALL [for load]', 'sw': '9000'},
                              {'name': 'LOAD (1/1)', 'sw': '9000', 'por_sw': '6A80'}],
                    'final_cntr': '0000000002'}

        with mock.patch.object(srv, '_ram_plan_run', fake_run):
            srv._ram_batch_handle_card(st, _fake_server('8970119000004600098',
                                                        _FakeStore([preset])))
        rec = st['attempts'][0]
        self.assertEqual(rec['result'], 'failed')
        self.assertEqual(rec['failed_step'], 'LOAD (1/1)')
        self.assertEqual(rec['por_sw'], '6A80')
        self.assertEqual(st['counts']['failed'], 1)

    def test_a_raised_error_is_a_skip_and_the_card_stays_seen(self):
        st = srv._ram_batch_new_state(_plan())
        preset = _preset('8970119000004600098')

        def fake_run(*a, **k):
            raise ValueError('TAR 000000 needs a default keyset (its MSL 16 '
                             'requires security)')

        with mock.patch.object(srv, '_ram_plan_run', fake_run):
            srv._ram_batch_handle_card(st, _fake_server('8970119000004600098',
                                                        _FakeStore([preset])))
        rec = st['attempts'][0]
        self.assertEqual(rec['result'], 'skipped')
        self.assertIn('needs a default keyset', rec['reason'])
        self.assertEqual(st['seen']['8970119000004600098'], 'skipped')

    def test_a_transport_error_mid_chain_is_a_failure(self):
        st = srv._ram_batch_new_state(_plan())
        preset = _preset('8970119000004600098')

        def fake_run(*a, **k):
            raise OSError('card removed')

        with mock.patch.object(srv, '_ram_plan_run', fake_run):
            srv._ram_batch_handle_card(st, _fake_server('8970119000004600098',
                                                        _FakeStore([preset])))
        rec = st['attempts'][0]
        # the chain started and died: a failure, not a skip
        self.assertEqual(rec['result'], 'failed')
        self.assertIn('card removed', rec['error'])
        self.assertEqual(st['counts']['failed'], 1)
        self.assertEqual(st['seen']['8970119000004600098'], 'failed')

    def test_a_new_install_clears_the_previous_notice(self):
        """The notice reports the last *finished* card: starting the next
        card's install must clear it (v3.28.1)."""
        st = srv._ram_batch_new_state(_plan())
        srv._ram_batch_record(st, {'iccid': '8970119000004600000', 'result': 'installed'})
        self.assertIn('installed', st['notice'])
        preset = _preset('8970119000004600098', 'Alfa')
        seen = {}

        def fake_run(server, body, plan, scc):
            seen['notice'] = st['notice']      # what the panel would render
            seen['current'] = st['current']
            return {'success': True}

        with mock.patch.object(srv, '_ram_plan_run', fake_run):
            srv._ram_batch_handle_card(st, _fake_server('8970119000004600098',
                                                        _FakeStore([preset])))
        self.assertEqual(seen['notice'], '')               # cleared while installing
        self.assertEqual(seen['current']['iccid'], '8970119000004600098')
        self.assertIn('installed', st['notice'])           # ...and set for the new outcome

    def test_a_card_already_processed_is_reported_not_reinstalled(self):
        st = srv._ram_batch_new_state(_plan())
        preset = _preset('8970119000004600098', 'Alfa')
        calls = []
        with mock.patch.object(srv, '_ram_plan_run',
                               lambda *a, **k: calls.append(1) or {'success': True}):
            srv._ram_batch_handle_card(st, _fake_server('8970119000004600098',
                                                        _FakeStore([preset]), session=1))
            srv._ram_batch_handle_card(st, _fake_server('8970119000004600098',
                                                        _FakeStore([preset]), session=2))
        self.assertEqual(len(calls), 1)                   # the second card never installs
        self.assertEqual(st['attempts'][1]['result'], 'skipped')
        self.assertIn('already processed in this batch (installed)',
                      st['attempts'][1]['reason'])
        self.assertEqual(st['attempts_total'], 2)

    def test_the_worker_waits_for_a_card_and_installs(self):
        """The loop reacts to a new card session, records the attempt and ends
        on the stop flag."""
        st = srv._ram_batch_new_state(_plan())
        preset = _preset('8970119000004600098', 'Alfa')
        srv._RAM_BATCH = st
        srv._server_ref = _fake_server('8970119000004600098', _FakeStore([preset]),
                                       session=7)
        calls = []
        with mock.patch.object(srv, '_CARD_CONNECTED', True), \
                mock.patch.object(srv, '_ram_plan_run',
                                  lambda *a, **k: calls.append(1) or {'success': True}):
            t = threading.Thread(target=srv._ram_batch_worker, args=(st,), daemon=True)
            t.start()
            for _ in range(60):
                if st['attempts']:
                    break
                time.sleep(0.05)
            self.assertTrue(srv._ram_batch_stop())
            t.join(timeout=3)
        srv._server_ref = None
        self.assertEqual(len(calls), 1)
        self.assertEqual(st['attempts'][0]['result'], 'installed')
        self.assertFalse(st['active'])
        self.assertEqual(st['phase'], 'stopped')
        # the summary /api/status carries
        self.assertEqual(srv._ram_batch_summary()['counts']['installed'], 1)
        self.assertFalse(srv._ram_batch_summary()['active'])


class RamInstallPlanTests(unittest.TestCase):
    """The shared card-independent plan (v3.28.0): the block-size/hex checks
    and the transport knobs the single install and the batch both use."""

    def test_the_plan_validates_the_input(self):
        with mock.patch.object(srv, '_cap_parse', lambda h: ('A1', 'A2', b'\x01')), \
                mock.patch.object(srv, '_cap_apdu_sequence', lambda *a, **k: ['80E60200']):
            plan = srv._ram_install_plan({'cap_hex': 'AA', 'load_block_size': '120'})
            self.assertEqual(plan['load_size'], 120)
            self.assertEqual(plan['spi2'], '01')
            self.assertEqual(plan['ram_format'], 'auto')
            self.assertEqual(plan['kind'], 'install-cap')
            self.assertEqual(plan['seq'][0]['name'], 'INSTALL [for load]')
            self.assertEqual(plan['seq'][0]['apdu'], '80E60200')
            self.assertNotIn('delete', plan['seq'][0])
            with self.assertRaises(ValueError):
                srv._ram_install_plan({'cap_hex': 'AA', 'load_block_size': 300})
            with self.assertRaises(ValueError):
                srv._ram_install_plan({'cap_hex': 'AA', 'load_block_size': 'x'})
            with self.assertRaises(ValueError):
                srv._ram_install_plan({'cap_hex': ''})

    def test_delete_package_prepends_the_cascade_delete(self):
        with mock.patch.object(srv, '_cap_parse',
                               lambda h: ('F0414C46416101', 'A2', b'\x01')), \
                mock.patch.object(srv, '_cap_apdu_sequence',
                                  lambda *a, **k: ['80E60200', '80E80000', '80E60000']):
            plan = srv._ram_install_plan({'cap_hex': 'AA', 'delete_package': True})
        self.assertEqual(len(plan['seq']), 4)
        self.assertEqual(plan['seq'][0]['name'], 'DELETE (package + applications)')
        self.assertEqual(plan['seq'][0]['apdu'], '80E40080094F07F0414C46416101')
        self.assertTrue(plan['seq'][0]['delete'])
        self.assertEqual(plan['seq'][1]['name'], 'INSTALL [for load]')
        self.assertFalse(plan['seq'][1].get('delete', False))


class RamDeletePlanTests(unittest.TestCase):
    """The batch delete plan (v3.29.0): one GP DELETE - P2=00 for an applet
    instance, P2=80 (cascade) for a package with its related objects."""

    def test_the_app_delete_is_object_only(self):
        plan = srv._ram_delete_plan({'aid': 'F0414C4641610101', 'target': 'app'})
        self.assertEqual(plan['kind'], 'delete')
        self.assertEqual(plan['seq'][0]['apdu'], '80E400000A4F08F0414C4641610101')
        self.assertTrue(plan['seq'][0]['delete'])
        self.assertIn('application', plan['seq'][0]['name'])
        self.assertEqual(plan['response'], {'aid': 'F0414C4641610101', 'target': 'app'})

    def test_the_package_delete_cascades(self):
        plan = srv._ram_delete_plan({'aid': 'F0 41 4C 46 41 61 01',
                                     'target': 'package'})
        self.assertEqual(plan['seq'][0]['apdu'], '80E40080094F07F0414C46416101')
        self.assertIn('package', plan['seq'][0]['name'])

    def test_the_target_defaults_to_the_application(self):
        plan = srv._ram_delete_plan({'aid': 'F0414C4641610101'})
        self.assertEqual(plan['seq'][0]['apdu'], '80E400000A4F08F0414C4641610101')

    def test_the_plan_validates_the_aid_and_the_target(self):
        for bad in ('', 'F041', 'F0414C464161010', 'ZZZZZZZZZZ'):
            with self.assertRaises(ValueError):
                srv._ram_delete_plan({'aid': bad})
        with self.assertRaises(ValueError):
            srv._ram_delete_plan({'aid': 'F0414C4641610101', 'target': 'applet'})


class RamReinstallPlanTests(unittest.TestCase):
    """The reinstall plan (v3.29.0): DELETE the instance, then INSTALL [for
    install] from the loaded package - no .cap, no LOAD.  The same plan serves
    the single INSTALL [for install] (kind 'install-app')."""

    def test_the_plan_deletes_the_instance_then_installs(self):
        plan = srv._ram_reinstall_plan({'loadfile_aid': 'F0414C46416101',
                                        'module_aid': 'F0414C4641610101',
                                        'privileges': '00'})
        self.assertEqual(plan['kind'], 'reinstall-app')
        self.assertEqual(plan['seq'][0]['name'], 'DELETE (application)')
        self.assertEqual(plan['seq'][0]['apdu'], '80E400000A4F08F0414C4641610101')
        self.assertTrue(plan['seq'][0]['delete'])
        self.assertEqual(plan['seq'][1]['name'], 'INSTALL [for install]')
        self.assertEqual(plan['seq'][1]['apdu'],
                         srv._cap_install_apdu('F0414C46416101', 'F0414C4641610101',
                                               instance_aid='F0414C4641610101',
                                               privileges='00'))
        self.assertEqual(srv._ram_reinstall_plan({'loadfile_aid': 'F0414C46416101',
                                                  'module_aid': 'F0414C4641610101'},
                                                 kind='install-app')['kind'],
                         'install-app')

    def test_the_plan_validates_the_aids_and_the_parameters(self):
        with self.assertRaises(ValueError):
            srv._ram_reinstall_plan({})
        with self.assertRaises(ValueError):
            srv._ram_reinstall_plan({'loadfile_aid': 'F0414C46416101'})
        with self.assertRaises(ValueError):
            srv._ram_reinstall_plan({'loadfile_aid': 'F0414C46416101',
                                     'module_aid': 'F0414C4641610101',
                                     'install_params': 'C900', 'nv_quota': 1})


class RamPlanRunTests(unittest.TestCase):
    """The runner's delete-step policy (v3.29.0): 'not found' (6A88/6A82)
    continues the chain - the delete's goal is met; any other delete failure
    stops it."""

    def _run(self, por_sw):
        sends = []

        def fake_send(server, scc, sp, state, step_name, apdu, silent=False):
            sends.append(step_name)
            if step_name.startswith('DELETE'):
                state['steps'].append({'name': step_name, 'por_status': 'por_ok',
                                       'por_sw': por_sw,
                                       'por_error': 'remote SW %s' % por_sw})
                state['failure']['error'] = '%s: remote SW %s' % (step_name, por_sw)
                return False
            state['steps'].append({'name': step_name, 'por_status': 'por_ok',
                                   'por_sw': '6101'})
            return True

        plan = {'kind': 'reinstall-app', 'ram_format': 'compact', 'spi2': '01',
                'include_cpi': True,
                'seq': [{'name': 'DELETE (application)', 'apdu': '80E400',
                         'delete': True},
                        {'name': 'INSTALL [for install]', 'apdu': '80E600'}]}
        with mock.patch.object(srv, '_op_keyset_resolve',
                               lambda server, body, tar: (3, '35', '35', None)), \
                mock.patch.object(srv, '_preset_counter_seed',
                                  lambda server, body, kvn: '0000000010'), \
                mock.patch.object(srv, '_ram_send_gp_apdu', fake_send):
            resp = srv._ram_plan_run(None, {'preset_id': 'p', 'spi1': '16',
                                            'tar': '000000'}, plan, None)
        return resp, sends

    def test_a_not_found_delete_proceeds(self):
        resp, sends = self._run('6A88')
        self.assertTrue(resp['success'])
        self.assertEqual(sends, ['DELETE (application)', 'INSTALL [for install]'])
        self.assertTrue(resp['steps'][0]['delete_not_found'])

    def test_another_delete_failure_stops_the_chain(self):
        resp, sends = self._run('6A80')
        self.assertFalse(resp['success'])
        self.assertEqual(sends, ['DELETE (application)'])
        self.assertEqual(resp['failed_step'], 1)
        self.assertIn('6A80', resp['error'])


class BatchKindTests(unittest.TestCase):
    """The batch start dispatches by kind (v3.29.0) and the status/summary
    carry it, so the panel's labels follow the operation."""

    def setUp(self):
        srv._RAM_BATCH = None

    def tearDown(self):
        srv._RAM_BATCH = None

    def _start(self, body, plan=None):
        with mock.patch.object(srv, '_ram_batch_worker', lambda state: None), \
                mock.patch.object(srv.threading, 'Thread') as th:
            th.return_value = SimpleNamespace(start=lambda: None)
            if plan is not None:
                with mock.patch.object(srv, '_ram_install_plan',
                                       lambda b: plan):
                    return srv._ram_batch_start(body)
            return srv._ram_batch_start(body)

    def test_the_delete_kind(self):
        st = self._start({'kind': 'delete', 'aid': 'F0414C4641610101'})
        self.assertEqual(st['plan']['kind'], 'delete')
        self.assertEqual(srv._ram_batch_status()['kind'], 'delete')
        self.assertEqual(srv._ram_batch_summary()['kind'], 'delete')

    def test_the_reinstall_kind(self):
        st = self._start({'kind': 'reinstall-app', 'loadfile_aid': 'F0414C46416101',
                          'module_aid': 'F0414C4641610101'})
        self.assertEqual(st['plan']['kind'], 'reinstall-app')
        self.assertEqual(len(st['plan']['seq']), 2)

    def test_the_default_kind_is_the_cap_install(self):
        st = self._start({'cap_hex': 'AA'}, plan=_plan())
        self.assertEqual(st['plan']['kind'], 'install-cap')

    def test_an_unknown_kind_is_refused(self):
        with self.assertRaises(ValueError):
            self._start({'kind': 'wipe'})

    def test_the_notice_verb_follows_the_kind(self):
        st = srv._ram_batch_new_state(_plan('delete'))
        srv._ram_batch_record(st, {'iccid': '8970119000004600098',
                                   'result': 'installed'})
        self.assertIn('deleted', st['notice'])
        st = srv._ram_batch_new_state(_plan('reinstall-app'))
        srv._ram_batch_record(st, {'iccid': '8970119000004600098',
                                   'result': 'installed'})
        self.assertIn('reinstalled', st['notice'])
        st = srv._ram_batch_new_state(_plan())
        srv._ram_batch_record(st, {'iccid': '8970119000004600098',
                                   'result': 'installed'})
        self.assertIn('installed', st['notice'])

    def test_the_steps_fields_carry_the_tail(self):
        result = {'steps': [{'name': 'INSTALL [for load]', 'sw': '9000'},
                            {'name': 'LOAD (1/1)', 'por_sw': '9000'},
                            {'name': 'INSTALL [for install]', 'por_sw': '6101'},
                            {'name': 'DELETE (application)', 'por_sw': '6A88',
                             'delete_not_found': True,
                             'por_error': 'remote SW 6A88'}]}
        fields = srv._ram_batch_steps_fields(result)
        self.assertEqual(fields['steps_total'], 4)
        self.assertEqual([x['name'] for x in fields['steps']],
                         ['LOAD (1/1)', 'INSTALL [for install]',
                          'DELETE (application)'])
        self.assertTrue(fields['steps'][-1]['not_found'])
        self.assertTrue(fields['steps'][-1]['ok'])


class BatchWorkerSourceTests(unittest.TestCase):
    def test_the_worker_waits_for_a_finished_equip(self):
        """_CARD_CONNECTED flips True (and the session bumps) before the ICCID
        is read: acting mid-equip would consume the card with a bogus 'no
        ICCID' skip.  The worker must wait for the ready marker (v3.28.0
        audit)."""
        st = srv._ram_batch_new_state(_plan())
        preset = _preset('8970119000004600098', 'Alfa')
        srv._RAM_BATCH = st
        # the session is bumped but the equip has not finished yet
        srv._server_ref = _fake_server('8970119000004600098', _FakeStore([preset]),
                                       session=9, ready=8)
        calls = []
        with mock.patch.object(srv, '_CARD_CONNECTED', True), \
                mock.patch.object(srv, '_ram_plan_run',
                                  lambda *a, **k: calls.append(1) or {'success': True}):
            t = threading.Thread(target=srv._ram_batch_worker, args=(st,), daemon=True)
            t.start()
            time.sleep(0.8)                     # a couple of polls mid-equip
            self.assertEqual(calls, [])
            self.assertEqual(st['attempts'], [])
            srv._server_ref.card_ready_session = 9   # the equip finished
            for _ in range(60):
                if st['attempts']:
                    break
                time.sleep(0.05)
            srv._ram_batch_stop()
            t.join(timeout=3)
        srv._server_ref = None
        self.assertEqual(len(calls), 1)
        self.assertEqual(st['attempts'][0]['result'], 'installed')

    def test_the_ready_marker_is_set_on_every_equip_path(self):
        """_apply_equipped_card marks the finished session, and the startup
        init (which never runs it) sets the same marker (v3.28.0 audit)."""
        import inspect
        src = inspect.getsource(srv._apply_equipped_card)
        self.assertIn('server.card_ready_session = server.card_session', src)
        main = (PROJECTS / 'simple' / 'pysim_simple_server' / '__main__.py').read_text(
            encoding='utf-8')
        self.assertIn('server.card_ready_session = server.card_session', main)

    def test_the_worker_reads_the_module_connection_state(self):
        """The server object has no `connected` attribute - the worker must
        read `_CARD_CONNECTED`, the global the rest of the server uses
        (v3.28.0 review)."""
        import inspect
        src = inspect.getsource(srv._ram_batch_worker)
        self.assertIn('_CARD_CONNECTED', src)
        self.assertNotIn("'connected'", src)


class BatchLifecycleTests(unittest.TestCase):
    def setUp(self):
        srv._RAM_BATCH = None

    def tearDown(self):
        srv._RAM_BATCH = None

    def test_start_refuses_a_second_batch_and_a_test_run(self):
        with mock.patch.object(srv, '_ram_install_plan', lambda body: _plan()), \
                mock.patch.object(srv, '_ram_batch_worker', lambda state: None):
            st = srv._ram_batch_start({'cap_hex': 'AA'})
            self.assertTrue(st['active'])
            with self.assertRaises(RuntimeError):
                srv._ram_batch_start({'cap_hex': 'AA'})
            srv._ram_batch_stop()
        # a test run blocks the start
        srv._RAM_BATCH = None
        with mock.patch.object(srv, '_ram_install_plan', lambda body: _plan()), \
                mock.patch.object(srv, '_TEST_RUNNING', True):
            with self.assertRaises(RuntimeError):
                srv._ram_batch_start({'cap_hex': 'AA'})

    def test_the_status_payload_shape(self):
        idle = srv._ram_batch_status()
        self.assertFalse(idle['active'])
        self.assertEqual(idle['phase'], 'idle')
        self.assertEqual(idle['attempts'], [])
        self.assertIsNone(srv._ram_batch_summary())
        with mock.patch.object(srv, '_ram_install_plan', lambda body: _plan()), \
                mock.patch.object(srv, '_ram_batch_worker', lambda state: None):
            srv._ram_batch_start({'cap_hex': 'AA'})
        srv._ram_batch_record(srv._RAM_BATCH, {'iccid': '8970119000004600098',
                                               'result': 'installed'})
        st = srv._ram_batch_status()
        self.assertTrue(st['active'])
        self.assertEqual(st['phase'], 'waiting_card')
        self.assertEqual(len(st['attempts']), 1)
        self.assertEqual(st['counts']['installed'], 1)
        self.assertEqual(srv._ram_batch_summary()['counts']['installed'], 1)
        # a stop request flips the phase; clear is refused while active
        self.assertTrue(srv._ram_batch_stop())
        self.assertEqual(srv._ram_batch_status()['phase'], 'stopping')
        self.assertFalse(srv._ram_batch_clear())
        srv._RAM_BATCH['active'] = False
        self.assertTrue(srv._ram_batch_clear())
        self.assertIsNone(srv._RAM_BATCH)


class BatchHttpTests(unittest.TestCase):
    """The endpoints: start needs no card, status is card-free (the panel polls
    it during an install), stop answers while the card lock is held."""

    def setUp(self):
        srv._RAM_BATCH = None
        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.card_presets = 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()
        srv._RAM_BATCH = None

    def _post(self, path, body, timeout=5):
        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=timeout) 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 _get(self, path, timeout=5):
        try:
            with urllib.request.urlopen(
                    'http://127.0.0.1:%d%s' % (self.port, path), timeout=timeout) 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_the_batch_endpoints(self):
        with mock.patch.object(srv, '_ram_install_plan', lambda body: _plan()), \
                mock.patch.object(srv, '_ram_batch_worker', lambda state: None):
            status, resp = self._post('/api/ram-batch/start', {'cap_hex': 'AA'})
            self.assertEqual(status, 200, resp)
            self.assertTrue(resp['active'])
            self.assertEqual(resp['phase'], 'waiting_card')
            # a second start is refused
            status, resp = self._post('/api/ram-batch/start', {'cap_hex': 'AA'})
            self.assertEqual(status, 409, resp)
            # the status is card-free
            status, resp = self._get('/api/ram-batch/status')
            self.assertEqual(status, 200, resp)
            self.assertTrue(resp['active'])
            # clear is refused while active
            status, resp = self._post('/api/ram-batch/clear', {})
            self.assertEqual(status, 409, resp)
            # the stop answers and flips the phase
            status, resp = self._post('/api/ram-batch/stop', {})
            self.assertEqual(status, 200, resp)
            self.assertEqual(resp['phase'], 'stopping')
            # a repeated stop is idempotent (the panel disables the button)
            status, resp = self._post('/api/ram-batch/stop', {})
            self.assertEqual(status, 200, resp)
        srv._RAM_BATCH['active'] = False
        status, resp = self._post('/api/ram-batch/clear', {})
        self.assertEqual(status, 200, resp)
        # with no run left, a stop is a 409
        status, resp = self._post('/api/ram-batch/stop', {})
        self.assertEqual(status, 409, resp)
        status, resp = self._get('/api/ram-batch/status')
        self.assertEqual(status, 200, resp)
        self.assertFalse(resp['active'])
        self.assertEqual(resp['phase'], 'idle')

    def test_a_bad_start_is_a_400_and_a_test_run_blocks_it(self):
        def boom(body):
            raise ValueError('No cap_hex provided')

        with mock.patch.object(srv, '_ram_install_plan', boom):
            status, resp = self._post('/api/ram-batch/start', {})
            self.assertEqual(status, 400, resp)
            self.assertIn('cap_hex', resp['error'])
        # a malformed .cap (a zip parse error) is a 400 too, never a dropped
        # connection
        status, resp = self._post('/api/ram-batch/start', {'cap_hex': '00'})
        self.assertEqual(status, 400, resp)
        self.assertIn('zip', resp['error'].lower())
        with mock.patch.object(srv, '_ram_install_plan', lambda body: _plan()), \
                mock.patch.object(srv, '_TEST_RUNNING', True):
            status, resp = self._post('/api/ram-batch/start', {'cap_hex': 'AA'})
            self.assertEqual(status, 409, resp)

    def test_a_test_run_is_refused_while_a_batch_runs(self):
        with mock.patch.object(srv, '_ram_install_plan', lambda body: _plan()), \
                mock.patch.object(srv, '_ram_batch_worker', lambda state: None):
            status, resp = self._post('/api/ram-batch/start', {'cap_hex': 'AA'})
            self.assertEqual(status, 200, resp)
        status, resp = self._post('/api/test/run', {})
        self.assertEqual(status, 409, resp)
        self.assertIn('batch install', resp['error'])

    def test_stop_answers_while_the_card_lock_is_held(self):
        """The panel's Stop must be effective during an install (the worker
        holds _CARD_LOCK for the whole chain)."""
        with mock.patch.object(srv, '_ram_install_plan', lambda body: _plan()), \
                mock.patch.object(srv, '_ram_batch_worker', lambda state: None):
            status, resp = self._post('/api/ram-batch/start', {'cap_hex': 'AA'})
            self.assertEqual(status, 200, resp)
        with _CARD_LOCK:
            status, resp = self._post('/api/ram-batch/stop', {}, timeout=2)
            self.assertEqual(status, 200, resp)
            self.assertEqual(resp['phase'], 'stopping')
            status, resp = self._get('/api/ram-batch/status', timeout=2)
            self.assertEqual(status, 200, resp)


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