v3.25.1: the run-ownership fix for the proactive auto-handling

A card that re-queues a command immediately after each accepted TR
(live 2026-10-11: the EP BAP card re-issued the same PROVIDE LOCAL
INFORMATION 1100 times) turned the background poll's FETCH/TR chain
into an unbounded loop that held _CARD_LOCK forever - every card
operation, a test run's start included, blocked behind it.

The chain now yields:
- _handle_proactive_chain aborts before the next FETCH while a test run
  owns the card (_TEST_RUNNING) - the pending command stays for the
  script - and takes max_rounds; the background poll passes
  _POLL_CHAIN_MAX = 10, so a chatty card cannot hold the lock.
- _do_status_poll re-arms on its run skip (a heartbeat that never
  fetches); _timer_expired and the BIP Data-available event wait during
  a run; _bip_flush_channel_events stays quiet too.
- _sim_resume_poll() resumes polling at every run end - the two worker
  bodies, their exception paths and the spawn failure.

Also the TR command-details qualifier echo (TS 102 223 6.8.1: the
object shall be identical to the command's): _build_tr,
testscript.build_tr and the interactive _menu_send_response echoed a
hardcoded 00 - harmless for the loop's qual-00 PLI, wrong for any
other qualifier.  The menu pendings now carry cmd_qual (parsed in the
on_fetch handler).

Tests: the chain bound/abort (a fake card re-queueing a command), the
poll/timer guards, the resume wiring, the qualifier echo in both
builders.  883 frontend + 865 Python; sw cache simple-v425.
This commit is contained in:
2026-10-11 12:23:12 +03:00
parent 87931afc41
commit af00932aca
10 changed files with 4743 additions and 32 deletions
+1 -1
View File
@@ -565,7 +565,7 @@
<section class="mb-10">
<h3 id="test-script" class="text-lg font-medium mb-2">8.9 Тест-скрипт</h3>
<p class="text-sm mb-2"><strong>Тест-скрипт</strong> &mdash; упорядоченный список <strong>действий</strong> и <strong>ожиданий</strong> проактивных команд, выполняемый на подключённой карте на стороне сервера. Пока запуск активен, карта принадлежит скрипту: остальные операции с картой отклоняются (HTTP&nbsp;409), фоновый опрос STATUS приостанавливается; отчёт о запуске хранится в памяти (в том числе после перезагрузки страницы) до нажатия <strong>Очистить отчёт</strong>. Скрипты хранятся в <strong>хранилище на сервере</strong> (<code class="font-mono text-sm">~/.pysim-simple-server/test_scripts.json</code>, <code class="font-mono text-sm">--test-scripts</code>), поэтому все браузеры работают с одним набором; изменения записываются в хранилище по мере правки (кнопки &laquo;Сохранить&raquo; нет &mdash; черновик без валидного шага остаётся локальным); экспорт/импорт JSON остаётся для резервных копий и переноса скриптов между стендами (старый экспорт из localStorage импортируется как есть).</p>
<p class="text-sm mb-2"><strong>Тест-скрипт</strong> &mdash; упорядоченный список <strong>действий</strong> и <strong>ожиданий</strong> проактивных команд, выполняемый на подключённой карте на стороне сервера. Пока запуск активен, карта принадлежит скрипту: остальные операции с картой отклоняются (HTTP&nbsp;409), фоновый опрос STATUS и автоматическая обработка проактивных команд приостанавливаются &mdash; объявленная картой команда остаётся ожидающей для скрипта, и оба механизма возобновляются по завершении запуска; отчёт о запуске хранится в памяти (в том числе после перезагрузки страницы) до нажатия <strong>Очистить отчёт</strong>. Скрипты хранятся в <strong>хранилище на сервере</strong> (<code class="font-mono text-sm">~/.pysim-simple-server/test_scripts.json</code>, <code class="font-mono text-sm">--test-scripts</code>), поэтому все браузеры работают с одним набором; изменения записываются в хранилище по мере правки (кнопки &laquo;Сохранить&raquo; нет &mdash; черновик без валидного шага остаётся локальным); экспорт/импорт JSON остаётся для резервных копий и переноса скриптов между стендами (старый экспорт из localStorage импортируется как есть).</p>
<p class="text-sm mb-2">У вкладки <strong>четыре представления</strong>: <strong>список наборов</strong> (по умолчанию; <em>Добавить набор</em> / <em>Редактировать</em> / <em>Запустить</em> / <em>Экспорт</em> / <em>Удалить</em>), <strong>редактор набора</strong> (скрипты набора в порядке выполнения &mdash; необязательный <em>setup</em> первым и <em>teardown</em> последним, с пометками ролей; у каждого скрипта <em>Запустить</em>/<em>Редактировать</em>/<em>Переместить</em>/<em>Скопировать</em>/<em>Удалить</em>, политика <em>при ошибке</em> <em>остановить</em> (по умолчанию) / <em>продолжить</em> и <em>Требуется ADM</em>), <strong>редактор скрипта</strong> (список шагов; кнопка <em>Старт</em> выполняет только этот скрипт и показывает результат под шагами &mdash; он не сохраняется) и <strong>отчёт набора</strong> (после запуска набора: статус каждого скрипта, таблица шагов с ожидаемым/фактическим, примечание об ошибке и выдержка из журнала запуска; сводка с успешно/предупреждения/ошибки/пропущено, время выполнения, счётчики SCP80 до&nbsp;&rarr;&nbsp;после по каждому keyset, пресет и предупреждение о смене сессии карты; экспорт в <em>JSON</em> или <em>Markdown</em>). <strong>Запуск набора</strong> выполняет его скрипты по порядку в <strong>одной сессии карты, без сбросов и без продолжения с места остановки</strong>; <em>при ошибке</em> решает, остановится ли набор на этом скрипте или запустится следующий, <strong>teardown выполняется всегда</strong> при остановке или ошибке (пока карта в ридере), чтобы следующий тест начался с чистого состояния, а сброс карты останавливает набор и отмечается в отчёте. <strong>Начать со сброса карты</strong> сбрасывает карту и выполняет полную переинициализацию перед запуском набора (скрипты затем работают со свежеинициализированной картой); пресет должен по-прежнему совпадать с ICCID карты, а требуемый ADM проверяется заново для новой сессии карты. Инициализация поглощает начальные proactive-команды карты, поэтому скрипт не поймает первую команду после сброса (они остаются в журнале proactive). <strong>Требуется ADM</strong> (у набора или скрипта) проверяет ключ ADM из подобранного пресета <strong>один раз</strong> перед запуском &mdash; ошибка отклоняет запуск (проверьте вручную через значок ADM; автоматических повторов нет). Скрипты хранятся в хранилище наборов (<code class="font-mono text-sm">~/.pysim-simple-server/test_suites.json</code>, <code class="font-mono text-sm">--test-suites</code>); каждый скрипт принадлежит ровно одному набору, и скрипт можно переместить или скопировать в другой набор (экспорт набора &mdash; это пакет с набором и его скриптами; старый экспорт только скриптов импортируется в автоматически созданный набор &laquo;Imported scripts&raquo;).</p>
<ul class="list-disc list-inside text-sm space-y-1 mb-3">
<li><strong>Действия</strong> &mdash; <em>Envelope</em> (Event Download: сырой тип события + данные, с опциональным переопределением источника &mdash; 82 терминал / 83 сеть) или <em>Event</em> (семантические формы: <strong>та же форма, что и в диалоге отправки на вкладке «Телефон»</strong>, для всех смоделированных событий &mdash; четыре события, которые собирает сервер (Location status, Access technology change, Network rejection, Data connection status change), передаются полями, которые он проверяет и собирает во время запуска, остальные несут собранный формой hex данных события (показывается под формой, с проверкой одного ENVELOPE); источник device identities берётся из события (например, MT call &rarr; сеть) или из его поля источника), <em>Menu Selection</em> (item id 1&ndash;255 или текст пункта с точным/частичным совпадением, по умолчанию без учёта регистра &mdash; выбор <em>«Учёт регистра»</em> в редакторе шагов включает точное совпадение; текст подбирается по меню карты, которое следует за параметрами установки апплета), <em>UPDATE/READ файла</em> (путь + режим; чтение проверяет данные файла), <em>сырой APDU</em> (без авто-обработчика), <em>SCP80</em> (C-APDU или готовый защищённый пакет, опциональные переопределения номера keyset (kvn) / TAR / SPI1 / SPI2 и формат команды &mdash; компактный или обёрнутый в шаблон Command Scripting TS 102 226 &sect;5.2.1 <em>expanded</em> (AA) / <em>expanded AE&nbsp;80</em>; ключи и счётчик берутся из пресета карты, подобранного по ICCID подключённой карты &mdash; запуск называет пресет в хранилище, поэтому всегда используются его keysets, TAR и счётчик &mdash; а сервер сохраняет потреблённый счётчик в пресет по ходу запуска), <em>STATUS</em> (при <em>попытках</em> &gt; 1 опрашивает, пока карта не объявит команду) и <em>Очистка очереди проактивных команд</em> (опрос STATUS и поглощение всех объявленных команд &mdash; результат TERMINAL RESPONSE для каждой, опционально другой результат для первой; граница попыток/интервала; опциональная проверка <em>require</em> по поглощённому. Пустой опрос проходит, финальный STATUS должен быть idle, а оставшаяся команда передаётся следующему шагу &mdash; используйте в начале, середине или конце скрипта, чтобы убедиться, что карта перестала слать команды; она же поглощает команду, оставшуюся от предыдущего шага (ответившего <code class="font-mono text-sm">91XX</code>)). У каждого действия есть проверка ответа: ожидаемые SW и данные (точно или маска с подстановкой <code class="font-mono text-sm">?</code> &mdash; <code class="font-mono text-sm">91??</code>) и проверка PoR для SCP80 (<em>любой / без PoR / por_ok</em> или своя проверка декодированного PoR &mdash; статус / SW / данные; транспорт через SEND SHORT MESSAGE работает так же).</li>
+1 -1
View File
@@ -564,7 +564,7 @@
<section class="mb-10">
<h3 id="test-script" class="text-lg font-medium mb-2">8.9 Test script</h3>
<p class="text-sm mb-2">A <strong>test script</strong> is an ordered list of <strong>actions</strong> and proactive-command <strong>expectations</strong> executed server-side on the equipped card. While a run is active the card belongs to the script: other card actions are refused (HTTP&nbsp;409) and background STATUS polling is paused; the run report is kept in memory (also after reloading the page) until <strong>Clear report</strong>. Scripts live in the <strong>server-side store</strong> (<code class="font-mono text-sm">~/.pysim-simple-server/test_scripts.json</code>, <code class="font-mono text-sm">--test-scripts</code>), so every browser manages the same set; edits are written to the store as you change them (there is no Save button &mdash; a draft without a valid step stays local until it is valid); JSON export/import remains for backup and for moving scripts between benches (the old localStorage export imports as-is).</p>
<p class="text-sm mb-2">A <strong>test script</strong> is an ordered list of <strong>actions</strong> and proactive-command <strong>expectations</strong> executed server-side on the equipped card. While a run is active the card belongs to the script: other card actions are refused (HTTP&nbsp;409), and background STATUS polling and the automatic proactive handling are suspended &mdash; a command the card announces stays pending for the script, and both resume when the run ends; the run report is kept in memory (also after reloading the page) until <strong>Clear report</strong>. Scripts live in the <strong>server-side store</strong> (<code class="font-mono text-sm">~/.pysim-simple-server/test_scripts.json</code>, <code class="font-mono text-sm">--test-scripts</code>), so every browser manages the same set; edits are written to the store as you change them (there is no Save button &mdash; a draft without a valid step stays local until it is valid); JSON export/import remains for backup and for moving scripts between benches (the old localStorage export imports as-is).</p>
<p class="text-sm mb-2">The pill has <strong>four views</strong>: the <strong>suite list</strong> (default; <em>Add suite</em> / <em>Edit</em> / <em>Run</em> / <em>Export</em> / <em>Delete</em>), the <strong>suite editor</strong> (the suite's scripts in run order &mdash; the optional <em>setup</em> first and the <em>teardown</em> last, marked with role badges; per script <em>Run</em>/<em>Edit</em>/<em>Move</em>/<em>Copy</em>/<em>Delete</em>, the per-member <em>on failure</em> policy <em>stop</em> (default) / <em>continue</em>, and <em>ADM required</em>), the <strong>script editor</strong> (the step list; its <em>Start</em> runs that single script and shows the result under the steps &mdash; not preserved) and the <strong>suite report</strong> (after a suite run: every script's status, its step table with expected/actual, the failing note and a run-log excerpt; the summary with passed/warnings/failed/skipped, wall time, the SCP80 counters before&nbsp;&rarr;&nbsp;after per keyset, the preset and a card-session-changed warning; export as <em>JSON</em> or <em>Markdown</em>). A <strong>suite run</strong> executes its scripts in order in <strong>one card session, without resets and without resume</strong>; <em>on failure</em> decides whether the suite stops at that script or runs the next one, the <strong>teardown always runs</strong> on a stop or a failure (while the card is in the reader) so the next test starts clean, and a card reset stops the suite and is flagged in the report. <strong>Start with card reset</strong> resets the card and runs the full re-initialization before the suite starts (the scripts then run against a freshly initialized card); the preset must still match the card's ICCID afterwards and a required ADM is verified again for the new card session. The initialization consumes the card's initial proactive commands, so a script cannot catch the first command after the reset (they stay in the proactive log). <strong>ADM required</strong> (on the suite or a script) verifies the matched preset's ADM key <strong>once</strong> before the start &mdash; a failure refuses the run (verify manually with the ADM badge; nothing is retried automatically). Scripts live in the suite's store (<code class="font-mono text-sm">~/.pysim-simple-server/test_suites.json</code>, <code class="font-mono text-sm">--test-suites</code>); every script belongs to exactly one suite, and a script can be moved or copied to another suite (a bundle export carries the suite plus its scripts; the old script-only export imports into an auto-created &ldquo;Imported scripts&rdquo; suite).</p>
<ul class="list-disc list-inside text-sm space-y-1 mb-3">
<li><strong>Actions</strong> &mdash; <em>Envelope</em> (Event Download: raw event type + data, with an optional source override &mdash; 82 terminal / 83 network) or <em>Event</em> (the semantic forms: the <strong>same form the Phone tab's send dialog uses</strong>, for every modelled event &mdash; the four events the server builds (Location status, Access technology change, Network rejection, Data connection status change) travel as fields it validates and builds at run time, every other event carries the form's built event-data hex (shown live under the form, with the one-ENVELOPE check); the device-identities source comes from the event (e.g. MT call &rarr; network) or its source field), <em>Menu Selection</em> (item id 1&ndash;255, or the item's text with an exact/contains match, case-insensitive by default &mdash; the step editor's <em>Match case</em> choice switches to an exact-case match; the text is resolved against the card's menu, which follows the applet's install parameters), <em>UPDATE/READ file</em> (path + mode; reads verify the file data), <em>raw APDU</em> (no auto-handler), <em>SCP80</em> (C-APDU or a pre-built secured packet, optional keyset number (kvn) / TAR / SPI1 / SPI2 overrides and a command format &mdash; compact, or the TS 102 226 &sect;5.2.1 Command Scripting template <em>expanded</em> (AA) / <em>expanded AE&nbsp;80</em> around the C-APDU; the keys and the counter come from the card preset matched by the equipped card's ICCID &mdash; the run names the stored preset, so it always uses its keysets, TARs and counter &mdash; the server persists the consumed counter into it as the run proceeds), <em>STATUS</em> (with <em>attempts</em> &gt; 1 it polls until the card announces a command) and <em>Proactive drain</em> (poll STATUS and consume every announced command &mdash; the TERMINAL RESPONSE result for each, optionally a different result for the first; the attempt/interval bound; an optional <em>require</em> assertion over what was drained. An empty poll passes, the final STATUS must be idle, and a still-pending command is handed to the next step &mdash; use it at the start, in the middle or at the end of a script to make sure the card has stopped sending; it also consumes a command a previous step left pending (a step that answered <code class="font-mono text-sm">91XX</code>). Every action has a response check: expected SW and data (exact, or a mask with <code class="font-mono text-sm">?</code> nibble wildcards &mdash; <code class="font-mono text-sm">91??</code>) plus a PoR check for SCP80 (<em>any / no PoR / por_ok</em>, or a custom assertion of the decoded PoR &mdash; status / SW / data; the SEND SHORT MESSAGE transport works the same).</li>
+1 -1
View File
@@ -1905,7 +1905,7 @@
// ===== Version =====
// Single source of truth for the PWA version: shown in the header and used
// by the server version check in pysimConnect().
const SIMPLE_VERSION = '3.25.0';
const SIMPLE_VERSION = '3.25.1';
document.getElementById('app-version').textContent = 'v' + SIMPLE_VERSION;
// ===== view history (the browser Back/Forward buttons) =====
+1 -1
View File
@@ -1,4 +1,4 @@
const CACHE = 'simple-v424';
const CACHE = 'simple-v425';
const URLS = [
'index.html',
'help.html',
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "pysim-simple-server"
version = "3.25.0"
version = "3.25.1"
description = "HTTP REST server wrapping pysim for the SIMple PWA"
requires-python = ">=3.8"
# pysim is a git-only dependency installed explicitly by setup.bat/setup.sh.
+75 -20
View File
@@ -36,7 +36,7 @@ from osmocom.tlv import BER_TLV_IE
from osmocom.utils import rpad
VERSION = '3.25.0'
VERSION = '3.25.1'
MAX_ENVELOPE_SEGMENTS = 5 # max SMS segments for outgoing C-APDU in ENVELOPE
@@ -2697,6 +2697,12 @@ _SCP81_PSK_LEGACY = None
# suspend it with POLLING OFF (see _POLL_DISABLED_BY_CARD).
_POLL_ENABLED = True
_POLL_INTERVAL = 30
# The background poll answers at most this many announced commands per cycle:
# a card that re-queues a command immediately (a chatty applet) must not turn
# the poll into an unbounded chain that holds _CARD_LOCK (v3.25.1 - live
# 2026-10-11: the EP BAP card re-queued the same PROVIDE LOCAL INFORMATION
# 1100 times, blocking every card operation).
_POLL_CHAIN_MAX = 10
_POLL_TIMER = None
# Set when the card sent POLLING OFF (TS 102 223 6.4.14): proactive polling
# stays suspended until a POLL INTERVAL re-enables it. Presence detection and
@@ -2785,6 +2791,9 @@ def _do_status_poll():
global _POLL_TIMER
_POLL_TIMER = None
if not _POLL_ENABLED or _POLL_DISABLED_BY_CARD or _TEST_RUNNING:
# A test run owns the card: keep ticking (never fetch) so polling
# resumes by itself when the run ends (v3.25.1).
_reset_poll_timer()
return
with _CARD_LOCK:
try:
@@ -2793,7 +2802,7 @@ def _do_status_poll():
st_data, st_sw = _send_status(scc)
sys.stderr.write('AUTO-STATUS -> %s\n' % st_sw)
if st_sw.startswith('91'):
_handle_proactive_chain(scc, st_sw)
_handle_proactive_chain(scc, st_sw, max_rounds=_POLL_CHAIN_MAX)
except Exception as e:
sys.stderr.write('AUTO-STATUS error: %s\n' % e)
_handle_card_disconnect(stale=_is_transport_fatal(e))
@@ -2810,6 +2819,15 @@ def _poll_enable():
_POLL_ENABLED = True
_reset_poll_timer()
def _sim_resume_poll():
"""Resume the background STATUS poll after a test run released the card
(no-op while polling is disabled or the card turned it off). Every run
end calls this - the spawn failure included, so a run that never starts
cannot leave polling dead (v3.25.1)."""
if _POLL_ENABLED and not _POLL_DISABLED_BY_CARD:
_reset_poll_timer()
_MENU_TIMEOUT = 60
_MENU_TIMER = None
@@ -3528,7 +3546,9 @@ def _bip_data_available(ch):
server = _server_ref
if not server or not _CARD_CONNECTED or not getattr(server, 'scc', None):
return False
if _PROACTIVE_BUSY or getattr(server, 'stk_pending', None):
if _PROACTIVE_BUSY or getattr(server, 'stk_pending', None) or _TEST_RUNNING:
# a test run owns the card: the bytes stay un-notified and the
# monitor retries on its next tick (v3.25.1)
return False
length = min(len(ch.rx), 0xFF)
tlv = bytes([0xB8, 0x02, 0x80 | (ch.id & 0x07), 0x00,
@@ -4265,8 +4285,10 @@ def _build_tr(scc, cmd_num, cmd_type, dev_src, dev_dst, cmd_qual, dt=None):
The object order follows TS 102 223 6.8.0: command details, device
identities, Result, then the command-specific data objects (the echoed
Duration for POLL INTERVAL, the PLI data for PROVIDE LOCAL INFORMATION).
`dt` is a test seam for the PLI date/time default (see _pli_data_hex)."""
base = bytes([0x81, 0x03, cmd_num, cmd_type, 0x00,
`dt` is a test seam for the PLI date/time default (see _pli_data_hex).
The command details echo the command's qualifier: TS 102 223 6.8.1
requires the object to be identical to the command's (v3.25.1)."""
base = bytes([0x81, 0x03, cmd_num, cmd_type, (cmd_qual or 0) & 0xFF,
0x82, 0x02, dev_dst, dev_src,
0x03, 0x01, 0x00])
if cmd_type == 0x03:
@@ -5080,6 +5102,9 @@ def _bip_flush_channel_events(scc):
global _FLUSHING_CHANNEL_EVENTS
if _FLUSHING_CHANNEL_EVENTS:
return
if _TEST_RUNNING:
# a test run owns the card: the events stay queued (v3.25.1)
return
if not _BIP_LINK_EVENTS:
_BIP.take_pending_events()
return
@@ -5210,6 +5235,11 @@ def _timer_expired(timer_id, elapsed):
server = _server_ref
if not _CARD_CONNECTED or not server:
return
if _TEST_RUNNING:
# A test run owns the card: the ENVELOPE would inject a proactive
# conversation into the script - retry after the run (v3.25.1).
_timer_start(timer_id, 5)
return
# Never inject the ENVELOPE while a fetched command awaits its TERMINAL
# RESPONSE (a paused STK menu); wait outside the card lock, then retry.
for _ in range(10):
@@ -5548,24 +5578,38 @@ def _parse_setup_menu_items(raw):
return _attach_nai(items, nai)
def _handle_proactive_chain(scc, sw91, on_fetch=None, status_poll=True):
def _handle_proactive_chain(scc, sw91, on_fetch=None, status_poll=True, max_rounds=None):
"""Run a FETCH/TERMINAL RESPONSE chain; marks the card as busy so that
terminal-initiated ENVELOPEs (Data available, Channel status, timers) wait.
``status_poll=False`` skips the trailing STATUS (used by the profile
switch, where the card is mid-switch and must not be queried further)."""
switch, where the card is mid-switch and must not be queried further).
``max_rounds`` bounds the answers per call (the background poll passes
_POLL_CHAIN_MAX so a card that re-queues a command cannot hold the card
lock forever); ``_TEST_RUNNING`` aborts the chain before the next FETCH -
a test run owns the card and the pending command stays for the script."""
global _PROACTIVE_BUSY
_PROACTIVE_BUSY = True
try:
return _run_proactive_chain(scc, sw91, on_fetch, status_poll)
return _run_proactive_chain(scc, sw91, on_fetch, status_poll, max_rounds)
finally:
_PROACTIVE_BUSY = False
def _run_proactive_chain(scc, sw91, on_fetch=None, status_poll=True):
def _run_proactive_chain(scc, sw91, on_fetch=None, status_poll=True, max_rounds=None):
sys.stderr.write('91XX chain: sw=%s\n' % sw91)
sw = sw91
paused = False
rounds = 0
while sw.startswith('91'):
if _TEST_RUNNING:
# A test run owns the card: stop before consuming another command
# - the pending one stays for the script (v3.25.1).
sys.stderr.write('91XX chain: stopping - a test run owns the card\n')
break
if max_rounds is not None and rounds >= max_rounds:
sys.stderr.write('91XX chain: %d rounds reached - yielding\n' % rounds)
break
rounds += 1
fetch_len = int(sw[2:], 16) if len(sw) == 4 else 0x100
rv = scc._tp.send_apdu('%s120000%02x' % (scc.cat_cla, fetch_len))
sys.stderr.write('FETCH(%s): %s -> %s\n' % (fetch_len, rv[0] if rv[0] else '(none)', rv[1]))
@@ -5766,11 +5810,21 @@ def _make_menu_fetch_handler(server, resp):
"""on_fetch callback for the menu chain: pauses on user-interactive commands
and stores the pending command so a TERMINAL RESPONSE can be sent later."""
def _on_menu_fetch(raw, cmd_num, cmd_type, dev_src, dev_dst):
# the TR's command details echo the command's qualifier (TS 102 223
# 6.8.1); the on_fetch callback signature does not carry it, so parse
# it here (v3.25.1)
cmd_qual = None
if raw:
try:
cmd_qual = _parse_proactive_header(raw)[4]
except Exception:
cmd_qual = None
if cmd_type == 0x21:
text = _parse_display_text(raw) if raw else None
if text:
server.stk_pending = {'type': 'display_text',
'cmd_num': cmd_num, 'cmd_type': cmd_type,
'cmd_qual': cmd_qual,
'dev_src': dev_src, 'dev_dst': dev_dst, 'text': text}
resp.update(type='display_text', text=text)
return 'pause'
@@ -5778,6 +5832,7 @@ def _make_menu_fetch_handler(server, resp):
items = _parse_select_item(raw) if raw else []
server.stk_pending = {'type': 'select_item',
'cmd_num': cmd_num, 'cmd_type': cmd_type,
'cmd_qual': cmd_qual,
'dev_src': dev_src, 'dev_dst': dev_dst, 'items': items}
resp.update(type='select_item', items=items)
return 'pause'
@@ -5785,6 +5840,7 @@ def _make_menu_fetch_handler(server, resp):
items = _parse_setup_menu_items(raw) if raw else []
server.stk_pending = {'type': 'select_item',
'cmd_num': cmd_num, 'cmd_type': cmd_type,
'cmd_qual': cmd_qual,
'dev_src': dev_src, 'dev_dst': dev_dst, 'items': items}
resp.update(type='select_item', items=items)
return 'pause'
@@ -5796,7 +5852,8 @@ def _make_menu_fetch_handler(server, resp):
if info:
kind = 'get_inkey' if cmd_type == 0x22 else 'get_input'
pending = dict(info, type=kind, cmd_num=cmd_num,
cmd_type=cmd_type, dev_src=dev_src, dev_dst=dev_dst)
cmd_type=cmd_type, cmd_qual=cmd_qual,
dev_src=dev_src, dev_dst=dev_dst)
dur = _find_duration(raw)
pending['duration_unit'] = dur[0] if dur else None
pending['shown_at'] = time.monotonic()
@@ -5892,7 +5949,8 @@ def _menu_send_response(server, result, item_id=None, text=None):
0x02: 10.0}.get(pd['duration_unit'], 1.0)))
value = max(1, min(255, value))
duration_tlv = bytes([0x04, 0x02, pd['duration_unit'], value])
cd = bytes([0x81, 0x03, pd['cmd_num'], pd['cmd_type'], 0x00])
cd = bytes([0x81, 0x03, pd['cmd_num'], pd['cmd_type'],
(pd.get('cmd_qual') or 0) & 0xFF])
di = bytes([0x82, 0x02, pd['dev_dst'], pd['dev_src']])
# TS 102 223 6.8.0 object order: Result (C), Duration (D), Text string (E),
# Item identifier (F).
@@ -6511,7 +6569,7 @@ def _test_terminal_response(cmd_num, cmd_type, dev_dst, dev_src, cmd_qual, respo
elif cmd_type == 0x03:
answer = '840201%02X' % _POLL_INTERVAL
return testscript.build_tr(cmd_num, cmd_type, dev_dst, dev_src, respond,
answer_tlvs=answer or None)
answer_tlvs=answer or None, cmd_qual=cmd_qual)
def _test_run_drain(server, step):
@@ -7301,8 +7359,7 @@ def _test_run_execute(server, script, preset):
else:
_TEST_RUN['status'] = 'ok'
_TEST_RUNNING = False
if _POLL_ENABLED and not _POLL_DISABLED_BY_CARD:
_reset_poll_timer()
_sim_resume_poll()
def _test_run_worker(server, script, preset):
@@ -7318,8 +7375,7 @@ def _test_run_worker(server, script, preset):
_TEST_RUN['finished'] = time.time()
if not _TEST_RUN['status']:
_TEST_RUN['status'] = 'error'
if _POLL_ENABLED and not _POLL_DISABLED_BY_CARD:
_reset_poll_timer()
_sim_resume_poll()
# ---- test suites (v3.22.0) -------------------------------------------------
@@ -7436,6 +7492,7 @@ def _test_run_spawn(server, target, args, thread_name):
# guard keys off _TEST_RUNNING, and no worker would ever clear it).
_TEST_THREAD = None
_TEST_RUNNING = False
_sim_resume_poll()
with _TEST_LOCK:
_TEST_RUN['running'] = False
_TEST_RUN['status'] = 'error'
@@ -7593,8 +7650,7 @@ def _test_suite_execute(server, suite, members, preset, preset_id):
_TEST_RUN['finished'] = time.time()
_TEST_RUN['log_prefix'] = '' # the report is done: no stale prefix
_TEST_RUNNING = False
if _POLL_ENABLED and not _POLL_DISABLED_BY_CARD:
_reset_poll_timer()
_sim_resume_poll()
def _test_suite_worker(server, suite, members, preset, preset_id):
@@ -7633,8 +7689,7 @@ def _test_suite_worker(server, suite, members, preset, preset_id):
}
finally:
_TEST_RUNNING = False
if _POLL_ENABLED and not _POLL_DISABLED_BY_CARD:
_reset_poll_timer()
_sim_resume_poll()
def _test_suite_state(suite, members, preset, preset_id, session, reset_info=None):
+7 -3
View File
@@ -700,15 +700,19 @@ def _encode_text(text, dcs):
return text.encode('gsm03.38')
def build_tr(cmd_num, cmd_type, dev_dst, dev_src, respond, answer_tlvs=None):
def build_tr(cmd_num, cmd_type, dev_dst, dev_src, respond, answer_tlvs=None,
cmd_qual=None):
"""Flat COMPREHENSION-TLV TERMINAL RESPONSE payload (TS 102 223 6.8) in the
6.8.0 object order: command details, device identities, Result, then the
command-specific objects (the terminal's own `answer_tlvs` - the PLI data
of PROVIDE LOCAL INFORMATION, the echoed Duration of POLL INTERVAL - the
Text string for GET INKEY/GET INPUT, the Item identifier for SELECT ITEM)
and any extra raw TLVs. Matches the interactive menu TR layout; the
runner passes the terminal objects the script did not override."""
out = bytearray([0x81, 0x03, cmd_num & 0xFF, cmd_type & 0xFF, 0x00])
runner passes the terminal objects the script did not override. The
command details echo the command's qualifier (TS 102 223 6.8.1: the
object shall be identical to the command's)."""
out = bytearray([0x81, 0x03, cmd_num & 0xFF, cmd_type & 0xFF,
(cmd_qual or 0) & 0xFF])
out += bytes([0x82, 0x02, dev_dst & 0xFF, dev_src & 0xFF])
result = int(respond.get('result', 0))
add_info = int(respond.get('additional_info', 0) or 0)
File diff suppressed because it is too large Load Diff
+19 -4
View File
@@ -863,6 +863,20 @@ class TestProactiveDecode(unittest.TestCase):
tr = _build_tr(None, 1, 0x26, 0x81, 0x82, 0x00)
self.assertEqual(tr.hex(), '81030126008202828103010093055210011000')
def test_build_tr_echoes_the_command_qualifier(self):
# TS 102 223 6.8.1: the TR's command details shall be identical to
# the command's (including the qualifier); without one they stay 00
self.assertTrue(_build_tr(None, 1, 0x26, 0x81, 0x82, 0x06)
.hex().startswith('8103012606'))
self.assertTrue(_build_tr(None, 1, 0x03, 0x81, 0x82, None)
.hex().startswith('8103010300'))
# the runner's builder echoes the same way
from pysim_simple_server import testscript as T
self.assertTrue(T.build_tr(1, 0x26, 0x82, 0x81, {}, cmd_qual=0x06)
.hex().startswith('8103012606'))
self.assertTrue(T.build_tr(1, 0x26, 0x82, 0x81, {})
.hex().startswith('8103012600'))
def test_build_tr_poll_interval_duration_after_result(self):
import pysim_simple_server.server as srv
tr = _build_tr(None, 1, 0x03, 0x81, 0x82, None)
@@ -884,11 +898,12 @@ class TestProactiveDecode(unittest.TestCase):
tr_west = _build_tr(None, 1, 0x26, 0x81, 0x82, 0x03, west)
finally:
srv._PLI_DATA[0x03] = saved
# the command details echo the qualifier (TS 102 223 6.8.1)
self.assertEqual(tr_east.hex(),
'810301260082028281030100260762900312550021')
'810301260382028281030100260762900312550021')
# negative zone: the sign bit of the time zone byte (0x08)
self.assertEqual(tr_west.hex(),
'810301260082028281030100260762105080709009')
'810301260382028281030100260762105080709009')
def test_build_tr_pli_dict_entry_overrides_host_clock(self):
import pysim_simple_server.server as srv
@@ -900,7 +915,7 @@ class TestProactiveDecode(unittest.TestCase):
datetime(2026, 9, 30, 21, 55, tzinfo=timezone.utc))
finally:
srv._PLI_DATA[0x03] = saved
self.assertEqual(tr.hex(), '810301260082028281030100130752f01000ff0001')
self.assertEqual(tr.hex(), '810301260382028281030100130752f01000ff0001')
def test_build_tr_pli_empty_other_qualifier_sends_no_data(self):
# a qualifier with neither an override nor a built-in default (02
@@ -913,7 +928,7 @@ class TestProactiveDecode(unittest.TestCase):
data = srv._pli_data_hex(0x02)
finally:
srv._PLI_DATA[0x02] = saved
self.assertEqual(tr.hex(), '810301260082028281030100')
self.assertEqual(tr.hex(), '810301260282028281030100')
self.assertEqual(data, '')
def test_pli_defaults_cover_the_common_qualifiers(self):
+91
View File
@@ -2,6 +2,7 @@
"""Tests for the background STATUS polling interval semantics."""
import sys
import types
import unittest
from pathlib import Path
from unittest import mock
@@ -156,5 +157,95 @@ class TestCardDrivenPolling(unittest.TestCase):
self.assertEqual(fields[0]['value'], '30 s')
class FakeChattyScc:
"""A card that re-queues the same PROVIDE LOCAL INFORMATION immediately
after every accepted TERMINAL RESPONSE - the live card's loop (v3.25.1)."""
cat_cla = '80'
def __init__(self):
self.sent = []
self._tp = types.SimpleNamespace(send_apdu=self._send)
def _send(self, hexstr):
self.sent.append(hexstr)
low = hexstr.lower()
if low.startswith('8012'): # FETCH: the PLI is pending
return ('D009810301260082028182', '9000')
if low.startswith('8014'): # TR: accepted
return ('', '9000')
if low.startswith('80f2'): # STATUS: re-announced
return ('', '910b')
return ('', '9000')
class TestProactiveChainOwnership(unittest.TestCase):
"""The auto chain must yield: a bounded number of answers per poll cycle
and no fetching at all while a test run owns the card (v3.25.1 - the live
1100-command loop held _CARD_LOCK and blocked a run's start)."""
def setUp(self):
self.saved = (S._TEST_RUNNING, S._server_ref, S._CARD_CONNECTED)
S._TEST_RUNNING = False
def tearDown(self):
S._TEST_RUNNING, S._server_ref, S._CARD_CONNECTED = self.saved
def _fetches(self, scc):
return [a for a in scc.sent if a.lower().startswith('8012')]
def test_the_poll_chain_yields_after_the_bound(self):
scc = FakeChattyScc()
S._handle_proactive_chain(scc, '910b', max_rounds=3)
self.assertEqual(len(self._fetches(scc)), 3)
def test_the_chain_stops_when_a_run_owns_the_card(self):
scc = FakeChattyScc()
S._TEST_RUNNING = True
S._handle_proactive_chain(scc, '910b')
self.assertEqual(scc.sent, [])
def test_the_poll_never_fetches_during_a_run_and_keeps_ticking(self):
scc = FakeChattyScc()
S._TEST_RUNNING = True
S._server_ref = types.SimpleNamespace(scc=scc)
S._POLL_ENABLED = True
S._POLL_DISABLED_BY_CARD = False
try:
with mock.patch.object(S, '_reset_poll_timer') as timer:
S._do_status_poll()
timer.assert_called_once()
finally:
S._POLL_ENABLED = False
self.assertEqual(scc.sent, [])
def test_the_timer_expiration_waits_for_a_run(self):
S._TEST_RUNNING = True
S._CARD_CONNECTED = True
S._server_ref = types.SimpleNamespace()
with mock.patch.object(S, '_timer_start') as start:
S._timer_expired(1, 60)
start.assert_called_once_with(1, 5)
def test_sim_resume_poll_rearms_only_when_enabled(self):
S._POLL_ENABLED = True
S._POLL_DISABLED_BY_CARD = False
with mock.patch.object(S, '_reset_poll_timer') as timer:
S._sim_resume_poll()
timer.assert_called_once()
S._POLL_DISABLED_BY_CARD = True
with mock.patch.object(S, '_reset_poll_timer') as timer:
S._sim_resume_poll()
timer.assert_not_called()
def test_every_run_end_resumes_the_poll(self):
# the two worker bodies, their two exception paths and the spawn
# failure all call the helper
import inspect
src = inspect.getsource(S)
self.assertEqual(src.count('\n _sim_resume_poll()'), 2)
self.assertEqual(src.count('\n _sim_resume_poll()'), 3)
if __name__ == '__main__':
unittest.main()