catarrh/simple
1
0
Fork 0
Code Issues Pull Requests Actions 8 Packages Projects Releases Wiki Activity

v3.16.0: test scripts move to a server-side store

Browse Source 
The scripts lived in the browser's localStorage although only the local
server can run them.  The new test_scripts.py store
(~/.pysim-simple-server/test_scripts.json, --test-scripts PATH) sits next to
the card presets: stable uuids and timestamps, atomic writes, and every
mutation is validated with the runner's own engine (the submitted shape is
kept - the normaliser's output is not re-normalisable: it fills defaults
like an 'any' PoR check that the check normaliser then rejects), so a stored
script cannot fail at run start for a validation reason.  Import accepts the
old localStorage export as-is (export/import remains the migration path; the
PWA import also takes an exported array).

- /api/test/scripts (GET card-free: path/count + scripts), POST create,
  /update, /delete and /import serve the store; the endpoints answer 503
  without a store and 500 on a write failure, like the preset endpoints.
- /api/test/run takes a stored script_id (resolved when no inline script is
  sent); the PWA sends the edited copy inline plus the id for the report,
  and the run snapshot carries script_id.
- The PWA's Test script pill loads from the store, writes through on edits
  (debounced name, immediate step/new/copy/delete), reports load/save/delete
  failures inline, and exports the portable {name, steps} shape.

Tests: 808 frontend + 701 Python; sw cache simple-v379.
This commit is contained in:
Антон Трошин
2026-10-04 17:24:21 +03:00
parent abc9ea2751
commit 8430dd1694
14 changed files with 1003 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pysim_simple_server/__main__.py
+14 -1
View File
@@ -15,7 +15,9 @@ from .shell import load_pysim_app
from . import fastinit
from . import gsmtap
from . import presets
from .server import PysimHandler, StderrApduTracer, _LoggingApduTracer, VERSION, _send_terminal_profile, _DefaultProactiveHandler, _handle_proactive_chain, _send_status, _init_proactive_session, _timing_on, _tlog, _set_menu_timeout, start_card_monitor, set_auto_equip, _read_iccid, _netstate_read, _netstate_install, _LineFilter
from . import test_scripts
from . import testscript
from .server import PysimHandler, StderrApduTracer, _LoggingApduTracer, VERSION, _send_terminal_profile, _DefaultProactiveHandler, _handle_proactive_chain, _send_status, _init_proactive_session, _timing_on, _tlog, _set_menu_timeout, start_card_monitor, set_auto_equip, _read_iccid, _netstate_read, _netstate_install, _LineFilter, _test_command_type
_server_start = 0
@@ -66,6 +68,8 @@ def main():
parser.add_argument('--log-requests', action='store_true', default=False, help='Log request/response payloads to stderr')
parser.add_argument('--card-presets', default=None, metavar='PATH',
help='Card preset store (default: ~/.pysim-simple-server/card_presets.json)')
parser.add_argument('--test-scripts', default=None, metavar='PATH',
help='Test script store (default: ~/.pysim-simple-server/test_scripts.json)')
parser.add_argument('--sms-oa', default='12345', metavar='DIGITS',
help='TP-Originating-Address (SMSC number) for the SMS-DELIVER TPDU (default: 12345)')
parser.add_argument('--sms-sm-sc', default='12345678912', metavar='DIGITS',
@@ -287,6 +291,15 @@ def main():
server.card_presets = presets.PresetStore(opts.card_presets)
sys.stderr.write('PRESETS: %s (%d presets)\n'
% (server.card_presets.path, len(server.card_presets.list())))
# Test scripts live server-side too (only the server can run them):
# --test-scripts overrides the default store, and the store validates a
# script with the same engine the runner uses, so a stored script can
# never fail at run start for a validation reason.
server.test_scripts = test_scripts.TestScriptStore(
opts.test_scripts,
validator=lambda raw: testscript.normalise_script(raw, _test_command_type))
sys.stderr.write('TESTSCRIPTS: %s (%d scripts)\n'
% (server.test_scripts.path, len(server.test_scripts.list())))
server.sim_menu = sim_menu
server.event_list = event_list
server.menu_active = False
pysim_simple_server/server.py
+139 -7
View File
@@ -22,6 +22,7 @@ from pysim_simple_server import esim
from pysim_simple_server import capmem
from pysim_simple_server import testscript
from pysim_simple_server import presets
from pysim_simple_server import test_scripts
from smartcard.CardMonitoring import CardMonitor, CardObserver
from cmd2.exceptions import CommandSetRegistrationError
@@ -32,7 +33,7 @@ from osmocom.tlv import BER_TLV_IE
from osmocom.utils import rpad
VERSION = '3.15.0'
VERSION = '3.16.0'
MAX_ENVELOPE_SEGMENTS = 5 # max SMS segments for outgoing C-APDU in ENVELOPE
@@ -5572,7 +5573,8 @@ _TEST_RUNNING = False
_TEST_RUN = {
'running': False, 'stop': False, 'name': None, 'status': None,
'session': None, 'index': 0, 'total': 0, 'steps': [],
'preset': None, 'scp80_counter': None, 'scp80_counters': {},
'preset': None, 'script_id': None,
'scp80_counter': None, 'scp80_counters': {},
'started': None, 'finished': None, 'error': None,
}
_TEST_LOCK = threading.Lock()
@@ -5606,7 +5608,7 @@ _CARD_FREE_GET = frozenset([
'/api/stk-status', '/api/poll-status', '/api/proactive-log',
'/api/menu', '/api/events', '/api/terminal-profile',
'/api/pli-qualifiers', '/api/pli-dict', '/api/commands',
'/api/net-state', '/api/mcc-mnc', '/api/presets',
'/api/net-state', '/api/mcc-mnc', '/api/presets', '/api/test/scripts',
'/api/scp81/status', '/api/bip/status', '/api/scp81/script',
'/api/scp81/log', '/api/bip/log',
])
@@ -5949,6 +5951,23 @@ def _test_run_action(server, step, ctx):
return result, pending
def _test_run_script_from_body(server, body):
"""The script of a run: the inline `script` body (the PWA always sends the
edited copy) or the stored script named by `script_id`. Returns
``(script_raw, script_id, error)``; error is None on success."""
script_raw = body.get('script')
script_id = str(body.get('script_id') or '').strip()
if isinstance(script_raw, dict):
return script_raw, script_id or None, None
if not script_id:
return None, None, None
store = getattr(server, 'test_scripts', None)
entry = store.get(script_id) if store else None
if entry is None:
return None, None, 'test script %s not found' % script_id
return {'name': entry['name'], 'steps': entry['steps']}, entry['id'], None
def _test_run_preset_from_body(server, body):
"""The preset of a run: the inline `preset` body (external API callers,
docs/api.md) or the stored preset named by `preset_id` - the PWA sends the
@@ -6259,7 +6278,7 @@ def _test_run_worker(server, script, preset):
_reset_poll_timer()
def _test_run_start(server, script, preset):
def _test_run_start(server, script, preset, script_id=None):
"""Start a run: suspend background polling, answer a paused interactive
command, own the card via the worker thread."""
global _TEST_THREAD, _TEST_RUNNING, _POLL_TIMER
@@ -6269,7 +6288,7 @@ def _test_run_start(server, script, preset):
'session': getattr(server, 'card_session', 0), 'index': 0,
'total': len(script['steps']), 'steps': [], 'started': time.time(),
'finished': None, 'error': None, 'scp80_counter': None,
'scp80_counters': {},
'scp80_counters': {}, 'script_id': script_id,
'preset': (preset or {}).get('name') or (preset or {}).get('iccid'),
})
_TEST_RUNNING = True
@@ -6335,6 +6354,24 @@ class PysimHandler(BaseHTTPRequestHandler):
self._send_json(resp, 500)
self._log_resp(resp)
def _test_scripts_store_or_503(self):
"""The test script store, or None after answering 503 (a server built
without one - e.g. a test harness - must not crash the endpoints)."""
store = getattr(self.server, 'test_scripts', None)
if store is None:
resp = {'error': 'test script store not initialized'}
self._send_json(resp, 503)
self._log_resp(resp)
return None
return store
def _test_scripts_store_error(self, e):
"""A store I/O failure (read-only home, disk full, ...) answers a JSON
500 instead of escaping the handler as a dropped connection."""
resp = {'error': 'test script store write failed: %s' % e}
self._send_json(resp, 500)
self._log_resp(resp)
def _keyset_guard(self, body, kic, kid):
"""The keyset-number checks shared by the RAM operations: the KIc/KID
numbers must agree (TS 102 225 A.2 - the card rejects a mismatch) and a
@@ -6711,6 +6748,15 @@ class PysimHandler(BaseHTTPRequestHandler):
resp = _scp81_script_state()
self._send_json(resp)
self._log_resp(resp)
elif self.path == '/api/test/scripts':
self._log_req()
store = self._test_scripts_store_or_503()
if store is None:
return
resp = store.info()
resp['scripts'] = store.list()
self._send_json(resp)
self._log_resp({'path': resp['path'], 'count': resp['count']})
elif self.path == '/api/presets':
self._log_req()
store = self._preset_store_or_503()
@@ -8149,6 +8195,86 @@ class PysimHandler(BaseHTTPRequestHandler):
resp = {'ok': True, 'seq': _BIP.seq}
self._send_json(resp)
self._log_resp(resp)
elif self.path == '/api/test/scripts':
body = self._read_body()
self._log_req(body)
store = self._test_scripts_store_or_503()
if store is None:
return
script = body.get('script') if isinstance(body.get('script'), dict) else body
try:
entry = store.add(script)
except test_scripts.TestScriptError as e:
resp = {'error': str(e)}
self._send_json(resp, 400)
self._log_resp(resp)
return
except OSError as e:
self._test_scripts_store_error(e)
return
resp = {'ok': True, 'script': entry}
self._send_json(resp)
self._log_resp({'ok': True, 'id': entry['id'], 'name': entry['name']})
elif self.path == '/api/test/scripts/update':
body = self._read_body()
self._log_req(body)
store = self._test_scripts_store_or_503()
if store is None:
return
fields = body.get('script') if isinstance(body.get('script'), dict) else body
try:
entry = store.update(body.get('id'), fields)
except test_scripts.TestScriptError as e:
resp = {'error': str(e)}
self._send_json(resp, 400)
self._log_resp(resp)
return
except OSError as e:
self._test_scripts_store_error(e)
return
if entry is None:
resp = {'error': 'unknown test script id'}
self._send_json(resp, 404)
self._log_resp(resp)
return
resp = {'ok': True, 'script': entry}
self._send_json(resp)
self._log_resp({'ok': True, 'id': entry['id'], 'name': entry['name']})
elif self.path == '/api/test/scripts/delete':
body = self._read_body()
self._log_req(body)
store = self._test_scripts_store_or_503()
if store is None:
return
try:
removed = store.remove(body.get('id'))
except OSError as e:
self._test_scripts_store_error(e)
return
resp = {'ok': True, 'removed': removed}
self._send_json(resp)
self._log_resp(resp)
elif self.path == '/api/test/scripts/import':
body = self._read_body()
self._log_req(body)
store = self._test_scripts_store_or_503()
if store is None:
return
items = body.get('scripts')
if not isinstance(items, list) or len(items) > 1000:
resp = {'error': 'scripts must be a list of at most 1000 entries'}
self._send_json(resp, 400)
self._log_resp(resp)
return
mode = body.get('mode') if body.get('mode') in ('merge', 'replace') else 'merge'
try:
resp = store.import_scripts(items, mode)
except OSError as e:
self._test_scripts_store_error(e)
return
resp['ok'] = True
self._send_json(resp)
self._log_resp(resp)
elif self.path == '/api/test/run':
body = self._read_body()
self._log_req(body)
@@ -8157,8 +8283,14 @@ class PysimHandler(BaseHTTPRequestHandler):
self._send_json(resp, 409)
self._log_resp(resp)
return
script_raw, script_id, script_err = _test_run_script_from_body(self.server, body)
if script_err:
resp = {'error': script_err}
self._send_json(resp, 404)
self._log_resp(resp)
return
try:
script = testscript.normalise_script(body.get('script'), _test_command_type)
script = testscript.normalise_script(script_raw, _test_command_type)
except testscript.ScriptError as e:
resp = {'error': str(e)}
self._send_json(resp, 400)
@@ -8179,7 +8311,7 @@ class PysimHandler(BaseHTTPRequestHandler):
self._log_resp(resp)
return
try:
_test_run_start(self.server, script, preset)
_test_run_start(self.server, script, preset, script_id or None)
except Exception as e:
resp = {'error': 'could not start the test run: %s' % e}
self._send_json(resp, 500)
pysim_simple_server/test_scripts.py
+279
View File
@@ -0,0 +1,279 @@
"""Server-side test script store: the card test scripts the runner executes.
Test scripts used to live in the browser's localStorage, although the runner
that executes them is server-side - the store makes the bench state one unit
(next to the card presets), lets every browser (or a plain API client) manage
the same set, and validates a script before it is saved, so what is stored is
what the runner accepts.
A stored script is the script JSON (``name`` + ``steps``, the same shape the
PWA exports) plus a stable uuid and timestamps::
{"version": 1, "scripts": [
{"id": "...", "name": "applet RFM update", "steps": [...],
"created": 1690000000.0, "updated": 1690000000.0}]}
Validation runs on every mutation: the basic shape always, and - when the
server passes its ``validator`` - the full ``testscript.normalise_script``
pass (the same one the runner uses at start). Import accepts the old
localStorage shape (an array of ``{name, steps}``); there is no automatic
migration - export/import is the path.
Layout (``--test-scripts`` overrides the file; ``Path.home()`` resolves the
same way on Linux, macOS and Windows)::
~/.pysim-simple-server/test_scripts.json
Writes are atomic (temp file + fsync + ``os.replace``) and every mutation
holds an ``RLock``. Scripts carry a stable uuid - never array indices.
"""
import copy
import json
import os
import re
import sys
import tempfile
import threading
import time
import uuid
from pathlib import Path
DEFAULT_DIR = '.pysim-simple-server'
DEFAULT_FILENAME = 'test_scripts.json'
SCHEMA_VERSION = 1
def default_path():
"""The default store: ``~/.pysim-simple-server/test_scripts.json``."""
return Path.home() / DEFAULT_DIR / DEFAULT_FILENAME
class TestScriptError(Exception):
"""A refused mutation (validation, unknown id, ...)."""
class TestScriptStore:
"""The test script store. All public methods are thread-safe and return
copies - callers can never mutate the store's own state."""
def __init__(self, path=None, validator=None):
self.path = Path(path) if path else default_path()
self._validator = validator
self._lock = threading.RLock()
self._scripts = []
self._load()
# ------------------------------------------------------------------ disk
def _load(self):
try:
raw = self.path.read_text(encoding='utf-8')
except FileNotFoundError:
self._scripts = []
return
except OSError as e:
sys.stderr.write('TESTSCRIPTS: cannot read %s: %s\n' % (self.path, e))
self._scripts = []
return
try:
data = json.loads(raw)
except ValueError as e:
# Never silently overwrite an unreadable store: keep the file for
# the user, start empty (the next save would destroy it).
bad = self.path.with_name(self.path.name + '.bad-%s'
% time.strftime('%Y%m%d-%H%M%S'))
try:
self.path.replace(bad)
sys.stderr.write('TESTSCRIPTS: %s is not valid JSON (%s) - kept as '
'%s, starting empty\n' % (self.path, e, bad))
except OSError:
sys.stderr.write('TESTSCRIPTS: %s is not valid JSON (%s)\n'
% (self.path, e))
self._scripts = []
return
items = data.get('scripts') if isinstance(data, dict) else None
if not isinstance(items, list):
items = []
loaded = []
for entry in items:
try:
loaded.append(self._normalise(entry, keep_meta=True))
except TestScriptError as e:
sys.stderr.write('TESTSCRIPTS: dropping an invalid entry: %s\n' % e)
self._scripts = loaded
def _save(self):
data = {'version': SCHEMA_VERSION, 'scripts': self._scripts}
self.path.parent.mkdir(parents=True, exist_ok=True)
tmp_fd, tmp_name = tempfile.mkstemp(prefix=self.path.name + '.',
dir=str(self.path.parent))
try:
with os.fdopen(tmp_fd, 'w', encoding='utf-8') as fh:
json.dump(data, fh, indent=2, ensure_ascii=False)
fh.write('\n')
fh.flush()
os.fsync(fh.fileno())
os.replace(tmp_name, self.path)
except Exception:
try:
os.unlink(tmp_name)
except OSError:
pass
raise
try:
# Make the rename durable. Best effort: opening a directory is not
# portable (Windows) - a failure here never fails the mutation.
dir_fd = os.open(str(self.path.parent), os.O_RDONLY)
try:
os.fsync(dir_fd)
finally:
os.close(dir_fd)
except OSError:
pass
# ------------------------------------------------------------- shaping
def _check(self, raw):
"""The script dict, validated: the basic shape always, the full engine
pass when a validator was provided. The *input* shape is kept - the
engine's normalised output is not meant to be re-normalised (it fills
defaults like an 'any' PoR check), and the runner normalises again at
start, so storing what was submitted keeps the store re-validatable and
hand-editable."""
if not isinstance(raw, dict):
raise TestScriptError('script must be an object')
name = raw.get('name')
if name is None:
name = ''
if not isinstance(name, str):
raise TestScriptError('script name must be a string')
steps = raw.get('steps')
if not isinstance(steps, list) or not steps:
raise TestScriptError('script must have at least one step')
for i, step in enumerate(steps):
if not isinstance(step, dict) or step.get('type') not in ('action', 'expect'):
raise TestScriptError('step %d must be an action or an expectation'
% (i + 1))
script = {'name': name.strip(), 'steps': copy.deepcopy(steps)}
if self._validator is not None:
try:
self._validator(copy.deepcopy(script))
except Exception as e:
# testscript.ScriptError is a ValueError: pass its message on
raise TestScriptError(str(e))
return script
def _normalise(self, entry, keep_meta=False):
"""A stored entry: the validated script plus id and timestamps. With
``keep_meta`` the stored id/timestamps are preserved (a load), else
fresh ones are made (an add)."""
if not isinstance(entry, dict):
raise TestScriptError('script must be an object')
script = self._check(entry)
now = time.time()
rid = str(entry.get('id') or '').strip().lower()
if not re.fullmatch(r'[0-9a-f]{32}', rid):
rid = uuid.uuid4().hex
if keep_meta:
created = entry.get('created')
updated = entry.get('updated')
created = created if isinstance(created, (int, float)) else now
updated = updated if isinstance(updated, (int, float)) else created
else:
created = updated = now
return {'id': rid, 'name': script['name'], 'steps': script['steps'],
'created': created, 'updated': updated}
def _find(self, sid):
sid = str(sid or '').strip().lower()
for s in self._scripts:
if s['id'] == sid:
return s
return None
# -------------------------------------------------------------- public
def list(self):
with self._lock:
return [copy.deepcopy(s) for s in self._scripts]
def get(self, sid):
with self._lock:
s = self._find(sid)
return copy.deepcopy(s) if s else None
def add(self, script):
with self._lock:
s = self._normalise(script or {})
if self._find(s['id']):
# a caller-supplied id that is already taken never wins: ids
# must stay unique (the same guard the import uses)
s['id'] = uuid.uuid4().hex
self._scripts.append(s)
self._save()
sys.stderr.write('TESTSCRIPTS: added %s (%s)\n'
% (s['name'] or '(unnamed)', s['id']))
return copy.deepcopy(s)
def update(self, sid, fields):
"""Partial or full update: name/steps replaced, id and created kept."""
with self._lock:
cur = self._find(sid)
if cur is None:
return None
merged = {'name': cur['name'], 'steps': copy.deepcopy(cur['steps'])}
if isinstance(fields, dict):
if 'name' in fields:
merged['name'] = fields['name']
if 'steps' in fields:
merged['steps'] = fields['steps']
script = self._check(merged)
new = {'id': cur['id'], 'name': script['name'],
'steps': script['steps'], 'created': cur['created'],
'updated': time.time()}
self._scripts[self._scripts.index(cur)] = new
self._save()
return copy.deepcopy(new)
def remove(self, sid):
with self._lock:
s = self._find(sid)
if s is None:
return False
self._scripts.remove(s)
self._save()
sys.stderr.write('TESTSCRIPTS: removed %s (%s)\n'
% (s['name'] or '(unnamed)', s['id']))
return True
def import_scripts(self, items, mode='merge'):
"""Import exported (or old localStorage) scripts: entries are validated
one by one; an already-used id gets a fresh one, invalid entries are
reported instead of failing the whole import. ``replace`` wipes the
store first."""
added, skipped, errors = 0, 0, []
with self._lock:
if mode == 'replace':
self._scripts = []
for i, entry in enumerate(items or []):
if not isinstance(entry, dict):
skipped += 1
continue
try:
s = self._normalise(entry)
if self._find(s['id']):
s['id'] = uuid.uuid4().hex
except TestScriptError as e:
skipped += 1
errors.append('%s: %s' % (entry.get('name') or ('#' + str(i + 1)), e))
continue
self._scripts.append(s)
added += 1
if added or mode == 'replace':
self._save()
sys.stderr.write('TESTSCRIPTS: import %s: %d added, %d skipped\n'
% (mode, added, skipped))
return {'added': added, 'skipped': skipped, 'errors': errors[:10]}
def info(self):
with self._lock:
return {'path': str(self.path), 'count': len(self._scripts),
'version': SCHEMA_VERSION}