dohertj2/lmxopcua
1
0
Fork 0
Code Issues 17 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(mqtt): bespoke passive #-observation browser (plain)

Browse Source 
MQTT has no discovery protocol, so the AdminUI address picker learns the topic
space the only way there is: subscribe to the wildcard and accumulate what
arrives. MqttBrowseSession serves that observation as a segment tree (split on
'/'); MqttDriverBrowser opens it with a distinct "{clientId}-browse-{guid8}"
identity under a 5-30 s clamped budget.

Browsing publishes NOTHING - the load-bearing safety property, since the picker
runs against a live plant broker. Every outbound message must route through the
single PublishAsync seam that counts into PublishCountForTest; P2's operator-
triggered Sparkplug rebirth is the one intended exception. Sparkplug mode fails
fast rather than serving a raw spBv1.0 topic tree that looks like a metric tree.

Deviation from the plan's ref list: the project also references .Driver so the
browse CONNECT reuses MqttConnection.BuildClientOptions (TLS posture, CA-pin
chain validator, credentials). A second copy would be a security divergence.

Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
This commit is contained in:
Joseph Doherty
2026-07-24 15:45:32 -04:00
parent a6370a26f8
commit 7980b34692
6 changed files with 896 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttBrowseSessionTests.cs
+321
View File
@@ -0,0 +1,321 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Commons.Browsing;
using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser;
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);
}
// ---------------------------------------------------------------- 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");
}
[Fact]
public async Task OpenAsync_SparkplugMode_FailsFastWithoutConnecting()
{
var browser = new MqttDriverBrowser();
const string cfg = """{"host":"broker.invalid","port":8883,"mode":"SparkplugB"}""";
await Should.ThrowAsync<NotSupportedException>(
async () => await browser.OpenAsync(cfg, TestContext.Current.CancellationToken));
}
[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" };
MqttConnection.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));
}
}
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests.csproj
+1
View File
@@ -22,6 +22,7 @@
<ItemGroup>
<ProjectReference Include="..\..\..\src\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts.csproj"/>
<ProjectReference Include="..\..\..\src\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Mqtt\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.csproj"/>
<ProjectReference Include="..\..\..\src\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser.csproj"/>
</ItemGroup>
</Project>