Files
Joseph Doherty b0ef23c087 test(mqtt): Sparkplug array metrics in the simulator + live gate (lmxopcua#770)
The simulator now packs Int32/Boolean/Float/Double/String arrays per the
Sparkplug B spec (independent of the driver codec) and seeds Profile, Flags
and Curve in the standalone catalog. SparkplugLiveTests gains a live test that
asserts typed int[]/bool[]/float[] values from the birth, a changed vector
(including a different Boolean length) from alias-only DATA, the read path,
and BadTypeMismatch for an array metric bound without isArray. Run against a
throwaway local Mosquitto; mutation-checked (scalar binding / seed removed).

Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
2026-11-09 18:15:13 -05:00

179 lines
7.3 KiB
C#

using Org.Eclipse.Tahu.Protobuf;
using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.SparkplugSimulator;
// ---------------------------------------------------------------------------
// Standalone entry point for the `sparkplug-sim` fixture container.
//
// Runs one or more simulated edge nodes against the fixture broker, forever: NBIRTH + DBIRTH once,
// then N/DDATA on a cadence, honouring rebirth NCMDs throughout. It exists so an operator (and Task
// 26's AdminUI `/run` verification) has a live Sparkplug plant to browse, NOT so the live tests have
// one — those drive `SparkplugEdgeNode` in process, because the §3.6 matrix needs an edge node it
// can command mid-test.
//
// Every setting comes from the environment; nothing is defaulted to a credential.
// ---------------------------------------------------------------------------
var host = Env("MQTT_SIM_HOST", "mosquitto");
var port = int.TryParse(Env("MQTT_SIM_PORT", "8883"), out var p) ? p : 8883;
var useTls = !string.Equals(Env("MQTT_SIM_USE_TLS", "true"), "false", StringComparison.OrdinalIgnoreCase);
var group = Env("MQTT_SIM_GROUP", "OtOpcUaSim");
var nodes = Env("MQTT_SIM_NODES", "EdgeA,EdgeB").Split(',', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries);
var interval = int.TryParse(Env("MQTT_SIM_INTERVAL_SECONDS", "2"), out var i) && i > 0 ? i : 2;
var username = Environment.GetEnvironmentVariable("MQTT_FIXTURE_USERNAME");
var password = Environment.GetEnvironmentVariable("MQTT_FIXTURE_PASSWORD");
var caPath = Environment.GetEnvironmentVariable("MQTT_SIM_CA_CERT");
var allowUntrusted = string.Equals(Env("MQTT_SIM_ALLOW_UNTRUSTED", "false"), "true", StringComparison.OrdinalIgnoreCase);
if (string.IsNullOrEmpty(password))
{
// Same fail-loud posture as the compose file's `${VAR:?}` guards: the broker runs
// allow_anonymous=false, so a blank password would produce an unexplained connect refusal.
Console.Error.WriteLine(
"sparkplug-sim: MQTT_FIXTURE_PASSWORD is not set. The fixture broker runs allow_anonymous=false; "
+ "supply the password gen-fixture-material.sh was run with.");
return 2;
}
using var stopping = new CancellationTokenSource();
Console.CancelKeyPress += (_, e) =>
{
e.Cancel = true;
stopping.Cancel();
};
AppDomain.CurrentDomain.ProcessExit += (_, _) => stopping.Cancel();
Console.WriteLine(
$"sparkplug-sim: broker={host}:{port} tls={useTls} group={group} nodes=[{string.Join(", ", nodes)}] "
+ $"interval={interval}s");
var running = new List<Task>();
var edges = new List<SparkplugEdgeNode>();
foreach (var node in nodes)
{
var edge = new SparkplugEdgeNode(
new SparkplugEdgeNodeOptions
{
Host = host,
Port = port,
UseTls = useTls,
CaCertificatePath = caPath,
AllowUntrustedServerCertificate = allowUntrusted,
Username = username,
Password = password,
GroupId = group,
EdgeNodeId = node,
},
Console.WriteLine);
// A small but genuinely heterogeneous catalog: a float, a double, a NEGATIVE int32 (the two's
// complement case), a bool and a string — plus one device, so the picker's
// Group → EdgeNode → Device → Metric tree has every level populated.
edge.SetNodeMetrics(
new SparkplugMetricSeed("Temperature", 1, DataType.Float, 21.5f),
new SparkplugMetricSeed("Pressure", 2, DataType.Double, 101.325d),
new SparkplugMetricSeed("Count", 3, DataType.Int32, -42),
new SparkplugMetricSeed("Running", 4, DataType.Boolean, true),
new SparkplugMetricSeed("Serial", 5, DataType.String, $"{node}-001"),
// Array metrics (lmxopcua#770): bind with "isArray": true.
new SparkplugMetricSeed("Profile", 6, DataType.Int32Array, new[] { 1, -2, 300000 }),
new SparkplugMetricSeed("Flags", 7, DataType.BooleanArray, new[] { true, false, true, true, false, false, true, false, true }),
new SparkplugMetricSeed("Curve", 8, DataType.FloatArray, new[] { 0.5f, 1.25f, -2.5f }));
edge.SetDeviceMetrics(
"Filler1",
new SparkplugMetricSeed("Temperature", 10, DataType.Float, 55.5f),
new SparkplugMetricSeed("FillCount", 11, DataType.Int64, 1000L),
new SparkplugMetricSeed("Jammed", 12, DataType.Boolean, false),
// Structured metrics for the Gitea #515 gate. They live on the DEVICE rather than the node
// deliberately: a DDATA template is the case a naive implementation gets wrong, because its
// metric carries the parent's ALIAS and no name, so the member names cannot be composed until
// the alias table has resolved the parent.
new SparkplugMetricSeed("Motor", 13, DataType.Template, BuildMotorTemplate()),
new SparkplugMetricSeed("Recipe", 14, DataType.DataSet, BuildRecipeDataSet()));
edges.Add(edge);
}
try
{
foreach (var edge in edges)
{
await edge.ConnectAsync(stopping.Token);
running.Add(edge.RunAsync(TimeSpan.FromSeconds(interval), stopping.Token));
}
await Task.WhenAll(running);
}
catch (OperationCanceledException)
{
Console.WriteLine("sparkplug-sim: stopping.");
}
finally
{
foreach (var edge in edges)
{
await edge.DisposeAsync();
}
}
return 0;
static string Env(string name, string fallback) =>
Environment.GetEnvironmentVariable(name) is { Length: > 0 } value ? value : fallback;
// A UDT instance: `Motor` with a `Speed` double, a `Running` bool and a nested `Stator` template.
// Flattens to `Motor/Speed`, `Motor/Running` and `Motor/Stator/Temp` (Gitea #515).
static Payload.Types.Template BuildMotorTemplate()
{
var stator = new Payload.Types.Template { Version = "1", IsDefinition = false };
stator.Metrics.Add(new Payload.Types.Metric
{
Name = "Temp", Datatype = (uint)DataType.Double, DoubleValue = 71.5,
});
var motor = new Payload.Types.Template { Version = "1", IsDefinition = false };
motor.Metrics.Add(new Payload.Types.Metric
{
Name = "Speed", Datatype = (uint)DataType.Double, DoubleValue = 1500,
});
motor.Metrics.Add(new Payload.Types.Metric
{
Name = "Running", Datatype = (uint)DataType.Boolean, BooleanValue = true,
});
// Nested one level: proves the recursion, and proves the `/`-joined path composes correctly rather
// than only working for flat templates.
motor.Metrics.Add(new Payload.Types.Metric
{
Name = "Stator", Datatype = (uint)DataType.Template, TemplateValue = stator,
});
return motor;
}
// A two-column, two-row table. Flattens to `Recipe/Step/0`, `Recipe/Target/0`, `Recipe/Step/1`,
// `Recipe/Target/1` — the row index is always present, so a table that grows a row never renames the
// tags already authored against it.
static Payload.Types.DataSet BuildRecipeDataSet()
{
var dataSet = new Payload.Types.DataSet { NumOfColumns = 2UL };
dataSet.Columns.Add("Step");
dataSet.Columns.Add("Target");
dataSet.Types_.Add((uint)DataType.Int32);
dataSet.Types_.Add((uint)DataType.Double);
for (var row = 0; row < 2; row++)
{
var r = new Payload.Types.DataSet.Types.Row();
r.Elements.Add(new Payload.Types.DataSet.Types.DataSetValue { IntValue = (uint)row });
r.Elements.Add(new Payload.Types.DataSet.Types.DataSetValue { DoubleValue = 20d + row });
dataSet.Rows.Add(r);
}
return dataSet;
}