catarrh
0231df905c
pySim's scc.read_binary/read_record call select_path(ef) first, so the ICCID, network-state, EF.ARR and EF.PBR readers - which had already made a parent-scoped selection and hold its FCP - sent a second SELECT per file and per record. _read_selected_ef() issues READ BINARY (chunked at max_cmd_len from the FCP file size) / READ RECORD (b2<rec>04<len> at the FCP record length) directly; the pySim helpers stay as the fallback when the FCP lacks the value. The startup TERMINAL PROFILE drain now breaks on the chain's idle SW instead of polling STATUS once more. Measured on the last startup: ~3.4 s traced init (netstate/ARR/PBR 271 APDUs / 1.26 s incl. 169 SELECTs; the card's own TP processing 1.58 s); the change removes ~0.7-0.8 s of the read phase without touching any data or the TP. v3.38.1; tests: tests/test_read_selected.py (direct reads, chunking, the fallback and the drain source pin) and the ARR/PBR/ICCID/netstate fakes. 1067 Python + 1075 frontend. sw cache simple-v482.
224 lines
8.1 KiB
Python
224 lines
8.1 KiB
Python
# coding=utf-8
|
|
"""EF.ARR session cache (v3.24.0): the equip-time read, the FCP '8B' backing
|
|
store, the /api/arr route and the write refresh."""
|
|
|
|
import json
|
|
import pathlib
|
|
import threading
|
|
import unittest
|
|
import urllib.request
|
|
from unittest import mock
|
|
|
|
from pysim_simple_server.__main__ import _build_http_server
|
|
from pysim_simple_server.server import (PysimHandler, _CARD_FREE_GET, _arr_fid,
|
|
_arr_refresh_after_write, _read_arr,
|
|
_refresh_arr_cache)
|
|
|
|
ROOT = pathlib.Path(__file__).resolve().parents[1]
|
|
SRC = (ROOT / 'pysim_simple_server' / 'server.py').read_text(encoding='utf-8')
|
|
|
|
|
|
class FakeFile:
|
|
def __init__(self, fid):
|
|
self.fid = fid
|
|
|
|
|
|
class FakeScc:
|
|
"""Serves READ RECORD for the selected fake ARR file."""
|
|
|
|
cla_byte = '00'
|
|
max_cmd_len = 255
|
|
|
|
def __init__(self, lchan):
|
|
self.lchan = lchan
|
|
self.sent = []
|
|
|
|
def send_apdu_checksw(self, pdu):
|
|
self.sent.append(pdu)
|
|
return (self.lchan.files[self.lchan._cur][int(pdu[4:6], 16) - 1], '9000')
|
|
|
|
|
|
class FakeLchan:
|
|
"""One selected ARR file: READ RECORD serves its records."""
|
|
|
|
def __init__(self, files):
|
|
self.files = files # path -> [record hex]
|
|
self._cur = None
|
|
self.selected_file = None
|
|
self.scc = FakeScc(self)
|
|
|
|
def selected_file_num_of_rec(self):
|
|
return len(self.files.get(self._cur) or [])
|
|
|
|
def selected_file_record_len(self):
|
|
return 6
|
|
|
|
def read_record(self, n):
|
|
return (self.files[self._cur][n - 1], '9000')
|
|
|
|
|
|
def _select(lchan, path, app):
|
|
if path not in lchan.files:
|
|
raise RuntimeError('file not found')
|
|
lchan._cur = path
|
|
lchan.selected_file = FakeFile(path.rsplit('/', 1)[-1])
|
|
return lchan.selected_file, None
|
|
|
|
|
|
def _app(files):
|
|
lchan = FakeLchan(files)
|
|
app = mock.Mock()
|
|
app.rs.lchan = [lchan]
|
|
return app, lchan
|
|
|
|
|
|
def _read(app, file_type='linear_fixed'):
|
|
with mock.patch('pysim_simple_server.server._select_path', _select), \
|
|
mock.patch('pysim_simple_server.server._get_file_type', return_value=file_type):
|
|
return _read_arr(app)
|
|
|
|
|
|
class ReadArrTests(unittest.TestCase):
|
|
def test_reads_both_copies(self):
|
|
app, _ = _app({'MF/2F06': ['800101A40683010A950108', '9000'],
|
|
'ADF.USIM/6F06': ['8001019000']})
|
|
arrs = _read(app)
|
|
self.assertEqual([a['fid'] for a in arrs], ['2F06', '6F06'])
|
|
self.assertEqual(arrs[0]['path'], 'MF/2F06')
|
|
self.assertEqual(arrs[0]['records'], ['800101A40683010A950108', '9000'])
|
|
self.assertEqual(arrs[0]['num_records'], 2)
|
|
self.assertEqual(arrs[0]['record_len'], 6)
|
|
self.assertEqual(arrs[1]['path'], 'ADF.USIM/6F06')
|
|
|
|
def test_reads_the_df_tree_copies(self):
|
|
app, _ = _app({'MF/2F06': ['9000'], 'MF/7F10/6F06': ['9000'],
|
|
'MF/7F20/6F06': ['9000'], 'ADF.USIM/6F06': ['9000']})
|
|
arrs = _read(app)
|
|
self.assertEqual([a['path'] for a in arrs],
|
|
['MF/2F06', 'MF/7F10/6F06', 'MF/7F20/6F06', 'ADF.USIM/6F06'])
|
|
|
|
def test_a_missing_copy_is_skipped(self):
|
|
app, _ = _app({'ADF.USIM/6F06': ['9000']})
|
|
arrs = _read(app)
|
|
self.assertEqual([a['fid'] for a in arrs], ['6F06'])
|
|
|
|
def test_no_arr_returns_none(self):
|
|
app, _ = _app({})
|
|
self.assertIsNone(_read(app))
|
|
|
|
def test_a_transparent_file_is_skipped(self):
|
|
app, _ = _app({'MF/2F06': ['9000']})
|
|
self.assertIsNone(_read(app, file_type='transparent'))
|
|
|
|
def test_a_failing_read_skips_the_file(self):
|
|
app, lchan = _app({'MF/2F06': ['9000'], 'ADF.USIM/6F06': ['9000']})
|
|
orig = lchan.scc.send_apdu_checksw
|
|
|
|
def flaky(pdu):
|
|
if lchan._cur == 'MF/2F06':
|
|
raise RuntimeError('6A82')
|
|
return orig(pdu)
|
|
lchan.scc.send_apdu_checksw = flaky
|
|
arrs = _read(app)
|
|
self.assertEqual([a['fid'] for a in arrs], ['6F06'])
|
|
|
|
def test_the_record_count_is_capped(self):
|
|
app, _ = _app({'MF/2F06': ['9000'] * 200})
|
|
arrs = _read(app)
|
|
self.assertEqual(arrs[0]['num_records'], 64)
|
|
|
|
def test_no_app_returns_none(self):
|
|
self.assertIsNone(_read_arr(None))
|
|
self.assertIsNone(_read_arr(mock.Mock(rs=None)))
|
|
|
|
|
|
class CacheTests(unittest.TestCase):
|
|
def test_arr_fid(self):
|
|
self.assertTrue(_arr_fid('2F06'))
|
|
self.assertTrue(_arr_fid('6f06'))
|
|
self.assertFalse(_arr_fid('6F07'))
|
|
self.assertFalse(_arr_fid(None))
|
|
|
|
def test_refresh_stores_and_clears(self):
|
|
s = mock.Mock()
|
|
with mock.patch('pysim_simple_server.server._read_arr',
|
|
return_value=[{'fid': '2F06', 'num_records': 2}]), \
|
|
mock.patch('pysim_simple_server.server._tlog'):
|
|
_refresh_arr_cache(s)
|
|
self.assertEqual(s.arr, [{'fid': '2F06', 'num_records': 2}])
|
|
with mock.patch('pysim_simple_server.server._read_arr', return_value=None):
|
|
_refresh_arr_cache(s)
|
|
self.assertIsNone(s.arr)
|
|
|
|
def test_write_refresh_only_for_arr_fids(self):
|
|
s = mock.Mock()
|
|
with mock.patch('pysim_simple_server.server._read_arr',
|
|
return_value=[{'fid': '2F06', 'num_records': 2}]) as m:
|
|
self.assertFalse(_arr_refresh_after_write(s, None, '6F07'))
|
|
m.assert_not_called()
|
|
self.assertTrue(_arr_refresh_after_write(s, None, '2f06'))
|
|
m.assert_called_once()
|
|
self.assertEqual(s.arr, [{'fid': '2F06', 'num_records': 2}])
|
|
|
|
def test_equip_and_disconnect_wiring(self):
|
|
# the equip hook reads the cache; a disconnect clears it; an EF.ARR
|
|
# write refreshes it
|
|
seg = SRC[SRC.index('def _apply_equipped_card'):SRC.index('def _card_reset_reinit')]
|
|
self.assertIn('_refresh_arr_cache(server)', seg)
|
|
seg = SRC[SRC.index('def _handle_card_disconnect'):SRC.index('def _ensure_transport')]
|
|
self.assertIn('_server_ref.arr = None', seg)
|
|
seg = SRC[SRC.index("elif self.path == '/api/write':"):SRC.index("elif self.path == '/api/tree':")]
|
|
self.assertIn('_arr_refresh_after_write', seg)
|
|
|
|
def test_the_startup_init_reads_the_arr(self):
|
|
# a card present at server start never runs _apply_equipped_card, so
|
|
# the startup init reads the ARR in its own CAT-free window (v3.24.0
|
|
# review - the cache was empty on a startup-equipped card)
|
|
src = (ROOT / 'pysim_simple_server' / '__main__.py').read_text(encoding='utf-8')
|
|
self.assertIn('arr_files = _read_arr(app)', src)
|
|
self.assertIn('server.arr = arr_files', src)
|
|
imp = src[src.index('from .server import'):]
|
|
self.assertIn('_read_arr', imp[:imp.index('\n')])
|
|
|
|
|
|
class ArrRouteTests(unittest.TestCase):
|
|
"""GET /api/arr serves the session cache without touching the card."""
|
|
|
|
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 _get(self, path):
|
|
with urllib.request.urlopen('http://127.0.0.1:%d%s' % (self.port, path),
|
|
timeout=5) as res:
|
|
return res.status, json.loads(res.read() or b'{}')
|
|
|
|
def test_arr_is_a_card_free_get(self):
|
|
self.assertIn('/api/arr', _CARD_FREE_GET)
|
|
|
|
def test_empty_cache_answers_not_found(self):
|
|
status, resp = self._get('/api/arr')
|
|
self.assertEqual(status, 200, resp)
|
|
self.assertFalse(resp['ok'])
|
|
self.assertEqual(resp['arrs'], [])
|
|
|
|
def test_the_cache_is_served_with_the_session(self):
|
|
self.server.arr = [{'fid': '2F06', 'path': 'MF/2F06', 'records': ['9000']}]
|
|
self.server.card_session = 7
|
|
status, resp = self._get('/api/arr')
|
|
self.assertEqual(status, 200, resp)
|
|
self.assertTrue(resp['ok'])
|
|
self.assertEqual(resp['session'], 7)
|
|
self.assertEqual(resp['arrs'][0]['fid'], '2F06')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|