catarrh
e8190c65f7
- A non-ValueError exception from the install run (a card removed mid-chain, a transport error) was recorded as a 'skipped' attempt with the error text - but the chain had started and died, which is a failure. Only a ValueError (a request/eligibility error: the preset's ISD keyset, the plan) is a skip now; anything else is a 'failed' record carrying its error. The PWA's failed row falls back to that error when there is no failing step / status word to show. - The panel's Clear button is disabled while a batch is active (the server refuses the clear with 409 anyway) - no pointless error alert. - api.md and the repo AGENTS document the two outcomes. Tests: a transport error mid-chain is a failure (and the card stays seen), the plan's input validation (the block size, a missing cap_hex), the failed row without a step, the shared ram_format field, and the Clear state. 915 frontend + 921 Python; sw cache simple-v438 (unchanged).
431 lines
20 KiB
Python
431 lines
20 KiB
Python
#!/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():
|
|
return {'seq': ['80E60200', '80E80000', '80E60000'], 'loadfile_aid': 'F0414C46416101',
|
|
'module_aid': 'F0414C4641610101', 'block_size': 240, 'block_size_req': None,
|
|
'spi2': '01', 'include_cpi': True, 'ram_format': 'compact'}
|
|
|
|
|
|
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, connected=True, session=1):
|
|
return SimpleNamespace(iccid=iccid, card_presets=store, connected=connected,
|
|
card_session=session, 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_install_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_install_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_install_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_install_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_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_install_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]),
|
|
connected=True, session=7)
|
|
calls = []
|
|
with mock.patch.object(srv, '_ram_install_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['block_size'], 120)
|
|
self.assertEqual(plan['spi2'], '01')
|
|
self.assertEqual(plan['ram_format'], 'auto')
|
|
self.assertEqual(plan['seq'], ['80E60200'])
|
|
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': ''})
|
|
|
|
|
|
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(None, {'cap_hex': 'AA'})
|
|
self.assertTrue(st['active'])
|
|
with self.assertRaises(RuntimeError):
|
|
srv._ram_batch_start(None, {'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(None, {'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(None, {'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()
|