Files
simple/tests/test_card_recovery.py
T
catarrh 8a08f642d4
release / release (push) Waiting to run
v3.32.2: the Windows card-swap auto-equip loop
After a card swap (or the first insertion after startup) on Windows the
auto-equip looped forever: "card inserted, initializing" / "card gone
during initialization", with no "retry 2/3" lines.

An equip with the card still present produced no card (app.card None),
and that branch reported "card gone" and returned True - a no-op success
- so the three retries never ran and the watchdog re-armed every ~5 s.

- The card-gone no-op is now only for a real removal (card_present
  False); a card-present init that produced no card raises a failure.
- Any failed attempt marks the transport stale, so the next attempt
  rebuilds the PC/SC link (a Windows swap can leave the old link unable
  to connect; _ensure_transport releases the old link before building
  the new one, so the 2026-09-24 stale-handle bug cannot recur).
- The error pySim swallowed (cmd2) is printed as "AUTO-EQUIP: init
  output:" with its traceback - the failure is diagnosable from the log.
- Docs: the READMEs' troubleshooting bullet.
- Version strings + the PWA cache bumped together (simple-v460).

Tests: a card-present init without a card is a failure, a mid-init
removal stays a silent no-op, any failure marks the transport, the
stale-handle traceback reaches the log, and an end-to-end retry: a
failed init, the old link released, a fresh one built, the retry
succeeds.  1012 Python + 953 frontend green.
2026-10-21 14:35:44 +03:00

482 lines
20 KiB
Python

# coding=utf-8
"""Tests for the PC/SC failure recovery: presence-monitor watchdog and
transport recreation after a service failure (pcscd restart)."""
import sys
import unittest
from io import StringIO
from types import SimpleNamespace
from unittest import mock
from pysim_simple_server import server
class PcscErrorTests(unittest.TestCase):
def test_pyscard_exception_counts_as_pcsc_error(self):
class FakePcscError(Exception):
def __init__(self):
self.hresult = 0x8010001D # SCARD_E_NO_SERVICE
self.assertTrue(server._is_pcsc_error(FakePcscError()))
def test_card_level_errors_do_not(self):
self.assertFalse(server._is_pcsc_error(RuntimeError('SW match failed')))
self.assertFalse(server._is_pcsc_error(Exception()))
class TransportFatalTests(unittest.TestCase):
@staticmethod
def _exc(hresult):
exc = RuntimeError('pcsc')
exc.hresult = hresult
return exc
def test_card_level_errors_do_not_rebuild_the_transport(self):
# A normal card swap (SCARD_W_REMOVED_CARD) is recovered by a plain
# reconnect on the existing link - it must not trigger a rebuild.
for hr in (0x80100069, 0x8010000C, 0x80100068, 0x80100066, 0x80100067):
self.assertFalse(server._is_transport_fatal(self._exc(hr)), hex(hr))
def test_sharing_violation_is_not_transport_fatal(self):
# Another process holds the card: a fresh context cannot free it, so
# rebuilding the transport would only burn the retry budget.
self.assertFalse(server._is_transport_fatal(self._exc(0x8010000B)))
def test_service_errors_rebuild_the_transport(self):
for hr in (0x8010001D, 0x8010001E, 0x8010002E, 0x80100003):
self.assertTrue(server._is_transport_fatal(self._exc(hr)), hex(hr))
def test_non_pcsc_errors_are_not_transport_fatal(self):
self.assertFalse(server._is_transport_fatal(RuntimeError('SW match failed')))
class EquipStateTests(unittest.TestCase):
@staticmethod
def _app(card='card', lchan='lchan', cmdsets=None, profile_sets=None):
rs = None
if profile_sets is not None:
rs = SimpleNamespace(profile=SimpleNamespace(shell_cmdsets=profile_sets))
app = SimpleNamespace(card=card, lchan=lchan, rs=rs)
if cmdsets is not None:
app.find_commandsets = lambda cls: cmdsets
return app
def test_card_and_channel_are_required(self):
self.assertFalse(server._app_equip_complete(None))
self.assertFalse(server._app_equip_complete(self._app(card=None)))
self.assertFalse(server._app_equip_complete(self._app(lchan=None)))
def test_profile_command_sets_must_be_installed(self):
# PysimApp.equip() sets card/rs before registering the command sets;
# an abort during registration must not pass as success.
cs = object()
app = self._app(cmdsets=[], profile_sets=[cs])
self.assertFalse(server._app_equip_complete(app))
app.find_commandsets = lambda cls: [cs]
self.assertTrue(server._app_equip_complete(app))
def test_unverifiable_state_is_trusted(self):
# No profile command sets / foreign app object: cannot verify.
self.assertTrue(server._app_equip_complete(self._app()))
class WatchdogTests(unittest.TestCase):
def test_alive_monitor_is_left_alone(self):
calls = []
rv = server._watchdog_tick('reader', alive_fn=lambda: True,
restart_fn=lambda name: calls.append(name))
self.assertEqual(rv, 'ok')
self.assertEqual(calls, [])
def test_stopped_monitor_is_restarted(self):
calls = []
rv = server._watchdog_tick('reader', alive_fn=lambda: False,
restart_fn=lambda name: calls.append(name))
self.assertEqual(rv, 'restarted')
self.assertEqual(calls, ['reader'])
class EnsureTransportTests(unittest.TestCase):
def setUp(self):
server._TRANSPORT_STALE = False
def tearDown(self):
server._TRANSPORT_STALE = False
@staticmethod
def make_server(factory):
app = SimpleNamespace(sl='old-sl')
return SimpleNamespace(sl='old-sl', scc='old-scc', card='old-card',
app=app, transport_factory=factory)
def test_fresh_transport_is_kept(self):
calls = []
srv = self.make_server(lambda: calls.append(1) or 'new-sl')
self.assertTrue(server._ensure_transport(srv))
self.assertEqual(calls, [])
self.assertEqual(srv.sl, 'old-sl')
def test_stale_transport_is_recreated(self):
srv = self.make_server(lambda: 'new-sl')
server._TRANSPORT_STALE = True
self.assertTrue(server._ensure_transport(srv))
self.assertEqual(srv.sl, 'new-sl')
self.assertEqual(srv.app.sl, 'new-sl')
self.assertIsNone(srv.scc)
self.assertIsNone(srv.card)
self.assertFalse(server._TRANSPORT_STALE)
def test_failed_reconnect_keeps_the_transport_stale(self):
def boom():
raise RuntimeError('service not available')
srv = self.make_server(boom)
server._TRANSPORT_STALE = True
self.assertFalse(server._ensure_transport(srv))
self.assertTrue(server._TRANSPORT_STALE)
self.assertEqual(srv.sl, 'old-sl')
def test_old_link_is_released_before_the_rebuild(self):
released = []
class FakeLink:
def disconnect(self):
released.append(True)
old = FakeLink()
app = SimpleNamespace(sl=old, card='card', rs='rs', lchan='lchan')
srv = SimpleNamespace(sl=old, scc='scc', card='card', app=app,
transport_factory=lambda: 'new-sl')
server._TRANSPORT_STALE = True
self.assertTrue(server._ensure_transport(srv))
self.assertEqual(released, [True])
self.assertEqual(srv.sl, 'new-sl')
self.assertEqual(app.sl, 'new-sl')
self.assertIsNone(app.card)
self.assertIsNone(app.rs)
self.assertIsNone(app.lchan)
def test_no_factory_clears_the_flag(self):
srv = SimpleNamespace(sl='old-sl', app=None, transport_factory=None)
server._TRANSPORT_STALE = True
self.assertTrue(server._ensure_transport(srv))
self.assertFalse(server._TRANSPORT_STALE)
def test_none_server_is_a_noop(self):
server._TRANSPORT_STALE = True
self.assertTrue(server._ensure_transport(None))
self.assertTrue(server._TRANSPORT_STALE)
class DisconnectTests(unittest.TestCase):
def setUp(self):
server._TRANSPORT_STALE = False
def tearDown(self):
server._TRANSPORT_STALE = False
def test_stale_disconnect_marks_the_transport(self):
server._handle_card_disconnect(stale=True)
self.assertTrue(server._TRANSPORT_STALE)
def test_plain_disconnect_leaves_the_transport_usable(self):
server._handle_card_disconnect()
self.assertFalse(server._TRANSPORT_STALE)
def test_disconnect_unequips_the_shell_first(self):
# PysimApp.equip(None, None) unregisters the old profile's command
# sets; without it the next equip fails with "CommandSet ... is
# already installed" and the file tree breaks.
calls = []
app = SimpleNamespace(rs='rs', card='card', lchan='lchan',
stdout=StringIO(),
equip=lambda c, r: calls.append((c, r)))
srv = SimpleNamespace(app=app, card='card', scc='scc', stk_pending=None,
menu_active=False, event_list=None, sim_menu=None,
iccid=None, net_state=None, equipping=False, card_session=1)
saved = server._server_ref
server._server_ref = srv
try:
server._handle_card_disconnect()
finally:
server._server_ref = saved
self.assertEqual(calls, [(None, None)])
self.assertIsNone(app.card)
self.assertIsNone(app.rs)
self.assertIsNone(app.lchan)
def test_disconnect_clears_the_app_card_state(self):
# The dead card must not stay reachable through app.card/app.rs:
# handlers using app.rs kept transmitting over the removed card and
# the old PC/SC link stayed connected (the v3.5.1 auto-equip bug).
app = SimpleNamespace(card='card', rs='rs', lchan='lchan')
srv = SimpleNamespace(app=app, card='card', scc='scc', stk_pending='x',
menu_active=True, event_list='e', sim_menu='m',
iccid='123', net_state='n', equipping=True, card_session=5)
saved = server._server_ref
server._server_ref = srv
try:
server._handle_card_disconnect()
finally:
server._server_ref = saved
self.assertIsNone(app.card)
self.assertIsNone(app.rs)
self.assertIsNone(app.lchan)
self.assertIsNone(srv.card)
self.assertIsNone(srv.scc)
class AutoEquipTests(unittest.TestCase):
def setUp(self):
self.saved = (server._server_ref, server._CARD_CONNECTED, server._TRANSPORT_STALE,
server._AUTO_EQUIP, server._AUTO_EQUIP_BUSY, server._AUTO_EQUIP_LAST,
server._AUTO_EQUIP_BACKOFF)
server._server_ref = None
server._CARD_CONNECTED = False
server._TRANSPORT_STALE = False
server._AUTO_EQUIP = True
server._AUTO_EQUIP_BUSY = False
server._AUTO_EQUIP_LAST = 0.0
server._AUTO_EQUIP_BACKOFF = server._AUTO_EQUIP_REARM_DELAY
def tearDown(self):
(server._server_ref, server._CARD_CONNECTED, server._TRANSPORT_STALE,
server._AUTO_EQUIP, server._AUTO_EQUIP_BUSY, server._AUTO_EQUIP_LAST,
server._AUTO_EQUIP_BACKOFF) = self.saved
@staticmethod
def make_server(onecmd):
app = SimpleNamespace(stdout=StringIO(), card=None, lchan='lchan')
app.onecmd_plus_hooks = onecmd
return SimpleNamespace(app=app, card=None, scc=None, card_present=True,
equipping=False, terminal_profile='tp', card_session=1)
def test_retries_after_a_transient_card_level_failure(self):
state = {'calls': 0}
srv = self.make_server(None)
def onecmd(cmd):
state['calls'] += 1
if state['calls'] == 1:
raise RuntimeError('Failed to transmit with protocol T0. Card was removed.')
srv.app.card = SimpleNamespace(_scc='scc')
srv.app.lchan = 'lchan' # the failed attempt unequipped the shell
srv.app.onecmd_plus_hooks = onecmd
applied = []
with mock.patch.object(server, '_ensure_transport', lambda s: True), \
mock.patch.object(server, '_apply_equipped_card', lambda s: applied.append(s)):
self.assertTrue(server._auto_equip_attempts(srv, 3, lambda s: None))
self.assertEqual(state['calls'], 2)
self.assertEqual(applied, [srv])
# the failed attempt left the handle suspect; the retry's
# _ensure_transport (patched here) would rebuild and clear the flag
self.assertTrue(server._TRANSPORT_STALE)
def test_a_failed_init_rebuilds_the_transport_before_the_retry(self):
# The Windows swap end to end: the first init fails with the card
# present, the retry starts from a fresh PC/SC link and succeeds -
# and the old link is released before the new one is built.
srv = self.make_server(None)
released = []
built = []
class Link:
def disconnect(self):
released.append(self)
old = Link()
srv.sl = old
srv.transport_factory = lambda: built.append(Link()) or built[-1]
calls = {'n': 0}
def onecmd(cmd):
calls['n'] += 1
if calls['n'] == 1:
raise RuntimeError('ProtocolError: Connection failed')
# a completed equip installs both (the failure unequipped the shell)
srv.app.card = SimpleNamespace(_scc='scc')
srv.app.lchan = 'lchan'
srv.app.onecmd_plus_hooks = onecmd
with mock.patch.object(server, '_apply_equipped_card', lambda s: None):
self.assertTrue(server._auto_equip_attempts(srv, 3, lambda s: None))
self.assertEqual(calls['n'], 2)
self.assertEqual(len(built), 1)
self.assertEqual(released, [old])
self.assertIs(srv.sl, built[0])
self.assertFalse(server._TRANSPORT_STALE)
def test_a_failed_attempt_marks_the_transport_for_the_rebuild(self):
# A card-level error used to keep the link: after a Windows card swap
# the link stayed unable to connect, every retry failed and the
# watchdog re-armed forever (live 2026-10-21). Any failed attempt now
# marks the transport stale so the retry rebuilds it (which releases
# the old link first - the 2026-09-24 bug).
srv = self.make_server(None)
def onecmd(cmd):
exc = RuntimeError('Card was removed.')
exc.hresult = 0x80100069
raise exc
srv.app.onecmd_plus_hooks = onecmd
with mock.patch.object(server, '_ensure_transport', lambda s: True):
self.assertFalse(server._auto_equip_attempt(srv))
self.assertTrue(server._TRANSPORT_STALE)
def test_transport_fatal_failure_marks_the_transport(self):
srv = self.make_server(None)
def onecmd(cmd):
exc = RuntimeError('service not available')
exc.hresult = 0x8010001D
raise exc
srv.app.onecmd_plus_hooks = onecmd
with mock.patch.object(server, '_ensure_transport', lambda s: True):
self.assertFalse(server._auto_equip_attempt(srv))
self.assertTrue(server._TRANSPORT_STALE)
def test_rearm_triggers_with_cooldown(self):
calls = []
server._server_ref = SimpleNamespace(card_present=True, equipping=False)
with mock.patch.object(server, '_auto_equip_trigger',
lambda: calls.append(1) or True):
self.assertTrue(server._auto_equip_rearm(now=100.0))
self.assertFalse(server._auto_equip_rearm(now=100.5))
self.assertTrue(server._auto_equip_rearm(now=106.0))
self.assertEqual(len(calls), 2)
def test_failed_worker_backs_off_the_rearm(self):
state = {'calls': 0}
srv = self.make_server(None)
def onecmd(cmd):
state['calls'] += 1
raise RuntimeError('card cannot be initialized')
srv.app.onecmd_plus_hooks = onecmd
server._server_ref = srv
with mock.patch.object(server, '_ensure_transport', lambda s: True), \
mock.patch.object(server, '_apply_equipped_card', lambda s: None), \
mock.patch.object(server, '_AUTO_EQUIP_ATTEMPTS', 1):
server._auto_equip_worker()
self.assertEqual(state['calls'], 1)
self.assertGreater(server._AUTO_EQUIP_BACKOFF, server._AUTO_EQUIP_REARM_DELAY)
def test_half_initialized_equip_is_a_failure(self):
# cmd2 swallows the equip exception, so a card without the new
# profile's command sets must not count as success (it broke
# /api/tree on 2026-09-24 while the worker reported done).
srv = self.make_server(None)
cs = object()
srv.app.rs = SimpleNamespace(profile=SimpleNamespace(shell_cmdsets=[cs]))
srv.app.find_commandsets = lambda cls: []
def onecmd(cmd):
srv.app.card = SimpleNamespace(_scc='scc') # set before the abort
srv.app.onecmd_plus_hooks = onecmd
with mock.patch.object(server, '_ensure_transport', lambda s: True), \
mock.patch.object(server, '_apply_equipped_card',
lambda s: self.fail('_apply_equipped_card must not run')):
self.assertFalse(server._auto_equip_attempt(srv))
self.assertTrue(server._TRANSPORT_STALE)
def test_captured_equip_traceback_is_a_failure(self):
# cmd2 prints swallowed exceptions (with cmd2 debug: a traceback) to
# sys.stderr, which the attempt captures.
srv = self.make_server(None)
def onecmd(cmd):
srv.app.card = SimpleNamespace(_scc='scc')
sys.stderr.write('Traceback (most recent call last):\nRuntimeError: boom\n')
srv.app.onecmd_plus_hooks = onecmd
with mock.patch.object(server, '_ensure_transport', lambda s: True), \
mock.patch.object(server, '_apply_equipped_card',
lambda s: self.fail('_apply_equipped_card must not run')):
self.assertFalse(server._auto_equip_attempt(srv))
def test_a_card_present_init_without_a_card_is_a_failure(self):
# The Windows loop: the presence monitor still reports the card, the
# init produced nothing - it must retry (and rebuild), never report a
# silent "card gone" success that re-arms forever (2026-10-21).
srv = self.make_server(None)
def onecmd(cmd):
pass # cmd2 swallowed the real error: no card built
srv.app.onecmd_plus_hooks = onecmd
with mock.patch.object(server, '_ensure_transport', lambda s: True):
self.assertFalse(server._auto_equip_attempt(srv))
self.assertTrue(server._TRANSPORT_STALE)
def test_a_card_removed_during_the_init_is_still_a_silent_no_op(self):
srv = self.make_server(None)
def onecmd(cmd):
srv.card_present = False # the monitor saw the removal mid-init
srv.app.onecmd_plus_hooks = onecmd
with mock.patch.object(server, '_ensure_transport', lambda s: True), \
mock.patch('sys.stderr', new=StringIO()) as err:
self.assertTrue(server._auto_equip_attempt(srv))
self.assertIn('card gone during initialization', err.getvalue())
self.assertFalse(server._TRANSPORT_STALE)
def test_the_swallowed_traceback_reaches_the_log(self):
srv = self.make_server(None)
def onecmd(cmd):
srv.app.card = SimpleNamespace(_scc='scc')
sys.stderr.write('Traceback (most recent call last):\n'
' File "x", line 1, in equip\n'
'ProtocolError: Connection failed\n')
srv.app.onecmd_plus_hooks = onecmd
with mock.patch.object(server, '_ensure_transport', lambda s: True), \
mock.patch.object(server, '_apply_equipped_card', lambda s: None), \
mock.patch('sys.stderr', new=StringIO()) as err:
self.assertFalse(server._auto_equip_attempt(srv))
out = err.getvalue()
self.assertIn('AUTO-EQUIP: init output:', out)
self.assertIn('ProtocolError: Connection failed', out)
def test_failed_attempt_unequips_the_half_initialized_shell(self):
calls = []
srv = self.make_server(None)
srv.app.rs = 'rs'
srv.app.equip = lambda c, r: calls.append((c, r))
def onecmd(cmd):
raise RuntimeError('CommandSet ShellCommands is already installed')
srv.app.onecmd_plus_hooks = onecmd
with mock.patch.object(server, '_ensure_transport', lambda s: True):
self.assertFalse(server._auto_equip_attempt(srv))
self.assertEqual(calls, [(None, None)])
self.assertIsNone(srv.app.card)
def test_busy_trigger_leaves_the_rearm_window_open(self):
server._server_ref = SimpleNamespace(card_present=True, equipping=False)
with mock.patch.object(server, '_auto_equip_trigger', lambda: False):
self.assertFalse(server._auto_equip_rearm(now=100.0))
self.assertEqual(server._AUTO_EQUIP_LAST, 0.0)
def test_rearm_needs_a_present_card_and_no_session(self):
calls = []
server._server_ref = SimpleNamespace(card_present=True, equipping=False)
with mock.patch.object(server, '_auto_equip_trigger',
lambda: calls.append(1) or True):
server._CARD_CONNECTED = True
self.assertFalse(server._auto_equip_rearm(now=100.0))
server._CARD_CONNECTED = False
server._server_ref.card_present = False
self.assertFalse(server._auto_equip_rearm(now=100.0))
server._server_ref.card_present = True
server._server_ref.equipping = True
self.assertFalse(server._auto_equip_rearm(now=100.0))
self.assertEqual(calls, [])