diff --git a/ZB.MOM.WW.OtOpcUa.slnx b/ZB.MOM.WW.OtOpcUa.slnx index 7c5147f0..17da9866 100644 --- a/ZB.MOM.WW.OtOpcUa.slnx +++ b/ZB.MOM.WW.OtOpcUa.slnx @@ -109,6 +109,7 @@ + diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/.gitignore b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/.gitignore index b69c261a..7e91479f 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/.gitignore +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/.gitignore @@ -5,6 +5,11 @@ # git; its output is regenerated per machine. secrets/ +# Build output of publish-simulator.sh — the `sparkplug-sim` service's bind-mounted app +# directory. Regenerated per machine from the SparkplugSimulator project, exactly like +# secrets/ is regenerated by gen-fixture-material.sh; the SCRIPT is the artifact in git. +simulator/app/ + # Belt-and-braces: catch key/cert material written directly into this directory by a # hand-run openssl command that forgot the secrets/ prefix. *.key diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/docker-compose.yml b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/docker-compose.yml index a1ecfec7..2c42a8fb 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/docker-compose.yml +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/docker-compose.yml @@ -1,4 +1,5 @@ -# MQTT integration-test fixture — Eclipse Mosquitto (auth + TLS) + a JSON publisher. +# MQTT integration-test fixture — Eclipse Mosquitto (auth + TLS) + a JSON publisher +# + an optional Sparkplug B edge-node simulator. # # PREREQUISITE — run this first, in this directory: # @@ -15,9 +16,10 @@ # # Endpoints: 10.100.0.35:8883 (TLS + auth) · 10.100.0.35:1883 (plaintext + auth, smoke) # -# Unlike the Modbus / S7 fixtures there are no compose profiles: both services are -# always wanted together (a broker with nothing publishing into it tests nothing), and -# they bind different ports so nothing contends. +# `mosquitto` + `publisher` carry no profile: they are always wanted together (a broker +# with nothing publishing into it tests nothing) and they bind different ports so nothing +# contends. `sparkplug-sim` DOES carry one (`--profile sparkplug`), because it is a +# manual-driving aid rather than a test dependency — see its own banner. services: mosquitto: image: eclipse-mosquitto:2.0.22 @@ -75,3 +77,49 @@ services: MQTT_FIXTURE_USERNAME: ${MQTT_FIXTURE_USERNAME:?set MQTT_FIXTURE_USERNAME} MQTT_FIXTURE_PASSWORD: ${MQTT_FIXTURE_PASSWORD:?set MQTT_FIXTURE_PASSWORD} entrypoint: ["/bin/sh", "/publish.sh"] + + # --------------------------------------------------------------------------- + # Sparkplug B edge-node simulator — the SAME `SparkplugEdgeNode` engine the live + # tests drive, run standalone (see SparkplugSimulator/Program.cs). + # + # WHAT IT IS FOR, AND WHAT IT IS NOT FOR. It exists so an operator (and the Task-26 + # AdminUI `/run` verification) has a live Sparkplug plant to point the address picker + # at: NBIRTH/DBIRTH then N/DDATA every couple of seconds, honouring rebirth NCMDs. + # `SparkplugLiveTests` does NOT use it and must not — the §3.6 matrix needs an edge + # node it can command mid-test ("now reuse that alias", "now skip a seq", "now die"), + # which a container cannot be. Hence the profile: the default `docker compose up` + # leaves it out, and nothing in the test suite depends on it. + # + # PREREQUISITE: `./publish-simulator.sh` — the app directory below is build output, + # gitignored and rsynced like the rest of this folder. Without it the container has + # nothing to run. + # --------------------------------------------------------------------------- + sparkplug-sim: + image: mcr.microsoft.com/dotnet/runtime:10.0 + container_name: otopcua-sparkplug-sim + profiles: ["sparkplug"] + restart: unless-stopped + labels: + project: lmxopcua + depends_on: + mosquitto: + condition: service_healthy + volumes: + # Framework-dependent publish output: portable IL, so it runs under the stock + # runtime image whatever this file was rsynced from. + - ./simulator/app:/app:ro + # The fixture CA, so the simulator PINS the broker exactly as the driver does + # rather than trusting anything. The cert's SAN list includes DNS:mosquitto (see + # gen-fixture-material.sh), which is the host name used below. + - ./secrets/ca.crt:/secrets/ca.crt:ro + environment: + MQTT_SIM_HOST: mosquitto + MQTT_SIM_PORT: "8883" + MQTT_SIM_USE_TLS: "true" + MQTT_SIM_CA_CERT: /secrets/ca.crt + MQTT_SIM_GROUP: ${MQTT_SIM_GROUP:-OtOpcUaSim} + MQTT_SIM_NODES: ${MQTT_SIM_NODES:-EdgeA,EdgeB} + MQTT_SIM_INTERVAL_SECONDS: ${MQTT_SIM_INTERVAL_SECONDS:-2} + MQTT_FIXTURE_USERNAME: ${MQTT_FIXTURE_USERNAME:?set MQTT_FIXTURE_USERNAME} + MQTT_FIXTURE_PASSWORD: ${MQTT_FIXTURE_PASSWORD:?set MQTT_FIXTURE_PASSWORD} + command: ["dotnet", "/app/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.SparkplugSimulator.dll"] diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/publish-simulator.sh b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/publish-simulator.sh new file mode 100755 index 00000000..0ab2fd95 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/Docker/publish-simulator.sh @@ -0,0 +1,41 @@ +#!/usr/bin/env sh +# --------------------------------------------------------------------------- +# Publishes the Sparkplug edge-node simulator into ./simulator/app/, which the +# compose file's `sparkplug-sim` service bind-mounts and runs. +# +# WHY A PUBLISHED DIRECTORY RATHER THAN AN IMAGE BUILD. This fixture is deployed by +# rsyncing THIS directory to /opt/otopcua-mqtt on the Docker host (see the compose +# banner and infra/README.md §1) — the host never sees the repo, so a Dockerfile whose +# build context is the solution root could not run there. A framework-dependent publish +# is portable IL: it runs unchanged under the stock dotnet/runtime image on the host, +# whatever this machine's architecture is. +# +# Usage, from a checkout (needs the .NET 10 SDK): +# +# ./publish-simulator.sh +# rsync -av --exclude 'secrets/' ./ dohertj2@10.100.0.35:/opt/otopcua-mqtt/ +# ssh dohertj2@10.100.0.35 'cd /opt/otopcua-mqtt \ +# && MQTT_FIXTURE_USERNAME=otopcua MQTT_FIXTURE_PASSWORD= \ +# docker compose --profile sparkplug up -d' +# +# ./simulator/app/ is gitignored: it is build output, regenerated per machine, exactly +# like ./secrets/. +# --------------------------------------------------------------------------- +set -eu + +SCRIPT_DIR=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd) +PROJECT="$SCRIPT_DIR/../SparkplugSimulator/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.SparkplugSimulator.csproj" +OUT="$SCRIPT_DIR/simulator/app" + +command -v dotnet >/dev/null 2>&1 || { echo "error: dotnet SDK not found on PATH." >&2; exit 2; } +[ -f "$PROJECT" ] || { echo "error: simulator project not found at $PROJECT" >&2; exit 2; } + +echo "==> publishing $(basename "$PROJECT") -> $OUT" +rm -rf "$OUT" +dotnet publish "$PROJECT" -c Release -o "$OUT" --nologo + +echo "==> done:" +ls -1 "$OUT" | head -20 +echo +echo "Next: rsync this directory to the Docker host and bring the stack up with" +echo " docker compose --profile sparkplug up -d" diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugLiveTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugLiveTests.cs new file mode 100644 index 00000000..1a37c0c8 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugLiveTests.cs @@ -0,0 +1,979 @@ +using System.Buffers; +using System.Collections.Concurrent; +using System.Text.Json; +using MQTTnet; +using Org.Eclipse.Tahu.Protobuf; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Browsing; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser; +using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug; +using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.SparkplugSimulator; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests; + +/// +/// The Sparkplug B live gate: the §3.6 state-machine matrix driven end-to-end over the real +/// Mosquitto fixture, against a real edge node () that encodes with +/// the same generated Tahu schema the driver decodes with. +/// +/// What the live gate adds over the offline suite. Every case here already has an +/// offline pin driven through SparkplugIngestor.Dispatch with hand-built decoded +/// payloads. Those prove the logic. They cannot prove the wire: that a negative +/// Int32 really does arrive as an unsigned two's-complement int_value, that a DATA +/// metric really does carry an alias and nothing else, that an NDEATH really is a broker-issued +/// Will with no seq, or that the NCMD this driver encodes is one an independent decoder +/// accepts as a rebirth request. A hand-built SparkplugPayload shaped by the same +/// assumptions as the decoder cannot falsify any of that; a byte stream through a broker can. +/// +/// +/// Env-gated + skip-clean, exactly as : without +/// MQTT_FIXTURE_ENDPOINT (and credentials) every test calls Assert.Skip. See +/// for the env-var list. +/// +/// +/// Every test uses its own Sparkplug group id. The fixture broker is shared and the +/// driver subscribes spBv1.0/{group}/#; a fixed group would let one test's edge node +/// feed another's driver, and — worse — let a previous run's traffic satisfy an assertion. +/// +/// +[Collection(MqttFixtureCollection.Name)] +[Trait("Category", "LiveIntegration")] +public sealed class SparkplugLiveTests(MqttFixture fixture) +{ + /// OPC UA Good. + private const uint Good = 0x00000000u; + + /// + /// OPC UA BadCommunicationError — the quality the ingest state machine stamps on a tag + /// whose source announced it is offline (NDEATH/DDEATH). + /// + private const uint BadCommunicationError = 0x80050000u; + + /// + /// Bound on every wait for broker traffic. Generous enough to absorb a TLS handshake plus a + /// rebirth round-trip; short enough that a genuinely dark path fails rather than hangs. + /// + private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(30); + + /// + /// How long a "this must NOT happen" assertion watches for. Long enough that the message would + /// have arrived (a broker round-trip on this rig is single-digit milliseconds), short enough not + /// to dominate the suite. + /// + private static readonly TimeSpan NegativeWindow = TimeSpan.FromSeconds(4); + + private readonly MqttFixture _fx = fixture; + + // ----------------------------------------------------------------------- + // birth → data → alias resolve → OnDataChange (+ §3.6 case iv: negative Int32) + // ----------------------------------------------------------------------- + + /// + /// The spine of the driver: an NBIRTH/DBIRTH builds the alias table, an alias-only NDATA/DDATA + /// resolves through it, and the value is published under the tag's RawPath — the v3 + /// driver wire reference, not the Sparkplug tuple and not the topic. + /// + /// It simultaneously carries §3.6 matrix case (iv), the negative Int32. The metric's + /// value of −1234 travels as int_value = 4294966062 because that is what Sparkplug + /// says; a driver that skipped ReinterpretSigned would publish 4294966062 at Good + /// quality — a plausible number, a wrong one, and invisible to any assertion that only + /// checks "a value arrived". + /// + /// + [Fact] + public async Task Birth_then_alias_only_data_publishes_under_rawpath_with_signed_ints_intact() + { + SkipIfUnconfigured(); + + var ct = TestContext.Current.CancellationToken; + var group = NewGroup("birth"); + const string node = "EdgeA"; + const string device = "Filler1"; + + const string tempPath = "Fixture/Sparkplug/EdgeA/Temperature"; + const string countPath = "Fixture/Sparkplug/EdgeA/Count"; + const string levelPath = "Fixture/Sparkplug/EdgeA/Filler1/Level"; + + await using var sim = new SparkplugEdgeNode(SimOptions(group, node)); + sim.SetNodeMetrics( + new SparkplugMetricSeed("Temperature", 1, DataType.Float, 21.5f), + new SparkplugMetricSeed("Count", 2, DataType.Int32, -42)); + sim.SetDeviceMetrics(device, new SparkplugMetricSeed("Level", 10, DataType.Double, 3.5d)); + + await sim.ConnectAsync(ct); + await sim.BirthAsync(ct); + + await using var driver = new MqttDriver( + SparkplugOptions( + group, + Tag(tempPath, group, node, null, "Temperature"), + Tag(countPath, group, node, null, "Count"), + Tag(levelPath, group, node, device, "Level")), + "mqtt-live-sp-birth"); + var recorder = new SparkplugRecorder(driver); + + await driver.InitializeAsync(string.Empty, ct); + await driver.SubscribeAsync([tempPath, countPath, levelPath], TimeSpan.FromSeconds(1), ct); + + // The birth is a full snapshot, so these are the BIRTH's own values. + var birthTemp = await recorder.WaitAsync(tempPath, v => Equals(v.Value, 21.5f), Timeout, ct); + birthTemp.FullReference.ShouldBe( + tempPath, + "the driver must publish under the tag's RawPath, never the Sparkplug tuple or the topic"); + birthTemp.Snapshot.StatusCode.ShouldBe(Good); + + await recorder.WaitAsync(countPath, v => Equals(v.Value, -42), Timeout, ct); + await recorder.WaitAsync(levelPath, v => Equals(v.Value, 3.5d), Timeout, ct); + + // …and now DATA: alias only, no name, no datatype — the shape a bind-by-alias driver survives + // and a bind-by-name driver must too. + await sim.PublishNodeDataAsync([("Temperature", 33.25f), ("Count", -1234)], ct); + await sim.PublishDeviceDataAsync(device, [("Level", 9.75d)], ct); + + var temp = await recorder.WaitAsync(tempPath, v => Equals(v.Value, 33.25f), Timeout, ct); + temp.Snapshot.StatusCode.ShouldBe(Good); + + // The predicate names the DATA value, not just "an int arrived": the birth already delivered a + // negative Int32 (−42), so a laxer wait would be satisfied by the earlier message and the + // assertion below would compare the wrong notification. + var count = await recorder.WaitAsync(countPath, v => Equals(v.Value, -1234), Timeout, ct); + count.Snapshot.StatusCode.ShouldBe(Good); + count.Snapshot.Value.ShouldBe( + -1234, + "a Sparkplug Int32 rides as two's complement in the UNSIGNED int_value field (−1234 goes on " + + "the wire as 4294966062); without ReinterpretSigned that is what would be published, at " + + "Good quality"); + + await recorder.WaitAsync(levelPath, v => Equals(v.Value, 9.75d), Timeout, ct); + + // The read path serves the same last values under the same keys. + var read = await driver.ReadAsync([tempPath, countPath, levelPath], ct); + read.Select(r => r.StatusCode).ShouldAllBe(s => s == Good); + read[1].Value.ShouldBe(-1234); + } + + // ----------------------------------------------------------------------- + // rebirth recovery (§3.6 #5) + // ----------------------------------------------------------------------- + + /// + /// Late join: the driver connects to a node that has already birthed and gone quiet, so it holds + /// no alias table at all. It must request a rebirth, and the request must be one a real edge + /// node accepts. + /// + /// This is the NCMD encode/decode symmetry check. The driver's + /// RebirthRequester.Build output is decoded here by the simulator — an independent + /// reader that looks for a Boolean metric literally named Node Control/Rebirth set to + /// true. Offline tests assert the driver's own bytes against the driver's own expectations; + /// only this one proves the command means what the spec says to a party that did not write it. + /// + /// + [Fact] + public async Task Late_join_requests_a_rebirth_the_edge_node_honours_and_metadata_recovers() + { + SkipIfUnconfigured(); + + var ct = TestContext.Current.CancellationToken; + var group = NewGroup("rebirth"); + const string node = "EdgeA"; + const string rawPath = "Fixture/Sparkplug/Rebirth/Temperature"; + + await using var sim = new SparkplugEdgeNode(SimOptions(group, node)); + sim.SetNodeMetrics(new SparkplugMetricSeed("Temperature", 7, DataType.Double, 64.5d)); + + // Connected and listening for NCMD, but DELIBERATELY not birthed: the driver joins a stream it + // has no metadata for, which is exactly the state a restart or a long outage leaves it in. + await sim.ConnectAsync(ct); + + await using var driver = new MqttDriver( + SparkplugOptions(group, Tag(rawPath, group, node, null, "Temperature")), + "mqtt-live-sp-rebirth"); + var recorder = new SparkplugRecorder(driver); + + await driver.InitializeAsync(string.Empty, ct); + await driver.SubscribeAsync([rawPath], TimeSpan.FromSeconds(1), ct); + + var honoured = await sim.WaitForRebirthAsync(1, Timeout, ct); + honoured.ShouldBeGreaterThanOrEqualTo( + 1, + "the driver must issue a late-join rebirth NCMD on connect (§3.6 #5), and it must be one a " + + "spec-reading edge node recognises"); + + var change = await recorder.WaitAsync(rawPath, v => Equals(v.Value, 64.5d), Timeout, ct); + change.Snapshot.StatusCode.ShouldBe( + Good, + "the rebirth's NBIRTH is a full snapshot — its values are what restores the tag"); + } + + // ----------------------------------------------------------------------- + // §3.6 case (i) — alias reuse across a rebirth + // ----------------------------------------------------------------------- + + /// + /// The single most dangerous Sparkplug defect, on the wire. An edge node may reassign an + /// alias to a different metric across a rebirth — legal, and something real edge nodes do + /// whenever their metric set changes. A driver that bound its tags to aliases keeps routing by + /// the old mapping and publishes Pressure's value into the Temperature tag: Good quality, + /// entirely plausible values, wrong tag, and nothing anywhere reports an error. + /// + /// The test swaps aliases 5 and 6 between Temperature and Pressure at the rebirth, then + /// publishes a DATA message carrying alias 5 only. Bind-by-name puts it in Pressure; + /// bind-by-alias puts it in Temperature. The assertion checks both tags, because "Pressure + /// got the value" alone would still pass if the driver had fed both. + /// + /// + [Fact] + public async Task Alias_reused_across_a_rebirth_routes_by_metric_name_not_by_alias() + { + SkipIfUnconfigured(); + + var ct = TestContext.Current.CancellationToken; + var group = NewGroup("alias"); + const string node = "EdgeA"; + const string tempPath = "Fixture/Sparkplug/Alias/Temperature"; + const string pressPath = "Fixture/Sparkplug/Alias/Pressure"; + + await using var sim = new SparkplugEdgeNode(SimOptions(group, node)); + sim.SetNodeMetrics( + new SparkplugMetricSeed("Temperature", 5, DataType.Double, 1d), + new SparkplugMetricSeed("Pressure", 6, DataType.Double, 2d)); + + await sim.ConnectAsync(ct); + await sim.BirthAsync(ct); + + await using var driver = new MqttDriver( + SparkplugOptions( + group, + Tag(tempPath, group, node, null, "Temperature"), + Tag(pressPath, group, node, null, "Pressure")), + "mqtt-live-sp-alias"); + var recorder = new SparkplugRecorder(driver); + + await driver.InitializeAsync(string.Empty, ct); + await driver.SubscribeAsync([tempPath, pressPath], TimeSpan.FromSeconds(1), ct); + + await recorder.WaitAsync(tempPath, v => Equals(v.Value, 1d), Timeout, ct); + await recorder.WaitAsync(pressPath, v => Equals(v.Value, 2d), Timeout, ct); + + // Sanity, pre-swap: alias 5 IS Temperature right now. + await sim.PublishNodeDataAsync([("Temperature", 11d)], ct); + await recorder.WaitAsync(tempPath, v => Equals(v.Value, 11d), Timeout, ct); + + // The rebirth: aliases 5 and 6 change hands, and the birth republishes both values. + sim.SetNodeMetrics( + new SparkplugMetricSeed("Temperature", 6, DataType.Double, 100d), + new SparkplugMetricSeed("Pressure", 5, DataType.Double, 200d)); + await sim.BirthAsync(ct); + + await recorder.WaitAsync(tempPath, v => Equals(v.Value, 100d), Timeout, ct); + await recorder.WaitAsync(pressPath, v => Equals(v.Value, 200d), Timeout, ct); + + // …and now the message that discriminates: alias 5, which is Pressure NOW and was Temperature + // a moment ago. + await sim.PublishNodeDataAsync([("Pressure", 777d)], ct); + + var pressure = await recorder.WaitAsync(pressPath, v => Equals(v.Value, 777d), Timeout, ct); + pressure.Snapshot.StatusCode.ShouldBe(Good); + + // The other half of the assertion. Without it, a driver that fed BOTH tags would pass. + var read = await driver.ReadAsync([tempPath], ct); + read[0].Value.ShouldBe( + 100d, + "alias 5 now belongs to Pressure; a driver still routing it to Temperature would show 777 " + + "here — Good quality, plausible value, wrong tag"); + } + + // ----------------------------------------------------------------------- + // §3.6 case (ii) — a stale-bdSeq NDEATH arriving after the reconnect NBIRTH + // ----------------------------------------------------------------------- + + /// + /// An edge node's link drops, it reconnects and births under a new bdSeq, and only + /// then does the broker get round to delivering the old connection's Will. Without the + /// bdSeq tie that stale death marks a live, freshly-born node dead and drives every one + /// of its tags Bad, with nothing coming to correct it. + /// + /// Both arms are asserted, and they must be: a driver that ignored every NDEATH would + /// pass the first half. So the matching-token death follows immediately and must land. + /// + /// + [Fact] + public async Task Ndeath_with_a_stale_bdseq_is_ignored_and_the_current_one_stales() + { + SkipIfUnconfigured(); + + var ct = TestContext.Current.CancellationToken; + var group = NewGroup("bdseq"); + const string node = "EdgeA"; + const string rawPath = "Fixture/Sparkplug/BdSeq/Temperature"; + + await using var sim = new SparkplugEdgeNode(SimOptions(group, node)); + sim.SetNodeMetrics(new SparkplugMetricSeed("Temperature", 3, DataType.Double, 10d)); + + await sim.ConnectAsync(ct); // session 0 + sim.BdSeq.ShouldBe(0UL); + await sim.BirthAsync(ct); + + await using var driver = new MqttDriver( + SparkplugOptions(group, Tag(rawPath, group, node, null, "Temperature")), + "mqtt-live-sp-bdseq"); + var recorder = new SparkplugRecorder(driver); + + await driver.InitializeAsync(string.Empty, ct); + await driver.SubscribeAsync([rawPath], TimeSpan.FromSeconds(1), ct); + await recorder.WaitAsync(rawPath, v => Equals(v.Value, 10d), Timeout, ct); + + // The node "reconnects": a clean DISCONNECT (so no Will fires) followed by a new session, whose + // bdSeq is 1, and a fresh birth carrying a distinguishable value. + sim.SetNodeMetrics(new SparkplugMetricSeed("Temperature", 3, DataType.Double, 20d)); + await sim.ConnectAsync(ct); // session 1 + sim.BdSeq.ShouldBe(1UL); + await sim.BirthAsync(ct); + await recorder.WaitAsync(rawPath, v => Equals(v.Value, 20d), Timeout, ct); + + // The old session's Will, delivered late. bdSeq 0 ≠ the live session's 1 ⇒ it must be discarded. + await sim.PublishNodeDeathAsync(bdSeq: 0UL, ct); + + var deadline = DateTime.UtcNow + NegativeWindow; + while (DateTime.UtcNow < deadline) + { + var sample = await driver.ReadAsync([rawPath], ct); + sample[0].StatusCode.ShouldBe( + Good, + "an NDEATH carrying a PREVIOUS session's bdSeq is a stale Will; acting on it kills a live node"); + await Task.Delay(200, ct); + } + + // …and the falsifiability control: the SAME message with the CURRENT token must stale the tag. + // Without this leg, a driver that ignored every death would pass the assertion above. + await sim.PublishNodeDeathAsync(bdSeq: 1UL, ct); + + var stale = await recorder.WaitAsync( + rawPath, + v => v.StatusCode == BadCommunicationError, + Timeout, + ct); + stale.Snapshot.Value.ShouldBeNull(); + } + + // ----------------------------------------------------------------------- + // death → STALE, through the real Last-Will path + // ----------------------------------------------------------------------- + + /// + /// The realistic death: the edge node's session ends and the broker publishes the NDEATH + /// Will it registered at CONNECT. Nothing in this test publishes the death message — which is + /// the point, because a Will is the only death an edge node that crashed can produce, and it is + /// also the one that carries no seq. + /// + /// The recovery half matters as much: §3.6 #4 says the next birth restores Good, so a driver + /// that latched Bad on death would be wrong in the direction nobody notices until a plant + /// comes back and its tags do not. + /// + /// + [Fact] + public async Task Broker_published_ndeath_will_stales_the_tags_and_the_next_birth_restores_them() + { + SkipIfUnconfigured(); + + var ct = TestContext.Current.CancellationToken; + var group = NewGroup("death"); + const string node = "EdgeA"; + const string rawPath = "Fixture/Sparkplug/Death/Temperature"; + + await using var sim = new SparkplugEdgeNode(SimOptions(group, node)); + sim.SetNodeMetrics(new SparkplugMetricSeed("Temperature", 4, DataType.Double, 42d)); + + await sim.ConnectAsync(ct); + await sim.BirthAsync(ct); + + await using var driver = new MqttDriver( + SparkplugOptions(group, Tag(rawPath, group, node, null, "Temperature")), + "mqtt-live-sp-death"); + var recorder = new SparkplugRecorder(driver); + + await driver.InitializeAsync(string.Empty, ct); + await driver.SubscribeAsync([rawPath], TimeSpan.FromSeconds(1), ct); + await recorder.WaitAsync(rawPath, v => Equals(v.Value, 42d), Timeout, ct); + + // The broker fires the Will — we publish nothing. + await sim.KillAsync(ct); + + var stale = await recorder.WaitAsync(rawPath, v => v.StatusCode == BadCommunicationError, Timeout, ct); + stale.FullReference.ShouldBe(rawPath); + stale.Snapshot.Value.ShouldBeNull("a staled tag must not keep serving the dead node's last value"); + + // §3.6 #4: the next birth restores Good. + sim.SetNodeMetrics(new SparkplugMetricSeed("Temperature", 4, DataType.Double, 43d)); + await sim.ConnectAsync(ct); + await sim.BirthAsync(ct); + + var revived = await recorder.WaitAsync(rawPath, v => Equals(v.Value, 43d), Timeout, ct); + revived.Snapshot.StatusCode.ShouldBe(Good); + } + + // ----------------------------------------------------------------------- + // §3.6 case (iii) — seq wrap 255 → 0, and its falsifiability control + // ----------------------------------------------------------------------- + + /// + /// 255 → 0 is the sequence continuing, not a gap. Read the wrap as a gap and every edge + /// node is commanded to rebirth once per 256 messages — a busy node would spend its life + /// republishing metadata. This drives a full lap of the counter over a real broker and asserts + /// that no NCMD appears on the wire. + /// + /// Two things make this test able to fail. First, the ingestor is built with + /// rebirthDebounce: TimeSpan.Zero: at the shipped 10 s default a wrap-triggered + /// request would be swallowed by the debounce and the test would pass for the wrong reason. + /// Second, the negative control — a deliberately skipped seq immediately afterwards + /// must produce an NCMD. Silence proves nothing unless noise is also proven. + /// + /// + /// The assertion is made on the wire, by a third MQTT client watching + /// spBv1.0/{group}/NCMD/# — not on a counter inside the driver. + /// + /// + [Fact] + public async Task Seq_wrap_255_to_0_requests_no_rebirth_but_a_real_gap_does() + { + SkipIfUnconfigured(); + + var ct = TestContext.Current.CancellationToken; + var group = NewGroup("seq"); + const string node = "EdgeA"; + const string rawPath = "Fixture/Sparkplug/Seq/Counter"; + + await using var watcher = await NcmdWatcher.StartAsync(_fx, group, ct); + + // QoS 1 for the 256-message lap: the seq semantics under test are identical at either QoS, and + // a broker-side drop at QoS 0 would surface as a "gap" — a real detection, but a flaky test. + await using var sim = new SparkplugEdgeNode(SimOptions(group, node) with { DataQos = 1 }); + sim.SetNodeMetrics(new SparkplugMetricSeed("Counter", 1, DataType.Double, 0d)); + + await sim.ConnectAsync(ct); + await sim.BirthAsync(ct); + + var options = SparkplugOptions(group, Tag(rawPath, group, node, null, "Counter")); + + // The production ingest path — AttachTo + EstablishAsync is exactly what MqttDriver does — but + // constructed directly so the rebirth debounce can be taken out of the picture. + await using var connection = new MqttConnection(options, "mqtt-live-sp-seq"); + var ingestor = new SparkplugIngestor( + options, + "mqtt-live-sp-seq", + logger: null, + maxPayloadBytes: options.MaxPayloadBytes, + rebirthDebounce: TimeSpan.Zero); + ingestor.Register(options.RawTags); + ingestor.AttachTo(connection); + + var recorder = new SparkplugRecorder(ingestor); + await connection.ConnectAsync(ct); + await ingestor.EstablishAsync(ct); + await ingestor.SubscribeAsync([rawPath], TimeSpan.FromSeconds(1), ct); + + // Establish issues the late-join NCMD; let the node answer it and settle before measuring. + await sim.WaitForRebirthAsync(1, Timeout, ct); + await recorder.WaitAsync(rawPath, v => v.StatusCode == Good, Timeout, ct); + await Task.Delay(500, ct); + + var beforeWrap = watcher.Count; + + // A full lap: the birth left seq at 0, so these carry seq 1…255 and then 0 — the wrap. + for (var i = 1; i <= 256; i++) + { + await sim.PublishNodeDataAsync([("Counter", (double)i)], ct); + } + + sim.LastSeq.ShouldBe((byte)0, "256 messages after a birth at seq 0 must land exactly on the wrap"); + + await recorder.WaitAsync(rawPath, v => Equals(v.Value, 256d), Timeout, ct); + await Task.Delay(500, ct); // any NCMD the wrap provoked would have arrived by now + + watcher.Count.ShouldBe( + beforeWrap, + $"a seq wrap 255 → 0 is the stream continuing, not a gap; NCMDs seen: " + + $"{string.Join(", ", watcher.Topics)}"); + + // ---- the falsifiability control ------------------------------------------------- + // Burn one sequence number without publishing it. That IS a lost message from the host's point + // of view, and it MUST produce a rebirth request — otherwise the silence above proves only that + // this driver never asks for anything. + sim.SkipSeq(); + await sim.PublishNodeDataAsync([("Counter", 999d)], ct); + + await watcher.WaitForCountAsync(beforeWrap + 1, Timeout, ct); + + watcher.Topics.ShouldAllBe(t => t == $"spBv1.0/{group}/NCMD/{node}"); + + // A gap resynchronizes rather than latching: the message that exposed it still lands. + await recorder.WaitAsync(rawPath, v => Equals(v.Value, 999d), Timeout, ct); + } + + // ----------------------------------------------------------------------- + // browser: passive window + // ----------------------------------------------------------------------- + + /// + /// Opening an address picker against a live plant must publish nothing. The browse + /// session counts its own publishes, and the offline suite asserts that counter — but a counter + /// can only see messages that went through the seam it guards. This asserts the property where + /// it actually matters: on the broker, with an independent client subscribed to the group's NCMD + /// topic while the picker opens, roots, expands every level and reads every attribute. + /// + [Fact] + public async Task Browser_open_and_full_tree_walk_publish_no_ncmd() + { + SkipIfUnconfigured(); + + var ct = TestContext.Current.CancellationToken; + var group = NewGroup("browse"); + + await using var watcher = await NcmdWatcher.StartAsync(_fx, group, ct); + await using var edgeA = await StartBrowseNodeAsync(group, "EdgeA", withDevice: true, ct); + await using var edgeB = await StartBrowseNodeAsync(group, "EdgeB", withDevice: false, ct); + + var browser = new MqttDriverBrowser(); + await using var session = await browser.OpenAsync(BrowseConfigJson(group), ct); + + // The window is passive, so it can only see births published AFTER it subscribed — Sparkplug + // births are never retained. This is the same round-trip the picker's "Request rebirth" button + // exists for; here the test plays the edge node's part directly, so the browser stays untouched. + await edgeA.BirthAsync(ct); + await edgeB.BirthAsync(ct); + + var roots = await WaitForRootAsync(session, group, ct); + roots.ShouldContain(n => n.NodeId == group); + + // Walk everything the picker could walk: every level expanded, every node's attributes read. + var visited = await WalkAsync(session, roots, ct); + + visited.ShouldContain($"{group}/EdgeA"); + visited.ShouldContain($"{group}/EdgeB"); + visited.ShouldContain($"{group}/EdgeA/Filler1"); + visited.ShouldContain($"{group}/EdgeA::Temperature"); + visited.ShouldContain($"{group}/EdgeA/Filler1::Level"); + + // Give anything the walk might have kicked off time to reach the broker before concluding. + await Task.Delay(NegativeWindow, ct); + + watcher.Count.ShouldBe( + 0, + "no browse path may publish — an operator opening a picker must not be able to command a " + + $"running plant. Published: {string.Join(", ", watcher.Topics)}"); + edgeA.RebirthCount.ShouldBe(0); + edgeB.RebirthCount.ShouldBe(0); + } + + // ----------------------------------------------------------------------- + // browser: explicit rebirth, node vs. group scope + // ----------------------------------------------------------------------- + + /// + /// The one browse-session member that publishes, and the enumeration it performs: an edge-node + /// scope addresses exactly that node, a group scope fans out to every edge node observed beneath + /// it. Asserted on the wire (which topics were published) and at the far end (which simulated + /// nodes actually honoured a rebirth), because "the call returned 2" says nothing about where + /// the commands went. + /// + [Fact] + public async Task Browser_request_rebirth_addresses_one_node_or_every_node_in_the_group() + { + SkipIfUnconfigured(); + + var ct = TestContext.Current.CancellationToken; + var group = NewGroup("rbscope"); + + await using var watcher = await NcmdWatcher.StartAsync(_fx, group, ct); + await using var edgeA = await StartBrowseNodeAsync(group, "EdgeA", withDevice: true, ct); + await using var edgeB = await StartBrowseNodeAsync(group, "EdgeB", withDevice: false, ct); + + var browser = new MqttDriverBrowser(); + await using var session = await browser.OpenAsync(BrowseConfigJson(group), ct); + var rebirthable = session.ShouldBeAssignableTo(); + + await edgeA.BirthAsync(ct); + await edgeB.BirthAsync(ct); + await WaitForRootAsync(session, group, ct); + + // Both edge nodes must be in the tree before the group scope is exercised — it enumerates what + // the window OBSERVED, so asserting the fan-out against a half-built tree would prove nothing. + await WaitForChildrenAsync(session, group, expected: 2, ct); + + // ---- node scope ---- + var nodeTargets = await rebirthable.RequestRebirthAsync($"{group}/EdgeA", ct); + nodeTargets.ShouldBe(1); + + await watcher.WaitForCountAsync(1, Timeout, ct); + watcher.Topics.ShouldBe([$"spBv1.0/{group}/NCMD/EdgeA"]); + await edgeA.WaitForRebirthAsync(1, Timeout, ct); + edgeB.RebirthCount.ShouldBe(0, "a node-scoped rebirth must not touch the node next to it"); + + // ---- group scope ---- + var groupTargets = await rebirthable.RequestRebirthAsync(group, ct); + groupTargets.ShouldBe(2); + + await watcher.WaitForCountAsync(3, Timeout, ct); + watcher.Topics.ShouldBe( + [ + $"spBv1.0/{group}/NCMD/EdgeA", + $"spBv1.0/{group}/NCMD/EdgeA", + $"spBv1.0/{group}/NCMD/EdgeB", + ], + "a group scope addresses every observed edge node, once each, in a deterministic order"); + + await edgeA.WaitForRebirthAsync(2, Timeout, ct); + await edgeB.WaitForRebirthAsync(1, Timeout, ct); + } + + // ----------------------------------------------------------------------- + // helpers + // ----------------------------------------------------------------------- + + private void SkipIfUnconfigured() + { + if (_fx.NotConfigured) + { + Assert.Skip(_fx.SkipReason!); + } + } + + /// + /// A Sparkplug group id unique to this run. The fixture broker is shared and the driver + /// subscribes the whole group, so a fixed id would let tests — and previous runs — feed each + /// other. + /// + private static string NewGroup(string label) => $"OtOpcUaLive-{label}-{Guid.NewGuid():N}"[..40]; + + /// Driver options for Sparkplug mode over the fixture's TLS listener. + private MqttDriverOptions SparkplugOptions(string groupId, params RawTagEntry[] rawTags) => new() + { + Host = _fx.TlsHost, + Port = _fx.TlsPort, + UseTls = true, + CaCertificatePath = _fx.CaCertificatePath, + AllowUntrustedServerCertificate = _fx.CaCertificatePath is null, + Username = _fx.Username, + Password = _fx.Password, + ClientId = $"otopcua-live-sp-{Guid.NewGuid():N}", + ConnectTimeoutSeconds = 15, + Mode = MqttMode.SparkplugB, + Sparkplug = new MqttSparkplugOptions { GroupId = groupId, RequestRebirthOnGap = true }, + RawTags = rawTags, + }; + + /// Simulator connection settings, pointed at the same listener with the same credentials. + private SparkplugEdgeNodeOptions SimOptions(string groupId, string edgeNodeId) => new() + { + Host = _fx.TlsHost, + Port = _fx.TlsPort, + UseTls = true, + CaCertificatePath = _fx.CaCertificatePath, + AllowUntrustedServerCertificate = _fx.CaCertificatePath is null, + Username = _fx.Username, + Password = _fx.Password, + GroupId = groupId, + EdgeNodeId = edgeNodeId, + }; + + /// An authored Sparkplug raw tag — the binding tuple, with the datatype left to the birth. + private static RawTagEntry Tag( + string rawPath, + string groupId, + string edgeNodeId, + string? deviceId, + string metricName) + { + var device = deviceId is null ? "" : $""" + "deviceId":"{deviceId}", + """; + return new RawTagEntry( + rawPath, + $$""" + {"groupId":"{{groupId}}","edgeNodeId":"{{edgeNodeId}}",{{device}}"metricName":"{{metricName}}"} + """, + WriteIdempotent: false); + } + + /// The browse configuration blob, parsed through the driver's ONE shared JSON options. + private string BrowseConfigJson(string groupId) => + JsonSerializer.Serialize(SparkplugOptions(groupId), MqttJson.Options); + + /// Brings up one simulated edge node for the browse legs, connected but not yet birthed. + private async Task StartBrowseNodeAsync( + string group, + string node, + bool withDevice, + CancellationToken ct) + { + var sim = new SparkplugEdgeNode(SimOptions(group, node)); + sim.SetNodeMetrics( + new SparkplugMetricSeed("Temperature", 1, DataType.Float, 21.5f), + new SparkplugMetricSeed("Count", 2, DataType.Int32, -42)); + + if (withDevice) + { + sim.SetDeviceMetrics("Filler1", new SparkplugMetricSeed("Level", 10, DataType.Double, 3.5d)); + } + + await sim.ConnectAsync(ct); + return sim; + } + + /// Polls RootAsync until the observation window has recorded the group. + private static async Task> WaitForRootAsync( + IBrowseSession session, + string group, + CancellationToken ct) + { + var deadline = DateTime.UtcNow + Timeout; + while (DateTime.UtcNow < deadline) + { + var roots = await session.RootAsync(ct); + if (roots.Any(n => n.NodeId == group)) + { + return roots; + } + + await Task.Delay(200, ct); + } + + throw new TimeoutException( + $"The browse window never observed Sparkplug group '{group}'. A birth must be published " + + "AFTER the window opens — Sparkplug births are not retained."); + } + + /// Polls until a node reports at least children. + private static async Task WaitForChildrenAsync( + IBrowseSession session, + string nodeId, + int expected, + CancellationToken ct) + { + var deadline = DateTime.UtcNow + Timeout; + while (DateTime.UtcNow < deadline) + { + var children = await session.ExpandAsync(nodeId, ct); + if (children.Count >= expected) + { + return; + } + + await Task.Delay(200, ct); + } + + throw new TimeoutException($"'{nodeId}' never reported {expected} children in the browse window."); + } + + /// Expands every node depth-first and reads every node's attributes; returns the node ids seen. + private static async Task> WalkAsync( + IBrowseSession session, + IReadOnlyList level, + CancellationToken ct) + { + var seen = new List(); + foreach (var node in level) + { + seen.Add(node.NodeId); + await session.AttributesAsync(node.NodeId, ct); + + var children = await session.ExpandAsync(node.NodeId, ct); + if (children.Count > 0) + { + seen.AddRange(await WalkAsync(session, children, ct)); + } + } + + return seen; + } + + // ----------------------------------------------------------------------- + // fixtures used by the tests + // ----------------------------------------------------------------------- + + /// + /// Records every OnDataChange raised by a driver or an ingestor and lets a test await a + /// specific RawPath reaching a specific state. + /// + /// Awaits a predicate, not merely an arrival. A Sparkplug tag is fed by births, + /// rebirths and data alike, so "a notification arrived for this RawPath" is satisfied by the + /// wrong message roughly half the time; every wait here names the value or quality it is + /// waiting for. It also pre-scans, so a test that registers after the message landed is + /// served rather than timing out. + /// + /// + private sealed class SparkplugRecorder + { + private readonly List _events = []; + private readonly Lock _gate = new(); + + public SparkplugRecorder(MqttDriver driver) => driver.OnDataChange += Record; + + public SparkplugRecorder(SparkplugIngestor ingestor) => ingestor.OnDataChange += Record; + + public async Task WaitAsync( + string rawPath, + Func predicate, + TimeSpan timeout, + CancellationToken cancellationToken) + { + var deadline = DateTime.UtcNow + timeout; + while (true) + { + lock (_gate) + { + foreach (var candidate in _events) + { + if (candidate.FullReference == rawPath && predicate(candidate.Snapshot)) + { + return candidate; + } + } + } + + if (DateTime.UtcNow >= deadline) + { + throw new TimeoutException( + $"No notification for RawPath '{rawPath}' satisfied the expected state within " + + $"{timeout.TotalSeconds:0}s. Recorded: {Describe()}"); + } + + await Task.Delay(100, cancellationToken).ConfigureAwait(false); + } + } + + private string Describe() + { + lock (_gate) + { + return _events.Count == 0 + ? "(nothing)" + : string.Join( + " | ", + _events + .GroupBy(e => e.FullReference) + .Select(g => + $"{g.Key}: [{string.Join(", ", g.Select(e => $"{e.Snapshot.Value ?? "null"}@0x{e.Snapshot.StatusCode:X8}"))}]")); + } + } + + private void Record(object? sender, DataChangeEventArgs args) + { + lock (_gate) + { + _events.Add(args); + } + } + } + + /// + /// An independent MQTT client subscribed to spBv1.0/{group}/NCMD/#, recording every + /// command published into the group. + /// + /// This is what makes the passivity assertions mean something. The browse session's + /// own PublishCountForTest can only observe messages that pass through the seam it + /// guards; this observes the broker. If a future change published an NCMD by some other + /// route — a raw client, a Last Will, a stray retained command — the counter would still + /// read zero and this would not. + /// + /// + private sealed class NcmdWatcher : IAsyncDisposable + { + private readonly IMqttClient _client; + private readonly ConcurrentQueue _topics = new(); + + private NcmdWatcher(IMqttClient client) => _client = client; + + /// Every NCMD topic observed, in arrival order. + public IReadOnlyList Topics => [.. _topics]; + + /// How many NCMDs have been observed. + public int Count => _topics.Count; + + public static async Task StartAsync(MqttFixture fixture, string group, CancellationToken ct) + { + var client = new MqttClientFactory().CreateMqttClient(); + var watcher = new NcmdWatcher(client); + + client.ApplicationMessageReceivedAsync += args => + { + // Only a genuine rebirth command counts. A malformed or unrelated NCMD would be a + // different defect, and lumping them together would make the count uninterpretable. + if (watcher.IsRebirth(args.ApplicationMessage)) + { + watcher._topics.Enqueue(args.ApplicationMessage.Topic); + } + + return Task.CompletedTask; + }; + + var options = new MqttClientOptionsBuilder() + .WithTcpServer(fixture.PlainHost, fixture.PlainPort) + .WithCredentials(fixture.Username, fixture.Password) + .WithClientId($"otopcua-live-ncmdwatch-{Guid.NewGuid():N}") + .WithTimeout(TimeSpan.FromSeconds(10)) + .Build(); + + await client.ConnectAsync(options, ct).ConfigureAwait(false); + await client.SubscribeAsync( + new MqttClientSubscribeOptionsBuilder() + .WithTopicFilter(f => f.WithTopic($"spBv1.0/{group}/NCMD/#")) + .Build(), + ct).ConfigureAwait(false); + + return watcher; + } + + public async Task WaitForCountAsync(int count, TimeSpan timeout, CancellationToken ct) + { + var deadline = DateTime.UtcNow + timeout; + while (Count < count) + { + if (DateTime.UtcNow >= deadline) + { + throw new TimeoutException( + $"Only {Count} rebirth NCMD(s) reached the broker within {timeout.TotalSeconds:0}s; " + + $"expected {count}. Seen: {string.Join(", ", Topics)}"); + } + + await Task.Delay(100, ct).ConfigureAwait(false); + } + } + + public async ValueTask DisposeAsync() + { + try + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5)); + await _client.DisconnectAsync(new MqttClientDisconnectOptions(), cts.Token).ConfigureAwait(false); + } + catch (Exception) + { + // Teardown is best-effort; the broker may already have gone. + } + + _client.Dispose(); + } + + private bool IsRebirth(MqttApplicationMessage message) + { + var body = message.Payload; + var bytes = body.IsSingleSegment ? body.FirstSpan.ToArray() : body.ToArray(); + + try + { + var payload = Payload.Parser.ParseFrom(bytes); + return payload.Metrics.Any(m => + m.Name == SparkplugEdgeNode.RebirthMetricName + && m.ValueCase == Payload.Types.Metric.ValueOneofCase.BooleanValue + && m.BooleanValue); + } + catch (Google.Protobuf.InvalidProtocolBufferException) + { + return false; + } + } + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/Program.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/Program.cs new file mode 100644 index 00000000..db045291 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/Program.cs @@ -0,0 +1,114 @@ +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(); +var edges = new List(); + +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")); + + 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)); + + 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; diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/SparkplugEdgeNode.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/SparkplugEdgeNode.cs new file mode 100644 index 00000000..0140af0d --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/SparkplugEdgeNode.cs @@ -0,0 +1,917 @@ +using System.Buffers; +using System.Collections.Concurrent; +using System.Globalization; +using System.Security.Cryptography.X509Certificates; +using Google.Protobuf; +using MQTTnet; +using MQTTnet.Protocol; +using Org.Eclipse.Tahu.Protobuf; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.SparkplugSimulator; + +/// +/// A controllable Sparkplug B edge node — the driver's counterparty on a real broker. +/// Registers an NDEATH Last Will at CONNECT, publishes NBIRTH/DBIRTH then N/DDATA, honours a +/// Node Control/Rebirth NCMD, and exposes the §3.6 pathologies (alias reuse across a +/// rebirth, a seq gap, a stale-bdSeq death) as explicit, on-demand operations. +/// +/// +/// +/// Encodes with the generated Tahu types, frames topics independently. The payloads are +/// built from — the same generated schema the driver decodes with, which +/// is what makes this fixture a genuine encode/decode symmetry check. The topic strings +/// are composed here from literals rather than through the driver's own +/// SparkplugTopic.Format: a simulator that borrowed the parser's formatter could not +/// detect a bug in it, because both sides of the comparison would be wrong together. +/// +/// +/// What "the way a real edge node does it" means, concretely, and why each detail matters: +/// +/// +/// +/// NDEATH is the broker-published Last Will, registered at CONNECT — not a message +/// this class publishes on the way out. It carries a bdSeq metric and no +/// seq, because it is not a member of the node's sequence. +/// makes the broker fire it. +/// +/// +/// bdSeq increments once per CONNECT and appears in both the NBIRTH and the +/// NDEATH of that session — that pairing is the only thing that lets a host tell a +/// just-delivered stale will from a real death. +/// +/// +/// NBIRTH restarts seq at 0; DBIRTH/NDATA/DDATA/DDEATH continue it, wrapping +/// 255 → 0. +/// +/// +/// DATA metrics carry an alias and nothing else — no name, no datatype. That is what +/// every real post-birth DATA message looks like, and a driver that binds tags to aliases +/// rather than names passes every test written against a friendlier simulator. +/// +/// +/// Signed integers ride as two's complement in the UNSIGNED proto field. A +/// DataType.Int32 metric holding −1234 goes on the wire as int_value = +/// 4294966062. Encoding it "helpfully" as a positive number would silently retire the +/// driver's ReinterpretSigned path from the live gate. +/// +/// +/// +/// Not thread-safe for concurrent publishes: seq is a single counter and interleaving +/// publishes would manufacture the very gaps the tests are measuring. Drive one node from one +/// test at a time; the NCMD-triggered rebirth is serialized against callers by +/// . +/// +/// +public sealed class SparkplugEdgeNode : IAsyncDisposable +{ + /// The well-known Sparkplug node-control metric an NCMD sets to trigger a rebirth. + public const string RebirthMetricName = "Node Control/Rebirth"; + + /// The Sparkplug session-token metric name, carried by NBIRTH and NDEATH. + public const string BdSeqMetricName = "bdSeq"; + + private const string Namespace = "spBv1.0"; + + private readonly SparkplugEdgeNodeOptions _options; + private readonly Action? _log; + private readonly SemaphoreSlim _publishGate = new(1, 1); + + /// Device id → its metric catalog. Ordered so a birth's metric order is deterministic. + private readonly ConcurrentDictionary> _devices = new(StringComparer.Ordinal); + + private IMqttClient? _client; + private IReadOnlyList _nodeMetrics = []; + private ulong _bdSeq; + private int _sessions; + private byte _seq; + private bool _seqStarted; + private int _rebirthCount; + private int _disposed; + private TaskCompletionSource _rebirthSignal = new(TaskCreationOptions.RunContinuationsAsynchronously); + + /// Initializes a new simulated edge node. Does no I/O — call . + /// Broker connection + Sparkplug identity. + /// Optional line sink for the standalone runner's console output. + public SparkplugEdgeNode(SparkplugEdgeNodeOptions options, Action? log = null) + { + ArgumentNullException.ThrowIfNull(options); + ArgumentException.ThrowIfNullOrWhiteSpace(options.GroupId); + ArgumentException.ThrowIfNullOrWhiteSpace(options.EdgeNodeId); + + _options = options; + _log = log; + } + + /// The Sparkplug group id. + public string GroupId => _options.GroupId; + + /// The Sparkplug edge-node id. + public string EdgeNodeId => _options.EdgeNodeId; + + /// The current session token — incremented on every . + public ulong BdSeq => Volatile.Read(ref _bdSeq); + + /// The seq of the most recently published sequenced message. + public byte LastSeq => Volatile.Read(ref _seq); + + /// How many rebirth NCMDs this node has honoured since construction. + public int RebirthCount => Volatile.Read(ref _rebirthCount); + + /// Whether the underlying MQTT client currently holds a session. + public bool IsConnected => _client?.IsConnected ?? false; + + /// The device ids this node currently publishes for. + public IReadOnlyCollection Devices => [.. _devices.Keys]; + + /// + /// Replaces this node's own (NBIRTH-level) metric catalog. Takes effect at the next birth — + /// which is exactly how an edge node reassigns an alias across a rebirth. + /// + /// The catalog. + public void SetNodeMetrics(params SparkplugMetricSeed[] metrics) => + _nodeMetrics = [.. metrics ?? []]; + + /// Replaces one device's (DBIRTH-level) metric catalog. Takes effect at the next birth. + /// The device id. + /// The catalog. + public void SetDeviceMetrics(string deviceId, params SparkplugMetricSeed[] metrics) + { + ArgumentException.ThrowIfNullOrWhiteSpace(deviceId); + _devices[deviceId] = [.. metrics ?? []]; + } + + /// + /// Connects to the broker with an NDEATH Last Will registered for this session, and subscribes + /// the node's NCMD topic so a rebirth request can be honoured. + /// + /// + /// The Will is registered here or not at all. MQTT fixes a client's will at CONNECT; a + /// "death message" published later by the client itself is a different thing entirely and would + /// never fire for the case that matters — the node dying without getting to say so. + /// + /// Cancellation for the connect. + /// A task that completes when the session is established and NCMD is subscribed. + public async Task ConnectAsync(CancellationToken cancellationToken) + { + ObjectDisposedException.ThrowIf(Volatile.Read(ref _disposed) != 0, this); + + await DropClientAsync(graceful: true).ConfigureAwait(false); + + // A NEW session ⇒ a new session token, starting at 0 and incrementing per CONNECT. Both the + // Will registered below and the next NBIRTH carry it, which is the pairing a host uses to + // discard a previous session's late-delivered will. + Volatile.Write(ref _bdSeq, (ulong)(Interlocked.Increment(ref _sessions) - 1)); + + var client = new MqttClientFactory().CreateMqttClient(); + client.ApplicationMessageReceivedAsync += OnMessageAsync; + + var clientId = _options.ClientId is { Length: > 0 } id ? id : $"sim-{Guid.NewGuid():N}"; + + var builder = new MqttClientOptionsBuilder() + .WithTcpServer(_options.Host, _options.Port) + .WithProtocolVersion(MQTTnet.Formatter.MqttProtocolVersion.V500) + .WithCleanSession(true) + .WithClientId(clientId) + .WithKeepAlivePeriod(TimeSpan.FromSeconds(30)) + .WithTimeout(TimeSpan.FromSeconds(_options.TimeoutSeconds)) + + // NDEATH: QoS 1, retain false, no seq, carries this session's bdSeq. Sparkplug B v3.0 §6. + .WithWillTopic(NodeTopic("NDEATH")) + .WithWillPayload(BuildDeathPayload(CurrentBdSeq()).ToByteArray()) + .WithWillQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) + .WithWillRetain(false); + + if (!string.IsNullOrEmpty(_options.Username)) + { + builder = builder.WithCredentials(_options.Username, _options.Password ?? string.Empty); + } + + builder = builder.WithTlsOptions(ConfigureTls); + + _client = client; + await client.ConnectAsync(builder.Build(), cancellationToken).ConfigureAwait(false); + + var subscribe = new MqttClientSubscribeOptionsBuilder() + .WithTopicFilter(f => f + .WithTopic(NodeTopic("NCMD")) + .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)) + .Build(); + await client.SubscribeAsync(subscribe, cancellationToken).ConfigureAwait(false); + + Log($"connected as '{clientId}' (bdSeq {CurrentBdSeq()}), listening on {NodeTopic("NCMD")}"); + } + + /// + /// Publishes the node's birth certificate — NBIRTH at seq = 0, then one DBIRTH per + /// device continuing the same sequence. + /// + /// Cancellation for the publishes. + /// A task that completes when every birth message has been published. + public async Task BirthAsync(CancellationToken cancellationToken) + { + await _publishGate.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + await BirthCoreAsync(cancellationToken).ConfigureAwait(false); + } + finally + { + _publishGate.Release(); + } + } + + /// + /// Publishes an NDATA carrying the named node-level metrics, encoded the way a real edge node + /// does: alias only, resolved from the current birth catalog. + /// + /// Metric name → new value. A name not in the catalog throws. + /// Cancellation for the publish. + /// A task that completes when the message has been published. + public Task PublishNodeDataAsync( + IReadOnlyList<(string Metric, object? Value)> values, + CancellationToken cancellationToken) => + PublishDataAsync(device: null, values, cancellationToken); + + /// Publishes a DDATA for one device — alias only, same sequence as the node's own stream. + /// The device id. + /// Metric name → new value. + /// Cancellation for the publish. + /// A task that completes when the message has been published. + public Task PublishDeviceDataAsync( + string deviceId, + IReadOnlyList<(string Metric, object? Value)> values, + CancellationToken cancellationToken) => + PublishDataAsync(deviceId, values, cancellationToken); + + /// + /// Publishes an NDATA whose metrics carry an alias the current birth does not declare — + /// the "unknown alias" arm of §3.6. + /// + /// The undeclared alias. + /// A double value, encoded as double_value. + /// Cancellation for the publish. + /// A task that completes when the message has been published. + public async Task PublishUnknownAliasDataAsync(ulong alias, double value, CancellationToken cancellationToken) + { + await _publishGate.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + var payload = NewSequencedPayload(); + payload.Metrics.Add(new Payload.Types.Metric + { + Alias = alias, + Timestamp = NowMs(), + DoubleValue = value, + }); + + await PublishAsync(NodeTopic("NDATA"), payload, _options.DataQos, cancellationToken).ConfigureAwait(false); + } + finally + { + _publishGate.Release(); + } + } + + /// + /// Burns sequence numbers without publishing them — the next message + /// therefore arrives with a seq the host was not expecting. This is how a lost message + /// looks from the receiving end, produced deterministically instead of by hoping for packet loss. + /// + /// How many sequence numbers to skip; must be at least 1. + public void SkipSeq(int count = 1) + { + ArgumentOutOfRangeException.ThrowIfLessThan(count, 1); + for (var i = 0; i < count; i++) + { + NextSeq(); + } + } + + /// + /// Publishes an NDEATH directly, carrying . Two uses: a stale + /// token (a previous session's will, delivered late) which a correct host must ignore, and the + /// current token, which it must act on. + /// + /// The session token to stamp; uses the current one. + /// Cancellation for the publish. + /// A task that completes when the message has been published. + /// + /// A real NDEATH is broker-issued (see ); this is the deterministic + /// equivalent for the ordering case a test cannot otherwise stage — a will that arrives + /// after the reconnect's NBIRTH. + /// + public async Task PublishNodeDeathAsync(ulong? bdSeq, CancellationToken cancellationToken) + { + await _publishGate.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + // QoS 1 and NO seq — an NDEATH is not a member of the node's sequence. + await PublishAsync( + NodeTopic("NDEATH"), + BuildDeathPayload(bdSeq ?? CurrentBdSeq()), + qos: 1, + cancellationToken).ConfigureAwait(false); + } + finally + { + _publishGate.Release(); + } + } + + /// Publishes a DDEATH for one device — sequenced, unlike an NDEATH. + /// The device id. + /// Cancellation for the publish. + /// A task that completes when the message has been published. + public async Task PublishDeviceDeathAsync(string deviceId, CancellationToken cancellationToken) + { + ArgumentException.ThrowIfNullOrWhiteSpace(deviceId); + + await _publishGate.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + await PublishAsync( + DeviceTopic("DDEATH", deviceId), + NewSequencedPayload(), + _options.DataQos, + cancellationToken).ConfigureAwait(false); + } + finally + { + _publishGate.Release(); + } + } + + /// + /// Ends the session in a way that makes the broker publish this node's registered + /// NDEATH Will — the realistic death path. + /// + /// Cancellation for the disconnect. + /// A task that completes once the client has gone. + /// + /// Uses the MQTT 5 DISCONNECT reason code 0x04 Disconnect with Will Message, which + /// is the protocol's own "I am going away, publish my will" signal. Deliberately chosen over + /// yanking the socket: both produce the will, but only this one does so deterministically + /// and immediately — a killed socket leaves the broker waiting out the keep-alive, which would + /// make the death test depend on a 30-second timer. + /// + public async Task KillAsync(CancellationToken cancellationToken) + { + var client = _client; + if (client is null) + { + return; + } + + var options = new MqttClientDisconnectOptionsBuilder() + .WithReason(MqttClientDisconnectOptionsReason.DisconnectWithWillMessage) + .Build(); + + try + { + await client.DisconnectAsync(options, cancellationToken).ConfigureAwait(false); + } + catch (Exception ex) + { + Log($"kill: disconnect threw ({ex.GetType().Name}); the socket is gone either way"); + } + + client.ApplicationMessageReceivedAsync -= OnMessageAsync; + client.Dispose(); + _client = null; + Log($"killed session (bdSeq {CurrentBdSeq()}); the broker owns the NDEATH now"); + } + + /// + /// Awaits the -th honoured rebirth NCMD. Pre-checks what has already + /// happened, so a caller that registers after the command landed is served immediately. + /// + /// The rebirth count to wait for. + /// Bound on the wait. + /// Caller cancellation. + /// The rebirth count once it has reached . + public async Task WaitForRebirthAsync(int count, TimeSpan timeout, CancellationToken cancellationToken) + { + var deadline = DateTime.UtcNow + timeout; + while (true) + { + var observed = RebirthCount; + if (observed >= count) + { + return observed; + } + + var remaining = deadline - DateTime.UtcNow; + if (remaining <= TimeSpan.Zero) + { + throw new TimeoutException( + $"Simulated edge node '{GroupId}/{EdgeNodeId}' honoured {observed} rebirth request(s) " + + $"within {timeout.TotalSeconds:0}s; expected at least {count}. Is the driver's " + + "requestRebirthOnGap on, and is the node's NCMD subscription live?"); + } + + var signal = Volatile.Read(ref _rebirthSignal); + try + { + await signal.Task.WaitAsync(remaining, cancellationToken).ConfigureAwait(false); + } + catch (TimeoutException) + { + // Loop; the deadline check above produces the diagnostic message. + } + } + } + + /// + /// The standalone (container) loop: birth once, then publish node + device data every + /// until cancelled. Rebirth NCMDs are honoured throughout. + /// + /// Publish cadence. + /// Stops the loop. + /// A task that completes when the loop is cancelled. + public async Task RunAsync(TimeSpan interval, CancellationToken cancellationToken) + { + await BirthAsync(cancellationToken).ConfigureAwait(false); + + var tick = 0; + while (!cancellationToken.IsCancellationRequested) + { + try + { + await Task.Delay(interval, cancellationToken).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + break; + } + + tick++; + try + { + await PublishNodeDataAsync( + [.. _nodeMetrics + .Where(m => m.Name is not BdSeqMetricName and not RebirthMetricName) + .Select(m => (m.Name, Advance(m, tick)))], + cancellationToken).ConfigureAwait(false); + + foreach (var (device, metrics) in _devices) + { + await PublishDeviceDataAsync( + device, + [.. metrics.Select(m => (m.Name, Advance(m, tick)))], + cancellationToken).ConfigureAwait(false); + } + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + Log($"publish failed: {ex.GetType().Name}: {ex.Message}"); + } + } + } + + /// + public async ValueTask DisposeAsync() + { + if (Interlocked.Exchange(ref _disposed, 1) != 0) + { + return; + } + + await DropClientAsync(graceful: true).ConfigureAwait(false); + _publishGate.Dispose(); + } + + // ---- publishing ---- + + private async Task BirthCoreAsync(CancellationToken cancellationToken) + { + // An NBIRTH restarts the sequence: seq = 0, by spec, on the birth itself. + Volatile.Write(ref _seq, 0); + _seqStarted = true; + + var nbirth = new Payload { Timestamp = NowMs(), Seq = 0UL }; + + // bdSeq first, as Tahu's own reference edge node emits it: it is the session token the + // matching NDEATH will carry. + nbirth.Metrics.Add(new Payload.Types.Metric + { + Name = BdSeqMetricName, + Datatype = (uint)DataType.Int64, + Timestamp = NowMs(), + LongValue = CurrentBdSeq(), + }); + + // Every conformant edge node declares the rebirth control point in its NBIRTH. + nbirth.Metrics.Add(new Payload.Types.Metric + { + Name = RebirthMetricName, + Datatype = (uint)DataType.Boolean, + Timestamp = NowMs(), + BooleanValue = false, + }); + + foreach (var metric in _nodeMetrics) + { + nbirth.Metrics.Add(EncodeBirthMetric(metric)); + } + + await PublishAsync(NodeTopic("NBIRTH"), nbirth, _options.DataQos, cancellationToken).ConfigureAwait(false); + + foreach (var (device, metrics) in _devices) + { + // A DBIRTH is a sequenced member of the edge node's stream — it does NOT restart seq, and + // it carries no bdSeq. + var dbirth = NewSequencedPayload(); + foreach (var metric in metrics) + { + dbirth.Metrics.Add(EncodeBirthMetric(metric)); + } + + await PublishAsync(DeviceTopic("DBIRTH", device), dbirth, _options.DataQos, cancellationToken) + .ConfigureAwait(false); + } + + Log($"birthed (bdSeq {CurrentBdSeq()}, {_nodeMetrics.Count} node metric(s), {_devices.Count} device(s))"); + } + + private async Task PublishDataAsync( + string? device, + IReadOnlyList<(string Metric, object? Value)> values, + CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(values); + + var catalog = device is null + ? _nodeMetrics + : _devices.TryGetValue(device, out var found) ? found : []; + + await _publishGate.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + var payload = NewSequencedPayload(); + foreach (var (name, value) in values) + { + var seed = catalog.FirstOrDefault(m => string.Equals(m.Name, name, StringComparison.Ordinal)) + ?? throw new InvalidOperationException( + $"Metric '{name}' is not in the current birth catalog for " + + $"'{GroupId}/{EdgeNodeId}{(device is null ? "" : "/" + device)}'. A DATA message can only " + + "carry metrics a birth declared — that is the whole point of the alias table."); + + payload.Metrics.Add(EncodeDataMetric(seed.WithValue(value))); + } + + var topic = device is null ? NodeTopic("NDATA") : DeviceTopic("DDATA", device); + await PublishAsync(topic, payload, _options.DataQos, cancellationToken).ConfigureAwait(false); + } + finally + { + _publishGate.Release(); + } + } + + private async Task PublishAsync(string topic, Payload payload, int qos, CancellationToken cancellationToken) + { + var client = _client ?? throw new InvalidOperationException( + $"Simulated edge node '{GroupId}/{EdgeNodeId}' is not connected; call ConnectAsync first."); + + var message = new MqttApplicationMessageBuilder() + .WithTopic(topic) + .WithPayload(payload.ToByteArray()) + .WithQualityOfServiceLevel((MqttQualityOfServiceLevel)Math.Clamp(qos, 0, 2)) + .WithRetainFlag(false) // Sparkplug never retains birth or data messages. + .Build(); + + using var deadline = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + deadline.CancelAfter(TimeSpan.FromSeconds(_options.TimeoutSeconds)); + await client.PublishAsync(message, deadline.Token).ConfigureAwait(false); + } + + // ---- inbound NCMD ---- + + private async Task OnMessageAsync(MqttApplicationMessageReceivedEventArgs args) + { + try + { + if (!IsRebirthCommand(args.ApplicationMessage)) + { + return; + } + + var honoured = Interlocked.Increment(ref _rebirthCount); + Log($"rebirth NCMD #{honoured} received; republishing metadata"); + + await _publishGate.WaitAsync(CancellationToken.None).ConfigureAwait(false); + try + { + await BirthCoreAsync(CancellationToken.None).ConfigureAwait(false); + } + finally + { + _publishGate.Release(); + } + + // Swap the signal AFTER the birth is on the wire, so a waiter that wakes on it can + // immediately assert against the republished state rather than racing it. + var previous = Interlocked.Exchange( + ref _rebirthSignal, + new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously)); + previous.TrySetResult(honoured); + } + catch (Exception ex) + { + // This runs on MQTTnet's dispatcher thread; a throw here stalls delivery for the client. + Log($"NCMD handling failed: {ex.GetType().Name}: {ex.Message}"); + } + } + + private bool IsRebirthCommand(MqttApplicationMessage message) + { + if (!string.Equals(message.Topic, NodeTopic("NCMD"), StringComparison.Ordinal)) + { + return false; + } + + var body = message.Payload; + var bytes = body.IsSingleSegment ? body.FirstSpan.ToArray() : body.ToArray(); + + Payload decoded; + try + { + decoded = Payload.Parser.ParseFrom(bytes); + } + catch (InvalidProtocolBufferException) + { + Log("NCMD payload was not a Sparkplug payload; ignored"); + return false; + } + + foreach (var metric in decoded.Metrics) + { + if (string.Equals(metric.Name, RebirthMetricName, StringComparison.Ordinal) + && metric.ValueCase == Payload.Types.Metric.ValueOneofCase.BooleanValue + && metric.BooleanValue) + { + return true; + } + } + + return false; + } + + // ---- encoding ---- + + /// + /// A birth metric: name, alias, datatype and value — everything a host needs to build its alias + /// table. + /// + private static Payload.Types.Metric EncodeBirthMetric(SparkplugMetricSeed seed) + { + var metric = new Payload.Types.Metric + { + Name = seed.Name, + Alias = seed.Alias, + Datatype = (uint)seed.DataType, + Timestamp = NowMs(), + }; + + ApplyValue(metric, seed.DataType, seed.Value); + return metric; + } + + /// + /// A DATA metric: alias only. No name, no datatype — which is how a real edge node + /// publishes after a birth, and the shape that makes bind-by-name load-bearing on the host. + /// + private static Payload.Types.Metric EncodeDataMetric(SparkplugMetricSeed seed) + { + var metric = new Payload.Types.Metric + { + Alias = seed.Alias, + Timestamp = NowMs(), + }; + + ApplyValue(metric, seed.DataType, seed.Value); + return metric; + } + + /// + /// Writes a value into the metric's oneof, in the field Sparkplug's datatype dictates. + /// + /// + /// Signed integers go on the wire as two's complement in the UNSIGNED field — Int8/16/32 + /// in int_value, Int64 in long_value. This is the encoding a host has to undo, and + /// writing a negative value any other way would quietly retire that code path from the gate. + /// + private static void ApplyValue(Payload.Types.Metric metric, DataType dataType, object? value) + { + if (value is null) + { + // An explicit null, not an omitted value: the two mean different things, and only the + // explicit form says "this metric exists and currently has no value". + metric.IsNull = true; + return; + } + + switch (dataType) + { + case DataType.Int8: + metric.IntValue = unchecked((uint)(sbyte)Convert.ToSByte(value, CultureInfo.InvariantCulture)); + break; + case DataType.Int16: + metric.IntValue = unchecked((uint)Convert.ToInt16(value, CultureInfo.InvariantCulture)); + break; + case DataType.Int32: + metric.IntValue = unchecked((uint)Convert.ToInt32(value, CultureInfo.InvariantCulture)); + break; + case DataType.Int64: + metric.LongValue = unchecked((ulong)Convert.ToInt64(value, CultureInfo.InvariantCulture)); + break; + case DataType.Uint8: + case DataType.Uint16: + case DataType.Uint32: + metric.IntValue = Convert.ToUInt32(value, CultureInfo.InvariantCulture); + break; + case DataType.Uint64: + metric.LongValue = Convert.ToUInt64(value, CultureInfo.InvariantCulture); + break; + case DataType.Float: + metric.FloatValue = Convert.ToSingle(value, CultureInfo.InvariantCulture); + break; + case DataType.Double: + metric.DoubleValue = Convert.ToDouble(value, CultureInfo.InvariantCulture); + break; + case DataType.Boolean: + metric.BooleanValue = Convert.ToBoolean(value, CultureInfo.InvariantCulture); + break; + case DataType.DateTime: + metric.LongValue = value is DateTime dt + ? (ulong)new DateTimeOffset(dt.ToUniversalTime()).ToUnixTimeMilliseconds() + : Convert.ToUInt64(value, CultureInfo.InvariantCulture); + break; + case DataType.String: + case DataType.Text: + case DataType.Uuid: + metric.StringValue = Convert.ToString(value, CultureInfo.InvariantCulture) ?? string.Empty; + break; + case DataType.Bytes: + case DataType.File: + metric.BytesValue = ByteString.CopyFrom((byte[])value); + break; + default: + throw new NotSupportedException( + $"The simulator does not encode Sparkplug datatype {dataType}; v1 of the driver does " + + "not consume it either."); + } + } + + private Payload NewSequencedPayload() + { + var payload = new Payload { Timestamp = NowMs(), Seq = NextSeq() }; + return payload; + } + + /// Advances the wrapping 0–255 sequence and returns the value to stamp. + private ulong NextSeq() + { + if (!_seqStarted) + { + // Nothing has birthed yet: start the stream at 0 rather than at 1, so a data-before-birth + // test still produces a legal-looking stream. + _seqStarted = true; + Volatile.Write(ref _seq, 0); + return 0UL; + } + + var next = unchecked((byte)(Volatile.Read(ref _seq) + 1)); + Volatile.Write(ref _seq, next); + return next; + } + + private Payload BuildDeathPayload(ulong bdSeq) + { + // NO seq: an NDEATH is the broker-published Last Will and is not a member of the node's + // sequence. Adding one here would be the single most plausible-looking way to make every + // death look like a gap on the host. + var payload = new Payload { Timestamp = NowMs() }; + payload.Metrics.Add(new Payload.Types.Metric + { + Name = BdSeqMetricName, + Datatype = (uint)DataType.Int64, + Timestamp = NowMs(), + LongValue = bdSeq, + }); + + return payload; + } + + // ---- helpers ---- + + private ulong CurrentBdSeq() => Volatile.Read(ref _bdSeq); + + private string NodeTopic(string type) => $"{Namespace}/{_options.GroupId}/{type}/{_options.EdgeNodeId}"; + + private string DeviceTopic(string type, string device) => + $"{Namespace}/{_options.GroupId}/{type}/{_options.EdgeNodeId}/{device}"; + + private static ulong NowMs() => (ulong)DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(); + + /// Nudges a seed's value so the standalone loop publishes something that visibly moves. + private static object? Advance(SparkplugMetricSeed seed, int tick) => seed.DataType switch + { + DataType.Float => 20f + (tick % 50) / 10f, + DataType.Double => 100d + (tick % 100) / 10d, + DataType.Int32 or DataType.Int16 or DataType.Int8 => -tick, + DataType.Int64 => (long)tick, + DataType.Uint8 or DataType.Uint16 or DataType.Uint32 or DataType.Uint64 => (uint)tick, + DataType.Boolean => tick % 2 == 0, + DataType.String or DataType.Text or DataType.Uuid => $"tick-{tick}", + _ => seed.Value, + }; + + private void ConfigureTls(MqttClientTlsOptionsBuilder tls) + { + if (!_options.UseTls) + { + tls.UseTls(false); + return; + } + + tls.UseTls(true).WithTargetHost(_options.Host); + + if (_options.AllowUntrustedServerCertificate) + { + tls.WithAllowUntrustedCertificates(true) + .WithIgnoreCertificateChainErrors(true) + .WithCertificateValidationHandler(static _ => true); + return; + } + + if (string.IsNullOrWhiteSpace(_options.CaCertificatePath)) + { + return; // OS trust store. + } + + var path = _options.CaCertificatePath; + var roots = new Lazy( + () => + { + try + { + var collection = new X509Certificate2Collection(); + collection.ImportFromPemFile(path); + return collection.Count == 0 ? null : collection; + } + catch (Exception) + { + return null; + } + }, + LazyThreadSafetyMode.ExecutionAndPublication); + + tls.WithCertificateValidationHandler(args => + { + if (roots.Value is not { Count: > 0 } trusted || args.Certificate is null) + { + return false; // Fail closed, exactly as the driver's own pin does. + } + + using var chain = new X509Chain(); + chain.ChainPolicy.TrustMode = X509ChainTrustMode.CustomRootTrust; + chain.ChainPolicy.CustomTrustStore.AddRange(trusted); + chain.ChainPolicy.RevocationMode = X509RevocationMode.NoCheck; + + var presented = args.Certificate as X509Certificate2 + ?? X509CertificateLoader.LoadCertificate(args.Certificate.Export(X509ContentType.Cert)); + + return chain.Build(presented); + }); + } + + private async Task DropClientAsync(bool graceful) + { + var client = _client; + _client = null; + if (client is null) + { + return; + } + + client.ApplicationMessageReceivedAsync -= OnMessageAsync; + + if (graceful) + { + try + { + using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5)); + await client.DisconnectAsync(new MqttClientDisconnectOptions(), cts.Token).ConfigureAwait(false); + } + catch (Exception) + { + // The broker may already be gone; teardown is best-effort. + } + } + + client.Dispose(); + } + + private void Log(string message) => _log?.Invoke($"[{GroupId}/{EdgeNodeId}] {message}"); +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/SparkplugEdgeNodeOptions.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/SparkplugEdgeNodeOptions.cs new file mode 100644 index 00000000..0a790ef8 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/SparkplugEdgeNodeOptions.cs @@ -0,0 +1,58 @@ +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.SparkplugSimulator; + +/// +/// Connection + identity settings for one simulated Sparkplug B edge node. +/// +/// +/// Deliberately its own shape rather than a reuse of the driver's MqttDriverOptions: an edge +/// node is the driver's counterparty, and reusing the driver's options record would drag the +/// driver assembly into this project and blur which side of the wire a setting belongs to. The TLS +/// knobs mirror the driver's by name so a rig configured for one is configured for the other. +/// +public sealed record SparkplugEdgeNodeOptions +{ + /// Broker hostname or IP. + public string Host { get; init; } = "localhost"; + + /// Broker TCP port. + public int Port { get; init; } = 8883; + + /// Connect over TLS. + public bool UseTls { get; init; } = true; + + /// + /// PEM CA file pinning the broker chain. /empty falls back to the OS + /// trust store (or to when that is set). + /// + public string? CaCertificatePath { get; init; } + + /// Accept any broker certificate. Fixture escape hatch; never a deployment posture. + public bool AllowUntrustedServerCertificate { get; init; } + + /// Broker username. The fixture broker runs allow_anonymous false. + public string? Username { get; init; } + + /// Broker password. + public string? Password { get; init; } + + /// The Sparkplug group id this node publishes under. + public string GroupId { get; init; } = "Plant1"; + + /// The Sparkplug edge-node id. + public string EdgeNodeId { get; init; } = "EdgeA"; + + /// + /// MQTT client id. Left unset a unique one is generated — two simulated nodes sharing a client + /// id would knock each other off the broker. + /// + public string? ClientId { get; init; } + + /// Bounded connect/publish deadline, in seconds. + public int TimeoutSeconds { get; init; } = 15; + + /// + /// QoS for NBIRTH/DBIRTH/NDATA/DDATA/DDEATH. Sparkplug B publishes all of these at QoS 0 + /// (only NDEATH — the broker-issued Will — and STATE are QoS 1), which is the default here. + /// + public int DataQos { get; init; } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/SparkplugMetricSeed.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/SparkplugMetricSeed.cs new file mode 100644 index 00000000..40f856b2 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/SparkplugMetricSeed.cs @@ -0,0 +1,32 @@ +using Org.Eclipse.Tahu.Protobuf; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.SparkplugSimulator; + +/// +/// One metric in a simulated edge node's (or device's) catalog: the name a birth certificate +/// declares, the alias it binds that name to for the life of the birth, the Sparkplug datatype, and +/// the value the birth publishes. +/// +/// The stable metric name — what an authored tag binds to. +/// +/// The per-birth alias. Mutable across rebirths by design: reassigning an alias to a +/// different metric on a rebirth is legal Sparkplug and is §3.6 matrix case (i), the one that +/// silently mis-routes data in a driver that binds by alias. +/// +/// The Sparkplug datatype the birth declares for this metric. +/// +/// The value the birth publishes. is encoded as an explicit +/// is_null metric rather than as an omitted value — those mean different things on the wire. +/// +public sealed record SparkplugMetricSeed(string Name, ulong Alias, DataType DataType, object? Value) +{ + /// Returns this seed with a different value, leaving name/alias/type alone. + /// The new value. + /// The updated seed. + public SparkplugMetricSeed WithValue(object? value) => this with { Value = value }; + + /// Returns this seed with a different alias — the alias-reuse case. + /// The new alias. + /// The updated seed. + public SparkplugMetricSeed WithAlias(ulong alias) => this with { Alias = alias }; +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.SparkplugSimulator.csproj b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.SparkplugSimulator.csproj new file mode 100644 index 00000000..fec0ea4e --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.SparkplugSimulator.csproj @@ -0,0 +1,46 @@ + + + + + Exe + net10.0 + enable + enable + latest + true + true + $(NoWarn);CS1591 + false + ZB.MOM.WW.OtOpcUa.Driver.Mqtt.SparkplugSimulator + ZB.MOM.WW.OtOpcUa.Driver.Mqtt.SparkplugSimulator + + + + + + + + + + + diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests.csproj b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests.csproj index f7b27445..2f8d349f 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests.csproj +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests.csproj @@ -33,9 +33,29 @@ + + + + + + + + + +