catarrh/simple
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
81b50199cc0a4f30fcf00dc0e10f7bae2970cb6e
simple/pysim_simple_server/netsim.py
T
catarrh 99183acc89 net-sim: skip files missing on the card; /api/write accepts a full path 
Live-testing against a UICC without the EPS files (no USIM service 85)
and without the Kc files showed the runner aborting when EPSLOCI/Kc were
absent.  write_binary/write_record now take an 'optional' flag: a missing
candidate file is logged as a skip step instead of failing the scenario
(store_epsnsc, real/dummy locations, churn and Kc writes).  Also allow
POST /api/write to select by 'path' like /api/select and /api/read (the
netsim live test needed it to restore the captured file values).
2026-09-19 22:53:24 +03:00

692 lines
24 KiB
Python
Raw Blame History

# coding=utf-8
"""Network-condition simulation for the SIMple lab.
Replays the card-facing write patterns of a real phone when the network
condition changes, following the trace study in ``projects/UICC_NAA.md``
(section 13): EPS attach, service loss / limited service, roaming denial,
2G fallback, SMS delivery, cell-broadcast reconfiguration and an
AUTHENTICATE exchange.
Only the observed vocabulary is used: UPDATE BINARY (D6), UPDATE RECORD (DC),
ENVELOPE (Event Download) and AUTHENTICATE. FPLMN and the 5GS location files
are never written (no observed phone does - UICC_NAA.md section 12).
The ``build_*`` functions are pure and unit-tested (tests/test_netsim.py).
``NetSimRunner`` applies a scenario to a live card through the helpers of
``pysim_simple_server.server``; the caller holds the card lock.
"""
import random
import re
import threading
import time
# Candidate paths per logical file. The first existing one is used, so a
# USIM card is served from ADF.USIM and a GSM SIM from DF.GSM/DF.TELECOM.
FILE_PATHS = {
'epsnsc': ['ADF.USIM/6FE4'],
'loci': ['ADF.USIM/6F7E', 'DF.GSM/6F7E'],
'psloci': ['ADF.USIM/6F73', 'DF.GSM/6F73'],
'epsloci': ['ADF.USIM/6FE3'],
'kc': ['ADF.USIM/4F20', 'DF.GSM/6F08'],
'kcgprs': ['ADF.USIM/4F52', 'DF.GSM/6F09'],
'smsstatus': ['ADF.USIM/6F43', 'DF.TELECOM/6F43', 'DF.GSM/6F43'],
'cbmi': ['ADF.USIM/6F45', 'DF.GSM/6F45'],
'cbmir': ['ADF.USIM/6F50', 'DF.GSM/6F50'],
}
# File status codes (LOCI/PSLOCI/EPSLOCI update status).
ST_UPDATED = 0x00
ST_NOT_UPDATED = 0x01
ST_PLMN_NOT_ALLOWED = 0x02
ST_LA_NOT_ALLOWED = 0x03
# Location status event (TS 102 223 section 8.27).
LOC_STATUS_NORMAL = 0x00
LOC_STATUS_LIMITED = 0x01
LOC_STATUS_NO_SERVICE = 0x02
EVENT_LOCATION_STATUS = 0x03
def _norm_hex(value, nbytes=None):
"""Normalize a hex string, optionally checking the byte length."""
h = re.sub(r'[^0-9a-fA-F]', '', value or '').upper()
if nbytes is not None and len(h) != nbytes * 2:
raise ValueError('expected %d hex bytes, got %r' % (nbytes, value))
return h
def _pad_ff(data, size):
"""Pad (or truncate) a byte string to `size` with 0xFF."""
if size is None or len(data) >= size:
return data[:size] if size else data
return data + b'\xFF' * (size - len(data))
def rand_hex(nbytes, rng=None):
r = rng or random
return ''.join(r.choice('0123456789ABCDEF') for _ in range(nbytes * 2))
def plmn_bcd(mcc, mnc):
"""MCC/MNC digits -> 3-byte PLMN (TS 24.008 10.5.1.13).
Nibble order: byte0 = MCC2 MCC1, byte1 = MNC3 MCC3, byte2 = MNC2 MNC1
(MNC3 = 'F' for two-digit MNCs). Cross-checked against the pySim
PLMNsel test vector 228/06 -> 22 F8 60.
"""
mcc = re.sub(r'\D', '', str(mcc or ''))
mnc = re.sub(r'\D', '', str(mnc or ''))
if len(mcc) != 3:
raise ValueError('MCC must have 3 digits, got %r' % mcc)
if len(mnc) not in (2, 3):
raise ValueError('MNC must have 2 or 3 digits, got %r' % mnc)
m = [int(c) for c in mcc]
n = [int(c) for c in mnc] + ([0xF] if len(mnc) == 2 else [])
return '%02X%02X%02X' % ((m[1] << 4) | m[0], (n[2] << 4) | m[2], (n[1] << 4) | n[0])
def _plmn_bytes(plmn_hex):
return bytes.fromhex(_norm_hex(plmn_hex, 3))
# ---- EPS NAS Security Context (EF.EPSNSC, TS 31.102 4.2.92) ----
def build_epsnsc(ksi, kasme_hex, ul, dl, algo, size=54):
"""One A0 TLV record, padded to the card's record size with FF."""
kasme = bytes.fromhex(_norm_hex(kasme_hex, 32))
inner = (bytes([0x80, 0x01, ksi & 0xFF])
+ bytes([0x81, 0x20]) + kasme
+ bytes([0x82, 0x04]) + (ul & 0xFFFFFFFF).to_bytes(4, 'big')
+ bytes([0x83, 0x04]) + (dl & 0xFFFFFFFF).to_bytes(4, 'big')
+ bytes([0x84, 0x01, algo & 0xFF]))
rec = bytes([0xA0, len(inner)]) + inner
return _pad_ff(rec, size).hex().upper()
def build_epsnsc_invalidate(size=54, kasme_hex=None, algo=0x00):
"""Invalid context: KSI 07, optional old KASME kept (UICC_NAA.md 5.2)."""
kasme = (bytes.fromhex(_norm_hex(kasme_hex, 32)) if kasme_hex
else b'\xFF' * 32)
inner = (bytes([0x80, 0x01, 0x07])
+ bytes([0x81, 0x20]) + kasme
+ bytes([0x82, 0x04]) + b'\xFF' * 4
+ bytes([0x83, 0x04]) + b'\xFF' * 4
+ bytes([0x84, 0x01, algo & 0xFF]))
rec = bytes([0xA0, len(inner)]) + inner
return _pad_ff(rec, size).hex().upper()
def parse_epsnsc_kasme(record_hex):
"""Extract the 32-byte KASME from an EPSNSC record, or None."""
try:
data = bytes.fromhex(_norm_hex(record_hex))
except ValueError:
return None
# The record is one A0 container; KASME is the 81 TLV inside it.
if len(data) >= 2 and data[0] == 0xA0:
data = data[2:2 + data[1]]
i = 0
while i + 2 <= len(data):
tag, ln = data[i], data[i + 1]
value = data[i + 2:i + 2 + ln]
if tag == 0x81 and ln == 32:
if all(b == 0xFF for b in value):
return None # wiped key = invalid context
return value.hex().upper()
i += 2 + ln
return None
# ---- Location files (TS 31.102 4.2.16 / 4.2.23 / 4.2.91) ----
def build_loci(tmsi_hex, plmn_hex, lac_hex, status=ST_UPDATED, rfu=0xFF):
"""LOCI: TMSI(4) + LAI(5) + RFU(1) + update status(1) = 11 bytes."""
return (bytes.fromhex(_norm_hex(tmsi_hex, 4)) + _plmn_bytes(plmn_hex)
+ bytes.fromhex(_norm_hex(lac_hex, 2))
+ bytes([rfu & 0xFF, status & 0xFF])).hex().upper()
def build_loci_dummy(plmn_hex):
"""Service lost: TMSI FF, PLMN kept, LAC FFFE, status 01."""
return build_loci('FFFFFFFF', plmn_hex, 'FFFE', ST_NOT_UPDATED)
def build_psloci(ptmsi_hex, sig_hex, plmn_hex, lac_hex, rac_hex,
status=ST_UPDATED):
"""PSLOCI: P-TMSI(4) + signature(3) + RAI(6) + status(1) = 14 bytes."""
return (bytes.fromhex(_norm_hex(ptmsi_hex, 4))
+ bytes.fromhex(_norm_hex(sig_hex, 3)) + _plmn_bytes(plmn_hex)
+ bytes.fromhex(_norm_hex(lac_hex, 2))
+ bytes.fromhex(_norm_hex(rac_hex, 1))
+ bytes([status & 0xFF])).hex().upper()
def build_psloci_dummy(plmn_hex):
return build_psloci('FFFFFFFF', 'FFFFFF', plmn_hex, 'FFFE', 'FF',
ST_NOT_UPDATED)
def build_epsloci(guti_hex, plmn_hex, tac_hex, status=ST_UPDATED):
"""EPSLOCI: GUTI(12) + TAI(5) + EPS update status(1) = 18 bytes."""
return (bytes.fromhex(_norm_hex(guti_hex, 12)) + _plmn_bytes(plmn_hex)
+ bytes.fromhex(_norm_hex(tac_hex, 2))
+ bytes([status & 0xFF])).hex().upper()
def build_epsloci_dummy(plmn_hex):
"""GUTI header 0B F6 + PLMN, identity wiped; TAI PLMN + FFFF; status 01."""
return build_epsloci('0BF6' + plmn_hex + 'FF' * 7, plmn_hex, 'FFFF',
ST_NOT_UPDATED)
# ---- Ciphering keys and CB/SMS files ----
def build_kc(kc_hex, algo, size):
"""Kc: 9-byte USIM form (Kc + algo) or the 33-byte GSM record form."""
kc = bytes.fromhex(_norm_hex(kc_hex, 8))
return _pad_ff(kc + bytes([algo & 0xFF]), size).hex().upper()
def build_kc_invalidate(size):
"""Invalid Kc: 1-byte 07, USIM 9-byte FF*8+07, GSM 33-byte 07+FF*32."""
if size <= 1:
return '07'
if size == 9:
return (b'\xFF' * 8 + b'\x07').hex().upper()
return (b'\x07' + b'\xFF' * (size - 1)).hex().upper()
def build_smsstatus(count):
"""EF.SMSstatus: 2-byte big-endian counter (TS 51.011 10.5.9)."""
return '%04X' % (count & 0xFFFF)
def parse_smsstatus(data_hex):
"""Current counter value, or None when unreadable/erased."""
try:
data = bytes.fromhex(_norm_hex(data_hex))
except ValueError:
return None
if len(data) < 2 or all(b == 0xFF for b in data):
return None
return int.from_bytes(data[:2], 'big')
def bump_smsstatus(count):
value = parse_smsstatus(count)
return build_smsstatus((value + 1) if value is not None else 0x01FF)
def build_cbmi(ids, size=None):
"""CBMI: 2-byte message IDs; pad to the file size with FF when known."""
out = b''.join((i & 0xFFFF).to_bytes(2, 'big') for i in ids)
return _pad_ff(out, size).hex().upper()
def build_cbmir(ranges, size=None):
"""CBMIR: 4-byte ranges (low, high), most-significant byte first."""
out = b''.join((lo & 0xFFFF).to_bytes(2, 'big')
+ (hi & 0xFFFF).to_bytes(2, 'big') for lo, hi in ranges)
return _pad_ff(out, size).hex().upper()
# ---- Event Download and AUTHENTICATE ----
def build_location_status_event(status, plmn_hex=None, lac_hex=None,
cell_id_hex=None):
"""Location status data object (TS 102 223 8.27), with the optional
Location information object (8.19) only for normal service."""
out = bytes([0x9B, 0x01, status & 0xFF])
if (status == LOC_STATUS_NORMAL and plmn_hex and lac_hex
and cell_id_hex):
info = (_plmn_bytes(plmn_hex) + bytes.fromhex(_norm_hex(lac_hex, 2))
+ bytes.fromhex(_norm_hex(cell_id_hex, 2)))
out += bytes([0x13, len(info)]) + info
return out.hex().upper()
def build_auth_apdu(rand_hex, autn_hex, cla='00'):
"""AUTHENTICATE (3G/EPS/5G, P2 81), TS 31.102 7.1.2.1.
Command data: L1 RAND L2 AUTN = 1+16+1+16 = 34 bytes (Lc 0x22).
"""
rand = bytes.fromhex(_norm_hex(rand_hex, 16))
autn = bytes.fromhex(_norm_hex(autn_hex, 16))
body = bytes([len(rand)]) + rand + bytes([len(autn)]) + autn
return '%s880081%02X%s' % (cla, len(body), body.hex().upper())
def parse_auth_response(data_hex):
"""Parse the AUTHENTICATE response: DB (success) or DC (sync failure)."""
try:
data = bytes.fromhex(_norm_hex(data_hex or ''))
except ValueError:
return None
if not data or data[0] not in (0xDB, 0xDC):
return None
out = {'type': 'success' if data[0] == 0xDB else 'synchronisation_failure'}
i = 1
while i + 2 <= len(data):
ln = data[i]
out.setdefault('objects', []).append(data[i + 1:i + 1 + ln].hex().upper())
i += 1 + ln
return out
# ---- scenario runner ----
class StepError(Exception):
"""A write/select step failed; the scenario stops at that step."""
def _default(value, rng=None):
return value if value not in (None, '') else None
def _hexint(value, default=0):
"""Accept '0x1f'/1f/'31'/'3' style parameters as an integer."""
if value in (None, ''):
return default
if isinstance(value, int):
return value
text = str(value).strip()
try:
return int(text, 16) if re.fullmatch(r'(0[xX])?[0-9a-fA-F]+', text) else int(text)
except ValueError:
return default
class NetSimRunner:
"""Apply one scenario to a live card. The caller holds _CARD_LOCK."""
def __init__(self, srv, app, params=None, event_list=None, sleep=time.sleep):
self.srv = srv
self.app = app
self.params = params or {}
self.event_list = list(event_list or [])
self.sleep = sleep
self.lchan = app.rs.lchan[0]
self.steps = []
self._cleanups = []
scc = getattr(getattr(srv, '_server_ref', None), 'scc', None)
self.scc = scc
# -- parameter helpers
def p(self, key, default=None):
v = self.params.get(key)
return default if v in (None, '') else v
@property
def mcc(self):
return self.p('mcc', '001')
@property
def mnc(self):
return self.p('mnc', '01')
@property
def plmn(self):
return self.p('plmn') or plmn_bcd(self.mcc, self.mnc)
@property
def lac(self):
return self.p('lac') or rand_hex(2)
@property
def cell_id(self):
return self.p('cell_id') or rand_hex(2)
@property
def tac(self):
return self.p('tac') or self.lac
@property
def rac(self):
return self.p('rac') or rand_hex(1)
@property
def kasme(self):
return self.p('kasme') or rand_hex(32)
def status(self, default):
v = self.p('location_status')
return int(v, 16) if isinstance(v, str) else (int(v) if v is not None else default)
# -- step plumbing
def _add(self, action, **kw):
step = {'action': action}
step.update(kw)
self.steps.append(step)
return step
def _cleanup(self):
for c in self._cleanups:
try:
c()
except Exception:
pass
self._cleanups = []
def select(self, path):
cur, cleanup = self.srv._select_path(self.lchan, path, self.app)
if cleanup:
self._cleanups.append(cleanup)
return cur
def _open(self, key):
last = None
for path in FILE_PATHS[key]:
try:
self.select(path)
return path
except Exception as e:
last = e
raise StepError('no candidate file for %s (%s)' % (key, last))
def _meta(self):
l = self.lchan
return {'file_size': l.selected_file_size(),
'record_len': l.selected_file_record_len(),
'num_of_rec': l.selected_file_num_of_rec()}
def _check(self, data, sw, **kw):
ok = sw == '9000'
step = self._add(**kw, data=data, sw=sw, ok=ok)
if not ok:
raise StepError('%s returned SW %s' % (kw.get('action'), sw))
return step
def write_binary(self, key, data_hex, pad=True, label=None, optional=False):
try:
path = self._open(key)
except StepError as e:
if optional:
self._add('skip', file=label or key, note=str(e))
return None
raise
size = self.lchan.selected_file_size()
data = data_hex
if pad and size:
data = _pad_ff(bytes.fromhex(data_hex), size).hex().upper()
_out, sw = self.lchan.update_binary(data)
return self._check(data, sw, action='update_binary', file=label or key,
path=path)
def write_record(self, key, data_hex, record=1, pad=True, label=None, optional=False):
try:
path = self._open(key)
except StepError as e:
if optional:
self._add('skip', file=label or key, note=str(e))
return None
raise
size = self.lchan.selected_file_record_len()
data = data_hex
if pad and size:
data = _pad_ff(bytes.fromhex(data_hex), size).hex().upper()
_out, sw = self.lchan.update_record(record, data)
return self._check(data, sw, action='update_record', file=label or key,
path=path, record=record)
def read_binary_current(self):
data, sw = self.lchan.read_binary()
return (data or ''), sw
def read_record_current(self, record=1):
data, sw = self.lchan.read_record(record)
return (data or ''), sw
def send_location_status(self, status):
if EVENT_LOCATION_STATUS not in self.event_list:
self._add('skip', file='location_status',
note='event 0x03 not in SET UP EVENT LIST')
return
data = build_location_status_event(
status,
self.plmn if status == LOC_STATUS_NORMAL else None,
self.lac if status == LOC_STATUS_NORMAL else None,
self.cell_id if status == LOC_STATUS_NORMAL else None)
if not self.scc:
raise StepError('card session not available')
_d, sw = self.srv._send_event_download(self.scc, EVENT_LOCATION_STATUS,
bytes.fromhex(data))
self._add('event', file='location_status', data=data, sw=sw,
ok=(sw == '9000'))
# -- scenario building blocks
def invalidate_epsnsc(self, keep_key=True):
try:
self._open('epsnsc')
except StepError as e:
self._add('skip', file='epsnsc', note=str(e))
return
size = self.lchan.selected_file_record_len() or 54
kasme = None
if keep_key:
data, sw = self.read_record_current(1)
if sw == '9000':
kasme = parse_epsnsc_kasme(data)
self.write_record('epsnsc', build_epsnsc_invalidate(size, kasme),
label='epsnsc')
def store_epsnsc(self):
self.write_record('epsnsc', build_epsnsc(
_hexint(self.p('ksi', '03'), 0x03),
self.kasme,
_hexint(self.p('ul', 0)), _hexint(self.p('dl', 0)),
_hexint(self.p('algo', '02'), 0x02)),
label='epsnsc', optional=True)
def write_real_locations(self, status=ST_UPDATED):
self.write_binary('loci', build_loci(
self.p('tmsi') or rand_hex(4), self.plmn, self.lac, status),
label='loci', optional=True)
self.write_binary('psloci', build_psloci(
self.p('ptmsi') or rand_hex(4), self.p('ptmsi_sig') or rand_hex(3),
self.plmn, self.lac, self.rac, status), label='psloci', optional=True)
self.write_binary('epsloci', build_epsloci(
self.p('guti') or rand_hex(12), self.plmn, self.tac, status),
label='epsloci', optional=True)
def write_dummy_locations(self, status=ST_NOT_UPDATED):
self.write_binary('loci', build_loci_dummy(self.plmn), label='loci',
optional=True)
self.write_binary('psloci', build_psloci_dummy(self.plmn), label='psloci',
optional=True)
self.write_binary('epsloci', build_epsloci_dummy(self.plmn),
label='epsloci', optional=True)
def invalidate_kc(self):
for key in ('kc', 'kcgprs'):
try:
self._open(key)
except StepError as e:
self._add('skip', file=key, note=str(e))
continue
size = self.lchan.selected_file_size() or 9
self.write_binary(key, build_kc_invalidate(size), pad=False,
label=key)
def write_real_kc(self):
kc = self.p('kc') or rand_hex(8)
algo = int(self.p('kc_algo', 1))
for key in ('kc', 'kcgprs'):
try:
self._open(key)
except StepError as e:
self._add('skip', file=key, note=str(e))
continue
size = self.lchan.selected_file_size() or 9
self.write_binary(key, build_kc(kc, algo, size), pad=False,
label=key)
def bump_sms_counter(self):
try:
self._open('smsstatus')
except StepError as e:
self._add('skip', file='smsstatus', note=str(e))
return
data, sw = self.read_binary_current()
if sw != '9000':
data = ''
self.write_binary('smsstatus', bump_smsstatus(data), label='smsstatus')
# -- scenarios (UICC_NAA.md section 13)
def sc_cold_boot(self):
self.invalidate_epsnsc(keep_key=False)
if self.p('dummy_locations', True):
self.write_dummy_locations()
def sc_attach_eps(self):
self.store_epsnsc()
self.write_real_locations(ST_UPDATED)
def sc_service_lost(self):
if self.p('send_event', True):
self.send_location_status(self.status(LOC_STATUS_NO_SERVICE))
if self.p('invalidate_epsnsc', True):
self.invalidate_epsnsc(keep_key=bool(self.p('keep_kasme', True)))
if self.p('dummy_locations', True):
self.write_dummy_locations()
if self.p('write_kc', True):
self.invalidate_kc()
def sc_limited_service(self):
if self.p('send_event', True):
self.send_location_status(self.status(LOC_STATUS_LIMITED))
if self.p('invalidate_epsnsc', True):
self.invalidate_epsnsc(keep_key=bool(self.p('keep_kasme', True)))
if self.p('dummy_locations', True):
self.write_dummy_locations()
def sc_roaming_denied(self):
if self.p('send_event', True):
self.send_location_status(self.status(LOC_STATUS_LIMITED))
if self.p('invalidate_epsnsc', True):
self.invalidate_epsnsc(keep_key=bool(self.p('keep_kasme', True)))
if self.p('dummy_locations', True):
self.write_dummy_locations(status=self.status(ST_PLMN_NOT_ALLOWED))
def sc_churn(self):
count = int(self.p('churn_count', 3))
delay = int(self.p('churn_delay_ms', 150)) / 1000.0
real = build_epsnsc(
_hexint(self.p('ksi', '01'), 0x01),
self.kasme,
_hexint(self.p('ul', 0)), _hexint(self.p('dl', 0)),
_hexint(self.p('algo', '02'), 0x02))
try:
self._open('epsnsc')
except StepError as e:
self._add('skip', file='epsnsc', note=str(e))
return
size = self.lchan.selected_file_record_len() or 54
invalid = build_epsnsc_invalidate(size, self.kasme)
for _i in range(max(1, count)):
self.write_record('epsnsc', real, label='epsnsc')
if delay:
self.sleep(delay)
self.write_record('epsnsc', invalid, label='epsnsc')
if delay:
self.sleep(delay)
def sc_attach_2g(self):
self.write_real_kc()
self.write_real_locations(ST_UPDATED)
def sc_sms_received(self):
self.bump_sms_counter()
if self.p('sms_location', True):
self.write_real_locations(ST_UPDATED)
def sc_cb_reconfig(self):
clear = bool(self.p('cb_clear', False))
try:
self._open('cbmi')
except StepError as e:
self._add('skip', file='cbmi', note=str(e))
clear = None
if clear is not None:
size = self.lchan.selected_file_size()
if clear:
self.write_binary('cbmi', 'FF' * (size or 20), pad=False,
label='cbmi')
else:
ids = self.p('cbmi_ids') or [0x111F, 0x1112]
self.write_binary('cbmi', build_cbmi(ids, size), pad=False,
label='cbmi')
try:
self._open('cbmir')
except StepError:
return
size = self.lchan.selected_file_size()
if clear:
self.write_binary('cbmir', 'FF' * (size or 40), pad=False,
label='cbmir')
else:
ranges = self.p('cbmir_ranges') or [[0x111F, 0x111F],
[0x1112, 0x1112]]
self.write_binary('cbmir', build_cbmir(ranges, size), pad=False,
label='cbmir')
def sc_authenticate(self):
rand = self.p('rand') or rand_hex(16)
autn = self.p('autn') or rand_hex(16)
if not self.scc:
raise StepError('card session not available')
apdu = build_auth_apdu(rand, autn)
data, sw = self.scc._tp.send_apdu(apdu)
if sw.startswith('61'):
data, sw = self.scc._tp.send_apdu('00C00000' + sw[2:4])
parsed = parse_auth_response(data)
self._add('authenticate', data=apdu, response=(data or '').upper(),
parsed=parsed, sw=sw, ok=True)
SCENARIOS = {
'cold_boot': sc_cold_boot,
'attach_eps': sc_attach_eps,
'service_lost': sc_service_lost,
'limited_service': sc_limited_service,
'roaming_denied': sc_roaming_denied,
'churn': sc_churn,
'attach_2g': sc_attach_2g,
'sms_received': sc_sms_received,
'cb_reconfig': sc_cb_reconfig,
'authenticate': sc_authenticate,
}
def run(self, scenario):
fn = self.SCENARIOS.get(scenario)
if not fn:
raise ValueError('unknown scenario %r' % scenario)
error = None
try:
fn(self)
except StepError as e:
error = str(e)
except Exception as e:
error = '%s: %s' % (type(e).__name__, e)
finally:
self._cleanup()
return {'scenario': scenario, 'success': error is None, 'error': error,
'steps': self.steps}
def run_scenario(srv, app, scenario, params=None, event_list=None,
sleep=time.sleep):
"""Run `scenario` against the equipped card; returns the step log."""
runner = NetSimRunner(srv, app, params=params, event_list=event_list,
sleep=sleep)
return runner.run(scenario)
Reference in New Issue View Git Blame Copy Permalink