secs-gem/interop/spool_persistence_test.py

"""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())