Files
lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttWriteEncoderTests.cs
T
Joseph Doherty 72be19225d 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
2026-07-31 06:33:23 -04:00

252 lines
10 KiB
C#

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;
}
}