eddf5f57a5
Driver.Mqtt.Browser-002/-003/-004/-005. -002: ObservePlainTopic's pre-pass judged segment LENGTH only, so a plain topic segment carrying a newline or an ANSI escape became the picker's displayed label — a rule ObserveSparkplugBirth already applied via IsUsableLabel. The control-character half is now single-sourced (HasControlCharacter) and the plain marker text converges with the Sparkplug one. An EMPTY topic level stays accepted on purpose: it is legal MQTT and its terminal leaf is a real committable address. -003: GetOrCreate read _nodeCount and incremented it later, so N observers could each see cap-1 and all insert. The slot is now reserved with one Interlocked.Increment and given back on both non-consuming paths. -004: Project ordered a level by Segment alone, but the level is keyed by Path precisely because two Sparkplug siblings CAN share a Segment — leaving the tie to ConcurrentDictionary enumeration order, which .NET's randomised string hashing makes differ per process. .ThenBy(Path) makes the sort total. -005: no behaviour change — DefaultNodeCap's doc now states the real per-session ceiling and records that the aggregate across sessions is bounded by the reaper's idle TTL, not by any session cap.
1096 lines
52 KiB
C#
1096 lines
52 KiB
C#
using Shouldly;
|
|
using Xunit;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Browsing;
|
|
using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser;
|
|
|
|
// See SparkplugTopicTests.cs for why this alias is redeclared locally in every consuming project
|
|
// rather than inherited: `SparkplugDataType` is a `global using` alias for the generated
|
|
// `Org.Eclipse.Tahu.Protobuf.DataType` (declared in Contracts/SparkplugDataType.cs), and `global using`
|
|
// scope does not cross a ProjectReference.
|
|
using SparkplugDataType = Org.Eclipse.Tahu.Protobuf.DataType;
|
|
|
|
namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests;
|
|
|
|
/// <summary>
|
|
/// Covers the passive <c>#</c>-observation browse session and the browser that opens it. The
|
|
/// session is exercised through its test seam (<c>ObserveTopicForTest</c>) rather than a live
|
|
/// broker: the accumulating tree is the whole of the P1 logic, and the MQTTnet plumbing that
|
|
/// feeds it is a one-line handler.
|
|
/// </summary>
|
|
public sealed class MqttBrowseSessionTests
|
|
{
|
|
// ---------------------------------------------------------------- tree shape
|
|
|
|
[Fact]
|
|
public async Task ExpandAsync_BuildsTopicSegmentTree_FromObservedTopics()
|
|
{
|
|
await using var s = new MqttBrowseSession(MqttMode.Plain);
|
|
s.ObserveTopicForTest("factory/line3/oven/temp");
|
|
|
|
var root = await s.RootAsync(TestContext.Current.CancellationToken);
|
|
root.ShouldContain(n => n.DisplayName == "factory");
|
|
|
|
var lvl2 = await s.ExpandAsync("factory", TestContext.Current.CancellationToken);
|
|
lvl2.ShouldContain(n => n.DisplayName == "line3");
|
|
|
|
var lvl3 = await s.ExpandAsync("factory/line3", TestContext.Current.CancellationToken);
|
|
lvl3.ShouldContain(n => n.DisplayName == "oven");
|
|
|
|
var lvl4 = await s.ExpandAsync("factory/line3/oven", TestContext.Current.CancellationToken);
|
|
var leaf = lvl4.ShouldHaveSingleItem();
|
|
leaf.DisplayName.ShouldBe("temp");
|
|
leaf.NodeId.ShouldBe("factory/line3/oven/temp");
|
|
leaf.Kind.ShouldBe(BrowseNodeKind.Leaf);
|
|
leaf.HasChildrenHint.ShouldBeFalse();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ObservedTopics_AccumulateAcrossMessages_AndDoNotDuplicate()
|
|
{
|
|
await using var s = new MqttBrowseSession(MqttMode.Plain);
|
|
s.ObserveTopicForTest("factory/line3/oven/temp");
|
|
s.ObserveTopicForTest("factory/line3/oven/temp"); // repeat — must not duplicate
|
|
s.ObserveTopicForTest("factory/line3/oven/pressure");
|
|
s.ObserveTopicForTest("factory/line4/mixer/rpm");
|
|
s.ObserveTopicForTest("plant/power");
|
|
|
|
var root = await s.RootAsync(TestContext.Current.CancellationToken);
|
|
root.Select(n => n.DisplayName).ShouldBe(["factory", "plant"]); // sorted, deduped
|
|
|
|
var lines = await s.ExpandAsync("factory", TestContext.Current.CancellationToken);
|
|
lines.Select(n => n.DisplayName).ShouldBe(["line3", "line4"]);
|
|
|
|
var oven = await s.ExpandAsync("factory/line3/oven", TestContext.Current.CancellationToken);
|
|
oven.Select(n => n.DisplayName).ShouldBe(["pressure", "temp"]);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task IntermediateNode_IsFolder_WithChildrenHint()
|
|
{
|
|
await using var s = new MqttBrowseSession(MqttMode.Plain);
|
|
s.ObserveTopicForTest("factory/line3/oven/temp");
|
|
|
|
var root = await s.RootAsync(TestContext.Current.CancellationToken);
|
|
var factory = root.ShouldHaveSingleItem();
|
|
factory.Kind.ShouldBe(BrowseNodeKind.Folder);
|
|
factory.HasChildrenHint.ShouldBeTrue();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ExpandAsync_UnknownNode_ReturnsEmpty()
|
|
{
|
|
await using var s = new MqttBrowseSession(MqttMode.Plain);
|
|
s.ObserveTopicForTest("factory/line3");
|
|
|
|
var children = await s.ExpandAsync("nope/not/here", TestContext.Current.CancellationToken);
|
|
children.ShouldBeEmpty();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RootAsync_BeforeAnyTraffic_IsEmpty()
|
|
{
|
|
// MQTT has no discovery protocol: an observation window that has seen nothing shows nothing.
|
|
await using var s = new MqttBrowseSession(MqttMode.Plain);
|
|
(await s.RootAsync(TestContext.Current.CancellationToken)).ShouldBeEmpty();
|
|
}
|
|
|
|
// ---------------------------------------------------------------- read-only guarantee
|
|
|
|
[Fact]
|
|
public async Task BrowseCalls_PublishNothing() // browse is read-only
|
|
{
|
|
await using var s = new MqttBrowseSession(MqttMode.Plain);
|
|
s.ObserveTopicForTest("factory/line3/oven/temp", "23.5");
|
|
|
|
await s.RootAsync(TestContext.Current.CancellationToken);
|
|
await s.ExpandAsync("factory", TestContext.Current.CancellationToken);
|
|
await s.AttributesAsync("factory/line3/oven/temp", TestContext.Current.CancellationToken);
|
|
|
|
s.PublishCountForTest.ShouldBe(0);
|
|
}
|
|
|
|
// ---------------------------------------------------------------- attributes
|
|
|
|
[Theory]
|
|
[InlineData("23.5", nameof(Core.Abstractions.DriverDataType.Float64))]
|
|
[InlineData("42", nameof(Core.Abstractions.DriverDataType.Int64))]
|
|
[InlineData("true", nameof(Core.Abstractions.DriverDataType.Boolean))]
|
|
[InlineData("{\"v\":1}", nameof(Core.Abstractions.DriverDataType.String))]
|
|
public async Task AttributesAsync_InfersTypeFromLastPayload(string payload, string expectedType)
|
|
{
|
|
await using var s = new MqttBrowseSession(MqttMode.Plain);
|
|
s.ObserveTopicForTest("factory/oven/temp", payload);
|
|
|
|
var attrs = await s.AttributesAsync("factory/oven/temp", TestContext.Current.CancellationToken);
|
|
var a = attrs.ShouldHaveSingleItem();
|
|
a.Name.ShouldBe("temp");
|
|
a.DriverDataType.ShouldBe(expectedType);
|
|
a.IsArray.ShouldBeFalse();
|
|
a.IsAlarm.ShouldBeFalse();
|
|
a.SecurityClass.ShouldContain(payload); // the last-payload snippet
|
|
}
|
|
|
|
[Fact]
|
|
public async Task AttributesAsync_LongPayload_IsTruncatedToASnippet()
|
|
{
|
|
await using var s = new MqttBrowseSession(MqttMode.Plain);
|
|
s.ObserveTopicForTest("t/long", new string('x', 500));
|
|
|
|
var a = (await s.AttributesAsync("t/long", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();
|
|
a.SecurityClass.Length.ShouldBeLessThanOrEqualTo(MqttBrowseSession.PayloadSnippetMaxChars + 1);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task AttributesAsync_UnobservedFolder_ReturnsEmpty()
|
|
{
|
|
await using var s = new MqttBrowseSession(MqttMode.Plain);
|
|
s.ObserveTopicForTest("factory/oven/temp", "1");
|
|
|
|
// "factory" is a synthesised path segment — nothing was ever published to it.
|
|
(await s.AttributesAsync("factory", TestContext.Current.CancellationToken)).ShouldBeEmpty();
|
|
(await s.AttributesAsync("no/such/node", TestContext.Current.CancellationToken)).ShouldBeEmpty();
|
|
}
|
|
|
|
// ---------------------------------------------------------------- concurrency
|
|
|
|
[Fact]
|
|
public async Task ObserveDuringBrowse_IsThreadSafe()
|
|
{
|
|
// Messages arrive on MQTTnet's dispatcher thread while the picker calls ExpandAsync.
|
|
await using var s = new MqttBrowseSession(MqttMode.Plain);
|
|
using var stop = new CancellationTokenSource();
|
|
|
|
var observed = 0;
|
|
var writer = Task.Run(() =>
|
|
{
|
|
var i = 0;
|
|
while (!stop.Token.IsCancellationRequested)
|
|
{
|
|
s.ObserveTopicForTest($"factory/line{i % 8}/dev{i % 32}/m{i}", i.ToString());
|
|
Interlocked.Increment(ref observed);
|
|
i++;
|
|
}
|
|
}, CancellationToken.None);
|
|
|
|
// The read loop is pure in-memory work and would otherwise finish before the writer thread
|
|
// is even scheduled, leaving the two never actually overlapping.
|
|
while (Volatile.Read(ref observed) == 0) await Task.Yield();
|
|
|
|
for (var i = 0; i < 400; i++)
|
|
{
|
|
await s.RootAsync(TestContext.Current.CancellationToken);
|
|
await s.ExpandAsync("factory", TestContext.Current.CancellationToken);
|
|
await s.ExpandAsync("factory/line3", TestContext.Current.CancellationToken);
|
|
}
|
|
|
|
await stop.CancelAsync();
|
|
await writer; // rethrows anything the observer thread hit
|
|
|
|
(await s.RootAsync(TestContext.Current.CancellationToken)).ShouldNotBeEmpty();
|
|
}
|
|
|
|
// ---------------------------------------------------------------- lifetime
|
|
|
|
[Fact]
|
|
public async Task LastUsedUtc_AdvancesOnEveryCall()
|
|
{
|
|
var s = new MqttBrowseSession(MqttMode.Plain);
|
|
var before = s.LastUsedUtc;
|
|
await Task.Delay(15, TestContext.Current.CancellationToken);
|
|
|
|
await s.RootAsync(TestContext.Current.CancellationToken);
|
|
s.LastUsedUtc.ShouldBeGreaterThan(before);
|
|
|
|
var afterRoot = s.LastUsedUtc;
|
|
await Task.Delay(15, TestContext.Current.CancellationToken);
|
|
await s.ExpandAsync("x", TestContext.Current.CancellationToken);
|
|
s.LastUsedUtc.ShouldBeGreaterThan(afterRoot);
|
|
|
|
await s.DisposeAsync();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task DisposeAsync_OnNeverOpenedSession_IsIdempotentAndDoesNotThrow()
|
|
{
|
|
var s = new MqttBrowseSession(MqttMode.Plain);
|
|
await Should.NotThrowAsync(async () => await s.DisposeAsync());
|
|
await Should.NotThrowAsync(async () => await s.DisposeAsync());
|
|
}
|
|
|
|
[Fact]
|
|
public async Task AfterDispose_BrowseCallsThrowObjectDisposed()
|
|
{
|
|
var s = new MqttBrowseSession(MqttMode.Plain);
|
|
s.ObserveTopicForTest("a/b");
|
|
await s.DisposeAsync();
|
|
|
|
await Should.ThrowAsync<ObjectDisposedException>(
|
|
async () => await s.RootAsync(TestContext.Current.CancellationToken));
|
|
await Should.ThrowAsync<ObjectDisposedException>(
|
|
async () => await s.ExpandAsync("a", TestContext.Current.CancellationToken));
|
|
await Should.ThrowAsync<ObjectDisposedException>(
|
|
async () => await s.AttributesAsync("a/b", TestContext.Current.CancellationToken));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ObserveAfterDispose_IsIgnored()
|
|
{
|
|
// A message already in MQTTnet's dispatcher when the reaper disposes the session.
|
|
var s = new MqttBrowseSession(MqttMode.Plain);
|
|
await s.DisposeAsync();
|
|
Should.NotThrow(() => s.ObserveTopicForTest("late/arrival", "1"));
|
|
}
|
|
|
|
// ---------------------------------------------------------------- node cap
|
|
|
|
[Fact]
|
|
public async Task NodeCap_StopsRecording_AndSurfacesATruncationMarker()
|
|
{
|
|
await using var s = new MqttBrowseSession(MqttMode.Plain, nodeCap: 4);
|
|
s.ObserveTopicForTest("a/b/c"); // 3 nodes
|
|
s.ObserveTopicForTest("d/e/f"); // would take it past 4
|
|
|
|
var root = await s.RootAsync(TestContext.Current.CancellationToken);
|
|
root.ShouldContain(n => n.NodeId == MqttBrowseSession.TruncatedNodeId);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task OversizedTopic_IsRejectedWhole_AndSurfacesAMarker()
|
|
{
|
|
// A truncated topic is a DIFFERENT topic — the picker would commit it as a tag address.
|
|
// Every segment here is 4 chars, well under MaxSegmentChars, so ONLY the whole-topic bound
|
|
// can reject it — the segment guard must not be what makes this test pass.
|
|
await using var s = new MqttBrowseSession(MqttMode.Plain);
|
|
s.ObserveTopicForTest("good/topic", "1");
|
|
|
|
var manyShortSegments = "huge/" + string.Join('/', Enumerable.Repeat("abc", MqttBrowseSession.MaxTopicChars / 4 + 8));
|
|
manyShortSegments.Length.ShouldBeGreaterThan(MqttBrowseSession.MaxTopicChars);
|
|
manyShortSegments.Split('/').ShouldAllBe(seg => seg.Length <= MqttBrowseSession.MaxSegmentChars);
|
|
s.ObserveTopicForTest(manyShortSegments, "1");
|
|
|
|
var root = await s.RootAsync(TestContext.Current.CancellationToken);
|
|
root.ShouldContain(n => n.NodeId == MqttBrowseSession.OversizedNodeId);
|
|
root.ShouldContain(n => n.DisplayName == "good");
|
|
root.ShouldNotContain(n => n.DisplayName == "huge"); // no half-built path left behind
|
|
}
|
|
|
|
[Fact]
|
|
public async Task OversizedSegment_IsRejectedWhole_EvenWhenTheTopicFits()
|
|
{
|
|
await using var s = new MqttBrowseSession(MqttMode.Plain);
|
|
s.ObserveTopicForTest("a/" + new string('y', MqttBrowseSession.MaxSegmentChars + 1) + "/b", "1");
|
|
|
|
s.OversizedRejected.ShouldBeTrue();
|
|
(await s.ExpandAsync("a", TestContext.Current.CancellationToken)).ShouldBeEmpty();
|
|
var root = await s.RootAsync(TestContext.Current.CancellationToken);
|
|
root.ShouldNotContain(n => n.DisplayName == "a");
|
|
}
|
|
|
|
[Fact]
|
|
public async Task TopicAtExactlyTheCap_IsAccepted()
|
|
{
|
|
await using var s = new MqttBrowseSession(MqttMode.Plain);
|
|
s.ObserveTopicForTest(new string('z', MqttBrowseSession.MaxSegmentChars), "1");
|
|
|
|
s.OversizedRejected.ShouldBeFalse();
|
|
(await s.RootAsync(TestContext.Current.CancellationToken)).ShouldHaveSingleItem();
|
|
}
|
|
|
|
/// <summary>
|
|
/// The two ingest paths must agree on what may become a tree label. MQTT forbids only
|
|
/// <c>U+0000</c> in a topic NAME, so a publisher can legally put a newline or an ANSI escape in
|
|
/// one — and a plain segment goes straight into the picker's displayed label, exactly where the
|
|
/// Sparkplug path already rejects one.
|
|
/// </summary>
|
|
/// <summary>
|
|
/// The node cap must hold under concurrent observation. MQTTnet delivers on a single dispatcher
|
|
/// per client today, so this is a guard on the shape rather than a live defect — but the cap is
|
|
/// the only thing standing between a <c>#</c> firehose and unbounded memory, and a read-then-
|
|
/// increment pair lets N threads each see <c>cap - 1</c> and all insert.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task NodeCap_HoldsUnderConcurrentObservation()
|
|
{
|
|
// One round would pass by luck; the overshoot is a race. Each round drives many threads at a
|
|
// cap of 1 through a barrier, so every thread reaches the cap check at once.
|
|
const int rounds = 40;
|
|
var workers = Math.Max(Environment.ProcessorCount * 2, 8);
|
|
|
|
for (var round = 0; round < rounds; round++)
|
|
{
|
|
await using var s = new MqttBrowseSession(MqttMode.Plain, nodeCap: 1);
|
|
using var barrier = new Barrier(workers);
|
|
|
|
// Dedicated threads, not Task.Run: every worker BLOCKS on the barrier, and blocking that
|
|
// many pool threads makes the pool inject new ones on its ~500 ms starvation timer — which
|
|
// both slows the test enormously and staggers the very arrivals it is trying to align.
|
|
var threads = Enumerable.Range(0, workers)
|
|
.Select(i => new Thread(() =>
|
|
{
|
|
barrier.SignalAndWait();
|
|
s.ObserveTopicForTest($"t{i}", "1");
|
|
}) { IsBackground = true })
|
|
.ToList();
|
|
|
|
foreach (var t in threads) t.Start();
|
|
foreach (var t in threads) t.Join();
|
|
|
|
var recorded = (await s.RootAsync(TestContext.Current.CancellationToken))
|
|
.Count(n => n.NodeId != MqttBrowseSession.TruncatedNodeId
|
|
&& n.NodeId != MqttBrowseSession.OversizedNodeId);
|
|
recorded.ShouldBeLessThanOrEqualTo(1, $"round {round} overshot the node cap");
|
|
}
|
|
}
|
|
|
|
[Fact]
|
|
public async Task PlainTopicSegmentWithAControlCharacter_IsRejectedWhole_AndSurfacesAMarker()
|
|
{
|
|
// Every segment here is well under MaxSegmentChars, so ONLY the control-character rule can
|
|
// reject it — the length guard must not be what makes this test pass.
|
|
await using var s = new MqttBrowseSession(MqttMode.Plain);
|
|
s.ObserveTopicForTest("good/topic", "1");
|
|
s.ObserveTopicForTest("bad/oven\nname/temp", "1");
|
|
|
|
s.OversizedRejected.ShouldBeTrue();
|
|
var root = await s.RootAsync(TestContext.Current.CancellationToken);
|
|
root.ShouldContain(n => n.NodeId == MqttBrowseSession.OversizedNodeId);
|
|
root.ShouldContain(n => n.DisplayName == "good");
|
|
root.ShouldNotContain(n => n.DisplayName == "bad"); // no half-built path left behind
|
|
}
|
|
|
|
[Theory]
|
|
[InlineData("\n")]
|
|
[InlineData("\r")]
|
|
[InlineData("\t")]
|
|
[InlineData("\u001b")] // ANSI escape
|
|
[InlineData("\u0000")]
|
|
public async Task PlainTopic_EachControlCharacterForm_IsRejected(string control)
|
|
{
|
|
await using var s = new MqttBrowseSession(MqttMode.Plain);
|
|
s.ObserveTopicForTest($"a/b{control}c/d", "1");
|
|
|
|
s.OversizedRejected.ShouldBeTrue();
|
|
(await s.RootAsync(TestContext.Current.CancellationToken))
|
|
.ShouldNotContain(n => n.DisplayName == "a");
|
|
}
|
|
|
|
/// <summary>
|
|
/// Falsification control for the rule above: the guard rejects CONTROL characters, not
|
|
/// "anything unusual". A topic segment is arbitrary UTF-8 and legitimately carries accents,
|
|
/// spaces and punctuation — rejecting those would drop real plant topics from the picker.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task PlainTopic_PrintableNonAsciiSegments_AreStillAccepted()
|
|
{
|
|
await using var s = new MqttBrowseSession(MqttMode.Plain);
|
|
s.ObserveTopicForTest("usiné/four n°3/temp %", "1");
|
|
|
|
s.OversizedRejected.ShouldBeFalse();
|
|
(await s.RootAsync(TestContext.Current.CancellationToken))
|
|
.ShouldHaveSingleItem().DisplayName.ShouldBe("usiné");
|
|
}
|
|
|
|
/// <summary>
|
|
/// An EMPTY topic level (<c>a//b</c>, or a trailing <c>/</c>) is legal MQTT and stays accepted —
|
|
/// deliberately NOT converged with <c>IsUsableLabel</c>'s non-empty rule. Its terminal leaf is a
|
|
/// real, committable address (the commit reads <c>AddressFields[topic]</c>, which carries the
|
|
/// node's full path), so rejecting the topic whole would cost the operator a legal address to
|
|
/// tidy a blank picker row. The control-character rule is the one that had to converge: a label
|
|
/// carrying a newline breaks the row it renders in.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task PlainTopic_WithAnEmptyLevel_IsStillRecorded()
|
|
{
|
|
await using var s = new MqttBrowseSession(MqttMode.Plain);
|
|
s.ObserveTopicForTest("a//b", "1");
|
|
|
|
s.OversizedRejected.ShouldBeFalse();
|
|
var leaf = (await s.ExpandAsync("a/", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();
|
|
leaf.NodeId.ShouldBe("a//b");
|
|
|
|
var attr = (await s.AttributesAsync("a//b", TestContext.Current.CancellationToken))
|
|
.ShouldHaveSingleItem();
|
|
attr.AddressFields.ShouldNotBeNull()[MqttTagConfigKeys.Topic].ShouldBe("a//b");
|
|
}
|
|
|
|
// ---------------------------------------------------------------- payload bounds
|
|
|
|
[Fact]
|
|
public async Task PayloadBeyondTheInspectionWindow_IsNeverExamined()
|
|
{
|
|
// The discriminating test for the byte cap: content past PayloadInspectMaxBytes cannot
|
|
// influence the result at all. With an unbounded decode the trailing scalar would be found
|
|
// and typed Int64, and would show up in the snippet.
|
|
await using var s = new MqttBrowseSession(MqttMode.Plain);
|
|
s.ObserveTopicForTest(
|
|
"t/tail",
|
|
new string(' ', MqttBrowseSession.PayloadInspectMaxBytes + 64) + "42");
|
|
|
|
var a = (await s.AttributesAsync("t/tail", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();
|
|
a.DriverDataType.ShouldBe(nameof(Core.Abstractions.DriverDataType.String));
|
|
a.SecurityClass.ShouldNotContain("42");
|
|
}
|
|
|
|
[Fact]
|
|
public async Task HugePayload_YieldsABoundedSnippetMarkedAsClipped()
|
|
{
|
|
// A '#' subscription receives multi-MB retained blobs; the node cap never engages for a
|
|
// large payload on an already-known topic, so the decode itself must be bounded.
|
|
await using var s = new MqttBrowseSession(MqttMode.Plain);
|
|
s.ObserveTopicForTest("t/blob", new string('a', 5_000_000));
|
|
|
|
var a = (await s.AttributesAsync("t/blob", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();
|
|
a.SecurityClass.Length.ShouldBeLessThanOrEqualTo(MqttBrowseSession.PayloadSnippetMaxChars + 1);
|
|
a.SecurityClass.ShouldEndWith("…");
|
|
a.DriverDataType.ShouldBe(nameof(Core.Abstractions.DriverDataType.String));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ClippedMultiByteUtf8_IsNotMislabelledAsBinary()
|
|
{
|
|
// The inspection window can land mid-sequence; backing off to a whole sequence keeps a
|
|
// perfectly good UTF-8 payload from being reported as binary.
|
|
await using var s = new MqttBrowseSession(MqttMode.Plain);
|
|
|
|
// 3 bytes per char, so 512 / 3 leaves a partial sequence at the window edge.
|
|
s.ObserveTopicForTest("t/utf8", new string('☃', 4000));
|
|
|
|
var a = (await s.AttributesAsync("t/utf8", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();
|
|
a.SecurityClass.ShouldStartWith("☃");
|
|
a.SecurityClass.ShouldNotContain("binary");
|
|
a.SecurityClass.ShouldNotContain("�"); // no replacement char from a split sequence
|
|
}
|
|
|
|
[Fact]
|
|
public async Task BinaryPayload_IsReportedAsOpaque_WithItsTrueLength()
|
|
{
|
|
await using var s = new MqttBrowseSession(MqttMode.Plain);
|
|
var binary = new byte[2048];
|
|
Array.Fill(binary, (byte)0xC0); // never legal UTF-8
|
|
s.Observe("t/bin", binary);
|
|
|
|
var a = (await s.AttributesAsync("t/bin", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();
|
|
a.SecurityClass.ShouldBe("(2048 bytes, binary)"); // full length, not the clipped window
|
|
}
|
|
|
|
[Fact]
|
|
public async Task EmbeddedControlCharacters_AreScrubbedToSpaces()
|
|
{
|
|
// Trim() only strips the ends; an embedded newline would break the picker's single-line row.
|
|
await using var s = new MqttBrowseSession(MqttMode.Plain);
|
|
s.ObserveTopicForTest("t/multiline", "line one\nline two\ttabbed[31m");
|
|
|
|
var a = (await s.AttributesAsync("t/multiline", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();
|
|
a.SecurityClass.ShouldNotContain("\n");
|
|
a.SecurityClass.ShouldNotContain("\t");
|
|
a.SecurityClass.ShouldNotContain("");
|
|
a.SecurityClass.ShouldStartWith("line one line two tabbed");
|
|
}
|
|
|
|
// ---------------------------------------------------------------- the browser
|
|
|
|
[Fact]
|
|
public void Ctor_IsConnectionFree_AndReportsTheMqttDriverType()
|
|
{
|
|
// The universal-browser CanBrowse pattern constructs a throwaway instance purely to ask
|
|
// what driver type it handles — the ctor must never touch the network.
|
|
var browser = new MqttDriverBrowser();
|
|
browser.DriverType.ShouldBe("Mqtt");
|
|
}
|
|
|
|
[Theory]
|
|
[InlineData(null, "#")]
|
|
[InlineData("", "#")]
|
|
[InlineData(" ", "#")]
|
|
[InlineData("factory", "factory/#")]
|
|
[InlineData("factory/", "factory/#")]
|
|
[InlineData("factory/line3", "factory/line3/#")]
|
|
public void BuildRootFilter_ScopesTheWildcardToTheConfiguredPrefix(string? prefix, string expected)
|
|
{
|
|
var opts = new MqttDriverOptions
|
|
{
|
|
Mode = MqttMode.Plain,
|
|
Plain = prefix is null ? null : new MqttPlainOptions { TopicPrefix = prefix },
|
|
};
|
|
MqttDriverBrowser.BuildRootFilter(opts).ShouldBe(expected);
|
|
}
|
|
|
|
[Fact]
|
|
public void BrowseClientIdSuffix_IsUnique_AndIsAppliedToTheConnectClientId()
|
|
{
|
|
// A broker disconnects an existing client when a new one CONNECTs with the same id —
|
|
// browsing must never knock the running driver offline.
|
|
var a = MqttDriverBrowser.BuildBrowseClientIdSuffix();
|
|
var b = MqttDriverBrowser.BuildBrowseClientIdSuffix();
|
|
a.ShouldNotBe(b);
|
|
a.ShouldStartWith("-browse-");
|
|
a.Length.ShouldBe("-browse-".Length + 8);
|
|
|
|
var opts = new MqttDriverOptions { ClientId = "otopcua-node1" };
|
|
MqttClientOptionsFactory.BuildClientOptions(opts, a).ClientId.ShouldBe("otopcua-node1" + a);
|
|
}
|
|
|
|
[Fact]
|
|
public void BrowseConnect_ForcesACleanSession()
|
|
{
|
|
// A transient browse identity must not leave queued-message state behind on the broker.
|
|
var opts = new MqttDriverOptions { CleanSession = false };
|
|
MqttDriverBrowser.ToBrowseOptions(opts).CleanSession.ShouldBeTrue();
|
|
}
|
|
|
|
[Theory]
|
|
[InlineData(1, 5)] // clamped up
|
|
[InlineData(15, 15)]
|
|
[InlineData(600, 30)] // clamped down
|
|
public void OpenBudget_IsClampedTo5To30Seconds(int configured, int expected)
|
|
{
|
|
var opts = new MqttDriverOptions { ConnectTimeoutSeconds = configured };
|
|
MqttDriverBrowser.OpenBudget(opts).ShouldBe(TimeSpan.FromSeconds(expected));
|
|
}
|
|
|
|
// ------------------------------------------------------- Sparkplug: the Last-Will landmine
|
|
|
|
[Fact]
|
|
public void MqttDriverOptions_CarryNothingWillShaped_SoABrowseSessionCanNeverFireAnNdeath()
|
|
{
|
|
// THE landmine (see the marked block on MqttDriverBrowser.ToBrowseOptions). A Sparkplug
|
|
// NDEATH is a Last Will: it is published by the BROKER, not by us, so it is invisible to
|
|
// PublishCountForTest — and any ungraceful end to a browse session would fire an NDEATH
|
|
// under the plant's real edge-node identity. ToBrowseOptions is where a will must be
|
|
// stripped; today there is nothing to strip, and this test is what makes that a checked
|
|
// fact rather than a claim: adding a will-shaped property anywhere in the options graph
|
|
// reddens it and forces the author to handle it in ToBrowseOptions.
|
|
static IEnumerable<string> Names(Type t) => t.GetProperties().Select(p => p.Name);
|
|
|
|
var suspects = Names(typeof(MqttDriverOptions))
|
|
.Concat(Names(typeof(MqttSparkplugOptions)))
|
|
.Concat(Names(typeof(MqttPlainOptions)))
|
|
.Where(n => n.Contains("will", StringComparison.OrdinalIgnoreCase)
|
|
|| n.Contains("testament", StringComparison.OrdinalIgnoreCase)
|
|
|| n.Contains("death", StringComparison.OrdinalIgnoreCase))
|
|
.ToList();
|
|
|
|
suspects.ShouldBeEmpty();
|
|
}
|
|
|
|
// ------------------------------------------------------- Sparkplug: birth-driven tree
|
|
|
|
[Fact]
|
|
public async Task SparkplugTree_FillsFromObservedBirths_BrowseNeverPublishes()
|
|
{
|
|
await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
|
|
s.ObserveBirthForTest("Plant1", "EdgeA", "Filler1", "Temperature", SparkplugDataType.Float);
|
|
|
|
var groups = await s.RootAsync(TestContext.Current.CancellationToken);
|
|
groups.ShouldContain(n => n.DisplayName == "Plant1");
|
|
|
|
var nodes = await s.ExpandAsync("Plant1", TestContext.Current.CancellationToken);
|
|
nodes.ShouldContain(n => n.DisplayName == "EdgeA");
|
|
|
|
var devices = await s.ExpandAsync("Plant1/EdgeA", TestContext.Current.CancellationToken);
|
|
var device = devices.ShouldHaveSingleItem();
|
|
device.DisplayName.ShouldBe("Filler1");
|
|
device.Kind.ShouldBe(BrowseNodeKind.Folder);
|
|
|
|
var metrics = await s.ExpandAsync("Plant1/EdgeA/Filler1", TestContext.Current.CancellationToken);
|
|
var metric = metrics.ShouldHaveSingleItem();
|
|
metric.DisplayName.ShouldBe("Temperature");
|
|
metric.Kind.ShouldBe(BrowseNodeKind.Leaf);
|
|
metric.NodeId.ShouldBe("Plant1/EdgeA/Filler1::Temperature");
|
|
|
|
await s.AttributesAsync(metric.NodeId, TestContext.Current.CancellationToken);
|
|
|
|
s.PublishCountForTest.ShouldBe(0); // passive — browsing a live plant broker publishes nothing
|
|
}
|
|
|
|
[Fact]
|
|
public async Task SparkplugTree_TemplateMetric_OffersItsMembersAndNotItself()
|
|
{
|
|
// Gitea #515. The parent has no value and no servable datatype, so a leaf for `Motor` would
|
|
// offer the operator an address that can only ever commit a tag which never reads — and the
|
|
// picker gives no hint that one leaf is real and the other is not.
|
|
await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
|
|
s.ObserveTemplateBirthForTest(
|
|
"Plant1", "EdgeA", "Filler1", "Motor",
|
|
("Speed", SparkplugDataType.Double),
|
|
("Torque", SparkplugDataType.Double));
|
|
|
|
var metrics = await s.ExpandAsync("Plant1/EdgeA/Filler1", TestContext.Current.CancellationToken);
|
|
|
|
metrics.Select(n => n.DisplayName).ShouldBe(["Motor/Speed", "Motor/Torque"]);
|
|
metrics.ShouldNotContain(n => n.DisplayName == "Motor");
|
|
metrics[0].Kind.ShouldBe(BrowseNodeKind.Leaf);
|
|
|
|
// The metric name travels INTACT as the node id — it is the binding key. Only the derived tag
|
|
// Name is later coerced to a legal RawPath segment, by RawBrowseCommitMapper (Gitea #509).
|
|
metrics[0].NodeId.ShouldBe("Plant1/EdgeA/Filler1::Motor/Speed");
|
|
|
|
s.PublishCountForTest.ShouldBe(0);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task SparkplugTree_NodeLevelMetrics_HangDirectlyUnderTheEdgeNode()
|
|
{
|
|
// An NBIRTH metric has no device. It is NOT parked under a synthesised device folder —
|
|
// a fake segment would be indistinguishable from a real device of the same name, and the
|
|
// authored address tuple carries deviceId = null for exactly these.
|
|
await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
|
|
s.ObserveBirthForTest("Plant1", "EdgeA", device: null, "Node Control/Rebirth", SparkplugDataType.Boolean);
|
|
s.ObserveBirthForTest("Plant1", "EdgeA", "Filler1", "Temperature", SparkplugDataType.Float);
|
|
|
|
var underEdge = await s.ExpandAsync("Plant1/EdgeA", TestContext.Current.CancellationToken);
|
|
underEdge.Select(n => n.DisplayName).ShouldBe(["Filler1", "Node Control/Rebirth"]);
|
|
|
|
var device = underEdge.Single(n => n.DisplayName == "Filler1");
|
|
device.Kind.ShouldBe(BrowseNodeKind.Folder);
|
|
|
|
// A metric name may itself contain '/'; it stays ONE leaf (the metric name is the tag's
|
|
// stable binding key, so a split would invite committing a partial name).
|
|
var nodeMetric = underEdge.Single(n => n.DisplayName == "Node Control/Rebirth");
|
|
nodeMetric.Kind.ShouldBe(BrowseNodeKind.Leaf);
|
|
nodeMetric.NodeId.ShouldBe("Plant1/EdgeA::Node Control/Rebirth");
|
|
(await s.ExpandAsync(nodeMetric.NodeId, TestContext.Current.CancellationToken)).ShouldBeEmpty();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task SparkplugTree_ADeviceAndANodeMetricSharingALabel_AreBothKept()
|
|
{
|
|
// The two are siblings under the edge node and CAN share a label. They are distinct nodes with
|
|
// distinct ids, so neither may displace the other — a tree keyed by label would silently drop
|
|
// whichever arrived second, and the picker would offer a folder where a metric should be.
|
|
await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
|
|
s.ObserveBirthForTest("Plant1", "EdgeA", "Temp", "Value", SparkplugDataType.Float); // device "Temp"
|
|
s.ObserveBirthForTest("Plant1", "EdgeA", null, "Temp", SparkplugDataType.Float); // metric "Temp"
|
|
|
|
var children = await s.ExpandAsync("Plant1/EdgeA", TestContext.Current.CancellationToken);
|
|
children.Count.ShouldBe(2);
|
|
children.ShouldContain(n => n.NodeId == "Plant1/EdgeA/Temp" && n.Kind == BrowseNodeKind.Folder);
|
|
children.ShouldContain(n => n.NodeId == "Plant1/EdgeA::Temp" && n.Kind == BrowseNodeKind.Leaf);
|
|
|
|
(await s.AttributesAsync("Plant1/EdgeA::Temp", TestContext.Current.CancellationToken))
|
|
.ShouldHaveSingleItem().Name.ShouldBe("Temp");
|
|
}
|
|
|
|
/// <summary>
|
|
/// Siblings that share a label must still have a TOTAL order. The level is projected from a
|
|
/// <c>ConcurrentDictionary</c> keyed by path — precisely because two Sparkplug siblings CAN share
|
|
/// a segment — and its enumeration order is not guaranteed, so ordering by segment alone leaves
|
|
/// the tie broken by whatever the dictionary happens to yield. The two rows are visually
|
|
/// identical, so a reorder between successive expands is a picker that changes what a click means
|
|
/// with nothing on screen to show it.
|
|
/// </summary>
|
|
[Fact]
|
|
public async Task SparkplugTree_SiblingsSharingALabel_AreOrderedDeterministically()
|
|
{
|
|
// MANY colliding pairs, not one. .NET string hashing is randomised per process, so the tie order
|
|
// a segment-only sort falls through to is whatever THIS run's dictionary yields — with a single
|
|
// pair it lands the right way round about half the time and the test would pass vacuously. With
|
|
// 24 pairs, every tie has to agree by accident for that to happen.
|
|
var names = Enumerable.Range(0, 24).Select(i => $"N{i:D2}").ToList();
|
|
|
|
await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
|
|
foreach (var name in names)
|
|
{
|
|
s.ObserveBirthForTest("Plant1", "EdgeA", null, name, SparkplugDataType.Float); // metric
|
|
s.ObserveBirthForTest("Plant1", "EdgeA", name, "Value", SparkplugDataType.Float); // device
|
|
}
|
|
|
|
// Segment first (the names, in ordinal order), then path — and Ordinal puts '/' (0x2F) before
|
|
// ':' (0x3A), so each device folder precedes the same-named node-level metric.
|
|
var expected = names.SelectMany(n => new[] { $"Plant1/EdgeA/{n}", $"Plant1/EdgeA::{n}" }).ToList();
|
|
|
|
var children = await s.ExpandAsync("Plant1/EdgeA", TestContext.Current.CancellationToken);
|
|
children.Select(n => n.NodeId).ShouldBe(expected);
|
|
|
|
// …and it is stable: expanding the same level again yields the same order.
|
|
(await s.ExpandAsync("Plant1/EdgeA", TestContext.Current.CancellationToken))
|
|
.Select(n => n.NodeId)
|
|
.ShouldBe(expected);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task SparkplugTree_EmptyBirth_StillRendersItsDeviceAsAFolder()
|
|
{
|
|
// A DBIRTH carrying no metrics leaves a childless device node; it must not masquerade as a
|
|
// committable metric leaf.
|
|
await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
|
|
s.ObserveEmptyBirthForTest("Plant1", "EdgeA", "Filler1");
|
|
|
|
var device = (await s.ExpandAsync("Plant1/EdgeA", TestContext.Current.CancellationToken))
|
|
.ShouldHaveSingleItem();
|
|
device.DisplayName.ShouldBe("Filler1");
|
|
device.Kind.ShouldBe(BrowseNodeKind.Folder);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task SparkplugObserve_NonBirthMessages_BuildNoTree()
|
|
{
|
|
// Only NBIRTH/DBIRTH are self-describing. A DDATA metric carries an alias and no name at
|
|
// all, so anything built from one would be a tree of unnamed nodes.
|
|
await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
|
|
s.ObserveRawForTest("spBv1.0/Plant1/DDATA/EdgeA/Filler1", [0x08, 0x01]);
|
|
s.ObserveRawForTest("spBv1.0/Plant1/NCMD/EdgeA", [0x08, 0x01]);
|
|
s.ObserveRawForTest("spBv1.0/Plant1/NDEATH/EdgeA", [0x08, 0x01]);
|
|
s.ObserveRawForTest("spBv1.0/STATE/host1", "ONLINE"u8.ToArray());
|
|
s.ObserveRawForTest("some/plain/topic", "23.5"u8.ToArray()); // not Sparkplug at all
|
|
|
|
(await s.RootAsync(TestContext.Current.CancellationToken)).ShouldBeEmpty();
|
|
s.PublishCountForTest.ShouldBe(0);
|
|
}
|
|
|
|
[Fact]
|
|
public void SparkplugObserve_GarbageBirthPayload_IsIgnoredAndNeverThrows()
|
|
{
|
|
// Runs on MQTTnet's dispatcher thread behind an unauthenticated firehose.
|
|
var s = new MqttBrowseSession(MqttMode.SparkplugB);
|
|
Should.NotThrow(() => s.ObserveRawForTest("spBv1.0/Plant1/NBIRTH/EdgeA", [0xFF, 0xFF, 0xFF, 0xFF]));
|
|
Should.NotThrow(() => s.ObserveRawForTest("spBv1.0/Plant1/NBIRTH/EdgeA", []));
|
|
}
|
|
|
|
// ------------------------------------------------------- Sparkplug: AttributesAsync
|
|
|
|
[Fact]
|
|
public async Task SparkplugAttributes_AreBirthDerived_NotPayloadSniffed()
|
|
{
|
|
await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
|
|
s.ObserveBirthForTest("Plant1", "EdgeA", "Filler1", "Temperature", SparkplugDataType.Float);
|
|
|
|
var a = (await s.AttributesAsync(
|
|
"Plant1/EdgeA/Filler1::Temperature", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();
|
|
|
|
a.Name.ShouldBe("Temperature");
|
|
a.DriverDataType.ShouldBe(nameof(Core.Abstractions.DriverDataType.Float32)); // from the birth
|
|
a.IsArray.ShouldBeFalse();
|
|
a.SecurityClass.ShouldContain("DBIRTH");
|
|
a.SecurityClass.ShouldContain("Float");
|
|
// The plain-mode payload-snippet machinery must not run: a Sparkplug body is protobuf, and
|
|
// sniffing it would report "(N bytes, binary)" for every metric in the plant.
|
|
a.SecurityClass.ShouldNotContain("binary");
|
|
a.SecurityClass.ShouldNotContain("bytes");
|
|
}
|
|
|
|
[Fact]
|
|
public async Task SparkplugAttributes_ArrayMetric_ReportsElementTypePlusIsArray()
|
|
{
|
|
await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
|
|
s.ObserveBirthForTest("Plant1", "EdgeA", null, "Profile", SparkplugDataType.Int32Array);
|
|
|
|
var a = (await s.AttributesAsync(
|
|
"Plant1/EdgeA::Profile", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();
|
|
a.DriverDataType.ShouldBe(nameof(Core.Abstractions.DriverDataType.Int32)); // element type
|
|
a.IsArray.ShouldBeTrue();
|
|
}
|
|
|
|
[Fact]
|
|
public async Task SparkplugAttributes_UnsupportedDataType_IsReportedNotDefaulted()
|
|
{
|
|
// ToDriverDataType() returns null for DataSet/Template/PropertySet/PropertySetList/Unknown.
|
|
// A null must never be coerced into a guessed DriverDataType — the operator has to see that
|
|
// the metric is not bindable.
|
|
await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
|
|
s.ObserveBirthForTest("Plant1", "EdgeA", null, "Curve", SparkplugDataType.DataSet);
|
|
|
|
var a = (await s.AttributesAsync(
|
|
"Plant1/EdgeA::Curve", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();
|
|
|
|
Enum.TryParse<Core.Abstractions.DriverDataType>(a.DriverDataType, ignoreCase: true, out _)
|
|
.ShouldBeFalse(); // not a guessed type — unmappable, and visibly so
|
|
a.SecurityClass.ShouldContain("unsupported");
|
|
}
|
|
|
|
[Fact]
|
|
public async Task SparkplugAttributes_StateTheBindingTuple_SoTheCommitNeverParsesTheNodeId()
|
|
{
|
|
// The address of a Sparkplug tag is (group, edgeNode, device?, metric) — NOT the node id. The
|
|
// session already holds the decomposition (it parsed the birth topic), so it states it; the
|
|
// AdminUI's browse-commit mapper reads these keys and never splits the id.
|
|
await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
|
|
s.ObserveBirthForTest("OtOpcUaSim", "EdgeA", "Filler1", "Temperature", SparkplugDataType.Float);
|
|
|
|
var a = (await s.AttributesAsync(
|
|
"OtOpcUaSim/EdgeA/Filler1::Temperature", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();
|
|
|
|
a.AddressFields.ShouldNotBeNull();
|
|
a.AddressFields![MqttTagConfigKeys.GroupId].ShouldBe("OtOpcUaSim");
|
|
a.AddressFields[MqttTagConfigKeys.EdgeNodeId].ShouldBe("EdgeA");
|
|
a.AddressFields[MqttTagConfigKeys.DeviceId].ShouldBe("Filler1");
|
|
a.AddressFields[MqttTagConfigKeys.MetricName].ShouldBe("Temperature");
|
|
}
|
|
|
|
[Fact]
|
|
public async Task SparkplugAttributes_NodeLevelMetric_StatesNoDeviceId()
|
|
{
|
|
// An NBIRTH metric belongs to the edge node itself. An invented/blank device id would be a
|
|
// device that does not exist — the same reason the tree hangs it directly under the edge node.
|
|
await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
|
|
s.ObserveBirthForTest("OtOpcUaSim", "EdgeA", null, "Uptime", SparkplugDataType.Int64);
|
|
|
|
var a = (await s.AttributesAsync(
|
|
"OtOpcUaSim/EdgeA::Uptime", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();
|
|
|
|
a.AddressFields.ShouldNotBeNull();
|
|
a.AddressFields!.ContainsKey(MqttTagConfigKeys.DeviceId).ShouldBeFalse();
|
|
a.AddressFields[MqttTagConfigKeys.EdgeNodeId].ShouldBe("EdgeA");
|
|
}
|
|
|
|
[Fact]
|
|
public async Task SparkplugAttributes_MetricNameContainingASlash_IsStatedWhole()
|
|
{
|
|
// 'Node Control/Rebirth' is a canonical Sparkplug metric name. Its node id is
|
|
// 'OtOpcUaSim/EdgeA::Node Control/Rebirth' — un-splittable back into the tuple, which is why the
|
|
// tuple travels rather than the id.
|
|
await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
|
|
s.ObserveBirthForTest("OtOpcUaSim", "EdgeA", null, "Node Control/Rebirth", SparkplugDataType.Boolean);
|
|
|
|
var a = (await s.AttributesAsync(
|
|
"OtOpcUaSim/EdgeA::Node Control/Rebirth", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();
|
|
|
|
a.AddressFields.ShouldNotBeNull();
|
|
a.AddressFields![MqttTagConfigKeys.MetricName].ShouldBe("Node Control/Rebirth");
|
|
a.AddressFields[MqttTagConfigKeys.EdgeNodeId].ShouldBe("EdgeA");
|
|
}
|
|
|
|
[Fact]
|
|
public async Task SparkplugAttributes_DeviceAndNodeLevelMetricsOfTheSameName_StateDifferentAddresses()
|
|
{
|
|
// The two are siblings sharing a label ('EdgeA/Temp' the device folder vs 'EdgeA::Temp' the
|
|
// node-level metric). Anything that re-derived the address from the label would collapse them.
|
|
await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
|
|
s.ObserveBirthForTest("OtOpcUaSim", "EdgeA", null, "Temp", SparkplugDataType.Float);
|
|
s.ObserveBirthForTest("OtOpcUaSim", "EdgeA", "Filler1", "Temp", SparkplugDataType.Float);
|
|
|
|
var nodeLevel = (await s.AttributesAsync(
|
|
"OtOpcUaSim/EdgeA::Temp", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();
|
|
var deviceLevel = (await s.AttributesAsync(
|
|
"OtOpcUaSim/EdgeA/Filler1::Temp", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();
|
|
|
|
nodeLevel.AddressFields!.ContainsKey(MqttTagConfigKeys.DeviceId).ShouldBeFalse();
|
|
deviceLevel.AddressFields![MqttTagConfigKeys.DeviceId].ShouldBe("Filler1");
|
|
}
|
|
|
|
[Fact]
|
|
public async Task PlainAttributes_StateTheTopic_AsTheAddress()
|
|
{
|
|
await using var s = new MqttBrowseSession(MqttMode.Plain);
|
|
s.ObserveTopicForTest("otopcua/fixture/oven/temp", "21.5");
|
|
|
|
var a = (await s.AttributesAsync(
|
|
"otopcua/fixture/oven/temp", TestContext.Current.CancellationToken)).ShouldHaveSingleItem();
|
|
|
|
a.AddressFields.ShouldNotBeNull();
|
|
a.AddressFields![MqttTagConfigKeys.Topic].ShouldBe("otopcua/fixture/oven/temp");
|
|
a.AddressFields.ContainsKey(MqttTagConfigKeys.MetricName).ShouldBeFalse(); // Plain has no metric
|
|
}
|
|
|
|
[Theory]
|
|
[InlineData(MqttMode.SparkplugB, true)]
|
|
[InlineData(MqttMode.Plain, false)]
|
|
public void RebirthAvailable_IsSparkplugOnly(MqttMode mode, bool expected)
|
|
{
|
|
// The picker draws its Request-rebirth affordance off this. A plain window has no re-announce
|
|
// action at all — RequestRebirthAsync refuses one — so offering the button would be a lie.
|
|
new MqttBrowseSession(mode).RebirthAvailable.ShouldBe(expected);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task SparkplugAttributes_ForAFolderNode_AreEmpty()
|
|
{
|
|
await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
|
|
s.ObserveBirthForTest("Plant1", "EdgeA", "Filler1", "Temperature", SparkplugDataType.Float);
|
|
|
|
(await s.AttributesAsync("Plant1", TestContext.Current.CancellationToken)).ShouldBeEmpty();
|
|
(await s.AttributesAsync("Plant1/EdgeA", TestContext.Current.CancellationToken)).ShouldBeEmpty();
|
|
(await s.AttributesAsync("Plant1/EdgeA/Filler1", TestContext.Current.CancellationToken)).ShouldBeEmpty();
|
|
(await s.AttributesAsync("no/such/node", TestContext.Current.CancellationToken)).ShouldBeEmpty();
|
|
}
|
|
|
|
// ------------------------------------------------------- Sparkplug: the one sanctioned publish
|
|
|
|
[Fact]
|
|
public async Task RequestRebirthAsync_ScopedToNode_PublishesOneNcmd()
|
|
{
|
|
await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
|
|
s.ObserveBirthForTest("Plant1", "EdgeA", "Filler1", "T", SparkplugDataType.Float);
|
|
|
|
var published = await s.RequestRebirthAsync("Plant1/EdgeA", TestContext.Current.CancellationToken);
|
|
|
|
published.ShouldBe(1);
|
|
s.PublishCountForTest.ShouldBe(1);
|
|
s.PublishedTopicsForTest.ShouldBe(["spBv1.0/Plant1/NCMD/EdgeA"]);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RequestRebirthAsync_ForANeverSeenEdgeNode_StillPublishes()
|
|
{
|
|
// A node that has never birthed is the PRIME rebirth target — refusing to address one
|
|
// because the observation window never saw it would defeat the whole feature.
|
|
await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
|
|
|
|
(await s.RequestRebirthAsync("Plant1/EdgeGhost", TestContext.Current.CancellationToken)).ShouldBe(1);
|
|
s.PublishedTopicsForTest.ShouldBe(["spBv1.0/Plant1/NCMD/EdgeGhost"]);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RequestRebirthAsync_FromADeviceOrMetricNodeId_ResolvesUpToItsEdgeNode()
|
|
{
|
|
// NCMD is node-addressed; there is no device-scoped rebirth. Selecting a metric and
|
|
// clicking Request-rebirth must address the metric's OWNING edge node — resolved from the
|
|
// tuple stored on the node, never re-parsed out of the id string.
|
|
await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
|
|
s.ObserveBirthForTest("Plant1", "EdgeA", "Filler1", "T", SparkplugDataType.Float);
|
|
|
|
(await s.RequestRebirthAsync("Plant1/EdgeA/Filler1", TestContext.Current.CancellationToken)).ShouldBe(1);
|
|
(await s.RequestRebirthAsync("Plant1/EdgeA/Filler1::T", TestContext.Current.CancellationToken)).ShouldBe(1);
|
|
|
|
s.PublishedTopicsForTest.ShouldBe(["spBv1.0/Plant1/NCMD/EdgeA", "spBv1.0/Plant1/NCMD/EdgeA"]);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RequestRebirthAsync_GroupScope_PublishesExactlyOneNcmdPerObservedEdgeNode()
|
|
{
|
|
await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
|
|
s.ObserveBirthForTest("Plant1", "EdgeB", null, "T", SparkplugDataType.Float);
|
|
s.ObserveBirthForTest("Plant1", "EdgeA", "Filler1", "T", SparkplugDataType.Float);
|
|
s.ObserveBirthForTest("Plant1", "EdgeA", "Filler2", "T", SparkplugDataType.Float); // same node
|
|
s.ObserveBirthForTest("Plant2", "EdgeZ", null, "T", SparkplugDataType.Float); // other group
|
|
|
|
var published = await s.RequestRebirthAsync("Plant1", TestContext.Current.CancellationToken);
|
|
|
|
published.ShouldBe(2);
|
|
s.PublishCountForTest.ShouldBe(2);
|
|
s.PublishedTopicsForTest.ShouldBe(["spBv1.0/Plant1/NCMD/EdgeA", "spBv1.0/Plant1/NCMD/EdgeB"]);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RequestRebirthAsync_GroupScope_BeyondTheCap_PublishesNothingAtAll()
|
|
{
|
|
// A 200-node group must not emit 200 NCMDs. The refusal is all-or-nothing: a partial fan-out
|
|
// would leave the operator unable to say which half of the plant was asked to rebirth.
|
|
await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
|
|
for (var i = 0; i <= MqttBrowseSession.MaxGroupRebirthNodes; i++)
|
|
s.ObserveBirthForTest("Plant1", $"Edge{i:D3}", null, "T", SparkplugDataType.Float);
|
|
|
|
var ex = await Should.ThrowAsync<InvalidOperationException>(
|
|
async () => await s.RequestRebirthAsync("Plant1", TestContext.Current.CancellationToken));
|
|
|
|
ex.Message.ShouldContain(MqttBrowseSession.MaxGroupRebirthNodes.ToString());
|
|
s.PublishCountForTest.ShouldBe(0);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RequestRebirthAsync_GroupScope_TruncatedBeforeAnyEdgeNode_PublishesNothing()
|
|
{
|
|
// The node cap can leave a group node with no edge node recorded beneath it. A group that
|
|
// resolves to zero targets is refused rather than quietly succeeding with a count of 0 — the
|
|
// operator clicked Request-rebirth and is entitled to know nothing was sent.
|
|
await using var s = new MqttBrowseSession(MqttMode.SparkplugB, nodeCap: 1);
|
|
s.ObserveBirthForTest("Plant1", "EdgeA", null, "T", SparkplugDataType.Float);
|
|
|
|
s.Truncated.ShouldBeTrue();
|
|
(await s.ExpandAsync("Plant1", TestContext.Current.CancellationToken)).ShouldBeEmpty();
|
|
|
|
await Should.ThrowAsync<InvalidOperationException>(
|
|
async () => await s.RequestRebirthAsync("Plant1", TestContext.Current.CancellationToken));
|
|
s.PublishCountForTest.ShouldBe(0);
|
|
}
|
|
|
|
[Theory]
|
|
[InlineData("")]
|
|
[InlineData(" ")]
|
|
[InlineData("Plant1")] // a group this window has never observed
|
|
[InlineData("Plant1/EdgeA/Filler1/Extra")] // unknown, and not resolvable to an edge node
|
|
public async Task RequestRebirthAsync_UnusableScope_PublishesNothing(string scope)
|
|
{
|
|
await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
|
|
|
|
await Should.ThrowAsync<ArgumentException>(
|
|
async () => await s.RequestRebirthAsync(scope, TestContext.Current.CancellationToken));
|
|
s.PublishCountForTest.ShouldBe(0);
|
|
}
|
|
|
|
[Theory]
|
|
[InlineData("Plant1/#")]
|
|
[InlineData("Plant1/+")]
|
|
[InlineData("+/EdgeA")]
|
|
[InlineData("#/EdgeA")]
|
|
public async Task RequestRebirthAsync_WildcardScope_PublishesNothing(string scope)
|
|
{
|
|
// SparkplugRebirthScope (the AdminUI's typed input) rejects a wildcard and its own remarks say
|
|
// it is "advisory, not the gate" because ResolveRebirthTargets "refuses an unusable one BEFORE
|
|
// any message leaves". That claim has to be true here: an unobserved `{group}/{edgeNode}` scope
|
|
// is honoured on purpose (a node that has not birthed is the prime rebirth target), and the
|
|
// label check behind it looked only at length and control characters. A wildcard segment would
|
|
// therefore have published `spBv1.0/Plant1/NCMD/#` — a PUBLISH the MQTT spec forbids, which a
|
|
// conformant broker answers by closing the browse connection.
|
|
await using var s = new MqttBrowseSession(MqttMode.SparkplugB);
|
|
|
|
await Should.ThrowAsync<ArgumentException>(
|
|
async () => await s.RequestRebirthAsync(scope, TestContext.Current.CancellationToken));
|
|
s.PublishCountForTest.ShouldBe(0);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RequestRebirthAsync_InPlainMode_IsRefused()
|
|
{
|
|
// P1's guarantee is unconditional: a plain-topic browse window publishes nothing, ever.
|
|
await using var s = new MqttBrowseSession(MqttMode.Plain);
|
|
s.ObserveTopicForTest("factory/oven/temp", "1");
|
|
|
|
await Should.ThrowAsync<NotSupportedException>(
|
|
async () => await s.RequestRebirthAsync("Plant1/EdgeA", TestContext.Current.CancellationToken));
|
|
s.PublishCountForTest.ShouldBe(0);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RequestRebirthAsync_AfterDispose_Throws()
|
|
{
|
|
var s = new MqttBrowseSession(MqttMode.SparkplugB);
|
|
await s.DisposeAsync();
|
|
|
|
await Should.ThrowAsync<ObjectDisposedException>(
|
|
async () => await s.RequestRebirthAsync("Plant1/EdgeA", TestContext.Current.CancellationToken));
|
|
s.PublishCountForTest.ShouldBe(0);
|
|
}
|
|
|
|
// ------------------------------------------------------- Sparkplug: the browser
|
|
|
|
[Fact]
|
|
public async Task OpenAsync_SparkplugMode_NoLongerRefuses_ButStillReachesTheBroker()
|
|
{
|
|
// Task 23 unseals the Sparkplug branch. The open still has to CONNECT, so an unreachable
|
|
// broker fails the way any other open does — what must NOT come back is the Task-10
|
|
// NotSupportedException seal.
|
|
var browser = new MqttDriverBrowser();
|
|
const string cfg = """
|
|
{"host":"broker.invalid","port":8883,"mode":"SparkplugB","connectTimeoutSeconds":5,
|
|
"sparkplug":{"groupId":"Plant1"}}
|
|
""";
|
|
|
|
var ex = await Should.ThrowAsync<Exception>(
|
|
async () => await browser.OpenAsync(cfg, TestContext.Current.CancellationToken));
|
|
ex.ShouldNotBeOfType<NotSupportedException>();
|
|
}
|
|
|
|
[Theory]
|
|
[InlineData("Plant1", "spBv1.0/Plant1/#")]
|
|
[InlineData("", "spBv1.0/#")] // group not authored yet — discover every group
|
|
[InlineData(" ", "spBv1.0/#")]
|
|
public void BuildRootFilter_SparkplugMode_ScopesToTheSparkplugNamespace(string groupId, string expected)
|
|
{
|
|
var opts = new MqttDriverOptions
|
|
{
|
|
Mode = MqttMode.SparkplugB,
|
|
Sparkplug = new MqttSparkplugOptions { GroupId = groupId },
|
|
Plain = new MqttPlainOptions { TopicPrefix = "ignored/in/sparkplug/mode" },
|
|
};
|
|
MqttDriverBrowser.BuildRootFilter(opts).ShouldBe(expected);
|
|
}
|
|
|
|
[Fact]
|
|
public void BuildRootFilter_SparkplugMode_WithNoSparkplugSection_DiscoversEveryGroup()
|
|
{
|
|
var opts = new MqttDriverOptions { Mode = MqttMode.SparkplugB, Sparkplug = null };
|
|
MqttDriverBrowser.BuildRootFilter(opts).ShouldBe("spBv1.0/#");
|
|
}
|
|
}
|