v3.18.0: the semantic event action for test scripts
The other-applet batch: a test script can send the common event downloads semantically instead of raw hex. - events.py: server-side mirrors of the PWA's event forms for Location status (0x03), Access technology change (0x0B), Network rejection (0x12 - the LA/RA/TA object selected by the registration group, the extended EMM cause and the extended information object) and Data connection status change (0x1D - status/type/TI plus the optional cause, host-clock 8.39 date-time, location information, access technology, APN and PDP/PDN/PDU type). Fields are validated before the run (a bad value is a script error, not a step failure); `src` overrides the device-identities source (default terminal, matching the Phone forms). - The engine gains the `event` action kind; the runner builds and sends it. The PWA's action-kind list stays in sync (drift-guarded - the guard itself caught its own regex bug on the first run). - Byte parity: the vectors are generated from the PWA's own builders and pinned in tests/test_events.py; the frontend suites assert the same values. PLMN and SCTS helpers are mirrored too. - The Phone tab's event form shows the built "Event data (hex)" (updated as the fields change), so every event without a builder can be used through the raw `envelope` action. - The test-step editor offers the event select + a JSON fields textarea; malformed JSON refuses the save with a clear message. - docs/test-scripts.md (the semantic-event table, an event example, the format stamp), docs/api.md and help EN/RU updated. Tests: 812 frontend + 717 Python; sw cache simple-v381.
This commit is contained in:
@@ -0,0 +1,140 @@
|
||||
"""Server-side event-data builders (the semantic test-script `event` action).
|
||||
|
||||
The byte vectors are generated from the PWA's own `EVENT_FORMS` builders (the
|
||||
Node generator run over `frontend/index.html`) and pinned here, so the two
|
||||
implementations cannot drift apart: every vector below is asserted by the
|
||||
frontend suite too (or was produced by the very code it asserts).
|
||||
"""
|
||||
|
||||
import time
|
||||
import unittest
|
||||
|
||||
from pysim_simple_server import events
|
||||
|
||||
|
||||
def st(year=2026, mon=9, day=30, hour=21, minute=55, sec=0):
|
||||
return time.struct_time((year, mon, day, hour, minute, sec, 0, 0, -1))
|
||||
|
||||
|
||||
class EventVectorTests(unittest.TestCase):
|
||||
def test_location_status(self):
|
||||
cases = [
|
||||
({'status': '0', 'mcc': '250', 'mnc': '01', 'lac': '00FF', 'cell': '0001'},
|
||||
'9B0100930752F01000FF0001'),
|
||||
({'status': '1', 'mcc': '250'}, '9B0101'),
|
||||
({'status': '0'}, '9B0100'),
|
||||
({'status': '0', 'mcc': '310', 'mnc': '410', 'lac': 'ABCD', 'cell': '1234'},
|
||||
'9B01009307130014ABCD1234'),
|
||||
]
|
||||
for fields, want in cases:
|
||||
self.assertEqual(events.build(0x03, fields).upper(), want)
|
||||
|
||||
def test_access_tech(self):
|
||||
for fields, want in [({'tech': '0'}, 'BF0100'), ({'tech': '8'}, 'BF0108'),
|
||||
({'tech': '12'}, 'BF010C')]:
|
||||
self.assertEqual(events.build(0x0B, fields).upper(), want)
|
||||
|
||||
def test_network_rejection(self):
|
||||
cases = [
|
||||
# location updating -> LAI, defaults filled
|
||||
({'reg_type': '0'}, 'F40100BF0100F50102930552F0100000'),
|
||||
# GPRS -> RAI
|
||||
({'reg_type': '3', 'mcc': '262', 'mnc': '02', 'lac': '00FF',
|
||||
'rac': '05', 'access_tech': '3', 'cause': '15'},
|
||||
'F40103BF0103F5010FF30662F22000FF05'),
|
||||
# EPS -> TAI (6-hex TAC) + the extended EMM cause
|
||||
({'reg_type': '9', 'mcc': '310', 'mnc': '260', 'tac': 'A1B2C3',
|
||||
'access_tech': '8', 'cause': '7', 'ext_cause': '22'},
|
||||
'F40109BF0108F501077D06130062A1B2C3570116'),
|
||||
# 5GS -> TAI (4-hex TAC)
|
||||
({'reg_type': '17', 'tac': '0001', 'access_tech': '10', 'cause': '2'},
|
||||
'F40111BF010AF501027D0552F0100001'),
|
||||
# the extended information object (CAG ID)
|
||||
({'reg_type': '9', 'access_tech': '8', 'cause': '2',
|
||||
'ext_info_type': '1', 'ext_info': '00000001'},
|
||||
'F40109BF0108F501027D0552F0100001F2050100000001'),
|
||||
]
|
||||
for fields, want in cases:
|
||||
self.assertEqual(events.build(0x12, fields).upper(), want)
|
||||
|
||||
def test_data_connection(self):
|
||||
cases = [
|
||||
({'status': '0', 'type': '0', 'ti': '00', 'loc_status': '0'},
|
||||
'9D0100AA01001C01009B0100'),
|
||||
({'status': '1', 'type': '2', 'cause': '26', 'ti': '85', 'mcc': '250',
|
||||
'mnc': '01', 'lac': '00FF', 'cell': '0001', 'tech': '8',
|
||||
'loc_status': '1', 'apn': 'internet', 'pdp_type': '3'},
|
||||
'9D0101AA0102AE011A1C0185130752F01000FF0001BF01089B0101'
|
||||
'C708696E7465726E65740B0103'),
|
||||
({'status': '2', 'type': '1', 'ti': '', 'mcc': '250', 'loc_status': '2'},
|
||||
'9D0102AA01011C0100130752F010000000019B0102'),
|
||||
({'status': '0', 'type': '2', 'ti': 'AA', 'mcc': '262', 'mnc': '01',
|
||||
'lac': '0001', 'cell': '0002', 'tech': '7', 'loc_status': '1',
|
||||
'apn': 'web.operator.example', 'pdp_type': '4'},
|
||||
'9D0100AA01021C01AA130762F21000010002BF01079B0101'
|
||||
'C7147765622E6F70657261746F722E6578616D706C650B0104'),
|
||||
]
|
||||
for fields, want in cases:
|
||||
self.assertEqual(events.build(0x1D, fields).upper(), want)
|
||||
|
||||
def test_the_host_clock_date_time_object(self):
|
||||
# the PWA encoder's vector: UTC+3, 2026-09-30 21:55:00
|
||||
self.assertEqual(events.scts_bytes(st(), 180).hex().upper(), '62900312550021')
|
||||
# UTC-2:30: 10 quarters with the sign bit
|
||||
self.assertEqual(events.scts_bytes(st(2026, 1, 5, 8, 7, 9), -150).hex().upper(),
|
||||
'62105080709009')
|
||||
# the data-connection builder embeds it as the 8.39 object 26 07 ...
|
||||
out = events.build(0x1D, {'status': '0', 'type': '0', 'ti': '00',
|
||||
'loc_status': '0', 'datetime': 'now'},
|
||||
now=st()).upper()
|
||||
self.assertRegex(out, r'^9D0100AA01001C01002607[0-9A-F]{14}9B0100$')
|
||||
|
||||
def test_plmn_encoding(self):
|
||||
self.assertEqual(events.plmn_hex('250', '01'), '52F010')
|
||||
self.assertEqual(events.plmn_hex('310', '410'), '130014')
|
||||
self.assertEqual(events.plmn_hex('262', '02'), '62F220')
|
||||
self.assertEqual(events.plmn_hex('310', '260'), '130062')
|
||||
|
||||
|
||||
class EventValidationTests(unittest.TestCase):
|
||||
def test_resolve_event_accepts_names_and_hex(self):
|
||||
self.assertEqual(events.resolve_event('location status'), 0x03)
|
||||
self.assertEqual(events.resolve_event('Location_Status'), 0x03)
|
||||
self.assertEqual(events.resolve_event('0B'), 0x0B)
|
||||
self.assertEqual(events.resolve_event('0x12'), 0x12)
|
||||
self.assertEqual(events.resolve_event(0x1D), 0x1D)
|
||||
self.assertIsNone(events.resolve_event('nope'))
|
||||
self.assertIsNone(events.resolve_event(300))
|
||||
self.assertIsNone(events.resolve_event(True))
|
||||
self.assertIsNone(events.resolve_event(''))
|
||||
|
||||
def test_field_validation(self):
|
||||
with self.assertRaises(events.EventError):
|
||||
events.build(0x03, {'status': '9'})
|
||||
with self.assertRaises(events.EventError):
|
||||
events.build(0x03, {'status': '0', 'mcc': '25'})
|
||||
with self.assertRaises(events.EventError):
|
||||
events.build(0x0B, {'tech': '300'})
|
||||
with self.assertRaises(events.EventError):
|
||||
events.build(0x12, {'reg_type': '0x12'})
|
||||
with self.assertRaises(events.EventError):
|
||||
events.build(0x12, {'reg_type': '9', 'tac': 'ABC'})
|
||||
with self.assertRaises(events.EventError):
|
||||
events.build(0x12, {'reg_type': '9', 'ext_info_type': '1', 'ext_info': 'A'})
|
||||
with self.assertRaises(events.EventError):
|
||||
events.build(0x1D, {'status': '0', 'apn': 'x' * 101})
|
||||
with self.assertRaises(events.EventError):
|
||||
events.build(0x1D, {'status': '0', 'ti': 'ZZ'})
|
||||
# an event without a semantic builder is refused with a hint
|
||||
with self.assertRaises(events.EventError) as ctx:
|
||||
events.normalise(0x05, {})
|
||||
self.assertIn('envelope', str(ctx.exception))
|
||||
|
||||
def test_normalise_returns_the_validated_fields(self):
|
||||
fields = {'status': '1', 'mcc': '250'}
|
||||
self.assertEqual(events.normalise(0x03, fields), fields)
|
||||
self.assertEqual(events.normalise(0x0B, None), {})
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -981,6 +981,19 @@ class PresetStoreHttpTests(unittest.TestCase):
|
||||
self.assertEqual(status, 503, resp)
|
||||
|
||||
|
||||
class TestActionKindsDriftTests(unittest.TestCase):
|
||||
"""The PWA's test-script action kinds must match the engine's (a missing
|
||||
kind in either side breaks scripts silently)."""
|
||||
|
||||
def test_the_pwa_and_the_server_share_the_action_kinds(self):
|
||||
from pysim_simple_server import testscript
|
||||
html = (ROOT / 'frontend' / 'index.html').read_text(encoding='utf-8')
|
||||
m = re.search(r"const TEST_ACTION_KINDS = \[(.*?)\];", html)
|
||||
self.assertIsNotNone(m, 'TEST_ACTION_KINDS not found in the PWA')
|
||||
pwa = re.findall(r"'([a-z0-9-]+)'", m.group(1))
|
||||
self.assertEqual(pwa, list(testscript.ACTION_KINDS))
|
||||
|
||||
|
||||
class TestScriptStoreHttpTests(unittest.TestCase):
|
||||
"""The test script endpoints (v3.16.0): the PWA's Test script pill talks
|
||||
to the server store instead of localStorage, and a run may name a stored
|
||||
|
||||
@@ -140,6 +140,34 @@ class TestValidation(unittest.TestCase):
|
||||
T.normalise_script({'steps': [{'type': 'action', 'kind': 'menu-select',
|
||||
'params': {'text': 'x', 'mode': 'nope'}}]}, _resolver)
|
||||
|
||||
def test_event_action_normalises_the_semantic_fields(self):
|
||||
script = T.normalise_script({'steps': [
|
||||
{'type': 'action', 'kind': 'event',
|
||||
'params': {'event': 'location status',
|
||||
'fields': {'status': 1, 'mcc': '250'}}},
|
||||
{'type': 'action', 'kind': 'event',
|
||||
'params': {'event': '0x1D', 'src': '83',
|
||||
'fields': {'status': '0', 'type': '0', 'ti': '00',
|
||||
'loc_status': '0'}}},
|
||||
]}, _resolver)
|
||||
p = script['steps'][0]['params']
|
||||
self.assertEqual(p['event'], 0x03)
|
||||
self.assertEqual(p['fields'], {'status': 1, 'mcc': '250'})
|
||||
self.assertNotIn('src', p)
|
||||
p = script['steps'][1]['params']
|
||||
self.assertEqual(p['event'], 0x1D)
|
||||
self.assertEqual(p['src'], '83')
|
||||
# an unknown name, a bad field and a bad source are refused before the run
|
||||
with self.assertRaises(T.ScriptError):
|
||||
T.normalise_script({'steps': [{'type': 'action', 'kind': 'event',
|
||||
'params': {'event': 'nope'}}]}, _resolver)
|
||||
with self.assertRaises(T.ScriptError):
|
||||
T.normalise_script({'steps': [{'type': 'action', 'kind': 'event',
|
||||
'params': {'event': '0x03', 'fields': {'status': 9}}}]}, _resolver)
|
||||
with self.assertRaises(T.ScriptError):
|
||||
T.normalise_script({'steps': [{'type': 'action', 'kind': 'event',
|
||||
'params': {'event': 3, 'src': 'zz'}}]}, _resolver)
|
||||
|
||||
def test_scp80_keyset_number_and_new_content_checks(self):
|
||||
# kvn (1..15) selects the preset's keyset; `por`/`files` are expect
|
||||
# content checks (the PoR contents / the REFRESH file list).
|
||||
@@ -482,6 +510,30 @@ class TestRunnerActions(RunnerTestCase):
|
||||
self.assertIn('2 items match', note)
|
||||
self.assertIn('use the item id', note)
|
||||
|
||||
def test_event_action_sends_the_built_object(self):
|
||||
scc = FakeScc()
|
||||
scc.push('80C2', '', '9000')
|
||||
run = self.run_script(FakeServer(scc), [
|
||||
{'type': 'action', 'kind': 'event',
|
||||
'params': {'event': 'location status',
|
||||
'fields': {'status': '0', 'mcc': '250', 'mnc': '01',
|
||||
'lac': '00FF', 'cell': '0001'}}},
|
||||
])
|
||||
self.assertEqual(run['status'], 'ok', run['steps'])
|
||||
apdu = scc.sent[0].upper()
|
||||
# event list + device identities (terminal default) + the built object
|
||||
self.assertIn('99010382028281' + '9B0100930752F01000FF0001', apdu)
|
||||
# the src override names the network as the source
|
||||
scc2 = FakeScc()
|
||||
scc2.push('80C2', '', '9000')
|
||||
run = self.run_script(FakeServer(scc2), [
|
||||
{'type': 'action', 'kind': 'event',
|
||||
'params': {'event': '0x0B', 'fields': {'tech': '8'}, 'src': '83'}},
|
||||
])
|
||||
self.assertEqual(run['status'], 'ok', run['steps'])
|
||||
apdu = scc2.sent[0].upper()
|
||||
self.assertIn('99010B82028381' + 'BF0108', apdu)
|
||||
|
||||
def test_scp80_uses_the_preset_and_advances_the_counter(self):
|
||||
scc = FakeScc()
|
||||
server = FakeServer(scc)
|
||||
|
||||
Reference in New Issue
Block a user