feat(mqtt): write-through (IWritable) — Plain publish + Sparkplug NCMD/DCMD (#508)
MqttDriver implemented no IWritable, so every MQTT node materialized read-only
regardless of what the operator authored. It now publishes.
Plain -> the tag's writeTopic; UTF-8 text (Raw/Scalar) or a bare JSON
literal (Json)
Sparkplug -> DCMD for a device metric, NCMD for an edge-node metric, carrying
the metric BY NAME with an explicit Datatype and no seq
Both at QoS 0, retain:false. Retain is deliberately not configurable — a retained
command is replayed by the broker on every future subscribe, including the
device's own reconnect, so one stale write would be re-applied forever.
writeTopic is a new, separate TagConfig key, and its ABSENCE is meaningful: the
write is refused BadNotWritable and the node is discovered ViewOnly. The issue's
"do not skip" warning drove the design — the one outcome worse than read-only
nodes is a node that looks writable, returns Good and publishes nothing. Reusing
the read topic would be worse still: it would echo back and read as a confirmed
device value. Browse-time writability is computed by the SAME encoder the write
path uses, so a node cannot advertise writability it would then refuse.
Good means published, not applied — MQTT has no application-level write ack, so
this is Galaxy's optimistic-Good posture and inherits its property: a failed
DEVICE write can never be surfaced. A failed PUBLISH is reported Bad; the
optimism covers what the device did with the command, never whether it was sent.
WriteIdempotent stays default-off.
The encoder is pure and lives in .Contracts, so both wire shapes are testable
without a broker. Unrepresentable values and types with no Sparkplug encoding are
refused rather than coerced — a Reference must not silently become a String on
the wire, and "not-a-number" must not become 0 on a live plant.
LIVE GATE PASSED against the real Mosquitto fixture on 10.100.0.35: 21/21
integration tests, including 6 new ones that observe the published message with
an INDEPENDENT broker subscriber — a fake asserting "publish was called" could
not tell the feature working from the feature publishing to the wrong topic.
The gate earned its keep: the NCMD test initially failed because NCMD is a SHARED
channel — the driver publishes its own rebirth request (Node Control/Rebirth) on
the very topic an edge-node metric write uses, and it arrived first. Correct
driver behaviour, wrong test; the watcher now matches on metric name. Documented,
because anything consuming that topic needs to know. DCMD has no contention.
Offline: 24 encoder + 9 driver tests. The four new status constants are NAMED so
StatusCodeParityTests can see them (#497/#519) — it went 258 -> 262 and validated
each against the pinned SDK.
Also corrects the Mqtt.md "Known gaps" table: prunes the six rows closed in this
round, and fixes the #507 row, which still described rediscovery as inert after
that issue was closed as superseded and its injection path deleted.
Claude-Session: https://claude.ai/code/session_015p7wGqy3YpZNCpDzTpGMKo
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,387 @@
|
||||
using System.Buffers;
|
||||
using System.Collections.Concurrent;
|
||||
using System.Text;
|
||||
using MQTTnet;
|
||||
using Org.Eclipse.Tahu.Protobuf;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using TahuDataType = Org.Eclipse.Tahu.Protobuf.DataType;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests;
|
||||
|
||||
/// <summary>
|
||||
/// Live validation of MQTT write-through (Gitea #508) against the real Mosquitto fixture: a Plain
|
||||
/// publish to the tag's write topic, and Sparkplug NCMD/DCMD commands, observed on the broker by an
|
||||
/// independent subscriber.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>Observed on the broker, not on a fake.</b> The whole risk this feature carries is a write that
|
||||
/// reports Good and reaches nothing, so the assertion has to be that a real message arrived on a
|
||||
/// real broker at the topic the operator authored. A test double asserting "publish was called"
|
||||
/// could not distinguish the feature working from the feature publishing to the wrong topic.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Env-gated + skip-clean</b>, like every suite here: without <c>MQTT_FIXTURE_ENDPOINT</c> and
|
||||
/// credentials each test calls <c>Assert.Skip</c>. See <see cref="MqttFixture"/>.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
[Collection(MqttFixtureCollection.Name)]
|
||||
[Trait("Category", "LiveIntegration")]
|
||||
public sealed class MqttWriteLiveTests(MqttFixture fixture)
|
||||
{
|
||||
private const uint Good = 0x00000000u;
|
||||
private const uint BadNotWritable = 0x803B0000u;
|
||||
|
||||
private static readonly TimeSpan MessageTimeout = TimeSpan.FromSeconds(30);
|
||||
|
||||
private readonly MqttFixture _fx = fixture;
|
||||
|
||||
private void SkipIfUnconfigured() => Assert.SkipWhen(_fx.SkipReason is not null, _fx.SkipReason ?? "");
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// Plain
|
||||
// -----------------------------------------------------------------------
|
||||
|
||||
[Fact]
|
||||
public async Task Plain_write_publishes_the_value_on_the_authored_write_topic()
|
||||
{
|
||||
SkipIfUnconfigured();
|
||||
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
const string rawPath = "Fixture/Mqtt/Line1/Setpoint";
|
||||
var readTopic = _fx.Topic($"write/{Guid.NewGuid():N}/state");
|
||||
var writeTopic = _fx.Topic($"write/{Guid.NewGuid():N}/set");
|
||||
|
||||
await using var watcher = await TopicWatcher.StartAsync(_fx, writeTopic, ct);
|
||||
|
||||
var tag = new RawTagEntry(
|
||||
rawPath,
|
||||
$$"""{"topic":"{{readTopic}}","writeTopic":"{{writeTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""",
|
||||
WriteIdempotent: false);
|
||||
|
||||
await using var driver = new MqttDriver(PlainOptions(tag), "mqtt-live-write");
|
||||
await driver.InitializeAsync(string.Empty, ct);
|
||||
|
||||
var results = await driver.WriteAsync([new WriteRequest(rawPath, 42.5)], ct);
|
||||
|
||||
results.ShouldHaveSingleItem().StatusCode.ShouldBe(Good);
|
||||
|
||||
var payload = await watcher.WaitForOneAsync(MessageTimeout, ct);
|
||||
Encoding.UTF8.GetString(payload).ShouldBe(
|
||||
"42.5",
|
||||
"the write must arrive on the authored write topic carrying the value, in invariant form");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Plain_write_never_lands_on_the_tags_read_topic()
|
||||
{
|
||||
SkipIfUnconfigured();
|
||||
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
const string rawPath = "Fixture/Mqtt/Line1/SetpointSplit";
|
||||
var suffix = Guid.NewGuid().ToString("N");
|
||||
var readTopic = _fx.Topic($"write/{suffix}/state");
|
||||
var writeTopic = _fx.Topic($"write/{suffix}/set");
|
||||
|
||||
// Both watchers are live BEFORE the write, so "nothing on the read topic" is a real observation
|
||||
// over the same window rather than an absence that was never looked for.
|
||||
await using var readWatcher = await TopicWatcher.StartAsync(_fx, readTopic, ct);
|
||||
await using var writeWatcher = await TopicWatcher.StartAsync(_fx, writeTopic, ct);
|
||||
|
||||
var tag = new RawTagEntry(
|
||||
rawPath,
|
||||
$$"""{"topic":"{{readTopic}}","writeTopic":"{{writeTopic}}","payloadFormat":"Scalar","dataType":"Float64"}""",
|
||||
WriteIdempotent: false);
|
||||
|
||||
await using var driver = new MqttDriver(PlainOptions(tag), "mqtt-live-write-split");
|
||||
await driver.InitializeAsync(string.Empty, ct);
|
||||
|
||||
(await driver.WriteAsync([new WriteRequest(rawPath, 7.25)], ct))[0].StatusCode.ShouldBe(Good);
|
||||
|
||||
// Positive control first: once the write HAS been seen, the read topic has had its chance.
|
||||
await writeWatcher.WaitForOneAsync(MessageTimeout, ct);
|
||||
readWatcher.Count.ShouldBe(
|
||||
0,
|
||||
"a write echoed onto the tag's own read topic would read back as a confirmed device value");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Plain_write_to_a_tag_with_no_write_topic_is_refused_end_to_end()
|
||||
{
|
||||
SkipIfUnconfigured();
|
||||
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
const string rawPath = "Fixture/Mqtt/Line1/ReadOnly";
|
||||
var tag = new RawTagEntry(
|
||||
rawPath,
|
||||
$$"""{"topic":"{{_fx.Topic("line1/temperature")}}","payloadFormat":"Json","jsonPath":"$.value","dataType":"Float64"}""",
|
||||
WriteIdempotent: false);
|
||||
|
||||
await using var driver = new MqttDriver(PlainOptions(tag), "mqtt-live-write-readonly");
|
||||
await driver.InitializeAsync(string.Empty, ct);
|
||||
|
||||
// Connected and resolvable — so BadNotWritable here is specifically "you never said where",
|
||||
// not a connection failure wearing the same status.
|
||||
(await driver.WriteAsync([new WriteRequest(rawPath, 1.0)], ct))[0].StatusCode.ShouldBe(BadNotWritable);
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// Sparkplug B
|
||||
// -----------------------------------------------------------------------
|
||||
|
||||
[Fact]
|
||||
public async Task Sparkplug_write_to_a_device_metric_publishes_a_DCMD()
|
||||
{
|
||||
SkipIfUnconfigured();
|
||||
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
var group = $"WriteGate{Guid.NewGuid():N}"[..20];
|
||||
const string node = "EdgeW";
|
||||
const string device = "Filler1";
|
||||
const string metric = "Setpoint";
|
||||
const string rawPath = "Fixture/Mqtt/Sp/Setpoint";
|
||||
|
||||
var topic = $"spBv1.0/{group}/DCMD/{node}/{device}";
|
||||
await using var watcher = await TopicWatcher.StartAsync(_fx, topic, ct);
|
||||
|
||||
var tag = new RawTagEntry(
|
||||
rawPath,
|
||||
$$"""{"groupId":"{{group}}","edgeNodeId":"{{node}}","deviceId":"{{device}}","metricName":"{{metric}}","dataType":"Float64"}""",
|
||||
WriteIdempotent: false);
|
||||
|
||||
await using var driver = new MqttDriver(SparkplugOptions(group, tag), "mqtt-live-write-dcmd");
|
||||
await driver.InitializeAsync(string.Empty, ct);
|
||||
|
||||
(await driver.WriteAsync([new WriteRequest(rawPath, 12.5)], ct))[0].StatusCode.ShouldBe(Good);
|
||||
|
||||
var bytes = await watcher.WaitForOneAsync(MessageTimeout, ct);
|
||||
var payload = Payload.Parser.ParseFrom(bytes);
|
||||
|
||||
payload.Metrics.ShouldHaveSingleItem();
|
||||
payload.Metrics[0].Name.ShouldBe(metric, "a command must address the metric by NAME, never by alias");
|
||||
payload.Metrics[0].Datatype.ShouldBe((uint)TahuDataType.Double);
|
||||
payload.Metrics[0].DoubleValue.ShouldBe(12.5);
|
||||
payload.HasSeq.ShouldBeFalse("seq belongs to the edge node's own outbound stream, not to a command");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Sparkplug_write_to_an_edge_node_metric_publishes_an_NCMD()
|
||||
{
|
||||
SkipIfUnconfigured();
|
||||
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
var group = $"WriteGate{Guid.NewGuid():N}"[..20];
|
||||
const string node = "EdgeW";
|
||||
const string metric = "NodeSetpoint";
|
||||
const string rawPath = "Fixture/Mqtt/Sp/NodeSetpoint";
|
||||
|
||||
await using var watcher = await TopicWatcher.StartAsync(_fx, $"spBv1.0/{group}/NCMD/{node}", ct);
|
||||
|
||||
var tag = new RawTagEntry(
|
||||
rawPath,
|
||||
$$"""{"groupId":"{{group}}","edgeNodeId":"{{node}}","metricName":"{{metric}}","dataType":"Int32"}""",
|
||||
WriteIdempotent: false);
|
||||
|
||||
await using var driver = new MqttDriver(SparkplugOptions(group, tag), "mqtt-live-write-ncmd");
|
||||
await driver.InitializeAsync(string.Empty, ct);
|
||||
|
||||
(await driver.WriteAsync([new WriteRequest(rawPath, 7)], ct))[0].StatusCode.ShouldBe(Good);
|
||||
|
||||
// NCMD is a SHARED channel: the driver publishes its own rebirth request
|
||||
// ("Node Control/Rebirth") on this exact topic when it connects, and the live gate caught it
|
||||
// arriving first. Wait for the command carrying OUR metric rather than for the first NCMD —
|
||||
// the DCMD case has no such contention, which is why only this one needed it.
|
||||
var payload = Payload.Parser.ParseFrom(
|
||||
await watcher.WaitForMatchAsync(
|
||||
bytes => Payload.Parser.ParseFrom(bytes).Metrics.Any(m => m.Name == metric),
|
||||
MessageTimeout,
|
||||
ct));
|
||||
|
||||
var written = payload.Metrics.ShouldHaveSingleItem();
|
||||
written.Name.ShouldBe(metric);
|
||||
written.Datatype.ShouldBe((uint)TahuDataType.Int32);
|
||||
written.IntValue.ShouldBe(7u);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_sparkplug_command_is_not_retained()
|
||||
{
|
||||
SkipIfUnconfigured();
|
||||
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
var group = $"WriteGate{Guid.NewGuid():N}"[..20];
|
||||
const string node = "EdgeW";
|
||||
const string rawPath = "Fixture/Mqtt/Sp/Retain";
|
||||
var topic = $"spBv1.0/{group}/NCMD/{node}";
|
||||
|
||||
await using var firstWatcher = await TopicWatcher.StartAsync(_fx, topic, ct);
|
||||
|
||||
var tag = new RawTagEntry(
|
||||
rawPath,
|
||||
$$"""{"groupId":"{{group}}","edgeNodeId":"{{node}}","metricName":"Cmd","dataType":"Boolean"}""",
|
||||
WriteIdempotent: false);
|
||||
|
||||
await using var driver = new MqttDriver(SparkplugOptions(group, tag), "mqtt-live-write-retain");
|
||||
await driver.InitializeAsync(string.Empty, ct);
|
||||
(await driver.WriteAsync([new WriteRequest(rawPath, true)], ct))[0].StatusCode.ShouldBe(Good);
|
||||
await firstWatcher.WaitForOneAsync(MessageTimeout, ct);
|
||||
|
||||
// A retained command would be replayed by the broker to every LATER subscriber — including the
|
||||
// device's own reconnect — so a single stale write would be re-applied forever. Subscribing
|
||||
// fresh, after the publish, is the only way to observe that.
|
||||
await using var lateWatcher = await TopicWatcher.StartAsync(_fx, topic, ct);
|
||||
await Task.Delay(TimeSpan.FromSeconds(2), ct);
|
||||
|
||||
// Counting OUR command only. The driver's own rebirth NCMD shares this topic, so a bare count
|
||||
// would make this assertion depend on rebirth timing rather than on the retain flag.
|
||||
lateWatcher.CountMatching(bytes => Payload.Parser.ParseFrom(bytes).Metrics.Any(m => m.Name == "Cmd"))
|
||||
.ShouldBe(0, "a retained NCMD/DCMD would drive a device into a permanent command loop");
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// Helpers
|
||||
// -----------------------------------------------------------------------
|
||||
|
||||
private MqttDriverOptions PlainOptions(params RawTagEntry[] rawTags) => new()
|
||||
{
|
||||
Host = _fx.PlainHost,
|
||||
Port = _fx.PlainPort,
|
||||
UseTls = false,
|
||||
Username = _fx.Username,
|
||||
Password = _fx.Password,
|
||||
ClientId = $"otopcua-live-{Guid.NewGuid():N}",
|
||||
ConnectTimeoutSeconds = 15,
|
||||
Mode = MqttMode.Plain,
|
||||
Plain = new MqttPlainOptions { DefaultQos = 1 },
|
||||
RawTags = rawTags,
|
||||
};
|
||||
|
||||
private MqttDriverOptions SparkplugOptions(string groupId, params RawTagEntry[] rawTags) => new()
|
||||
{
|
||||
Host = _fx.PlainHost,
|
||||
Port = _fx.PlainPort,
|
||||
UseTls = false,
|
||||
Username = _fx.Username,
|
||||
Password = _fx.Password,
|
||||
ClientId = $"otopcua-live-{Guid.NewGuid():N}",
|
||||
ConnectTimeoutSeconds = 15,
|
||||
Mode = MqttMode.SparkplugB,
|
||||
Sparkplug = new MqttSparkplugOptions { GroupId = groupId },
|
||||
RawTags = rawTags,
|
||||
};
|
||||
|
||||
/// <summary>
|
||||
/// An independent broker subscriber that records the payloads seen on one concrete topic — the
|
||||
/// third party that makes "the write actually reached the broker" an observation rather than an
|
||||
/// assumption about the driver's own code.
|
||||
/// </summary>
|
||||
private sealed class TopicWatcher : IAsyncDisposable
|
||||
{
|
||||
private readonly IMqttClient _client;
|
||||
private readonly ConcurrentQueue<byte[]> _payloads = new();
|
||||
|
||||
private TopicWatcher(IMqttClient client) => _client = client;
|
||||
|
||||
/// <summary>How many messages have been observed on the watched topic.</summary>
|
||||
public int Count => _payloads.Count;
|
||||
|
||||
public static async Task<TopicWatcher> StartAsync(MqttFixture fixture, string topic, CancellationToken ct)
|
||||
{
|
||||
var client = new MqttClientFactory().CreateMqttClient();
|
||||
var watcher = new TopicWatcher(client);
|
||||
|
||||
client.ApplicationMessageReceivedAsync += args =>
|
||||
{
|
||||
watcher._payloads.Enqueue(args.ApplicationMessage.Payload.ToArray());
|
||||
return Task.CompletedTask;
|
||||
};
|
||||
|
||||
var options = new MqttClientOptionsBuilder()
|
||||
.WithTcpServer(fixture.PlainHost, fixture.PlainPort)
|
||||
.WithCredentials(fixture.Username, fixture.Password)
|
||||
.WithClientId($"otopcua-live-watch-{Guid.NewGuid():N}")
|
||||
.WithTimeout(TimeSpan.FromSeconds(10))
|
||||
.Build();
|
||||
|
||||
await client.ConnectAsync(options, ct).ConfigureAwait(false);
|
||||
await client.SubscribeAsync(
|
||||
new MqttClientSubscribeOptionsBuilder().WithTopicFilter(f => f.WithTopic(topic)).Build(),
|
||||
ct).ConfigureAwait(false);
|
||||
|
||||
return watcher;
|
||||
}
|
||||
|
||||
/// <summary>How many observed messages satisfy <paramref name="predicate"/>.</summary>
|
||||
/// <param name="predicate">The payload test.</param>
|
||||
/// <returns>The matching count.</returns>
|
||||
public int CountMatching(Func<byte[], bool> predicate) => _payloads.Count(p => Safe(predicate, p));
|
||||
|
||||
/// <summary>
|
||||
/// Waits for a message satisfying <paramref name="predicate"/>. Needed where the topic is
|
||||
/// shared — an NCMD topic also carries the driver's own rebirth requests.
|
||||
/// </summary>
|
||||
/// <param name="predicate">The payload test.</param>
|
||||
/// <param name="timeout">Bound on the wait.</param>
|
||||
/// <param name="ct">Caller cancellation.</param>
|
||||
/// <returns>The first matching payload.</returns>
|
||||
public async Task<byte[]> WaitForMatchAsync(Func<byte[], bool> predicate, TimeSpan timeout, CancellationToken ct)
|
||||
{
|
||||
var deadline = DateTime.UtcNow + timeout;
|
||||
while (true)
|
||||
{
|
||||
if (_payloads.FirstOrDefault(p => Safe(predicate, p)) is { } match)
|
||||
{
|
||||
return match;
|
||||
}
|
||||
|
||||
if (DateTime.UtcNow >= deadline)
|
||||
{
|
||||
throw new TimeoutException(
|
||||
$"No matching message arrived on the watched topic within {timeout} " +
|
||||
$"({_payloads.Count} message(s) seen, none matching).");
|
||||
}
|
||||
|
||||
await Task.Delay(100, ct).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Applies a payload predicate, treating an undecodable payload as "no match".</summary>
|
||||
/// <param name="predicate">The payload test.</param>
|
||||
/// <param name="payload">The observed bytes.</param>
|
||||
/// <returns><see langword="true"/> when the payload matches.</returns>
|
||||
private static bool Safe(Func<byte[], bool> predicate, byte[] payload)
|
||||
{
|
||||
try { return predicate(payload); }
|
||||
catch { return false; }
|
||||
}
|
||||
|
||||
public async Task<byte[]> WaitForOneAsync(TimeSpan timeout, CancellationToken ct)
|
||||
{
|
||||
var deadline = DateTime.UtcNow + timeout;
|
||||
while (!_payloads.TryPeek(out _))
|
||||
{
|
||||
if (DateTime.UtcNow >= deadline)
|
||||
{
|
||||
throw new TimeoutException(
|
||||
$"No message arrived on the watched topic within {timeout}. The write reported success " +
|
||||
"but nothing reached the broker — the exact failure this gate exists to catch.");
|
||||
}
|
||||
|
||||
await Task.Delay(100, ct).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
_payloads.TryDequeue(out var payload);
|
||||
return payload!;
|
||||
}
|
||||
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
try { await _client.DisconnectAsync(cancellationToken: default).ConfigureAwait(false); }
|
||||
catch { /* teardown must not mask a test failure */ }
|
||||
|
||||
_client.Dispose();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,160 @@
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// Unit tests for <see cref="MqttDriver"/>'s <see cref="IWritable"/> leg (Gitea #508) — the batch
|
||||
/// contract and the per-slot refusals, up to (not including) the broker publish.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// The publish itself is the live gate's job: it needs a broker, and asserting it here would only
|
||||
/// prove that a fake was called. What IS worth pinning offline is everything that decides whether a
|
||||
/// write should reach a broker at all — because that is where a silent no-op would hide.
|
||||
/// </remarks>
|
||||
[Trait("Category", "Unit")]
|
||||
public sealed class MqttDriverWriteTests
|
||||
{
|
||||
private const string Path = "Plant/Mqtt/dev1/Setpoint";
|
||||
private const string ReadOnlyPath = "Plant/Mqtt/dev1/Temp";
|
||||
|
||||
private static RawTagEntry PlainTag(string rawPath, string? writeTopic)
|
||||
{
|
||||
var writeKey = writeTopic is null ? "" : $$""","writeTopic":"{{writeTopic}}" """;
|
||||
return new RawTagEntry(
|
||||
rawPath,
|
||||
$$"""{"topic":"factory/oven/temp","payloadFormat":"Scalar","dataType":"Float64"{{writeKey}}}""",
|
||||
WriteIdempotent: false);
|
||||
}
|
||||
|
||||
private static MqttDriver PlainDriver(params RawTagEntry[] tags)
|
||||
=> new(new MqttDriverOptions { Mode = MqttMode.Plain, RawTags = tags }, "d", null);
|
||||
|
||||
[Fact]
|
||||
public void MqttDriver_implements_IWritable()
|
||||
// The interface's presence is the feature: without it every MQTT node materialises read-only,
|
||||
// regardless of what the operator authored on the tag.
|
||||
=> PlainDriver().ShouldBeAssignableTo<IWritable>();
|
||||
|
||||
[Fact]
|
||||
public async Task An_unknown_reference_is_refused_in_its_own_slot()
|
||||
{
|
||||
var results = await PlainDriver(PlainTag(Path, "oven/set"))
|
||||
.WriteAsync([new WriteRequest("Plant/Mqtt/dev1/NotATag", 1.0)], default);
|
||||
|
||||
results.Count.ShouldBe(1);
|
||||
results[0].StatusCode.ShouldBe(0x80340000u); // BadNodeIdUnknown
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_tag_with_no_write_topic_is_refused_as_not_writable()
|
||||
{
|
||||
// NOT BadCommunicationError and NOT Good: the tag is addressable and the broker is irrelevant —
|
||||
// the operator simply never said where a write should go. The status has to say that, or the
|
||||
// only diagnosis available is "writes don't work".
|
||||
var results = await PlainDriver(PlainTag(ReadOnlyPath, writeTopic: null))
|
||||
.WriteAsync([new WriteRequest(ReadOnlyPath, 1.0)], default);
|
||||
|
||||
results[0].StatusCode.ShouldBe(0x803B0000u); // BadNotWritable
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_writable_tag_on_a_disconnected_driver_reports_a_communication_failure()
|
||||
{
|
||||
// Reported, never queued. A write buffered now and flushed at an unbounded later moment is the
|
||||
// replay hazard WriteIdempotent exists to prevent — and this would be the driver deciding it on
|
||||
// the operator's behalf.
|
||||
var results = await PlainDriver(PlainTag(Path, "oven/set"))
|
||||
.WriteAsync([new WriteRequest(Path, 42.5)], default);
|
||||
|
||||
results[0].StatusCode.ShouldBe(0x80050000u); // BadCommunicationError
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task One_bad_request_does_not_fail_the_others_in_the_batch()
|
||||
{
|
||||
// Mirrors ReadAsync: failing the whole call would let one stale reference reject every other
|
||||
// write in the same OPC UA request.
|
||||
var driver = PlainDriver(PlainTag(Path, "oven/set"), PlainTag(ReadOnlyPath, writeTopic: null));
|
||||
|
||||
var results = await driver.WriteAsync(
|
||||
[
|
||||
new WriteRequest("Plant/Mqtt/dev1/NotATag", 1.0),
|
||||
new WriteRequest(ReadOnlyPath, 1.0),
|
||||
new WriteRequest(Path, 42.5),
|
||||
],
|
||||
default);
|
||||
|
||||
// Order-preserving, one status per request, each naming its own distinct cause.
|
||||
results.Count.ShouldBe(3);
|
||||
results[0].StatusCode.ShouldBe(0x80340000u); // BadNodeIdUnknown
|
||||
results[1].StatusCode.ShouldBe(0x803B0000u); // BadNotWritable
|
||||
results[2].StatusCode.ShouldBe(0x80050000u); // BadCommunicationError — resolved and encodable
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task An_empty_batch_is_answered_with_an_empty_result()
|
||||
=> (await PlainDriver().WriteAsync([], default)).ShouldBeEmpty();
|
||||
|
||||
[Fact]
|
||||
public async Task WriteAsync_rejects_a_null_batch()
|
||||
=> await Should.ThrowAsync<ArgumentNullException>(
|
||||
async () => await PlainDriver().WriteAsync(null!, default));
|
||||
|
||||
// ---- Discovery advertises writability consistently with the write path ------------------------
|
||||
|
||||
[Fact]
|
||||
public async Task A_tag_with_a_write_topic_is_discovered_as_an_operate_node()
|
||||
{
|
||||
var builder = new WriteCapturingBuilder();
|
||||
|
||||
await PlainDriver(PlainTag(Path, "oven/set")).DiscoverAsync(builder, default);
|
||||
|
||||
builder.Variables.ShouldHaveSingleItem();
|
||||
builder.Variables[0].SecurityClass.ShouldBe(SecurityClassification.Operate);
|
||||
// Default-off regardless: MQTT gives no application-level write ack, and a Sparkplug command is
|
||||
// not necessarily idempotent. Opting in is the operator's call, per tag.
|
||||
builder.Variables[0].WriteIdempotent.ShouldBeFalse();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_tag_without_a_write_topic_stays_view_only()
|
||||
{
|
||||
// The browse-time answer must match the runtime one — a node advertised Operate that then
|
||||
// refuses every write would be the silent-no-op failure in a new costume.
|
||||
var builder = new WriteCapturingBuilder();
|
||||
|
||||
await PlainDriver(PlainTag(ReadOnlyPath, writeTopic: null)).DiscoverAsync(builder, default);
|
||||
|
||||
builder.Variables[0].SecurityClass.ShouldBe(SecurityClassification.ViewOnly);
|
||||
}
|
||||
|
||||
/// <summary>Minimal capturing builder — this project does not reference Commons' own.</summary>
|
||||
private sealed class WriteCapturingBuilder : IAddressSpaceBuilder
|
||||
{
|
||||
public List<DriverAttributeInfo> Variables { get; } = [];
|
||||
|
||||
public IAddressSpaceBuilder Folder(string browseName, string displayName) => this;
|
||||
|
||||
public IVariableHandle Variable(string browseName, string displayName, DriverAttributeInfo attributeInfo)
|
||||
{
|
||||
Variables.Add(attributeInfo);
|
||||
return new Handle(attributeInfo.FullName);
|
||||
}
|
||||
|
||||
public void AddProperty(string browseName, DriverDataType dataType, object? value) { }
|
||||
|
||||
private sealed class Handle(string fullRef) : IVariableHandle
|
||||
{
|
||||
public string FullReference => fullRef;
|
||||
|
||||
public IAlarmConditionSink MarkAsAlarmCondition(AlarmConditionInfo info) => new NullSink();
|
||||
}
|
||||
|
||||
private sealed class NullSink : IAlarmConditionSink
|
||||
{
|
||||
public void OnTransition(AlarmEventArgs args) { }
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,251 @@
|
||||
using System.Text;
|
||||
using Google.Protobuf;
|
||||
using Org.Eclipse.Tahu.Protobuf;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug;
|
||||
using TahuDataType = Org.Eclipse.Tahu.Protobuf.DataType;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// Unit tests for <see cref="MqttWriteEncoder"/> — the pure half of MQTT write-through (Gitea #508):
|
||||
/// what a write encodes to, and which writes are refused before anything reaches a broker.
|
||||
/// </summary>
|
||||
[Trait("Category", "Unit")]
|
||||
public sealed class MqttWriteEncoderTests
|
||||
{
|
||||
private static MqttTagDefinition Plain(
|
||||
string? writeTopic,
|
||||
MqttPayloadFormat format = MqttPayloadFormat.Scalar,
|
||||
DriverDataType dataType = DriverDataType.Float64)
|
||||
=> new(
|
||||
Name: "Plant/Mqtt/dev1/Setpoint",
|
||||
Topic: "factory/oven/temp",
|
||||
PayloadFormat: format,
|
||||
JsonPath: "$",
|
||||
DataType: dataType,
|
||||
Qos: null,
|
||||
RetainSeed: true,
|
||||
WriteTopic: writeTopic);
|
||||
|
||||
private static MqttTagDefinition Sparkplug(
|
||||
string? deviceId,
|
||||
DriverDataType dataType = DriverDataType.Float64,
|
||||
string metricName = "Speed")
|
||||
=> new(
|
||||
Name: "Plant/Mqtt/dev1/Speed",
|
||||
Topic: "",
|
||||
PayloadFormat: MqttPayloadFormat.Json,
|
||||
JsonPath: "$",
|
||||
DataType: dataType,
|
||||
Qos: null,
|
||||
RetainSeed: true,
|
||||
GroupId: "OtOpcUaSim",
|
||||
EdgeNodeId: "EdgeA",
|
||||
DeviceId: deviceId,
|
||||
MetricName: metricName);
|
||||
|
||||
// ---- The refusal that is the whole point of the feature -------------------------------------
|
||||
|
||||
[Fact]
|
||||
public void A_plain_tag_with_no_write_topic_is_refused_rather_than_published_nowhere()
|
||||
{
|
||||
// Before #508 there was no IWritable at all — honest read-only. The one outcome that would be
|
||||
// WORSE is a node that accepts a write, reports Good and publishes nothing. So this must be a
|
||||
// refusal, and specifically the "no destination" one, which the driver maps to BadNotWritable.
|
||||
var error = MqttWriteEncoder.TryEncode(MqttMode.Plain, Plain(writeTopic: null), 42.5, out var encoded);
|
||||
|
||||
error.ShouldBe(MqttWriteEncodeError.NoWriteDestination);
|
||||
encoded.Topic.ShouldBeNull();
|
||||
encoded.Payload.ShouldBeNull();
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData("")]
|
||||
[InlineData(" ")]
|
||||
public void A_blank_write_topic_is_treated_as_absent(string writeTopic)
|
||||
=> MqttWriteEncoder.TryEncode(MqttMode.Plain, Plain(writeTopic), 1, out _)
|
||||
.ShouldBe(MqttWriteEncodeError.NoWriteDestination);
|
||||
|
||||
[Fact]
|
||||
public void A_sparkplug_tag_with_an_incomplete_tuple_has_no_command_address()
|
||||
{
|
||||
// A Sparkplug tag's command address IS its binding tuple, so an incomplete one is the same
|
||||
// failure as a missing write topic, not a different kind.
|
||||
var def = Sparkplug(deviceId: null) with { MetricName = null };
|
||||
|
||||
MqttWriteEncoder.TryEncode(MqttMode.SparkplugB, def, 1.0, out _)
|
||||
.ShouldBe(MqttWriteEncodeError.NoWriteDestination);
|
||||
}
|
||||
|
||||
// ---- Plain encoding --------------------------------------------------------------------------
|
||||
|
||||
[Fact]
|
||||
public void Plain_publishes_to_the_write_topic_not_the_read_topic()
|
||||
{
|
||||
// The two-field split is the point: publishing a command onto the tag's own read topic would
|
||||
// echo back and read as a confirmed device value.
|
||||
var def = Plain("factory/oven/temp/set");
|
||||
|
||||
MqttWriteEncoder.TryEncode(MqttMode.Plain, def, 42.5, out var encoded)
|
||||
.ShouldBe(MqttWriteEncodeError.None);
|
||||
|
||||
encoded.Topic.ShouldBe("factory/oven/temp/set");
|
||||
encoded.Topic.ShouldNotBe(def.Topic);
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData(MqttPayloadFormat.Scalar)]
|
||||
[InlineData(MqttPayloadFormat.Raw)]
|
||||
public void Scalar_and_raw_encode_the_value_as_utf8_text(MqttPayloadFormat format)
|
||||
{
|
||||
MqttWriteEncoder.TryEncode(MqttMode.Plain, Plain("t/set", format), 42.5, out var encoded)
|
||||
.ShouldBe(MqttWriteEncodeError.None);
|
||||
|
||||
Encoding.UTF8.GetString(encoded.Payload).ShouldBe("42.5");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Numbers_are_encoded_with_the_invariant_culture()
|
||||
{
|
||||
// A comma decimal separator would publish a number the device cannot parse — and it would only
|
||||
// reproduce on a host with a European locale, which is the worst kind of bug to chase.
|
||||
var previous = Thread.CurrentThread.CurrentCulture;
|
||||
Thread.CurrentThread.CurrentCulture = new System.Globalization.CultureInfo("de-DE");
|
||||
try
|
||||
{
|
||||
MqttWriteEncoder.TryEncode(MqttMode.Plain, Plain("t/set"), 42.5, out var encoded)
|
||||
.ShouldBe(MqttWriteEncodeError.None);
|
||||
|
||||
Encoding.UTF8.GetString(encoded.Payload).ShouldBe("42.5");
|
||||
}
|
||||
finally
|
||||
{
|
||||
Thread.CurrentThread.CurrentCulture = previous;
|
||||
}
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Json_encodes_a_bare_json_literal()
|
||||
{
|
||||
MqttWriteEncoder.TryEncode(
|
||||
MqttMode.Plain, Plain("t/set", MqttPayloadFormat.Json), 42.5, out var encoded)
|
||||
.ShouldBe(MqttWriteEncodeError.None);
|
||||
|
||||
Encoding.UTF8.GetString(encoded.Payload).ShouldBe("42.5");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Json_quotes_a_string_value()
|
||||
=> Encoding.UTF8.GetString(
|
||||
EncodeOrThrow(MqttMode.Plain, Plain("t/set", MqttPayloadFormat.Json, DriverDataType.String), "on"))
|
||||
.ShouldBe("\"on\"");
|
||||
|
||||
[Fact]
|
||||
public void A_null_value_is_refused()
|
||||
// A zero-length MQTT payload is a legitimate clear-the-retained-message instruction — an
|
||||
// entirely different operation. Encoding null as "no bytes" would conflate the two.
|
||||
=> MqttWriteEncoder.TryEncode(MqttMode.Plain, Plain("t/set"), null, out _)
|
||||
.ShouldBe(MqttWriteEncodeError.ValueNotRepresentable);
|
||||
|
||||
// ---- Sparkplug command encoding ---------------------------------------------------------------
|
||||
|
||||
[Fact]
|
||||
public void A_device_metric_encodes_as_a_DCMD_addressed_to_that_device()
|
||||
{
|
||||
// Addressing a device metric with NCMD would command the NODE, which does not own it.
|
||||
MqttWriteEncoder.TryEncode(MqttMode.SparkplugB, Sparkplug("Filler1"), 12.5, out var encoded)
|
||||
.ShouldBe(MqttWriteEncodeError.None);
|
||||
|
||||
encoded.Topic.ShouldBe("spBv1.0/OtOpcUaSim/DCMD/EdgeA/Filler1");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void An_edge_node_metric_encodes_as_an_NCMD()
|
||||
=> EncodeTopic(MqttMode.SparkplugB, Sparkplug(deviceId: null))
|
||||
.ShouldBe("spBv1.0/OtOpcUaSim/NCMD/EdgeA");
|
||||
|
||||
[Fact]
|
||||
public void The_command_carries_the_metric_by_name_with_an_explicit_datatype()
|
||||
{
|
||||
// By NAME, never by alias: an alias may be reused for a different metric across a rebirth.
|
||||
// Datatype explicitly, because a node that treats it as ambiguous drops the command SILENTLY —
|
||||
// the exact failure mode this feature must not have.
|
||||
var bytes = EncodeOrThrow(MqttMode.SparkplugB, Sparkplug("Filler1"), 12.5);
|
||||
var payload = Payload.Parser.ParseFrom(bytes);
|
||||
|
||||
payload.Metrics.Count.ShouldBe(1);
|
||||
var metric = payload.Metrics[0];
|
||||
metric.Name.ShouldBe("Speed");
|
||||
metric.Alias.ShouldBe(0ul);
|
||||
metric.Datatype.ShouldBe((uint)TahuDataType.Double);
|
||||
metric.DoubleValue.ShouldBe(12.5);
|
||||
metric.Timestamp.ShouldBeGreaterThan(0ul);
|
||||
payload.Timestamp.ShouldBeGreaterThan(0ul);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void The_command_carries_no_sequence_number()
|
||||
{
|
||||
// seq belongs to the messages an edge node itself publishes. A strict node validating a
|
||||
// host-supplied one against its own counter would have grounds to reject the command.
|
||||
var payload = Payload.Parser.ParseFrom(EncodeOrThrow(MqttMode.SparkplugB, Sparkplug("Filler1"), 1.0));
|
||||
|
||||
payload.HasSeq.ShouldBeFalse();
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData(DriverDataType.Boolean, true, (uint)TahuDataType.Boolean)]
|
||||
[InlineData(DriverDataType.Int32, 7, (uint)TahuDataType.Int32)]
|
||||
[InlineData(DriverDataType.Int64, 7L, (uint)TahuDataType.Int64)]
|
||||
[InlineData(DriverDataType.UInt32, 7, (uint)TahuDataType.Uint32)]
|
||||
[InlineData(DriverDataType.Float32, 1.5f, (uint)TahuDataType.Float)]
|
||||
[InlineData(DriverDataType.String, "x", (uint)TahuDataType.String)]
|
||||
public void The_declared_data_type_selects_the_wire_datatype(
|
||||
DriverDataType dataType, object value, uint expectedWireType)
|
||||
{
|
||||
var payload = Payload.Parser.ParseFrom(
|
||||
EncodeOrThrow(MqttMode.SparkplugB, Sparkplug("Filler1", dataType), value));
|
||||
|
||||
payload.Metrics[0].Datatype.ShouldBe(expectedWireType);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void An_unrepresentable_value_is_refused_rather_than_coerced()
|
||||
// "not-a-number" into a Float64 tag must not silently become 0 on a live plant.
|
||||
=> MqttWriteEncoder.TryEncode(MqttMode.SparkplugB, Sparkplug("Filler1"), "not-a-number", out _)
|
||||
.ShouldBe(MqttWriteEncodeError.ValueNotRepresentable);
|
||||
|
||||
[Fact]
|
||||
public void A_type_with_no_command_encoding_is_refused_rather_than_stringified()
|
||||
// Reference has no Sparkplug wire representation. Falling back to String would put a value on
|
||||
// the wire that is not what the tag's type says it is.
|
||||
=> MqttWriteEncoder.TryEncode(
|
||||
MqttMode.SparkplugB, Sparkplug("Filler1", DriverDataType.Reference), "ref", out _)
|
||||
.ShouldBe(MqttWriteEncodeError.UnsupportedDataType);
|
||||
|
||||
[Fact]
|
||||
public void The_mode_decides_the_shape_not_the_blob()
|
||||
{
|
||||
// A tag can legitimately carry both shapes' keys (the editor preserves unknown keys across a
|
||||
// retype), so sniffing would make one authored tag mean two different things.
|
||||
var both = Sparkplug("Filler1") with { WriteTopic = "plain/set" };
|
||||
|
||||
EncodeTopic(MqttMode.Plain, both).ShouldBe("plain/set");
|
||||
EncodeTopic(MqttMode.SparkplugB, both).ShouldBe("spBv1.0/OtOpcUaSim/DCMD/EdgeA/Filler1");
|
||||
}
|
||||
|
||||
private static string EncodeTopic(MqttMode mode, MqttTagDefinition def)
|
||||
{
|
||||
MqttWriteEncoder.TryEncode(mode, def, 1.0, out var encoded).ShouldBe(MqttWriteEncodeError.None);
|
||||
return encoded.Topic;
|
||||
}
|
||||
|
||||
private static byte[] EncodeOrThrow(MqttMode mode, MqttTagDefinition def, object value)
|
||||
{
|
||||
MqttWriteEncoder.TryEncode(mode, def, value, out var encoded).ShouldBe(MqttWriteEncodeError.None);
|
||||
return encoded.Payload;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user