raphael/secs-gem
Public Access
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
secs-gem/tests/test_daemon_service.cpp
T
raphael 2218b854ce fix(daemon)+test: accurate duplicate-ARRIVED message; broaden E90/E157 + names coverage 
The duplicate-ARRIVED fix from the previous commit returned INVALID_OBJECT
with the message "no substrate 'X'" — a lie, since the substrate exists.
Rewrite ReportSubstrate so ARRIVED has its own ack mapping: a duplicate is
CANNOT_DO_NOW with "substrate 'X' already exists" (a state conflict, not a
missing object), and we never silently re-create over live FSM state.

Coverage gaps closed:
- C++: ARRIVED records carrier_id/slot (now asserted); module NOT_EXECUTING
  reset transition; duplicate-ARRIVED expects CANNOT_DO_NOW.
- Interop: @eq.command now drives the real host S2F41 path (was @eq.on, so
  the headline decorator had zero wire coverage); @eq.command NameError on
  unknown name; eq.names var/alarm + dir() + typo-suggestion; replaced the
  two `check(..., True)` tautologies with full E90 journey + AT_DESTINATION
  and real error paths (ghost wafer raises, illegal module jump raises).

All 8 daemon test cases (248 assertions) and 24 pyclient interop checks pass.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-26 22:01:20 +02:00

840 lines
31 KiB
C++
Raw Permalink Blame History

#define DOCTEST_CONFIG_IMPLEMENT_WITH_MAIN
#include <doctest/doctest.h>
#include <grpcpp/grpcpp.h>
#include <atomic>
#include <chrono>
#include <random>
#include <string>
#include <thread>
#include <vector>
#include "secsgem/daemon/equipment_service.hpp"
#include "secsgem/gem/default_handlers.hpp"
#include "secsgem/gem/messages.hpp"
using namespace secsgem;
namespace gem = secsgem::gem;
namespace s2 = secsgem::secs2;
namespace pb = secsgem::v1;
namespace dmn = secsgem::daemon;
#ifndef SECSGEM_DATA_DIR
#error "SECSGEM_DATA_DIR not defined; see CMakeLists.txt"
#endif
static gem::EquipmentRuntime::Config test_config() {
gem::EquipmentRuntime::Config c;
c.equipment_yaml = SECSGEM_DATA_DIR "/equipment.yaml";
c.control_state_yaml = SECSGEM_DATA_DIR "/control_state.yaml";
c.process_job_yaml = SECSGEM_DATA_DIR "/process_job_state.yaml";
c.control_job_yaml = SECSGEM_DATA_DIR "/control_job_state.yaml";
c.port = 0; // ephemeral; the engine isn't run() here, only poll()ed
return c;
}
// Exercises the real gRPC service over an in-process channel: client stub ->
// service -> EquipmentRuntime, proving the RPCs move data, not just compile.
TEST_CASE("Equipment gRPC service over an in-process channel") {
gem::EquipmentRuntime rt(test_config());
dmn::EquipmentService svc(rt);
grpc::ServerBuilder builder;
builder.RegisterService(&svc);
std::unique_ptr<grpc::Server> server(builder.BuildAndStart());
REQUIRE(server);
auto stub = pb::Equipment::NewStub(server->InProcessChannel(grpc::ChannelArguments{}));
SUBCASE("GetControlState returns the initial control state") {
grpc::ClientContext ctx;
pb::Empty req;
pb::ControlState resp;
auto st = stub->GetControlState(&ctx, req, &resp);
CHECK(st.ok());
CHECK(resp.state() == pb::ControlState::HOST_OFFLINE);
}
SUBCASE("SetVariables converts to the variable's declared wire format") {
// ChamberPressure is declared F4 and WaferCounter U4 in equipment.yaml;
// the daemon must honour those, not write F8/I8, or the host sees values
// whose format contradicts the S1F11/S1F21 namelists.
grpc::ClientContext ctx;
pb::VariableUpdate req;
pb::Ack resp;
(*req.mutable_values())["ChamberPressure"].set_real(2.5);
(*req.mutable_values())["WaferCounter"].set_integer(7);
auto st = stub->SetVariables(&ctx, req, &resp);
CHECK(st.ok());
CHECK(resp.code() == pb::Ack::ACCEPT);
rt.poll(); // drain the posted set_variable onto this thread
CHECK(rt.model().dvids.value(101) == s2::Item::f4(2.5f));
CHECK(rt.model().dvids.value(100) == s2::Item::u4(uint32_t{7}));
}
SUBCASE("SetVariables rejects an unknown variable name") {
grpc::ClientContext ctx;
pb::VariableUpdate req;
pb::Ack resp;
(*req.mutable_values())["definitely_not_a_var"].set_real(1.0);
auto st = stub->SetVariables(&ctx, req, &resp);
CHECK(st.ok());
CHECK(resp.code() == pb::Ack::PARAMETER_INVALID);
}
SUBCASE("SetVariables rejects a Value with no kind set (silent-'' guard)") {
grpc::ClientContext ctx;
pb::VariableUpdate req;
pb::Ack resp;
(*req.mutable_values())["ChamberPressure"]; // map entry, kind never set
auto st = stub->SetVariables(&ctx, req, &resp);
CHECK(st.ok());
CHECK(resp.code() == pb::Ack::PARAMETER_INVALID);
// The previous value must be untouched (no silent ASCII "" write).
rt.poll();
CHECK(rt.model().dvids.value(101)->format() == s2::Format::F4);
}
SUBCASE("property: every configured variable keeps its declared format") {
// Iterate ALL SVIDs+DVIDs from the live config: set each via gRPC with a
// type-appropriate plain value and assert the stored Item keeps the
// declared wire format. Catches any future variable whose format the
// conversion table doesn't handle — not just the two pinned above.
auto check_all = [&](const auto& entries) {
for (const auto& v : entries) {
const auto declared = v.value.format();
grpc::ClientContext ctx;
pb::VariableUpdate req;
pb::Ack resp;
pb::Value val;
switch (declared) {
case s2::Format::ASCII: case s2::Format::JIS8:
val.set_text("x"); break;
case s2::Format::Boolean:
val.set_boolean(true); break;
case s2::Format::Binary:
val.set_binary("\x01"); break;
case s2::Format::F4: case s2::Format::F8:
val.set_real(1.5); break;
default: // all integer widths
val.set_integer(1); break;
}
(*req.mutable_values())[v.name] = val;
REQUIRE(stub->SetVariables(&ctx, req, &resp).ok());
REQUIRE_MESSAGE(resp.code() == pb::Ack::ACCEPT, v.name);
rt.poll();
auto stored = rt.model().vid_value(v.id);
REQUIRE_MESSAGE(stored.has_value(), v.name);
CHECK_MESSAGE(stored->format() == declared,
v.name, ": declared ", static_cast<int>(declared),
" stored ", static_cast<int>(stored->format()));
}
};
check_all(rt.model().svids.all());
check_all(rt.model().dvids.all());
}
SUBCASE("SetAlarm / ClearAlarm by config name, by stringified id, unknown rejected") {
auto call = [&](auto method, const std::string& name) {
grpc::ClientContext ctx;
pb::Alarm req;
pb::Ack resp;
req.set_name(name);
REQUIRE((stub.get()->*method)(&ctx, req, &resp).ok());
return resp.code();
};
// By config name.
CHECK(call(&pb::Equipment::Stub::SetAlarm, "chiller_temp_high") == pb::Ack::ACCEPT);
rt.poll();
CHECK(rt.model().alarms.active(1));
CHECK(call(&pb::Equipment::Stub::ClearAlarm, "chiller_temp_high") == pb::Ack::ACCEPT);
rt.poll();
CHECK_FALSE(rt.model().alarms.active(1));
// By stringified ALID — always works, even for unnamed alarms.
CHECK(call(&pb::Equipment::Stub::SetAlarm, "2") == pb::Ack::ACCEPT);
rt.poll();
CHECK(rt.model().alarms.active(2));
// Unknown name.
CHECK(call(&pb::Equipment::Stub::SetAlarm, "no_such_alarm") == pb::Ack::PARAMETER_INVALID);
}
SUBCASE("FireEvent accepts a known event and rejects an unknown one") {
{
grpc::ClientContext ctx;
pb::Event req;
pb::Ack resp;
req.set_name("ProcessStarted");
CHECK(stub->FireEvent(&ctx, req, &resp).ok());
CHECK(resp.code() == pb::Ack::ACCEPT);
}
{
grpc::ClientContext ctx;
pb::Event req;
pb::Ack resp;
req.set_name("NoSuchEvent");
CHECK(stub->FireEvent(&ctx, req, &resp).ok());
CHECK(resp.code() == pb::Ack::PARAMETER_INVALID);
}
}
server->Shutdown();
}
// GetVariables needs a live io thread (read_sync posts onto it), so this case
// runs the engine in run_async() — the daemon's PRODUCTION threading mode —
// with real concurrency between the gRPC handler thread and the io thread.
TEST_CASE("GetVariables round-trip under run_async (production threading mode)") {
gem::EquipmentRuntime rt(test_config());
dmn::EquipmentService svc(rt); // snapshot BEFORE the io thread starts
rt.run_async();
grpc::ServerBuilder builder;
builder.RegisterService(&svc);
std::unique_ptr<grpc::Server> server(builder.BuildAndStart());
REQUIRE(server);
auto stub = pb::Equipment::NewStub(server->InProcessChannel(grpc::ChannelArguments{}));
// Write through the API, then read back through the API: exercises BOTH
// conversions (Value->Item with declared formats, Item->Value) end to end.
{
grpc::ClientContext ctx;
pb::VariableUpdate req;
pb::Ack resp;
(*req.mutable_values())["ChamberPressure"].set_real(2.5);
(*req.mutable_values())["WaferCounter"].set_integer(7);
REQUIRE(stub->SetVariables(&ctx, req, &resp).ok());
REQUIRE(resp.code() == pb::Ack::ACCEPT);
}
{
grpc::ClientContext ctx;
pb::VariableQuery req;
pb::VariableSnapshot resp;
req.add_names("ChamberPressure");
req.add_names("WaferCounter");
auto st = stub->GetVariables(&ctx, req, &resp);
REQUIRE(st.ok());
REQUIRE(resp.values().count("ChamberPressure") == 1);
REQUIRE(resp.values().count("WaferCounter") == 1);
CHECK(resp.values().at("ChamberPressure").real() == doctest::Approx(2.5));
CHECK(resp.values().at("WaferCounter").integer() == 7);
}
SUBCASE("empty query returns every configured variable") {
grpc::ClientContext ctx;
pb::VariableQuery req;
pb::VariableSnapshot resp;
REQUIRE(stub->GetVariables(&ctx, req, &resp).ok());
// The io thread is live here — model reads must go through read_sync
// (the first violation of that contract was caught by the TSan lane in
// exactly this line).
auto expected = rt.read_sync([&rt] {
return rt.model().svids.size() + rt.model().dvids.all().size();
});
REQUIRE(expected.has_value());
CHECK(resp.values().size() == *expected);
}
SUBCASE("unknown name is INVALID_ARGUMENT, naming the offender") {
grpc::ClientContext ctx;
pb::VariableQuery req;
pb::VariableSnapshot resp;
req.add_names("definitely_not_a_var");
auto st = stub->GetVariables(&ctx, req, &resp);
CHECK(st.error_code() == grpc::StatusCode::INVALID_ARGUMENT);
CHECK(st.error_message().find("definitely_not_a_var") != std::string::npos);
}
SUBCASE("RequestControlState walks the E30 table; WatchHealth pushes the change") {
// Open the health stream first: the initial snapshot arrives immediately.
grpc::ClientContext health_ctx;
pb::Empty empty;
auto reader = stub->WatchHealth(&health_ctx, empty);
pb::Health h;
REQUIRE(reader->Read(&h));
CHECK(h.link() == pb::Health::DISCONNECTED); // no HSMS host in this test
CHECK(h.control_state() == pb::ControlState::HOST_OFFLINE);
CHECK(h.spool_depth() == 0);
auto request = [&](pb::ControlState::State desired) {
grpc::ClientContext ctx;
pb::ControlStateRequest req;
pb::Ack resp;
req.set_desired(desired);
REQUIRE(stub->RequestControlState(&ctx, req, &resp).ok());
return resp;
};
// HostOffline --operator_online--> (AttemptOnline) --> OnlineLocal.
CHECK(request(pb::ControlState::ONLINE_LOCAL).code() == pb::Ack::ACCEPT);
CHECK(rt.control_state() == gem::ControlState::OnlineLocal);
// The stream pushes the change (well inside its 500ms poll interval).
REQUIRE(reader->Read(&h));
CHECK(h.control_state() == pb::ControlState::ONLINE_LOCAL);
// OnlineLocal --operator_remote--> OnlineRemote.
CHECK(request(pb::ControlState::ONLINE_REMOTE).code() == pb::Ack::ACCEPT);
CHECK(rt.control_state() == gem::ControlState::OnlineRemote);
// Transient state is not requestable.
CHECK(request(pb::ControlState::ATTEMPT_ONLINE).code() == pb::Ack::PARAMETER_INVALID);
// The shipped table has NO operator path to EquipmentOffline:
// operator_offline lands HostOffline — the API must say so honestly.
auto resp = request(pb::ControlState::EQUIPMENT_OFFLINE);
CHECK(resp.code() == pb::Ack::CANNOT_DO_NOW);
CHECK(resp.message().find("HostOffline") != std::string::npos);
CHECK(rt.control_state() == gem::ControlState::HostOffline);
// Requesting HOST_OFFLINE (the state operators actually get) succeeds —
// idempotently, since we're already there.
CHECK(request(pb::ControlState::HOST_OFFLINE).code() == pb::Ack::ACCEPT);
health_ctx.TryCancel();
pb::Health drain;
while (reader->Read(&drain)) {}
(void)reader->Finish(); // CANCELLED — expected
}
server->Shutdown();
rt.stop();
}
// The Subscribe command stream, per the HCACK-4 contract: a real S2F41 is
// dispatched through the full default-handler router ON the io thread, while
// the gRPC tool client consumes the stream — the daemon's production shape.
TEST_CASE("Subscribe: S2F41 -> stream -> HCACK 4; declarative fallback without subscriber") {
gem::EquipmentRuntime rt(test_config());
gem::register_default_handlers(rt); // the real S2F41/F21/F49 router path
dmn::EquipmentService svc(rt); // registers the forwarding handlers
rt.run_async();
grpc::ServerBuilder builder;
builder.RegisterService(&svc);
std::unique_ptr<grpc::Server> server(builder.BuildAndStart());
REQUIRE(server);
auto stub = pb::Equipment::NewStub(server->InProcessChannel(grpc::ChannelArguments{}));
// Dispatch a wire-shaped S2F41 on the io thread (the model's owner) and
// hand back the parsed S2F42 ack.
auto send_s2f41 = [&](const std::string& rcmd,
std::vector<gem::CommandParameter> params) {
auto reply = rt.read_sync([&rt, &rcmd, &params]() {
return rt.router().dispatch(gem::s2f41_host_command(rcmd, params));
});
REQUIRE(reply.has_value()); // read_sync answered
REQUIRE(reply->has_value()); // router produced an S2F42
auto parsed = gem::parse_s2f42(**reply);
REQUIRE(parsed.has_value());
return parsed->hcack;
};
// --- no subscriber: declarative ack (START is Accept in equipment.yaml) ---
CHECK(send_s2f41("START", {}) == gem::HostCmdAck::Accept);
// --- subscriber connected: HCACK 4 + the command arrives on the stream ----
grpc::ClientContext sub_ctx;
pb::SubscribeRequest sreq;
sreq.set_client("test-tool");
auto reader = stub->Subscribe(&sub_ctx, sreq);
// Subscription registration races the dispatch below only in this test
// (the registry insert happens on the gRPC server thread); give it a beat.
std::this_thread::sleep_for(std::chrono::milliseconds(100));
CHECK(send_s2f41("START", {{"PPID", s2::Item::ascii("RECIPE-A")}}) ==
gem::HostCmdAck::AcceptedWillFinishLater);
pb::HostRequest hr;
REQUIRE(reader->Read(&hr));
REQUIRE(hr.has_command());
CHECK(hr.command().name() == "START");
CHECK_FALSE(hr.command().id().empty());
REQUIRE(hr.command().params().count("PPID") == 1);
CHECK(hr.command().params().at("PPID").text() == "RECIPE-A");
// --- CompleteCommand: known id accepted, unknown rejected ------------------
{
grpc::ClientContext ctx;
pb::CommandResult res;
pb::Ack ack;
res.set_id(hr.command().id());
res.mutable_ack()->set_code(pb::Ack::ACCEPT);
REQUIRE(stub->CompleteCommand(&ctx, res, &ack).ok());
CHECK(ack.code() == pb::Ack::ACCEPT);
}
{
grpc::ClientContext ctx;
pb::CommandResult res;
pb::Ack ack;
res.set_id("no-such-id");
REQUIRE(stub->CompleteCommand(&ctx, res, &ack).ok());
CHECK(ack.code() == pb::Ack::PARAMETER_INVALID);
}
// --- unsubscribe: the fallback returns ------------------------------------
sub_ctx.TryCancel();
pb::HostRequest drain;
while (reader->Read(&drain)) {}
(void)reader->Finish(); // CANCELLED — expected
// The server-side Subscribe loop notices the cancel within its 500ms poll.
std::this_thread::sleep_for(std::chrono::milliseconds(700));
CHECK(send_s2f41("START", {}) == gem::HostCmdAck::Accept);
server->Shutdown();
rt.stop();
}
// Phase D — GEM300 in-the-loop (observe-and-report): job lifecycle, recipe
// downloads, and EC writes flow host -> engine -> Subscribe stream; the tool
// reports physical progress back and the engine drives the E40 FSM.
TEST_CASE("Phase D: jobs, recipes, and EC changes on the stream; ReportProcessJob") {
gem::EquipmentRuntime rt(test_config());
gem::register_default_handlers(rt);
dmn::EquipmentService svc(rt);
rt.run_async();
grpc::ServerBuilder builder;
builder.RegisterService(&svc);
std::unique_ptr<grpc::Server> server(builder.BuildAndStart());
REQUIRE(server);
auto stub = pb::Equipment::NewStub(server->InProcessChannel(grpc::ChannelArguments{}));
grpc::ClientContext sub_ctx;
pb::SubscribeRequest sreq;
auto reader = stub->Subscribe(&sub_ctx, sreq);
std::this_thread::sleep_for(std::chrono::milliseconds(100)); // registration
auto dispatch = [&](s2::Message msg) {
auto reply = rt.read_sync([&rt, &msg]() { return rt.router().dispatch(msg); });
REQUIRE(reply.has_value());
REQUIRE(reply->has_value());
};
// Host creates the PJ (S16F11), the tool sets up, the host starts it.
gem::PRJobCreateRequest pj{"PJ-D-1", gem::MaterialFlag::Substrate,
gem::ProcessRecipeMethod::RecipeOnly,
gem::RecipeSpec{"RECIPE-A", {}}, {"WFR-D-1"}, {}};
dispatch(gem::s16f11_pr_job_create(pj));
auto setup = rt.read_sync([&rt]() {
rt.model().process_jobs.fire_internal("PJ-D-1", gem::ProcessJobEvent::Select);
return rt.model().process_jobs.fire_internal("PJ-D-1",
gem::ProcessJobEvent::SetupComplete);
});
REQUIRE(setup.has_value());
REQUIRE(*setup);
dispatch(gem::s16f5_pr_job_command("PJ-D-1", "PJSTART"));
// The tool's stream receives the job with recipe + material bindings.
pb::HostRequest hr;
REQUIRE(reader->Read(&hr));
REQUIRE(hr.has_process_job());
CHECK(hr.process_job().job_id() == "PJ-D-1");
CHECK(hr.process_job().action() == pb::ProcessJob::START);
CHECK(hr.process_job().recipe() == "RECIPE-A");
REQUIRE(hr.process_job().carriers_size() == 1);
CHECK(hr.process_job().carriers(0) == "WFR-D-1");
// The tool finishes the physical work and reports; the engine's FSM follows.
{
grpc::ClientContext ctx;
pb::ProcessJobState rep;
pb::Ack ack;
rep.set_job_id("PJ-D-1");
rep.set_state(pb::ProcessJobState::COMPLETE);
REQUIRE(stub->ReportProcessJob(&ctx, rep, &ack).ok());
CHECK(ack.code() == pb::Ack::ACCEPT);
}
auto state = rt.read_sync(
[&rt]() { return rt.model().process_jobs.state("PJ-D-1"); });
REQUIRE(state.has_value());
CHECK(*state == gem::ProcessJobState::ProcessComplete);
// Reporting against an unknown job / an illegal transition is rejected.
{
grpc::ClientContext ctx;
pb::ProcessJobState rep;
pb::Ack ack;
rep.set_job_id("PJ-GHOST");
rep.set_state(pb::ProcessJobState::COMPLETE);
REQUIRE(stub->ReportProcessJob(&ctx, rep, &ack).ok());
CHECK(ack.code() == pb::Ack::INVALID_OBJECT);
}
{
grpc::ClientContext ctx;
pb::ProcessJobState rep;
pb::Ack ack;
rep.set_job_id("PJ-D-1"); // already ProcessComplete
rep.set_state(pb::ProcessJobState::COMPLETE);
REQUIRE(stub->ReportProcessJob(&ctx, rep, &ack).ok());
CHECK(ack.code() == pb::Ack::CANNOT_DO_NOW);
}
// Host downloads a recipe (S7F3) -> ProcessProgram on the stream.
dispatch(gem::s7f3_process_program_send("RECIPE-NEW", "step1\nstep2\n"));
REQUIRE(reader->Read(&hr));
REQUIRE(hr.has_process_program());
CHECK(hr.process_program().ppid() == "RECIPE-NEW");
CHECK(hr.process_program().body() == "step1\nstep2\n");
// Host writes an EC (S2F15) -> ConstantChange with the configured name.
dispatch(gem::s2f15_ec_send({{10, s2::Item::u4(uint32_t{1})}}));
REQUIRE(reader->Read(&hr));
REQUIRE(hr.has_constant());
CHECK(hr.constant().name() == "TimeFormat");
CHECK(hr.constant().value().integer() == 1);
sub_ctx.TryCancel();
pb::HostRequest drain;
while (reader->Read(&drain)) {}
(void)reader->Finish();
server->Shutdown();
rt.stop();
}
// D10 + E16: carriers on the stream, Describe / FlushSpool / terminal.
TEST_CASE("carriers (D10) and the operations RPCs (E16)") {
gem::EquipmentRuntime rt(test_config());
gem::register_default_handlers(rt);
dmn::EquipmentService svc(rt);
rt.run_async();
grpc::ServerBuilder builder;
builder.RegisterService(&svc);
std::unique_ptr<grpc::Server> server(builder.BuildAndStart());
REQUIRE(server);
auto stub = pb::Equipment::NewStub(server->InProcessChannel(grpc::ChannelArguments{}));
grpc::ClientContext sub_ctx;
pb::SubscribeRequest sreq;
auto reader = stub->Subscribe(&sub_ctx, sreq);
std::this_thread::sleep_for(std::chrono::milliseconds(100));
auto report_carrier = [&](const std::string& cid, pb::CarrierState::State st,
std::vector<bool> slots = {}) {
grpc::ClientContext ctx;
pb::CarrierState req;
pb::Ack ack;
req.set_carrier_id(cid);
req.set_port(2);
req.set_state(st);
for (bool b : slots) req.add_slots(b);
REQUIRE(stub->ReportCarrier(&ctx, req, &ack).ok());
return ack.code();
};
// Tool announces an arrived FOUP with a slot map.
CHECK(report_carrier("CAR-D-1", pb::CarrierState::WAITING,
{true, true, false}) == pb::Ack::ACCEPT);
auto created = rt.read_sync([&rt]() {
const auto* c = rt.model().carriers.get("CAR-D-1");
return c && c->slots.size() == 3 && c->slots[0].state == 1 &&
c->slots[2].state == 0 && c->port_id == 2;
});
REQUIRE(created.has_value());
CHECK(*created);
// Host says ProceedWithCarrier (S3F17) -> tool stream gets PROCEED.
auto reply = rt.read_sync([&rt]() {
return rt.router().dispatch(
gem::s3f17_carrier_action(0u, "ProceedWithCarrier", "CAR-D-1", {}));
});
REQUIRE(reply.has_value());
REQUIRE(reply->has_value());
pb::HostRequest hr;
REQUIRE(reader->Read(&hr));
REQUIRE(hr.has_carrier());
CHECK(hr.carrier().carrier_id() == "CAR-D-1");
CHECK(hr.carrier().port() == 2);
CHECK(hr.carrier().action() == pb::CarrierAction::PROCEED);
// Tool drives access: begin + end; reporting against an unknown id fails.
CHECK(report_carrier("CAR-D-1", pb::CarrierState::IN_ACCESS) == pb::Ack::ACCEPT);
CHECK(report_carrier("CAR-D-1", pb::CarrierState::COMPLETE) == pb::Ack::ACCEPT);
CHECK(report_carrier("CAR-GHOST", pb::CarrierState::IN_ACCESS) ==
pb::Ack::INVALID_OBJECT);
// Host cancels CAR-D-1 (Confirmed -> NotConfirmed is a real transition, so
// the observer fires) -> CANCEL on the stream. NOTE: a CancelCarrier on a
// still-NotConfirmed carrier is a self-transition the FSM doesn't signal,
// so the tool isn't told — a known E87 edge (see DAEMON_ROADMAP).
(void)rt.read_sync([&rt]() {
return rt.router().dispatch(
gem::s3f17_carrier_action(0u, "CancelCarrier", "CAR-D-1", {}));
});
REQUIRE(reader->Read(&hr));
REQUIRE(hr.has_carrier());
CHECK(hr.carrier().action() == pb::CarrierAction::CANCEL);
// ---- E16: Describe / FlushSpool / SendTerminalMessage --------------------
{
grpc::ClientContext ctx;
pb::Empty req;
pb::EquipmentDescription d;
REQUIRE(stub->Describe(&ctx, req, &d).ok());
CHECK(d.model_name() == "SECSGEM-SIM");
auto has = [](const auto& list, const std::string& want) {
for (const auto& e : list)
if (e == want) return true;
return false;
};
CHECK(has(d.variables(), "ChamberPressure"));
CHECK(has(d.events(), "ProcessStarted"));
CHECK(has(d.alarms(), "chiller_temp_high"));
CHECK(has(d.commands(), "START"));
CHECK(has(d.constants(), "TimeFormat"));
}
{
// No host: a fired event spools (stream 6 is spoolable); purge empties it.
grpc::ClientContext fctx;
pb::Event ev;
pb::Ack ack;
ev.set_name("ProcessStarted");
// Enable the event first so emit isn't suppressed.
(void)rt.read_sync([&rt]() {
return rt.model().enable_events(true, {300});
});
REQUIRE(stub->FireEvent(&fctx, ev, &ack).ok());
std::this_thread::sleep_for(std::chrono::milliseconds(100));
auto depth = rt.read_sync([&rt]() { return rt.model().spool.size(); });
REQUIRE(depth.has_value());
CHECK(*depth == 1);
grpc::ClientContext pctx;
pb::SpoolFlushRequest freq;
pb::Ack fack;
freq.set_purge(true);
REQUIRE(stub->FlushSpool(&pctx, freq, &fack).ok());
CHECK(fack.code() == pb::Ack::ACCEPT);
depth = rt.read_sync([&rt]() { return rt.model().spool.size(); });
CHECK(*depth == 0);
}
{
// Stream 10 is not spoolable and no host is connected -> honest refusal.
grpc::ClientContext ctx;
pb::TerminalMessage req;
pb::Ack ack;
req.set_tid(0);
req.set_text("hello fab");
REQUIRE(stub->SendTerminalMessage(&ctx, req, &ack).ok());
CHECK(ack.code() == pb::Ack::CANNOT_DO_NOW);
}
sub_ctx.TryCancel();
pb::HostRequest drain;
while (reader->Read(&drain)) {}
(void)reader->Finish();
server->Shutdown();
rt.stop();
}
// Randomized concurrent stress: several client threads fire a seeded random
// mix of RPCs against the live service while the io thread runs — the
// strongest TSan target we have, and a probe for ordering/lifetime bugs the
// scenario tests can't reach. Failures print the seed for reproduction.
TEST_CASE("randomized concurrent RPC stress (seeded)") {
const unsigned seed = static_cast<unsigned>(
std::random_device{}()); // logged below; rerun by hardcoding it
INFO("stress seed = " << seed);
gem::EquipmentRuntime rt(test_config());
gem::register_default_handlers(rt);
dmn::EquipmentService svc(rt);
rt.run_async();
grpc::ServerBuilder builder;
builder.RegisterService(&svc);
std::unique_ptr<grpc::Server> server(builder.BuildAndStart());
REQUIRE(server);
auto stub = pb::Equipment::NewStub(server->InProcessChannel(grpc::ChannelArguments{}));
constexpr int kThreads = 4;
constexpr int kOpsPerThread = 250;
std::atomic<int> failures{0};
auto worker = [&](unsigned tseed) {
std::mt19937 rng(tseed);
std::uniform_int_distribution<int> op(0, 7);
std::uniform_real_distribution<double> val(0.0, 10.0);
for (int i = 0; i < kOpsPerThread; ++i) {
grpc::ClientContext ctx;
pb::Ack ack;
grpc::Status st = grpc::Status::OK;
switch (op(rng)) {
case 0: {
pb::VariableUpdate r;
(*r.mutable_values())["ChamberPressure"].set_real(val(rng));
st = stub->SetVariables(&ctx, r, &ack);
break;
}
case 1: {
pb::VariableQuery r;
pb::VariableSnapshot snap;
st = stub->GetVariables(&ctx, r, &snap);
if (st.ok() && snap.values().empty()) ++failures; // config never empty
break;
}
case 2: {
pb::Event r;
r.set_name("ProcessStarted");
st = stub->FireEvent(&ctx, r, &ack);
break;
}
case 3: {
pb::Alarm r;
r.set_name("chiller_temp_high");
st = stub->SetAlarm(&ctx, r, &ack);
break;
}
case 4: {
pb::Alarm r;
r.set_name("chiller_temp_high");
st = stub->ClearAlarm(&ctx, r, &ack);
break;
}
case 5: {
pb::Empty r;
pb::ControlState cs;
st = stub->GetControlState(&ctx, r, &cs);
break;
}
case 6: {
pb::Empty r;
pb::EquipmentDescription d;
st = stub->Describe(&ctx, r, &d);
if (st.ok() && d.variables_size() == 0) ++failures;
break;
}
case 7: {
// Subscribe/cancel churn: exercises subscriber add/remove under load.
grpc::ClientContext sctx;
pb::SubscribeRequest sr;
auto rd = stub->Subscribe(&sctx, sr);
sctx.TryCancel();
pb::HostRequest hr;
while (rd->Read(&hr)) {}
(void)rd->Finish();
break;
}
}
if (!st.ok()) ++failures;
}
};
std::vector<std::thread> threads;
for (int t = 0; t < kThreads; ++t) threads.emplace_back(worker, seed + t);
for (auto& th : threads) th.join();
CHECK(failures.load() == 0);
// The engine must still be fully responsive afterwards.
grpc::ClientContext ctx;
pb::Empty req;
pb::ControlState cs;
CHECK(stub->GetControlState(&ctx, req, &cs).ok());
server->Shutdown();
rt.stop();
}
// Phase 16 tail: E90 substrate + E157 module reporting (observe-and-report).
TEST_CASE("ReportSubstrate (E90) and ReportModule (E157) drive the FSMs") {
gem::EquipmentRuntime rt(test_config());
gem::register_default_handlers(rt);
dmn::EquipmentService svc(rt);
rt.run_async();
grpc::ServerBuilder builder;
builder.RegisterService(&svc);
std::unique_ptr<grpc::Server> server(builder.BuildAndStart());
REQUIRE(server);
auto stub = pb::Equipment::NewStub(server->InProcessChannel(grpc::ChannelArguments{}));
auto sub = [&](const std::string& sid, pb::SubstrateReport::Milestone m) {
grpc::ClientContext ctx;
pb::SubstrateReport req;
pb::Ack ack;
req.set_substrate_id(sid);
req.set_milestone(m);
REQUIRE(stub->ReportSubstrate(&ctx, req, &ack).ok());
return ack.code();
};
auto sub_with = [&](const std::string& sid, pb::SubstrateReport::Milestone m,
const std::string& cid, uint32_t slot) {
grpc::ClientContext ctx;
pb::SubstrateReport req;
pb::Ack ack;
req.set_substrate_id(sid);
req.set_milestone(m);
req.set_carrier_id(cid);
req.set_slot(slot);
REQUIRE(stub->ReportSubstrate(&ctx, req, &ack).ok());
return ack.code();
};
// A wafer's journey: arrive -> picked up -> process -> done -> deposited.
// ARRIVED also records where the wafer came from (carrier + slot).
CHECK(sub_with("WFR-1", pb::SubstrateReport::ARRIVED, "FOUP-7", 3) == pb::Ack::ACCEPT);
auto origin = rt.read_sync([&rt]() {
const auto* s = rt.model().substrates.get("WFR-1");
return s ? std::make_pair(s->carrierid, s->slot)
: std::make_pair(std::string{}, uint8_t{0});
});
REQUIRE(origin.has_value());
CHECK(origin->first == "FOUP-7");
CHECK(static_cast<int>(origin->second) == 3);
CHECK(sub("WFR-1", pb::SubstrateReport::AT_WORK) == pb::Ack::ACCEPT);
CHECK(sub("WFR-1", pb::SubstrateReport::PROCESSING) == pb::Ack::ACCEPT);
CHECK(sub("WFR-1", pb::SubstrateReport::PROCESSED) == pb::Ack::ACCEPT);
CHECK(sub("WFR-1", pb::SubstrateReport::AT_DESTINATION) == pb::Ack::ACCEPT);
auto loc = rt.read_sync([&rt]() {
const auto* s = rt.model().substrates.get("WFR-1");
return s ? s->fsm->location_state() : gem::SubstrateState::NoState;
});
REQUIRE(loc.has_value());
CHECK(*loc == gem::SubstrateState::AtDestination);
// Reporting on a substrate that never ARRIVED is rejected (no such object).
CHECK(sub("WFR-GHOST", pb::SubstrateReport::AT_WORK) == pb::Ack::INVALID_OBJECT);
// Duplicate ARRIVED is a conflict, not a missing object: CANNOT_DO_NOW so we
// never silently wipe the existing wafer's state/history.
CHECK(sub("WFR-1", pb::SubstrateReport::ARRIVED) == pb::Ack::CANNOT_DO_NOW);
auto mod = [&](const std::string& mid, pb::ModuleReport::State st) {
grpc::ClientContext ctx;
pb::ModuleReport req;
pb::Ack ack;
req.set_module_id(mid);
req.set_state(st);
REQUIRE(stub->ReportModule(&ctx, req, &ack).ok());
return ack.code();
};
// Module: auto-created, then walked General -> Step -> StepCompleted -> Reset.
CHECK(mod("MOD-1", pb::ModuleReport::GENERAL_EXECUTING) == pb::Ack::ACCEPT);
CHECK(mod("MOD-1", pb::ModuleReport::STEP_EXECUTING) == pb::Ack::ACCEPT);
CHECK(mod("MOD-1", pb::ModuleReport::STEP_COMPLETED) == pb::Ack::ACCEPT);
auto mstate = rt.read_sync([&rt]() {
const auto* m = rt.model().modules.get("MOD-1");
return m ? m->fsm->state() : gem::ModuleState::NotExecuting;
});
REQUIRE(mstate.has_value());
CHECK(*mstate == gem::ModuleState::StepCompleted);
// NOT_EXECUTING resets the module back to idle (re-usable for the next wafer).
CHECK(mod("MOD-1", pb::ModuleReport::NOT_EXECUTING) == pb::Ack::ACCEPT);
auto reset = rt.read_sync([&rt]() {
const auto* m = rt.model().modules.get("MOD-1");
return m ? m->fsm->state() : gem::ModuleState::NoState;
});
REQUIRE(reset.has_value());
CHECK(*reset == gem::ModuleState::NotExecuting);
// An illegal jump (StepExecuting straight from a fresh NotExecuting module)
// is rejected by the E157 table.
CHECK(mod("MOD-2", pb::ModuleReport::STEP_EXECUTING) == pb::Ack::CANNOT_DO_NOW);
server->Shutdown();
rt.stop();
}
Reference in New Issue View Git Blame Copy Permalink