E40 Process Jobs + E94 Control Jobs + E30 communication state

GEM300 layer: SEMI E40-0705 Process Job and E94-0705 Control Job state machines, plus the E30 §6.1 communication-state machine that sits between HSMS SELECT and full GEM communication. Data-driven via data/process_job_state.yaml and data/control_job_state.yaml, mirroring the existing control_state.yaml pattern. Wire coverage: S14F9/F10 CreateObject (CJ) host -> equipment S14F11/F12 DeleteObject (CJ) host -> equipment S16F5/F6 PRJobCommand host -> equipment S16F9 PRJobAlert equipment -> host S16F11/F12 PRJobCreate (simplified body) host -> equipment S16F13/F14 PRJobDequeue host -> equipment S16F27/F28 CJobCommand host -> equipment Process Job FSM exposes 8 states matching PRJOBSTATE bytes (E40 §10.3.2); HOQ is reorder-aware (move-to-head against an insertion-order vector); Stop/Abort on a Queued PJ routes through ABORTING so the host observes PRJOBSTATE=7 on the wire (§6.3); alert_enabled is settable per-PJ for PRALERT control; FSM dispatches through ProcessJobStore::on_change_ dynamically so a late set_state_change_handler() reaches existing PJs. Hardening: loader rejects NoState (sentinel) as initial/from/to and rejects `on: created` rows; static_asserts pin enum values to wire bytes; ProcessJobStore is non-movable to keep the per-PJ this-capture safe. Server simulator cascades the full CJ -> PJ lifecycle on CJSTART so the wire trace exercises every legal state. CEIDs 400/401 fire on CJ state changes via the existing event-report pipeline. Tests: 60+ new assertions across test_process_jobs, test_control_jobs, test_communication_state, test_hsms_connection, plus loader and messages round-trip coverage. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-06-07 21:00:32 +02:00
parent 1f67aad985
commit 90c177b7ce
33 changed files with 3122 additions and 62 deletions
@@ -0,0 +1,231 @@
+// Integration tests for the HSMS Connection state machine.
+//
+// We open a pair of connected TCP sockets on loopback, wrap one in a real
+// passive Connection (the system under test), and use the other as a raw
+// "peer" socket that hand-builds wire frames. This lets us assert exactly
+// what bytes the Connection produces in response to specific stimuli — in
+// particular the E37 §7.2 / §7.4 / §7.7 corner cases that aren't exercised
+// by the happy-path demo: SelectReq while already SELECTED, DeselectReq
+// while NOT_SELECTED, and Reject.req emission for unsupported SType / PType.
+//
+// Both ends share the same io_context, so all socket I/O on the "peer"
+// side has to be async too — running an asio::read synchronously on the
+// peer would block the thread that also has to drive the Connection's
+// own async reads, deadlocking the test.
+
+#include <doctest/doctest.h>
+
+#include <asio.hpp>
+#include <array>
+#include <chrono>
+#include <cstdint>
+#include <memory>
+#include <thread>
+#include <utility>
+#include <vector>
+
+#include "secsgem/hsms/connection.hpp"
+#include "secsgem/hsms/header.hpp"
+
+using namespace secsgem::hsms;
+
+namespace {
+
+// Pair of TCP sockets connected over loopback; both ends share `io`.
+struct SocketPair {
+  asio::io_context io;
+  asio::ip::tcp::socket a{io};   // the "system under test" side
+  asio::ip::tcp::socket b{io};   // the "raw peer" side
+
+  SocketPair() {
+    asio::ip::tcp::acceptor acc(io, asio::ip::tcp::endpoint(
+                                        asio::ip::address_v4::loopback(), 0));
+    const auto port = acc.local_endpoint().port();
+
+    std::error_code ec_accept;
+    bool accepted = false;
+    acc.async_accept(a, [&](std::error_code ec) { ec_accept = ec; accepted = true; });
+
+    std::error_code ec_connect;
+    bool connected = false;
+    b.async_connect(asio::ip::tcp::endpoint(asio::ip::address_v4::loopback(), port),
+                    [&](std::error_code ec) { ec_connect = ec; connected = true; });
+
+    while (!(accepted && connected)) {
+      if (io.stopped()) io.restart();
+      if (io.poll() == 0) std::this_thread::sleep_for(std::chrono::milliseconds(1));
+    }
+    REQUIRE_FALSE(ec_accept);
+    REQUIRE_FALSE(ec_connect);
+  }
+};
+
+// Run the io_context until `pred()` returns true or `budget` is exhausted.
+// We drain all currently-ready handlers with poll(), then sleep briefly
+// before re-checking — run_one_for() can block for its full timeout even
+// when ready work exists, which made earlier iterations of this helper
+// look hung.
+template <typename Pred>
+void pump_until(asio::io_context& io, Pred pred,
+                std::chrono::milliseconds budget = std::chrono::seconds(5)) {
+  const auto deadline = std::chrono::steady_clock::now() + budget;
+  while (!pred()) {
+    if (std::chrono::steady_clock::now() > deadline) FAIL("pump_until budget exceeded");
+    if (io.stopped()) io.restart();
+    const std::size_t n = io.poll();
+    if (n == 0) std::this_thread::sleep_for(std::chrono::milliseconds(1));
+  }
+}
+
+// Async write a buffer; pumps the io_context until it completes.
+void send_bytes(SocketPair& sp, std::vector<uint8_t> bytes) {
+  auto buf = std::make_shared<std::vector<uint8_t>>(std::move(bytes));
+  bool done = false;
+  asio::async_write(sp.b, asio::buffer(*buf),
+                    [buf, &done](std::error_code ec, std::size_t) {
+                      REQUIRE_FALSE(ec);
+                      done = true;
+                    });
+  pump_until(sp.io, [&] { return done; });
+}
+
+// Async read one full HSMS frame from the peer socket; pumps the io_context.
+Frame recv_frame(SocketPair& sp) {
+  auto lenbuf = std::make_shared<std::array<uint8_t, 4>>();
+  bool len_done = false;
+  asio::async_read(sp.b, asio::buffer(*lenbuf),
+                   [lenbuf, &len_done](std::error_code ec, std::size_t) {
+                     REQUIRE_FALSE(ec);
+                     len_done = true;
+                   });
+  pump_until(sp.io, [&] { return len_done; });
+
+  const uint32_t len = (uint32_t((*lenbuf)[0]) << 24) | (uint32_t((*lenbuf)[1]) << 16) |
+                       (uint32_t((*lenbuf)[2]) << 8) | uint32_t((*lenbuf)[3]);
+  auto payload = std::make_shared<std::vector<uint8_t>>(len);
+  bool payload_done = false;
+  asio::async_read(sp.b, asio::buffer(*payload),
+                   [payload, &payload_done](std::error_code ec, std::size_t) {
+                     REQUIRE_FALSE(ec);
+                     payload_done = true;
+                   });
+  pump_until(sp.io, [&] { return payload_done; });
+  return Frame::decode(payload->data(), payload->size());
+}
+
+Timers default_timers() {
+  Timers t;
+  t.linktest = std::chrono::milliseconds(0);  // disabled in tests
+  return t;
+}
+
+}  // namespace
+
+TEST_CASE("Select.req while already SELECTED returns AlreadyActive (E37 §7.2)") {
+  SocketPair sp;
+  auto conn = std::make_shared<Connection>(std::move(sp.a), Connection::Mode::Passive,
+                                           /*device_id=*/0, default_timers());
+  bool selected = false;
+  conn->set_selected_handler([&] { selected = true; });
+  conn->start();
+
+  // First Select.req: should be answered with Ok (status=0) and transition
+  // the connection into SELECTED.
+  send_bytes(sp, Frame(Header::control(SType::SelectReq, /*sys=*/1)).encode());
+  Frame rsp1 = recv_frame(sp);
+  CHECK(rsp1.header.stype == SType::SelectRsp);
+  CHECK(rsp1.header.byte3 == static_cast<uint8_t>(SelectStatus::Ok));
+  pump_until(sp.io, [&] { return selected; });
+
+  // Second Select.req while already SELECTED: must reply AlreadyActive (1)
+  // and must NOT re-fire the selected handler.
+  selected = false;
+  send_bytes(sp, Frame(Header::control(SType::SelectReq, /*sys=*/2)).encode());
+  Frame rsp2 = recv_frame(sp);
+  CHECK(rsp2.header.stype == SType::SelectRsp);
+  CHECK(rsp2.header.byte3 == static_cast<uint8_t>(SelectStatus::AlreadyActive));
+  CHECK_FALSE(selected);
+
+  conn->close("test done");
+}
+
+TEST_CASE("Deselect.req while NOT_SELECTED returns NotEstablished (E37 §7.4)") {
+  SocketPair sp;
+  auto conn = std::make_shared<Connection>(std::move(sp.a), Connection::Mode::Passive,
+                                           /*device_id=*/0, default_timers());
+  conn->start();
+
+  send_bytes(sp, Frame(Header::control(SType::DeselectReq, /*sys=*/1)).encode());
+  Frame rsp = recv_frame(sp);
+  CHECK(rsp.header.stype == SType::DeselectRsp);
+  CHECK(rsp.header.byte3 == static_cast<uint8_t>(DeselectStatus::NotEstablished));
+  CHECK(conn->state() == Connection::State::NotSelected);
+
+  conn->close("test done");
+}
+
+TEST_CASE("Unsupported SType triggers Reject.req(StypeNotSupported) (E37 §7.7)") {
+  SocketPair sp;
+  auto conn = std::make_shared<Connection>(std::move(sp.a), Connection::Mode::Passive,
+                                           /*device_id=*/0, default_timers());
+  conn->start();
+
+  // Standalone Header doesn't let us produce an unknown SType via the
+  // constructors, so we build the wire bytes by hand.
+  std::vector<uint8_t> bad = {0x00, 0x00, 0x00, 0x0A,
+                              // header: session=0xFFFF, byte2=0, byte3=0,
+                              // ptype=0, stype=10, sys=5
+                              0xFF, 0xFF, 0x00, 0x00, 0x00, 0x0A,
+                              0x00, 0x00, 0x00, 0x05};
+  send_bytes(sp, std::move(bad));
+
+  Frame rej = recv_frame(sp);
+  CHECK(rej.header.stype == SType::RejectReq);
+  CHECK(rej.header.system_bytes == 5);
+  CHECK(rej.header.byte2 == 10);  // offending SType echoed in byte2
+  CHECK(rej.header.byte3 == static_cast<uint8_t>(RejectReason::StypeNotSupported));
+
+  conn->close("test done");
+}
+
+TEST_CASE("Non-zero PType triggers Reject.req(PtypeNotSupported) (E37 §7.7)") {
+  SocketPair sp;
+  auto conn = std::make_shared<Connection>(std::move(sp.a), Connection::Mode::Passive,
+                                           /*device_id=*/0, default_timers());
+  conn->start();
+
+  // Linktest.req-like frame but with PType=7 (unsupported).
+  std::vector<uint8_t> bad = {0x00, 0x00, 0x00, 0x0A,
+                              0xFF, 0xFF, 0x00, 0x00,
+                              /*ptype=*/0x07, /*stype=LinktestReq*/0x05,
+                              0x00, 0x00, 0x00, 0x09};
+  send_bytes(sp, std::move(bad));
+
+  Frame rej = recv_frame(sp);
+  CHECK(rej.header.stype == SType::RejectReq);
+  CHECK(rej.header.system_bytes == 9);
+  CHECK(rej.header.byte2 == 7);  // offending PType echoed in byte2
+  CHECK(rej.header.byte3 == static_cast<uint8_t>(RejectReason::PtypeNotSupported));
+
+  conn->close("test done");
+}
+
+TEST_CASE("Data frame while NOT_SELECTED triggers Reject.req(EntityNotSelected)") {
+  SocketPair sp;
+  auto conn = std::make_shared<Connection>(std::move(sp.a), Connection::Mode::Passive,
+                                           /*device_id=*/0, default_timers());
+  conn->start();
+
+  // Primary S1F1 W=1 before selecting — equipment must Reject with
+  // EntityNotSelected.  Construct via the header API: Data with an empty
+  // body is a valid wire frame.
+  Frame data(Header::data_message(/*session=*/1, /*stream=*/1, /*function=*/1,
+                                  /*reply_expected=*/true, /*sys=*/42));
+  send_bytes(sp, data.encode());
+
+  Frame rej = recv_frame(sp);
+  CHECK(rej.header.stype == SType::RejectReq);
+  CHECK(rej.header.byte3 == static_cast<uint8_t>(RejectReason::EntityNotSelected));
+
+  conn->close("test done");
+}