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

v3.22.0: the test suites (scripts grouped, ordered, with setup/teardown)

Browse Source 
A suite is the root object: every script belongs to exactly one suite, the
suite holds the ordered member list by reference - the optional setup first,
the members, the optional teardown last - and runs them in one card session,
without resets and without resume.

- test_suites.py: the suite store (test_suites.json, --test-suites), validated
  and atomic like the script store: roles normalised (setup first, teardown
  last, at most one of each, a script appears once per suite), per-member
  on_fail stop/continue and the suite-level require_adm.
- test_scripts.py schema v2: suite_id/require_adm; list_by_suite()/orphans();
  the import reports the id_map a bundle import remaps with.  v3.21.0 scripts
  without a suite are attached once to an auto-created "Imported scripts"
  suite (migrate_scripts_to_suites, called from __main__).
- API: /api/test/suites (+ update|delete|import - delete refused while
  non-empty; the bundle import imports the scripts first, remaps the suite
  references and skips unresolvable ones); the scripts endpoints gain
  suite_id/require_adm plus move|copy (delete detaches from the suite first;
  a suite update refuses a membership change).
- runner: /api/test/run {suite_id} resolves the members and the preset, runs
  the ADM gate, then setup -> members (per-member on_fail) -> teardown.  The
  teardown always runs on a stop or a failure (skipped only when the card is
  gone; it clears the stop flag for its duration); a card reset stops the
  suite, skips the remaining members and is flagged.  Members' live steps land
  in the shared snapshot, their records (status/steps/note/log excerpt/
  counters) in _TEST_RUN['suite']; the summary carries the counts, wall time,
  preset, the SCP80 counters before/after per keyset and the session flags.
  _test_log buffers the semantic lines (bounded) for the report's per-member
  excerpts.
- ADM gate (_ensure_adm): a suite/script require_adm verifies the matched
  preset's key once per card session (the /api/verify-adm badge sets the same
  latch); a failure refuses the start, stamps the session (no automatic
  retry) and needs a manual verify.
- PWA: the Test script pill becomes four views - the suite list (default),
  the suite editor (the script list: role badges, on_fail, up/down, per-script
  Run/Edit/Move/Copy/Delete, Add script with the templates), the script editor
  (the step list; a single run shows its result under the steps and is not
  preserved) and the suite report (per-member blocks with the step table,
  notes and run log; summary; export JSON/Markdown).  Suite bundles export/
  import; the legacy script export imports into "Imported scripts".  EN/RU.
- tests: the suite store/migration, the runner orchestration (order, on_fail,
  setup failure, stop, card reset, teardown, per-member logs) and the ADM gate
  (Python); the suite HTTP paths; the PWA report/Markdown/wiring (Node).
  Tests: 833 frontend + 794 Python; sw cache simple-v397.
This commit is contained in:
Антон Трошин
2026-10-06 00:26:50 +03:00
parent e6ced47323
commit 5976ef89b4
17 changed files with 3178 additions and 288 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_fetch_routes.py
+228 -6
View File
@@ -13,6 +13,7 @@ import json
import pathlib
import re
import threading
import time
import unittest
import urllib.error
import urllib.request
@@ -1158,6 +1159,7 @@ class TestScriptStoreHttpTests(unittest.TestCase):
def setUp(self):
import tempfile
from pysim_simple_server import test_scripts
from pysim_simple_server import test_suites
from pysim_simple_server import testscript
from pysim_simple_server import server as srv
self.srv = srv
@@ -1165,15 +1167,24 @@ class TestScriptStoreHttpTests(unittest.TestCase):
self.store = test_scripts.TestScriptStore(
pathlib.Path(self.tmp.name) / 'test_scripts.json',
validator=lambda raw: testscript.normalise_script(raw, srv._test_command_type))
self.suites = test_suites.TestSuiteStore(
pathlib.Path(self.tmp.name) / 'test_suites.json')
self.server = _build_http_server('127.0.0.1', 0, PysimHandler)
self.server.log_requests = False
self.server.app = None
self.server.sl = None
self.server.scc = object()
self.server.test_scripts = self.store
self.server.test_suites = self.suites
threading.Thread(target=self.server.serve_forever, daemon=True).start()
self.port = self.server.server_address[1]
def _suite(self, **kw):
"""A stored suite (the script store's scripts belong to one)."""
s = {'name': 'suite'}
s.update(kw)
return self.suites.add(s)
def tearDown(self):
self.server.shutdown()
self.server.server_close()
@@ -1210,20 +1221,29 @@ class TestScriptStoreHttpTests(unittest.TestCase):
def test_the_get_route_is_card_free_and_reports_the_store(self):
self.assertIn('/api/test/scripts', _CARD_FREE_GET)
self.assertIn('/api/test/suites', _CARD_FREE_GET)
with _CARD_LOCK:
status, resp = self._get('/api/test/scripts')
self.assertEqual(status, 200, resp)
self.assertEqual(resp['scripts'], [])
self.assertEqual(resp['path'], str(self.store.path))
status, resp = self._get('/api/test/suites')
self.assertEqual(status, 200, resp)
self.assertEqual(resp['suites'], [])
self.assertEqual(resp['path'], str(self.suites.path))
def test_create_update_delete_round_trip(self):
status, resp = self._post('/api/test/scripts', self._script())
suite = self._suite()
status, resp = self._post('/api/test/scripts',
self._script(suite_id=suite['id']))
self.assertEqual(status, 200, resp)
sid = resp['script']['id']
self.assertTrue(sid)
self.assertEqual(self.suites.get(suite['id'])['scripts'],
[{'script_id': sid, 'role': 'member', 'on_fail': 'stop'}])
status, resp = self._get('/api/test/scripts')
self.assertEqual(len(resp['scripts']), 1)
self.assertEqual(resp['scripts'][0]['id'], sid)
self.assertEqual(resp['scripts'][0]['suite_id'], suite['id'])
status, resp = self._post('/api/test/scripts/update',
{'id': sid, 'script': {'name': 'renamed'}})
@@ -1238,16 +1258,130 @@ class TestScriptStoreHttpTests(unittest.TestCase):
self.assertEqual(status, 200, resp)
self.assertTrue(resp['removed'])
self.assertEqual(self.store.list(), [])
# the suite's member list no longer references it (no dangling id)
self.assertEqual(self.suites.get(suite['id'])['scripts'], [])
def test_create_requires_a_suite(self):
status, resp = self._post('/api/test/scripts', self._script())
self.assertEqual(status, 400, resp)
self.assertIn('belongs to a suite', resp['error'])
status, resp = self._post('/api/test/scripts',
self._script(suite_id='deadbeefdeadbeefdeadbeefdeadbeef'))
self.assertEqual(status, 400, resp)
self.assertIn('suite_id of an existing suite', resp['error'])
self.assertEqual(self.store.list(), [])
def test_create_validates_with_the_engine(self):
status, resp = self._post('/api/test/scripts', {'name': 'x', 'steps': []})
suite = self._suite()
sid = suite['id']
status, resp = self._post('/api/test/scripts',
{'name': 'x', 'suite_id': sid, 'steps': []})
self.assertEqual(status, 400, resp)
self.assertIn('at least one step', resp['error'])
status, resp = self._post('/api/test/scripts', {'name': 'x', 'steps': [
{'type': 'action', 'kind': 'nonsense', 'params': {}}]})
status, resp = self._post('/api/test/scripts',
{'name': 'x', 'suite_id': sid, 'steps': [
{'type': 'action', 'kind': 'nonsense', 'params': {}}]})
self.assertEqual(status, 400, resp)
self.assertIn('unknown action kind', resp['error'])
def test_suite_crud_and_the_delete_guard(self):
status, resp = self._get('/api/test/suites')
self.assertEqual(status, 200, resp)
self.assertEqual(resp['suites'], [])
status, resp = self._post('/api/test/suites', {})
self.assertEqual(status, 200, resp)
sid = resp['suite']['id']
status, resp = self._post('/api/test/suites/update',
{'id': sid, 'suite': {'name': 'renamed',
'require_adm': True}})
self.assertEqual(status, 200, resp)
self.assertEqual(resp['suite']['name'], 'renamed')
self.assertTrue(resp['suite']['require_adm'])
# a non-empty suite refuses deletion
status, resp = self._post('/api/test/scripts', self._script(suite_id=sid))
self.assertEqual(status, 200, resp)
status, resp = self._post('/api/test/suites/delete', {'id': sid})
self.assertEqual(status, 400, resp)
self.assertIn('still holds 1 script', resp['error'])
self._post('/api/test/scripts/delete', {'id': self.store.list()[0]['id']})
status, resp = self._post('/api/test/suites/delete', {'id': sid})
self.assertEqual(status, 200, resp)
self.assertTrue(resp['removed'])
def test_suite_update_refuses_a_membership_change(self):
a = self._suite(name='a')
b = self._suite(name='b')
status, resp = self._post('/api/test/scripts', self._script(suite_id=a['id']))
sid = resp['script']['id']
# the suite's own list may be reordered/retagged, but the member set is
# owned by the script endpoints
status, resp = self._post('/api/test/suites/update', {
'id': a['id'], 'suite': {'scripts': [
{'script_id': sid, 'role': 'setup', 'on_fail': 'stop'}]}})
self.assertEqual(status, 200, resp)
self.assertEqual(resp['suite']['scripts'][0]['role'], 'setup')
other = self._suite()['id']
status, resp = self._post('/api/test/scripts', self._script(suite_id=other))
sid2 = resp['script']['id']
status, resp = self._post('/api/test/suites/update', {
'id': a['id'], 'suite': {'scripts': [
{'script_id': sid, 'role': 'member', 'on_fail': 'stop'},
{'script_id': sid2, 'role': 'member', 'on_fail': 'stop'}]}})
self.assertEqual(status, 400, resp)
self.assertIn('member set changes through the script endpoints', resp['error'])
def test_move_and_copy_between_suites(self):
a = self._suite(name='a')
b = self._suite(name='b')
status, resp = self._post('/api/test/scripts', self._script(suite_id=a['id']))
sid = resp['script']['id']
# a move keeps the on_fail policy and updates both suites
self.suites.update(a['id'], {'scripts': [
{'script_id': sid, 'role': 'member', 'on_fail': 'continue'}]})
status, resp = self._post('/api/test/scripts/move',
{'id': sid, 'suite_id': b['id']})
self.assertEqual(status, 200, resp)
self.assertEqual(resp['script']['suite_id'], b['id'])
self.assertEqual(self.suites.get(a['id'])['scripts'], [])
self.assertEqual(self.suites.get(b['id'])['scripts'],
[{'script_id': sid, 'role': 'member', 'on_fail': 'continue'}])
# a copy lands in the target as a member with a fresh id
status, resp = self._post('/api/test/scripts/copy',
{'id': sid, 'suite_id': a['id'], 'name': 'cloned'})
self.assertEqual(status, 200, resp)
copy_id = resp['script']['id']
self.assertNotEqual(copy_id, sid)
self.assertEqual(resp['script']['name'], 'cloned')
self.assertEqual(self.suites.get(a['id'])['scripts'][0]['script_id'], copy_id)
self.assertEqual(len(self.store.list()), 2)
def test_bundle_import_remaps_script_references(self):
# an exported bundle: one suite + its script
s = self.store.add({'name': 'one', 'steps': [
{'type': 'action', 'kind': 'status', 'params': {'attempts': 1}}]})
bundle_suite = self.suites.add({'name': 'bundle', 'scripts': [
{'script_id': s['id'], 'role': 'setup', 'on_fail': 'stop'}]})
bundle = {'suites': [dict(bundle_suite)], 'scripts': [dict(self.store.get(s['id']))]}
# importing into the same store duplicates both (new ids), and the
# suite references the copies
status, resp = self._post('/api/test/suites/import', bundle)
self.assertEqual(status, 200, resp)
self.assertEqual(resp['scripts']['added'], 1)
self.assertEqual(resp['added'], 1)
new_scripts = [x for x in self.store.list() if x['id'] != s['id']]
self.assertEqual(len(new_scripts), 1)
new_suite = [x for x in self.suites.list() if x['id'] != bundle_suite['id']][0]
self.assertEqual(new_suite['scripts'][0]['script_id'], new_scripts[0]['id'])
self.assertEqual(new_suite['scripts'][0]['role'], 'setup')
self.assertEqual(new_scripts[0]['suite_id'], new_suite['id'])
# a reference that resolves to nothing is reported, not imported
status, resp = self._post('/api/test/suites/import', {
'suites': [{'name': 'bad', 'scripts': [
{'script_id': 'a' * 32, 'role': 'member', 'on_fail': 'stop'}]}]})
self.assertEqual(status, 200, resp)
self.assertEqual(resp['added'], 0)
self.assertIn('unknown script', ' '.join(resp.get('errors') or []))
def test_import_accepts_the_old_localstorage_export(self):
status, resp = self._post('/api/test/scripts/import', {'scripts': [
{'name': 'A', 'steps': [{'type': 'action', 'kind': 'status',
@@ -1260,25 +1394,113 @@ class TestScriptStoreHttpTests(unittest.TestCase):
self.assertEqual(self.store.list()[0]['name'], 'A')
status, resp = self._post('/api/test/scripts/import', {'scripts': 'nope'})
self.assertEqual(status, 400, resp)
# an import into a named suite attaches the imported scripts there
suite = self._suite(name='target')
status, resp = self._post('/api/test/scripts/import', {
'suite_id': suite['id'],
'scripts': [{'name': 'B', 'steps': [
{'type': 'action', 'kind': 'status', 'params': {'attempts': 1}}]}]})
self.assertEqual(status, 200, resp)
self.assertEqual(resp['added'], 1)
self.assertEqual(len(self.suites.get(suite['id'])['scripts']), 1)
def test_the_run_rejects_an_unknown_script_id(self):
status, resp = self._post('/api/test/run', {'script_id': 'nope'})
self.assertEqual(status, 404, resp)
self.assertIn('not found', resp['error'])
def test_suite_run_starts_and_reports_members(self):
suite = self._suite(name='regression')
status, resp = self._post('/api/test/scripts', self._script(suite_id=suite['id']))
self.assertEqual(status, 200, resp)
status, resp = self._post('/api/test/run', {'suite_id': suite['id']})
self.assertEqual(status, 200, resp)
self.assertIsNotNone(resp['suite'])
self.assertEqual(resp['suite']['name'], 'regression')
self.assertEqual(len(resp['suite']['members']), 1)
self.assertEqual(resp['suite']['members'][0]['role'], 'member')
snap = resp
for _ in range(100):
status, snap = self._get('/api/test/status')
if not snap['running']:
break
time.sleep(0.05)
self.assertFalse(snap['running'], snap)
# the fake scc has no transport: the member fails, the report closes
self.assertEqual(len(snap['suite']['members']), 1)
self.assertIn(snap['suite']['members'][0]['status'],
('ok', 'warning', 'error'))
self.assertIsNotNone(snap['suite']['summary'])
self.assertIn('wall_ms', snap['suite']['summary'])
# the report carries the run-log buffer and the member's slice
self.assertIn('log', snap)
self.assertIn('steps', snap['suite']['members'][0])
def test_suite_run_refuses_missing_and_unknown(self):
status, resp = self._post('/api/test/run',
{'suite_id': 'deadbeefdeadbeefdeadbeefdeadbeef'})
self.assertEqual(status, 404, resp)
self.assertIn('unknown test suite', resp['error'])
suite = self._suite(name='empty')
status, resp = self._post('/api/test/run', {'suite_id': suite['id']})
self.assertEqual(status, 400, resp)
self.assertIn('no scripts', resp['error'])
# a shape-valid but dangling reference (a hand edit) is refused
status, resp = self._post('/api/test/scripts', self._script(suite_id=suite['id']))
self.assertEqual(status, 200, resp)
sid = resp['script']['id']
self.suites.update(suite['id'], {'scripts': [
{'script_id': 'f' * 32, 'role': 'member', 'on_fail': 'stop'}]})
status, resp = self._post('/api/test/run', {'suite_id': suite['id']})
self.assertEqual(status, 400, resp)
self.assertIn('unknown test script', resp['error'])
# restore: run with the real member is fine
self.suites.update(suite['id'], {'scripts': [
{'script_id': sid, 'role': 'member', 'on_fail': 'stop'}]})
def test_suite_run_adm_gate_refuses_without_a_key(self):
self.server.app = object() # the gate needs a card (app + scc)
suite = self._suite(name='gated', require_adm=True)
status, resp = self._post('/api/test/scripts', self._script(suite_id=suite['id']))
self.assertEqual(status, 200, resp)
status, resp = self._post('/api/test/run', {
'suite_id': suite['id'], 'preset': {'name': 'no-adm'}})
self.assertEqual(status, 400, resp)
self.assertIn('no ADM key', resp['error'])
# a failed verification is not retried automatically
with mock.patch.object(self.srv, '_verify_adm',
return_value={'ok': False, 'sw': '63C2',
'attempts_left': 3}):
status, resp = self._post('/api/test/run', {
'suite_id': suite['id'], 'preset': {'name': 'with-adm', 'adm': 'AABBCCDD'}})
self.assertEqual(status, 400, resp)
self.assertIn('63C2', resp['error'])
self.assertIn('3 attempt(s) left', resp['error'])
with mock.patch.object(self.srv, '_verify_adm') as v:
status, resp = self._post('/api/test/run', {
'suite_id': suite['id'], 'preset': {'name': 'with-adm', 'adm': 'AABBCCDD'}})
self.assertEqual(status, 400, resp)
self.assertIn('verify it manually', resp['error'])
v.assert_not_called()
def test_a_store_write_failure_answers_a_json_500(self):
suite = self._suite()
with mock.patch.object(self.store, 'add',
side_effect=OSError('read-only file system')):
status, resp = self._post('/api/test/scripts', self._script())
status, resp = self._post('/api/test/scripts',
self._script(suite_id=suite['id']))
self.assertEqual(status, 500, resp)
self.assertIn('test script store write failed', resp['error'])
def test_the_store_endpoints_answer_503_without_a_store(self):
self.server.test_scripts = None
self.server.test_suites = None
status, resp = self._get('/api/test/scripts')
self.assertEqual(status, 503, resp)
status, resp = self._post('/api/test/scripts', self._script())
self.assertEqual(status, 503, resp)
status, resp = self._get('/api/test/suites')
self.assertEqual(status, 503, resp)
if __name__ == '__main__':
tests/test_test_scripts.py
+64
View File
@@ -221,3 +221,67 @@ class StoreInfoTests(unittest.TestCase):
self.assertEqual(info['path'], str(store.path))
self.assertEqual(info['count'], 1)
self.assertEqual(info['version'], test_scripts.SCHEMA_VERSION)
class SuiteAttachmentTests(unittest.TestCase):
"""v3.22.0: a script belongs to a suite (suite_id), declares require_adm,
and the import reports the ids a bundle import needs."""
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.path = pathlib.Path(self.tmp.name) / 'test_scripts.json'
self.store = test_scripts.TestScriptStore(self.path)
def tearDown(self):
self.tmp.cleanup()
def test_suite_id_and_require_adm_round_trip(self):
sid = 'a' * 32
s = self.store.add(dict(script('one'), suite_id=sid, require_adm=True))
self.assertEqual(s['suite_id'], sid)
self.assertTrue(s['require_adm'])
got = test_scripts.TestScriptStore(self.path).get(s['id'])
self.assertEqual(got['suite_id'], sid)
self.assertTrue(got['require_adm'])
with self.assertRaises(test_scripts.TestScriptError):
self.store.add(dict(script('two'), suite_id='short'))
def test_orphans_and_list_by_suite(self):
sid = 'a' * 32
s1 = self.store.add(dict(script('one'), suite_id=sid))
s2 = self.store.add(script('two'))
self.assertEqual([x['id'] for x in self.store.list_by_suite(sid)], [s1['id']])
self.assertEqual([x['id'] for x in self.store.orphans()], [s2['id']])
# a move is an update
moved = self.store.update(s2['id'], {'suite_id': sid})
self.assertEqual(moved['suite_id'], sid)
self.assertEqual(self.store.orphans(), [])
self.assertEqual(len(self.store.list_by_suite(sid)), 2)
def test_import_reports_the_id_map_and_attaches_to_a_suite(self):
sid = 'a' * 32
s = self.store.add(script('one'))
result = self.store.import_scripts([self.store.get(s['id'])])
self.assertEqual(result['added'], 1)
self.assertEqual(result['id_map'][s['id']], result['added_ids'][0])
self.assertNotEqual(result['added_ids'][0], s['id'])
# import into a named suite: every added script belongs to it
result = self.store.import_scripts(
[script('two'), script('three')], suite_id=sid)
self.assertEqual(result['added'], 2)
for sid_added in result['added_ids']:
self.assertEqual(self.store.get(sid_added)['suite_id'], sid)
def test_a_v1_store_loads_with_no_suite(self):
self.store.add(script('one'))
data = json.loads(self.path.read_text())
data['version'] = 1
for e in data['scripts']:
e.pop('suite_id', None)
e.pop('require_adm', None)
self.path.write_text(json.dumps(data))
again = test_scripts.TestScriptStore(self.path)
got = again.list()[0]
self.assertIsNone(got['suite_id'])
self.assertFalse(got['require_adm'])
self.assertEqual(len(again.orphans()), 1)
tests/test_test_suites.py
+227
View File
@@ -0,0 +1,227 @@
"""Test suite store tests (server-side suites, v3.22.0).
A suite groups the test scripts: the root object, holding the ordered member
list by reference (setup first, members, teardown last - the store
normalises the order), the per-member on_fail policy and the suite-level
require_adm flag. The cross-store rules live in the HTTP layer; this file
covers the store's own shape, ordering, persistence and import behaviour.
"""
import json
import pathlib
import tempfile
import unittest
from pysim_simple_server import test_suites
SID_A = 'a' * 32
SID_B = 'b' * 32
SID_C = 'c' * 32
def make_store(tmp):
return test_suites.TestSuiteStore(pathlib.Path(tmp) / 'test_suites.json')
def suite(name='demo', scripts=None, **kw):
s = {'name': name, 'scripts': scripts if scripts is not None else []}
s.update(kw)
return s
class StoreCrudTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.store = make_store(self.tmp.name)
def tearDown(self):
self.tmp.cleanup()
def test_add_list_get_and_remove(self):
s = self.store.add(suite())
self.assertEqual(len(s['id']), 32)
self.assertEqual(s['name'], 'demo')
self.assertEqual(s['require_adm'], False)
self.assertEqual(self.store.list()[0]['id'], s['id'])
self.assertTrue(self.store.remove(s['id']))
self.assertEqual(self.store.list(), [])
self.assertFalse(self.store.remove(s['id']))
self.assertIsNone(self.store.get('nope'))
def test_callers_never_mutate_the_store(self):
s = self.store.add(suite(scripts=[
{'script_id': SID_A, 'role': 'member', 'on_fail': 'stop'}]))
s['scripts'][0]['on_fail'] = 'continue'
s['name'] = 'hacked'
self.assertEqual(self.store.get(s['id'])['name'], 'demo')
self.assertEqual(self.store.get(s['id'])['scripts'][0]['on_fail'], 'stop')
def test_roles_are_ordered_setup_first_teardown_last(self):
# a caller may send any order; the store normalises it and marks the
# roles - setup first, teardown last, members keep their order
s = self.store.add(suite(scripts=[
{'script_id': SID_C, 'role': 'teardown', 'on_fail': 'stop'},
{'script_id': SID_A, 'role': 'member', 'on_fail': 'continue'},
{'script_id': SID_B, 'role': 'member', 'on_fail': 'stop'},
{'script_id': SID_A.replace('a', 'd'), 'role': 'setup', 'on_fail': 'stop'},
]))
ids = [e['script_id'] for e in s['scripts']]
self.assertEqual(ids[0], 'd' * 32)
self.assertEqual(ids[-1], SID_C)
self.assertEqual(ids[1:3], [SID_A, SID_B])
roles = [e['role'] for e in s['scripts']]
self.assertEqual(roles, ['setup', 'member', 'member', 'teardown'])
self.assertEqual(s['scripts'][1]['on_fail'], 'continue')
def test_shape_is_validated(self):
with self.assertRaises(test_suites.TestSuiteError):
self.store.add('nope')
with self.assertRaises(test_suites.TestSuiteError):
self.store.add(suite(scripts='nope'))
with self.assertRaises(test_suites.TestSuiteError):
self.store.add(suite(scripts=[{'script_id': 'short'}]))
with self.assertRaises(test_suites.TestSuiteError):
self.store.add(suite(scripts=[{'script_id': SID_A, 'role': 'boss'}]))
with self.assertRaises(test_suites.TestSuiteError):
self.store.add(suite(scripts=[{'script_id': SID_A, 'on_fail': 'maybe'}]))
# one script appears once per suite
with self.assertRaises(test_suites.TestSuiteError):
self.store.add(suite(scripts=[
{'script_id': SID_A}, {'script_id': SID_A}]))
# at most one setup / teardown
with self.assertRaises(test_suites.TestSuiteError):
self.store.add(suite(scripts=[
{'script_id': SID_A, 'role': 'setup'},
{'script_id': SID_B, 'role': 'setup'}]))
with self.assertRaises(test_suites.TestSuiteError):
self.store.add(suite(scripts=[
{'script_id': SID_A, 'role': 'teardown'},
{'script_id': SID_B, 'role': 'teardown'}]))
def test_update_keeps_id_and_created(self):
s = self.store.add(suite(name='one'))
updated = self.store.update(s['id'], {
'name': 'two', 'require_adm': True,
'scripts': [{'script_id': SID_A, 'role': 'member', 'on_fail': 'stop'}]})
self.assertEqual(updated['id'], s['id'])
self.assertEqual(updated['created'], s['created'])
self.assertEqual(updated['name'], 'two')
self.assertTrue(updated['require_adm'])
self.assertEqual(len(updated['scripts']), 1)
self.assertIsNone(self.store.update('nope', {'name': 'x'}))
def test_find_by_name(self):
self.store.add(suite(name='alpha'))
self.assertEqual(self.store.find_by_name('alpha')['name'], 'alpha')
self.assertIsNone(self.store.find_by_name('beta'))
class StorePersistenceTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.path = pathlib.Path(self.tmp.name) / 'test_suites.json'
self.store = test_suites.TestSuiteStore(self.path)
def tearDown(self):
self.tmp.cleanup()
def test_round_trip_keeps_ids_and_timestamps(self):
s = self.store.add(suite(require_adm=True, scripts=[
{'script_id': SID_A, 'role': 'member', 'on_fail': 'stop'}]))
again = test_suites.TestSuiteStore(self.path)
got = again.get(s['id'])
self.assertEqual(got['name'], s['name'])
self.assertEqual(got['created'], s['created'])
self.assertTrue(got['require_adm'])
self.assertEqual(got['scripts'], s['scripts'])
def test_the_file_carries_the_schema_version(self):
self.store.add(suite())
data = json.loads(self.path.read_text())
self.assertEqual(data['version'], test_suites.SCHEMA_VERSION)
def test_an_entry_with_a_dangling_reference_is_kept(self):
# a referenced script removed by hand (or an import that lost it): the
# suite keeps loading - it is served (the editor shows the missing
# reference) and the HTTP layer refuses a save/run until it is fixed
self.store.add(suite(scripts=[{'script_id': SID_A, 'role': 'member',
'on_fail': 'stop'}]))
again = test_suites.TestSuiteStore(self.path)
self.assertEqual(len(again.list()), 1)
self.assertEqual(again.list()[0]['scripts'][0]['script_id'], SID_A)
def test_a_shape_invalid_entry_is_dropped_with_a_log(self):
# an unreadable entry (a malformed script id) is not servable: unlike
# the engine-validated scripts, the store's own shape check drops it
self.store.add(suite())
raw = json.loads(self.path.read_text())
raw['suites'][0]['scripts'] = [{'script_id': 'short', 'role': 'member',
'on_fail': 'stop'}]
self.path.write_text(json.dumps(raw))
again = test_suites.TestSuiteStore(self.path)
self.assertEqual(again.list(), [])
class StoreImportTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.store = make_store(self.tmp.name)
def tearDown(self):
self.tmp.cleanup()
def test_import_merge_and_replace(self):
self.store.add(suite(name='old'))
result = self.store.import_suites([suite(name='new', scripts=[
{'script_id': SID_A, 'role': 'member', 'on_fail': 'stop'}])])
self.assertEqual(result['added'], 1)
self.assertEqual(result['skipped'], 0)
self.assertEqual([s['name'] for s in self.store.list()], ['old', 'new'])
result = self.store.import_suites([suite(name='third')], 'replace')
self.assertEqual(result['added'], 1)
self.assertEqual([s['name'] for s in self.store.list()], ['third'])
def test_import_keeps_a_given_id_and_reassigns_a_duplicate(self):
s = self.store.add(suite(name='one'))
entry = dict(s, name='copy')
result = self.store.import_suites([entry])
self.assertEqual(result['added'], 1)
ids = sorted(x['id'] for x in self.store.list())
self.assertEqual(len(set(ids)), 2)
self.assertIn(s['id'], ids)
def test_import_reports_invalid_entries(self):
result = self.store.import_suites([suite(name='bad', scripts=[
{'script_id': 'nope'}]), suite(name='ok')])
self.assertEqual(result['added'], 1)
self.assertEqual(result['skipped'], 1)
self.assertTrue(result['errors'])
class MigrationTests(unittest.TestCase):
"""v3.21.0 -> v3.22.0: scripts without a suite are attached to an
auto-created "Imported scripts" suite, once (idempotent)."""
def test_orphans_are_attached_to_the_imported_suite(self):
from pysim_simple_server import server as srv
from pysim_simple_server import test_scripts
with tempfile.TemporaryDirectory() as tmp:
scripts = test_scripts.TestScriptStore(pathlib.Path(tmp) / 's.json')
suites = test_suites.TestSuiteStore(pathlib.Path(tmp) / 'u.json')
step = {'type': 'action', 'kind': 'status', 'params': {'attempts': 1}}
a = scripts.add({'name': 'a', 'steps': [step]})
b = scripts.add({'name': 'b', 'steps': [step]})
self.assertEqual(srv.migrate_scripts_to_suites(scripts, suites), 2)
suite = suites.find_by_name('Imported scripts')
self.assertIsNotNone(suite)
self.assertEqual([e['script_id'] for e in suite['scripts']],
[a['id'], b['id']])
self.assertEqual([e['role'] for e in suite['scripts']],
['member', 'member'])
self.assertEqual(scripts.get(a['id'])['suite_id'], suite['id'])
# idempotent: nothing left to attach
self.assertEqual(srv.migrate_scripts_to_suites(scripts, suites), 0)
if __name__ == '__main__':
unittest.main()
tests/test_testscript.py
+251 -2
View File
@@ -404,6 +404,7 @@ class FakeServer:
self.sms_oa = '12345'
self.sms_sc = '12345678912'
self.card_session = 1
self.card_present = True
self.stk_pending = None
self.menu_active = False
@@ -428,7 +429,8 @@ class RunnerTestCase(unittest.TestCase):
S._TEST_RUN.update({'running': True, 'stop': stop, 'name': 'test',
'status': None, 'steps': [], 'index': 0,
'total': len(script['steps']), 'scp80_counter': None,
'started': time.time(), 'finished': None, 'error': None})
'started': time.time(), 'finished': None, 'error': None,
'suite': None, 'log': [], 'log_prefix': '', 'adm': None})
S._TEST_RUNNING = True
try:
S._test_run_execute(server, script, preset or {})
@@ -1022,7 +1024,8 @@ class TestRunnerActions(RunnerTestCase):
self.assertIsNone(err)
# script_id alone resolves from the store
raw, sid, err = S._test_run_script_from_body(server, {'script_id': entry['id']})
self.assertEqual(raw, {'name': 'stored', 'steps': entry['steps']})
self.assertEqual(raw, {'name': 'stored', 'steps': entry['steps'],
'require_adm': False})
self.assertEqual(sid, entry['id'])
self.assertIsNone(err)
raw, sid, err = S._test_run_script_from_body(server, {'script_id': 'nope'})
@@ -1485,5 +1488,251 @@ class TestRunnerActions(RunnerTestCase):
self.assertEqual(S._increment_counter_hex(''), '')
OK_STEP = {'type': 'action', 'kind': 'apdu', 'params': {'apdu': '00A40000023F00'},
'check': {'sw': '9000'}}
BAD_STEP = {'type': 'action', 'kind': 'apdu', 'params': {'apdu': '00B0000000'},
'check': {'sw': '9000'}} # no queue -> 6D00
class SuiteRunnerTestCase(RunnerTestCase):
"""Suite runner tests (v3.22.0): setup -> members (per-member on_fail) ->
teardown, one card session, no resume."""
def members(self, specs):
"""specs: [(role, ok, on_fail)] -> the resolved member list."""
out = []
for i, spec in enumerate(specs):
role, ok = spec[0], spec[1]
on_fail = spec[2] if len(spec) > 2 else 'stop'
step = dict(OK_STEP if ok else BAD_STEP)
out.append(({'script_id': '%032x' % (i + 1), 'role': role,
'on_fail': on_fail},
T.normalise_script({'name': 'script %d' % (i + 1),
'steps': [step]}, S._test_command_type)))
return out
def suite_run(self, server, members, suite=None, preset=None, preset_id=None):
suite = suite or {'name': 'suite'}
S._TEST_RUN.clear()
S._TEST_RUN.update({'running': True, 'stop': False, 'name': 'suite',
'status': None, 'session': server.card_session,
'index': 0, 'total': 0, 'steps': [],
'started': time.time(), 'finished': None, 'error': None,
'scp80_counter': None, 'scp80_counters': {},
'script_id': None, 'preset': None,
'suite': S._test_suite_state(suite, members,
preset or {}, preset_id,
server.card_session),
'log': [], 'log_prefix': '', 'adm': None})
S._TEST_RUNNING = True
try:
S._test_suite_execute(server, suite, members, preset or {}, preset_id)
finally:
S._TEST_RUNNING = False
return S._TEST_RUN
def push_ok(self, scc, n):
for _ in range(n):
scc.push('00A4', '', '9000')
class TestSuiteRunner(SuiteRunnerTestCase):
def test_setup_members_teardown_run_in_order(self):
scc = FakeScc()
self.push_ok(scc, 4)
members = self.members([('setup', True), ('member', True),
('member', True), ('teardown', True)])
run = self.suite_run(FakeServer(scc), members)
self.assertEqual(run['status'], 'ok')
state = run['suite']
self.assertEqual([m['role'] for m in state['members']],
['setup', 'member', 'member', 'teardown'])
self.assertEqual([m['status'] for m in state['members']],
['ok', 'ok', 'ok', 'ok'])
self.assertEqual(state['summary']['setup'], 'ok')
self.assertEqual(state['summary']['teardown'], 'ok')
self.assertEqual(state['summary']['ok'], 4)
self.assertEqual(state['summary']['error'], 0)
self.assertFalse(state['summary']['stopped'])
self.assertEqual(state['summary']['counters_before'], {})
# the member steps are copied into the report (the live view is reset
# between members)
self.assertEqual(len(state['members'][1]['steps']), 1)
self.assertEqual(state['members'][1]['steps'][0]['status'], 'ok')
# the run log carries the member prefix and is sliced per member
joined = '\n'.join(state['members'][1]['log'])
self.assertIn('[2/4 member script 2]', joined)
def test_on_fail_stop_skips_the_rest_and_still_runs_teardown(self):
scc = FakeScc()
self.push_ok(scc, 1) # setup ok
self.push_ok(scc, 1) # first member ok
# the failing member gets no queue entry -> 6D00 (error)
self.push_ok(scc, 1) # teardown ok
members = self.members([('setup', True), ('member', True),
('member', False, 'stop'), ('member', True),
('teardown', True)])
run = self.suite_run(FakeServer(scc), members)
self.assertEqual(run['status'], 'error')
statuses = [m['status'] for m in run['suite']['members']]
self.assertEqual(statuses, ['ok', 'ok', 'error', 'skipped', 'ok'])
self.assertEqual(run['suite']['members'][3]['note'],
'a previous script failed (on_fail: stop)')
# no member ran after the error: the queue holds exactly the teardown's
self.assertEqual(sum(1 for a in scc.sent if a.startswith('00A4')), 3)
def test_on_fail_continue_runs_the_next_member(self):
scc = FakeScc()
self.push_ok(scc, 1) # member 1 ok
# member 2 fails (no queue entry)
self.push_ok(scc, 1) # member 3 ok
members = self.members([('member', True), ('member', False, 'continue'),
('member', True)])
run = self.suite_run(FakeServer(scc), members)
self.assertEqual(run['status'], 'error') # the failure is reported
self.assertEqual([m['status'] for m in run['suite']['members']],
['ok', 'error', 'ok'])
def test_setup_failure_skips_members_but_runs_teardown(self):
scc = FakeScc()
# setup fails (no queue), members never run, teardown ok
self.push_ok(scc, 1)
members = self.members([('setup', False), ('member', True),
('teardown', True)])
run = self.suite_run(FakeServer(scc), members)
self.assertEqual([m['status'] for m in run['suite']['members']],
['error', 'skipped', 'ok'])
self.assertEqual(run['suite']['members'][1]['note'],
'the setup script failed')
def test_stop_skips_the_members_and_runs_the_teardown(self):
scc = FakeScc()
self.push_ok(scc, 1) # setup ok (runs first)
self.push_ok(scc, 1) # teardown ok
members = self.members([('setup', True), ('member', True),
('teardown', True)])
server = FakeServer(scc)
real = S._test_run_script
calls = []
def wrapped(srv, script, preset):
ctx, stopped = real(srv, script, preset)
calls.append(script['name'])
if srv is server and len(calls) == 1:
S._TEST_RUN['stop'] = True # the operator stops here
return ctx, stopped
with mock.patch.object(S, '_test_run_script', wrapped):
run = self.suite_run(server, members)
statuses = [m['status'] for m in run['suite']['members']]
self.assertEqual(statuses, ['ok', 'skipped', 'ok'])
self.assertTrue(run['suite']['summary']['stopped'])
self.assertEqual(run['status'], 'stopped')
def test_a_card_reset_stops_the_suite_and_flags_it(self):
scc = FakeScc()
self.push_ok(scc, 1) # member 1 ok
self.push_ok(scc, 1) # teardown ok (new session)
members = self.members([('member', True), ('member', True),
('teardown', True)])
server = FakeServer(scc)
real = S._test_run_script
def wrapped(srv, script, preset):
ctx, stopped = real(srv, script, preset)
if srv is server and len(scc.sent) <= 1:
srv.card_session += 1 # the card reset mid-suite
return ctx, stopped
with mock.patch.object(S, '_test_run_script', wrapped):
run = self.suite_run(server, members)
state = run['suite']
self.assertEqual([m['status'] for m in state['members']],
['ok', 'skipped', 'ok'])
self.assertEqual(state['members'][1]['note'],
'card session changed (card removed or re-equipped)')
self.assertTrue(state['session_changed'])
self.assertEqual(state['session_start'], 1)
self.assertEqual(state['session_end'], 2)
self.assertTrue(state['summary']['session_changed'])
def test_teardown_is_skipped_when_the_card_is_gone(self):
scc = FakeScc()
self.push_ok(scc, 1)
members = self.members([('member', True), ('teardown', True)])
server = FakeServer(scc)
server.card_present = False
run = self.suite_run(server, members)
self.assertEqual([m['status'] for m in run['suite']['members']],
['ok', 'skipped'])
self.assertEqual(run['suite']['members'][1]['note'],
'no card in the reader')
class TestAdmGate(unittest.TestCase):
"""The run prerequisite: a script/suite flagged `require_adm` runs only
with the ADM verified for the current card session - the preset's key is
tried once, never retried automatically."""
def setUp(self):
self.server = FakeServer(FakeScc())
self.server.app = object()
def test_not_needed_passes(self):
ok, info, err = S._ensure_adm(self.server, {}, False)
self.assertTrue(ok)
self.assertIsNone(err)
self.assertFalse(info['required'])
def test_verified_latch_passes(self):
self.server.adm_session = 1
ok, info, err = S._ensure_adm(self.server, {}, True)
self.assertTrue(ok)
self.assertTrue(info['verified'])
def test_verifies_once_with_the_preset_key(self):
with mock.patch.object(S, '_verify_adm', return_value={'ok': True,
'sw': '9000'}) as v:
ok, info, err = S._ensure_adm(self.server, {'adm': 'AABBCCDD'}, True)
self.assertTrue(ok, err)
self.assertTrue(info['verified'])
self.assertTrue(info['checked'])
self.assertEqual(self.server.adm_session, self.server.card_session)
v.assert_called_once()
# the second gate in the same session does not verify again
with mock.patch.object(S, '_verify_adm', return_value={'ok': True}) as v2:
ok, _info, _err = S._ensure_adm(self.server, {'adm': 'AABBCCDD'}, True)
self.assertTrue(ok)
v2.assert_not_called()
def test_a_failure_is_not_retried_automatically(self):
with mock.patch.object(S, '_verify_adm', return_value={
'ok': False, 'sw': '63C2', 'attempts_left': 2}) as v:
ok, info, err = S._ensure_adm(self.server, {'adm': 'AABBCCDD'}, True)
self.assertFalse(ok)
self.assertIn('63C2', err)
self.assertIn('2 attempt(s) left', err)
self.assertEqual(info['attempts_left'], 2)
self.assertEqual(self.server.adm_failed_session, self.server.card_session)
# the next gate refuses without touching the card again
with mock.patch.object(S, '_verify_adm') as v2:
ok, _info, err = S._ensure_adm(self.server, {'adm': 'AABBCCDD'}, True)
self.assertFalse(ok)
self.assertIn('verify it manually', err)
v2.assert_not_called()
def test_no_key_in_the_preset_refuses(self):
ok, _info, err = S._ensure_adm(self.server, {}, True)
self.assertFalse(ok)
self.assertIn('no ADM key', err)
def test_a_new_session_clears_the_failure_stamp(self):
self.server.adm_failed_session = 1
self.server.card_session = 2
with mock.patch.object(S, '_verify_adm', return_value={'ok': True}):
ok, _info, _err = S._ensure_adm(self.server, {'adm': 'AABBCCDD'}, True)
self.assertTrue(ok)
if __name__ == '__main__':
unittest.main()