"""End-to-end persistent-spool restart test. Exercises SpoolStore's opt-in file-backed journal by: 1. Connecting to the C++ server (which has been started with --spool-dir /spool, a docker named volume). 2. Force-spooling a CEID 300 emission via the SPOOL_ON / START / SPOOL_OFF RCMD sequence. 3. Disconnecting cleanly (separate). 4. The caller (a shell driver) then issues `docker compose restart server-spool`, preserving the /spool volume. 5. We reconnect. If persistence works, the server should auto-emit S6F25 (spool-data-ready) shortly after SELECT. We then issue S6F23 RSDC=Transmit and expect the queued S6F11 to drain. Two modes: python3 spool_persistence_test.py --phase enqueue --host server-spool python3 spool_persistence_test.py --phase drain --host server-spool The driver runs phase 1, restarts the container, runs phase 2. Each phase returns 0 on success. """ from __future__ import annotations import argparse import logging import sys import threading import time import secsgem.common import secsgem.gem import secsgem.hsms import secsgem.secs.functions as F from secsgem.secs.functions.base import SecsStreamFunction LOG = logging.getLogger("interop.spool-restart") # secsgem-py 0.3.0 ships no S6F25/F26 — we register them inline so the # protocol layer can decode the spool-data-ready notification. class S06F25(SecsStreamFunction): _stream = 6 _function = 25 _to_host = True _to_equipment = False _has_reply = True _is_reply_required = True _is_multi_block = False _data_format = "< DATAID >" # carries the queued-message count as a U4 class S06F26(SecsStreamFunction): _stream = 6 _function = 26 _to_host = False _to_equipment = True _has_reply = False _is_reply_required = False _is_multi_block = False _data_format = "< ACKC6 >" def connect(host: str, port: int, session_id: int) -> secsgem.gem.GemHostHandler: settings = secsgem.hsms.HsmsSettings( address=host, port=port, session_id=session_id, connect_mode=secsgem.hsms.HsmsConnectMode.ACTIVE, device_type=secsgem.common.DeviceType.HOST, ) client = secsgem.gem.GemHostHandler(settings) client.enable() if not client.waitfor_communicating(timeout=15): raise RuntimeError("failed to communicate within 15s") return client def rcmd(client, name: str) -> int: rsp = client.send_and_waitfor_response( F.SecsS02F41({"RCMD": name, "PARAMS": []}) ) body = client.settings.streams_functions.decode(rsp).get() return body.get("HCACK") if isinstance(body, dict) else -1 def phase_enqueue(host: str, port: int, session_id: int) -> int: """Enqueue one spooled S6F11 then disconnect.""" client = connect(host, port, session_id) try: # Define + link + enable CEID 300 so the equipment actually emits S6F11. client.send_and_waitfor_response( F.SecsS02F33({"DATAID": 1, "DATA": [{"RPTID": 1, "VID": [2]}]}) ) client.send_and_waitfor_response( F.SecsS02F35({"DATAID": 1, "DATA": [{"CEID": 300, "RPTID": [1]}]}) ) client.send_and_waitfor_response( F.SecsS02F37({"CEED": True, "CEID": [300]}) ) # Force the spool on, fire START (emits CEID 300, goes to spool), # then turn force off. The S6F11 is now persisted on disk. ack = rcmd(client, "SPOOL_ON") if ack != 0: LOG.error("SPOOL_ON ack=%s", ack); return 1 ack = rcmd(client, "START") if ack != 0: LOG.error("START ack=%s", ack); return 1 ack = rcmd(client, "SPOOL_OFF") if ack != 0: LOG.error("SPOOL_OFF ack=%s", ack); return 1 LOG.info("phase=enqueue: spooled one S6F11 (CEID=300)") # Brief pause so the post-RCMD spool write completes before we close. time.sleep(0.3) return 0 finally: try: client.disable() except Exception: pass time.sleep(0.2) def phase_drain(host: str, port: int, session_id: int) -> int: """After server restart: drain via S6F23 and verify the replayed S6F11. Note: the C++ server also auto-emits S6F25 (spool-data-ready) on re-SELECT, but secsgem-py 0.3.0 doesn't ship S6F25/F26 in its catalog and decoding ours runs into SFDL grammar trouble; the persistence claim doesn't actually depend on F25 (it's just a hint to the host), so we skip the F25 wait and request transmit directly. If the spool was empty after restart, S6F23(Transmit) would ack but no S6F11 would arrive — that's the real test. """ drained_ceids: list[int] = [] drained_event = threading.Event() settings = secsgem.hsms.HsmsSettings( address=host, port=port, session_id=session_id, connect_mode=secsgem.hsms.HsmsConnectMode.ACTIVE, device_type=secsgem.common.DeviceType.HOST, ) client = secsgem.gem.GemHostHandler(settings) # Register S6F25 / S6F26 so the protocol layer can at least decode # the unsolicited notification cleanly while we wait on the drain. client.settings.streams_functions.update(S06F25) client.settings.streams_functions.update(S06F26) def on_s6f25(_handler, message): LOG.info("[evt] S6F25 spool-data-ready (server announces queue)") client.send_response(S06F26(0), message.header.system) def on_s6f11(_handler, message): decoded = client.settings.streams_functions.decode(message) body = decoded.get() ceid = body.get("CEID") if isinstance(body, dict) else None if ceid is not None: drained_ceids.append(int(ceid)) LOG.info("[evt] replayed S6F11 CEID=%s", ceid) drained_event.set() client.send_response(F.SecsS06F12(0), message.header.system) client.register_stream_function(6, 25, on_s6f25) client.register_stream_function(6, 11, on_s6f11) client.enable() try: if not client.waitfor_communicating(timeout=15): LOG.error("waitfor_communicating timed out"); return 1 # Brief pause so any auto-emitted S6F25 lands before we drain. time.sleep(0.5) # Ask the server to drain. rsp = client.send_and_waitfor_response(F.SecsS06F23(0)) # RSDC=Transmit body = client.settings.streams_functions.decode(rsp).get() rsda = body if isinstance(body, int) else -1 LOG.info("S6F23 -> S6F24 RSDA=%s", rsda) if rsda != 0: LOG.error("RSDA != Accept"); return 1 # Wait for the replayed S6F11 (CEID=300). If persistence broke, # there'd be no replay here even though the ack said Accept. if not drained_event.wait(timeout=3.0): LOG.error("no S6F11 replayed within 3s — spool empty after restart?") return 1 if 300 not in drained_ceids: LOG.error("replayed CEIDs=%s, expected 300", drained_ceids) return 1 LOG.info("phase=drain: spool replayed cleanly — persistence works") return 0 finally: try: client.disable() except Exception: pass time.sleep(0.2) def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--phase", choices=["enqueue", "drain"], required=True) ap.add_argument("--host", default="server-spool") ap.add_argument("--port", type=int, default=5000) ap.add_argument("--session-id", type=int, default=0) ap.add_argument("--log-level", default="INFO") args = ap.parse_args() logging.basicConfig(level=args.log_level, format="%(asctime)s %(levelname)s %(name)s: %(message)s") if args.log_level.upper() != "DEBUG": logging.getLogger("communication").setLevel(logging.WARNING) logging.getLogger("hsms_connection").setLevel(logging.WARNING) if args.phase == "enqueue": return phase_enqueue(args.host, args.port, args.session_id) return phase_drain(args.host, args.port, args.session_id) if __name__ == "__main__": sys.exit(main())