fix(alarms,acl): one condition event per Part 9 method call; gate Server-object model-change/audit events (#587, #588)
#587 — a client Acknowledge/Confirm/AddComment/Shelve/Enable/Disable emitted no condition event: the SDK's ReportStateChange is gated on AutoReportStateChanges (never set), and the engine re-projection found the node already in the new state and was suppressed by the delta-gate. WireMethodTransitionReporting now wraps each method's SDK handler and, on Good, reports exactly once via ReportConditionEvent, under Lock so the routed re-projection cannot race in ahead of the SDK applying the state. Independent of the SDK's imprecise AreEventsMonitored bookkeeping. The shelve methods' state is applied by the wrapper (the SDK never applies it). Native conditions' Confirmed/Shelving are method-owned (NativeAlarmProjector's constants would re-report every ack and undo every shelve). MaxTimeShelved set to int.MaxValue ms at materialise — Create() left it 0 and every TimedShelve failed BadShelvingTimeOutOfRange. Wrong "SDK auto-fires (E2)" comments fixed. #588 — with ACL on, GeneralModelChange and AuditWriteUpdate events no longer go to Server.ReportEvent (ungated to every i=2253 item): ReportToServerObject delivers them per item, a model change narrowed to the Affected nodes the session may Browse (dropped when none), an audit write only with Read on its source plus the SDK audit/encryption guard. Flag off: unchanged. #586 review: materialise raises only the new condition's monitored unit (was a full O(N) reconcile per condition ⇒ O(N²) deploys under Lock). Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
+5
-4
@@ -650,11 +650,12 @@ public sealed class AclEventSubscriptionEnforcementTests
|
||||
/// including a condition rebuilt by a redeploy while the item was live.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <c>ConditionState.EventsMonitored()</c> gates whether the SDK reports a condition's method-driven
|
||||
/// state changes at all. The SDK base raised it by attaching the item to every root notifier; #586
|
||||
/// <c>ConditionState.EventsMonitored()</c> gates whether the SDK reports a condition's method audit
|
||||
/// events. The SDK base raised it by attaching the item to every root notifier; #586
|
||||
/// attaches to nothing, so the first cut left it false on a server whose only subscribers are
|
||||
/// Server-object items — the machine-consumer default. Asserted server-side because the consequence on
|
||||
/// the wire depends on a separate, still-open reporting gap.
|
||||
/// Server-object items — the machine-consumer default. Asserted server-side: method-driven state-change
|
||||
/// events no longer depend on it (Gitea #587, <c>ConditionMethodEventTests</c>), and audit events only
|
||||
/// leave the server while it is auditing.
|
||||
/// </remarks>
|
||||
[Fact]
|
||||
public async Task A_server_object_item_alone_marks_conditions_monitored_until_it_is_deleted()
|
||||
|
||||
+201
@@ -0,0 +1,201 @@
|
||||
using Opc.Ua;
|
||||
using Opc.Ua.Client;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using static ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.ServerEventRig;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #588 over the wire — with <c>Security:Acl:Enabled</c> on, the server-wide events this node manager
|
||||
/// raises (<c>GeneralModelChangeEvent</c>s, <c>AuditWriteUpdateEvent</c>s) reach a Server-object
|
||||
/// (<c>i=2253</c>) item only when its session may see the node the event names.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// They used to go straight to <c>Server.ReportEvent</c> and the core node manager's Server object, which
|
||||
/// queues to every all-events item with no ACL decision — so a session with no grant learned, from a
|
||||
/// model-change event, the RawPath of every node added, deleted or retyped under it.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// The withheld session is granted <see cref="ServerEventRig.EqA"/> only, so the absence of the raw
|
||||
/// event on its item is paired with a positive control <b>on the same item</b>: the sentinel event about
|
||||
/// <see cref="ServerEventRig.EqA"/> is reported after the withheld one, through the same path, and must
|
||||
/// arrive.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class AclServerObjectModelChangeTests
|
||||
{
|
||||
private const string RawVariable = "Plant/Modbus/dev1/Speed";
|
||||
|
||||
private const NodePermissions See = NodePermissions.Browse | NodePermissions.Read | NodePermissions.Subscribe;
|
||||
|
||||
/// <summary>
|
||||
/// A NodeAdded / NodeDeleted / DataTypeChanged model-change event naming a raw node reaches the
|
||||
/// cluster-granted session and not the session granted only an equipment folder — which still receives
|
||||
/// the one naming its own folder.
|
||||
/// </summary>
|
||||
[Theory]
|
||||
[InlineData("NodeAdded")]
|
||||
[InlineData("NodeDeleted")]
|
||||
[InlineData("DataTypeChanged")]
|
||||
public async Task A_model_change_naming_a_node_the_session_cannot_browse_is_withheld(string verb)
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
var rig = await StartAsync($"AclModelChange{verb}", encrypted: false, ct);
|
||||
try
|
||||
{
|
||||
rig.Enforce(true, [Key()], ClusterGrant(KeyPrincipal, "acl-key", See), EquipmentGrant(EqA, See));
|
||||
rig.EnsureEquipment();
|
||||
rig.Sink.EnsureFolder(RawDeviceFolder, parentNodeId: null, displayName: "dev1", realm: AddressSpaceRealm.Raw);
|
||||
rig.Sink.EnsureVariable(RawVariable, RawDeviceFolder, "Speed", "Int32", writable: false, AddressSpaceRealm.Raw);
|
||||
|
||||
var (granted, grantedEvents, grantedItem, grantedSubscription) = await ServerObjectAsync(rig, anonymous: false, ct);
|
||||
var (narrow, narrowEvents, narrowItem, narrowSubscription) = await ServerObjectAsync(rig, anonymous: true, ct);
|
||||
|
||||
switch (verb)
|
||||
{
|
||||
case "NodeAdded": rig.Sink.RaiseNodesAddedModelChange(RawDeviceFolder, AddressSpaceRealm.Raw); break;
|
||||
case "NodeDeleted": rig.Sink.RemoveVariableNode(RawVariable, AddressSpaceRealm.Raw).ShouldBeTrue(); break;
|
||||
default:
|
||||
rig.Sink.UpdateTagAttributes(RawVariable, false, null, "Double", false, null, AddressSpaceRealm.Raw).ShouldBeTrue();
|
||||
break;
|
||||
}
|
||||
|
||||
rig.Sink.RaiseNodesAddedModelChange(EqA, AddressSpaceRealm.Uns);
|
||||
|
||||
await RecordedEvents.WaitUntilAsync(() =>
|
||||
grantedEvents.Of(grantedItem).Count(e => NamesEqA(e)) == 1
|
||||
&& narrowEvents.Of(narrowItem).Any(NamesEqA));
|
||||
|
||||
var rawNames = verb == "NodeAdded" ? RawDeviceFolder : RawVariable;
|
||||
grantedEvents.Of(grantedItem).Count(e => Names(e, rawNames)).ShouldBe(1, "the granted control receives the raw change");
|
||||
narrowEvents.Of(narrowItem).Where(e => Names(e, rawNames)).ShouldBeEmpty(
|
||||
"a session that cannot browse the node must not learn its name from a model-change event");
|
||||
narrowEvents.Of(narrowItem).Count.ShouldBe(1, "only the change to its own equipment folder");
|
||||
|
||||
await narrowSubscription.DeleteAsync(true, ct);
|
||||
await grantedSubscription.DeleteAsync(true, ct);
|
||||
narrow.Dispose();
|
||||
granted.Dispose();
|
||||
}
|
||||
finally { await rig.DisposeAsync(); }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// With enforcement off nothing changes: a session holding no grant at all still receives the raw
|
||||
/// model-change event through the Server object, exactly once.
|
||||
/// </summary>
|
||||
/// <remarks>Without this the test above also passes on a build that withholds every model change from everyone.</remarks>
|
||||
[Fact]
|
||||
public async Task With_acl_enforcement_off_every_server_object_item_receives_the_model_change()
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
var rig = await StartAsync("AclModelChangeOff", encrypted: false, ct);
|
||||
try
|
||||
{
|
||||
rig.Enforce(false, [Key()], ClusterGrant(KeyPrincipal, "acl-key", See));
|
||||
rig.Sink.EnsureFolder(RawDeviceFolder, parentNodeId: null, displayName: "dev1", realm: AddressSpaceRealm.Raw);
|
||||
|
||||
var (session, events, item, subscription) = await ServerObjectAsync(rig, anonymous: true, ct);
|
||||
|
||||
rig.Sink.RaiseNodesAddedModelChange(RawDeviceFolder, AddressSpaceRealm.Raw);
|
||||
await RecordedEvents.WaitUntilAsync(() => events.Of(item).Any(e => Names(e, RawDeviceFolder)));
|
||||
await Task.Delay(TimeSpan.FromMilliseconds(300), ct);
|
||||
events.Of(item).Count(e => Names(e, RawDeviceFolder)).ShouldBe(1);
|
||||
|
||||
await subscription.DeleteAsync(true, ct);
|
||||
session.Dispose();
|
||||
}
|
||||
finally { await rig.DisposeAsync(); }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// With auditing on over an encrypted channel, a rejected device write's <c>AuditWriteUpdateEvent</c> —
|
||||
/// which carries the written value and the node's prior one — reaches a session granted <c>Read</c> on
|
||||
/// the node and not one granted only an equipment folder.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Auditing is off by default (<c>ServerConfiguration.AuditingEnabled</c>, which this repo never sets),
|
||||
/// and audit events only ever leave over SignAndEncrypt — so both sessions are encrypted here, or the
|
||||
/// withheld one would be denied by the channel rather than by the ACL and the test would prove nothing.
|
||||
/// </remarks>
|
||||
[Fact]
|
||||
public async Task An_audit_write_event_reaches_only_sessions_that_may_read_the_node()
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
var rig = await StartAsync("AclAuditWrite", encrypted: true, ct);
|
||||
try
|
||||
{
|
||||
rig.Enforce(
|
||||
true, [Key()],
|
||||
ClusterGrant(KeyPrincipal, "acl-key", See | NodePermissions.WriteOperate), EquipmentGrant(EqA, See));
|
||||
rig.SetAuditing(true);
|
||||
rig.EnsureEquipment();
|
||||
rig.Sink.EnsureFolder(RawDeviceFolder, parentNodeId: null, displayName: "dev1", realm: AddressSpaceRealm.Raw);
|
||||
rig.Sink.EnsureVariable(RawVariable, RawDeviceFolder, "Speed", "Int32", writable: true, AddressSpaceRealm.Raw);
|
||||
rig.Nm.NodeWriteGateway = new RejectingGateway();
|
||||
|
||||
var (granted, grantedEvents, grantedItem, grantedSubscription) =
|
||||
await ServerObjectAsync(rig, anonymous: false, ct, encrypted: true);
|
||||
var (narrow, narrowEvents, narrowItem, narrowSubscription) =
|
||||
await ServerObjectAsync(rig, anonymous: true, ct, encrypted: true);
|
||||
|
||||
var variable = new NodeId(RawVariable, (ushort)granted.NamespaceUris.GetIndex(V3NodeIds.RawNamespaceUri));
|
||||
var write = await granted.WriteAsync(
|
||||
null,
|
||||
[new WriteValue { NodeId = variable, AttributeId = Attributes.Value, Value = new DataValue(new Variant(42)) }],
|
||||
ct);
|
||||
StatusCode.IsGood(write.Results[0]).ShouldBeTrue();
|
||||
|
||||
await RecordedEvents.WaitUntilAsync(() =>
|
||||
grantedEvents.HasType(grantedItem, ObjectTypeIds.AuditWriteUpdateEventType));
|
||||
|
||||
// Reported after the audit event through the same delivery, so it queues behind it on the item.
|
||||
rig.Sink.RaiseNodesAddedModelChange(EqA, AddressSpaceRealm.Uns);
|
||||
await RecordedEvents.WaitUntilAsync(() => narrowEvents.Of(narrowItem).Any(NamesEqA));
|
||||
|
||||
narrowEvents.HasType(narrowItem, ObjectTypeIds.AuditWriteUpdateEventType).ShouldBeFalse(
|
||||
"a session that may not read the node must not receive its written values through an audit event");
|
||||
|
||||
await narrowSubscription.DeleteAsync(true, ct);
|
||||
await grantedSubscription.DeleteAsync(true, ct);
|
||||
narrow.Dispose();
|
||||
granted.Dispose();
|
||||
}
|
||||
finally { await rig.DisposeAsync(); }
|
||||
}
|
||||
|
||||
// ───────────────────────────── helpers ─────────────────────────────
|
||||
|
||||
private static async Task<(ISession Session, RecordedEvents Events, MonitoredItem Item, Subscription Subscription)>
|
||||
ServerObjectAsync(ServerEventRig rig, bool anonymous, CancellationToken ct, bool encrypted = false)
|
||||
{
|
||||
var session = await rig.ConnectAsync(anonymous, ct, encrypted);
|
||||
var events = new RecordedEvents();
|
||||
var subscription = await RecordedEvents.SubscribeAsync(session, events, ct);
|
||||
var item = RecordedEvents.AddEventItem(subscription, ObjectIds.Server);
|
||||
await subscription.ApplyChangesAsync(ct);
|
||||
item.Status.Error.ShouldBeNull();
|
||||
return (session, events, item, subscription);
|
||||
}
|
||||
|
||||
private static bool NamesEqA(VariantCollection e) => Names(e, EqA);
|
||||
|
||||
/// <summary>Whether a model-change event's <c>Changes</c> name the node whose identifier is <paramref name="id"/>.</summary>
|
||||
private static bool Names(VariantCollection e, string id) =>
|
||||
e[RecordedEvents.ChangesField].Value is ExtensionObject[] changes
|
||||
&& changes.Any(c => c.Body is ModelChangeStructureDataType { Affected: { } affected }
|
||||
&& affected.Identifier as string == id);
|
||||
|
||||
/// <summary>Every device write fails, so the node manager reverts and audits it.</summary>
|
||||
private sealed class RejectingGateway : IOpcUaNodeWriteGateway
|
||||
{
|
||||
/// <inheritdoc />
|
||||
public Task<NodeWriteOutcome> WriteAsync(string nodeId, object? value, AddressSpaceRealm realm, CancellationToken ct) =>
|
||||
Task.FromResult(new NodeWriteOutcome(false, "rejected by the test device"));
|
||||
}
|
||||
}
|
||||
+306
@@ -0,0 +1,306 @@
|
||||
using Opc.Ua;
|
||||
using Opc.Ua.Client;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using static ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests.ServerEventRig;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #587 over the wire — every accepted Part 9 method call (Acknowledge, Confirm, AddComment,
|
||||
/// OneShotShelve, TimedShelve, Unshelve, Enable, Disable) on a native or scripted condition reaches a
|
||||
/// folder event item and a Server-object (<c>i=2253</c>) item as <b>exactly one</b> condition event
|
||||
/// carrying the new state.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// Before #587 a client Acknowledge produced no condition event at all: the SDK's own report is gated
|
||||
/// on <c>AutoReportStateChanges</c> (never set here), and the engine's re-projection found the node
|
||||
/// already acknowledged and suppressed its own. The code comments claiming "the SDK auto-fires its own
|
||||
/// event" were wrong — and a fix that only enabled the SDK's report would still double-report a shelve
|
||||
/// (the SDK reports before any shelving state is applied, then the re-projection reports the change).
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Each case runs the engine's (or, for a native alarm, the driver's) answer to the routed command as
|
||||
/// a re-projection from a pool thread, so "exactly one" covers the duplicate the re-projection could
|
||||
/// add, not only the method path. The count is read after a later <b>sentinel</b> transition arrives on
|
||||
/// the same item, so a late duplicate would already be in the queue ahead of it.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class ConditionMethodEventTests
|
||||
{
|
||||
private const NodePermissions All =
|
||||
NodePermissions.Browse | NodePermissions.Read | NodePermissions.Subscribe | NodePermissions.AlarmRead
|
||||
| NodePermissions.AlarmAcknowledge | NodePermissions.AlarmConfirm | NodePermissions.AlarmShelve
|
||||
| NodePermissions.MethodCall;
|
||||
|
||||
/// <summary>
|
||||
/// One accepted method call ⇒ one condition event, to the folder item and to the Server-object item,
|
||||
/// with the new state — while a session holding no grant receives none of it through the Server object.
|
||||
/// </summary>
|
||||
[Theory]
|
||||
[InlineData("Acknowledge", false, true)]
|
||||
[InlineData("Confirm", false, true)]
|
||||
[InlineData("AddComment", false, true)]
|
||||
[InlineData("OneShotShelve", false, true)]
|
||||
[InlineData("TimedShelve", false, true)]
|
||||
[InlineData("Unshelve", false, true)]
|
||||
[InlineData("Disable", false, true)]
|
||||
[InlineData("Enable", false, true)]
|
||||
[InlineData("Acknowledge", true, true)]
|
||||
[InlineData("Confirm", true, true)]
|
||||
[InlineData("AddComment", true, true)]
|
||||
[InlineData("OneShotShelve", true, true)]
|
||||
[InlineData("TimedShelve", true, true)]
|
||||
[InlineData("Unshelve", true, true)]
|
||||
// Flag off: the server-wide AlarmAck role admits the call; delivery is ungated.
|
||||
[InlineData("Acknowledge", false, false)]
|
||||
[InlineData("Acknowledge", true, false)]
|
||||
[InlineData("OneShotShelve", true, false)]
|
||||
public async Task An_accepted_method_call_reports_exactly_one_condition_event(string method, bool native, bool aclOn)
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
var rig = await StartAsync($"CondMethod{method}{native}{aclOn}", encrypted: false, ct);
|
||||
try
|
||||
{
|
||||
rig.Enforce(aclOn, [Key()], ClusterGrant(KeyPrincipal, "acl-key", All));
|
||||
rig.MaterialiseNative();
|
||||
rig.MaterialiseScripted();
|
||||
|
||||
var current = PreState(method, native);
|
||||
WireSimulatedOwners(rig, native, () => current, s => current = s);
|
||||
|
||||
using var session = await rig.ConnectAsync(anonymous: false, ct);
|
||||
var events = new RecordedEvents();
|
||||
var subscription = await RecordedEvents.SubscribeAsync(session, events, ct);
|
||||
var folderItem = RecordedEvents.AddEventItem(subscription, UnsNodeId(session, EqA));
|
||||
var serverItem = RecordedEvents.AddEventItem(subscription, ObjectIds.Server);
|
||||
await subscription.ApplyChangesAsync(ct);
|
||||
folderItem.Status.Error.ShouldBeNull();
|
||||
|
||||
using var ungranted = await rig.ConnectAsync(anonymous: true, ct);
|
||||
var ungrantedEvents = new RecordedEvents();
|
||||
var ungrantedSubscription = await RecordedEvents.SubscribeAsync(ungranted, ungrantedEvents, ct);
|
||||
var ungrantedItem = RecordedEvents.AddEventItem(ungrantedSubscription, ObjectIds.Server);
|
||||
await ungrantedSubscription.ApplyChangesAsync(ct);
|
||||
|
||||
var condition = ConditionNodeId(session, native);
|
||||
|
||||
// Every transition the test drives also moves the simulated owner's state, so its re-projections
|
||||
// carry the node's current message.
|
||||
void ProjectAs(string message)
|
||||
{
|
||||
current = current with { Message = message };
|
||||
rig.Project(native, current);
|
||||
}
|
||||
|
||||
// A native condition's Confirmed and Shelving are reached only through the methods.
|
||||
ProjectAs("setup");
|
||||
await RecordedEvents.WaitUntilAsync(() => events.Has(folderItem, "setup"));
|
||||
if (native && method == "Confirm")
|
||||
await CallAsync(session, condition, "Acknowledge", events.LatestEventId(folderItem, "setup"), ct);
|
||||
if (native && method == "Unshelve")
|
||||
await CallAsync(session, condition, "OneShotShelve", events.LatestEventId(folderItem, "setup"), ct);
|
||||
await rig.DrainProjectionsAsync();
|
||||
|
||||
ProjectAs("base");
|
||||
await RecordedEvents.WaitUntilAsync(() => events.Has(folderItem, "base") && events.Has(serverItem, "base"));
|
||||
|
||||
await CallAsync(session, condition, method, events.LatestEventId(folderItem, "base"), ct);
|
||||
await rig.DrainProjectionsAsync();
|
||||
|
||||
ProjectAs("sentinel");
|
||||
await RecordedEvents.WaitUntilAsync(() =>
|
||||
events.Has(folderItem, "sentinel") && events.Has(serverItem, "sentinel")
|
||||
&& (aclOn || ungrantedEvents.Has(ungrantedItem, "sentinel")));
|
||||
|
||||
foreach (var item in new[] { folderItem, serverItem })
|
||||
{
|
||||
var baseEvents = events.ConditionEvents(item, "base");
|
||||
baseEvents.Count.ShouldBe(2, $"the base transition plus exactly one {method} event ({item.StartNodeId})");
|
||||
AssertNewState(method, baseEvents[1]);
|
||||
}
|
||||
|
||||
if (aclOn)
|
||||
{
|
||||
ungrantedEvents.Of(ungrantedItem).ShouldBeEmpty(
|
||||
"a session with no grant receives no condition event through the Server object");
|
||||
}
|
||||
else
|
||||
{
|
||||
ungrantedEvents.ConditionEvents(ungrantedItem, "base").Count.ShouldBe(2, "flag off: ungated, still once");
|
||||
}
|
||||
|
||||
await ungrantedSubscription.DeleteAsync(true, ct);
|
||||
await subscription.DeleteAsync(true, ct);
|
||||
}
|
||||
finally { await rig.DisposeAsync(); }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A refused call reports nothing: Disable on a native condition is <c>BadNotSupported</c>, so no
|
||||
/// transition happened and no condition event may claim one.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task A_refused_method_call_reports_no_condition_event()
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
var rig = await StartAsync("CondMethodRefused", encrypted: false, ct);
|
||||
try
|
||||
{
|
||||
rig.Enforce(true, [Key()], ClusterGrant(KeyPrincipal, "acl-key", All));
|
||||
rig.MaterialiseNative();
|
||||
var current = PreState("Disable", native: true);
|
||||
|
||||
using var session = await rig.ConnectAsync(anonymous: false, ct);
|
||||
var events = new RecordedEvents();
|
||||
var subscription = await RecordedEvents.SubscribeAsync(session, events, ct);
|
||||
var folderItem = RecordedEvents.AddEventItem(subscription, UnsNodeId(session, EqA));
|
||||
var serverItem = RecordedEvents.AddEventItem(subscription, ObjectIds.Server);
|
||||
await subscription.ApplyChangesAsync(ct);
|
||||
|
||||
rig.Project(native: true, current = current with { Message = "base" });
|
||||
await RecordedEvents.WaitUntilAsync(() => events.Has(serverItem, "base"));
|
||||
|
||||
var refused = await Should.ThrowAsync<ServiceResultException>(
|
||||
CallAsync(session, ConditionNodeId(session, native: true), "Disable", null, ct));
|
||||
refused.StatusCode.ShouldBe(StatusCodes.BadNotSupported);
|
||||
|
||||
rig.Project(native: true, current with { Message = "sentinel" });
|
||||
await RecordedEvents.WaitUntilAsync(() => events.Has(folderItem, "sentinel") && events.Has(serverItem, "sentinel"));
|
||||
events.ConditionEvents(folderItem, "base").Count.ShouldBe(1);
|
||||
events.ConditionEvents(serverItem, "base").Count.ShouldBe(1);
|
||||
|
||||
await subscription.DeleteAsync(true, ct);
|
||||
}
|
||||
finally { await rig.DisposeAsync(); }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Materialising N conditions while a Server-object item is live costs N monitored-bookkeeping visits,
|
||||
/// not N²/2 — and every one of them still ends up marked monitored (#586 review, performance note).
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task Materialising_conditions_under_a_live_server_object_item_visits_each_once()
|
||||
{
|
||||
const int n = 40;
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
var rig = await StartAsync("CondMonitoredCost", encrypted: false, ct);
|
||||
try
|
||||
{
|
||||
rig.EnsureEquipment();
|
||||
using var session = await rig.ConnectAsync(anonymous: true, ct);
|
||||
var events = new RecordedEvents();
|
||||
var subscription = await RecordedEvents.SubscribeAsync(session, events, ct);
|
||||
RecordedEvents.AddEventItem(subscription, ObjectIds.Server);
|
||||
await subscription.ApplyChangesAsync(ct);
|
||||
|
||||
var before = rig.Nm.AllEventsMonitoringConditionVisits;
|
||||
for (var i = 0; i < n; i++)
|
||||
rig.Sink.MaterialiseAlarmCondition($"bulk-{i}", EqA, $"bulk {i}", "AlarmCondition", 500, AddressSpaceRealm.Uns);
|
||||
|
||||
(rig.Nm.AllEventsMonitoringConditionVisits - before).ShouldBe(n);
|
||||
for (var i = 0; i < n; i++)
|
||||
rig.Nm.ConditionEventsMonitored($"bulk-{i}", AddressSpaceRealm.Uns).ShouldBeTrue();
|
||||
|
||||
await subscription.DeleteAsync(true, ct);
|
||||
for (var i = 0; i < n; i++)
|
||||
rig.Nm.ConditionEventsMonitored($"bulk-{i}", AddressSpaceRealm.Uns).ShouldBeFalse(
|
||||
"the last item's delete still gives every unit back through the full reconcile");
|
||||
}
|
||||
finally { await rig.DisposeAsync(); }
|
||||
}
|
||||
|
||||
// ───────────────────────────── helpers ─────────────────────────────
|
||||
|
||||
/// <summary>The condition state a case starts from — one in which its method is accepted.</summary>
|
||||
private static AlarmConditionSnapshot PreState(string method, bool native)
|
||||
{
|
||||
var s = new AlarmConditionSnapshot(
|
||||
Active: true, Acknowledged: false, Confirmed: native, Enabled: true,
|
||||
Shelving: AlarmShelvingKind.Unshelved, Severity: 700, Message: "pre");
|
||||
if (native) return s; // the driver projection's constants: Confirmed, Unshelved
|
||||
|
||||
return method switch
|
||||
{
|
||||
"Confirm" => s with { Acknowledged = true, Confirmed = false },
|
||||
"Unshelve" => s with { Shelving = AlarmShelvingKind.OneShot },
|
||||
"Enable" => s with { Enabled = false },
|
||||
_ => s,
|
||||
};
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Stands in for the scripted engine and the native driver: each routed command comes back as a
|
||||
/// re-projection of the state it produces, from a pool thread.
|
||||
/// </summary>
|
||||
private static void WireSimulatedOwners(
|
||||
ServerEventRig rig, bool native, Func<AlarmConditionSnapshot> current, Action<AlarmConditionSnapshot> set)
|
||||
{
|
||||
rig.Nm.NativeAlarmAckRouter = _ =>
|
||||
{
|
||||
// NativeAlarmProjector's answer to the driver's ack echo: Acked, with its Confirmed/Unshelved constants.
|
||||
var next = current() with { Acknowledged = true, Confirmed = true, Shelving = AlarmShelvingKind.Unshelved };
|
||||
set(next);
|
||||
rig.ProjectLater(native: true, next);
|
||||
};
|
||||
rig.Nm.AlarmCommandRouter = cmd =>
|
||||
{
|
||||
if (native || cmd.AlarmId != ScriptedAlarmId) return; // the engine owns only scripted alarms
|
||||
var s = current();
|
||||
var next = cmd.Operation switch
|
||||
{
|
||||
"Acknowledge" => s with { Acknowledged = true },
|
||||
"Confirm" => s with { Confirmed = true },
|
||||
"OneShotShelve" => s with { Shelving = AlarmShelvingKind.OneShot },
|
||||
"TimedShelve" => s with { Shelving = AlarmShelvingKind.Timed },
|
||||
"Unshelve" => s with { Shelving = AlarmShelvingKind.Unshelved },
|
||||
"Enable" => s with { Enabled = true },
|
||||
"Disable" => s with { Enabled = false },
|
||||
_ => s, // AddComment: no field of the snapshot changes
|
||||
};
|
||||
set(next);
|
||||
rig.ProjectLater(native: false, next);
|
||||
};
|
||||
}
|
||||
|
||||
private static void AssertNewState(string method, VariantCollection e)
|
||||
{
|
||||
switch (method)
|
||||
{
|
||||
case "Acknowledge": e[RecordedEvents.AckedField].Value.ShouldBe(true); break;
|
||||
case "Confirm": e[RecordedEvents.ConfirmedField].Value.ShouldBe(true); break;
|
||||
case "AddComment": (e[RecordedEvents.CommentField].Value as LocalizedText)?.Text.ShouldBe("note"); break;
|
||||
case "OneShotShelve": e[RecordedEvents.ShelvingField].Value.ShouldBe(ObjectIds.ShelvedStateMachineType_OneShotShelved); break;
|
||||
case "TimedShelve": e[RecordedEvents.ShelvingField].Value.ShouldBe(ObjectIds.ShelvedStateMachineType_TimedShelved); break;
|
||||
case "Unshelve": e[RecordedEvents.ShelvingField].Value.ShouldBe(ObjectIds.ShelvedStateMachineType_Unshelved); break;
|
||||
case "Disable": e[RecordedEvents.EnabledField].Value.ShouldBe(false); break;
|
||||
case "Enable": e[RecordedEvents.EnabledField].Value.ShouldBe(true); break;
|
||||
default: throw new ArgumentOutOfRangeException(nameof(method), method, null);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Calls one Part 9 method on the condition by its type's method id (the SDK resolves it against the
|
||||
/// condition's method declarations — the shelve methods included); throws on a bad result.
|
||||
/// </summary>
|
||||
private static async Task CallAsync(ISession session, NodeId condition, string method, byte[]? eventId, CancellationToken ct)
|
||||
{
|
||||
var (methodId, args) = method switch
|
||||
{
|
||||
"Acknowledge" => (MethodIds.AcknowledgeableConditionType_Acknowledge, new object[] { eventId!, new LocalizedText("ack") }),
|
||||
"Confirm" => (MethodIds.AcknowledgeableConditionType_Confirm, [eventId!, new LocalizedText("confirm")]),
|
||||
"AddComment" => (MethodIds.ConditionType_AddComment, [eventId!, new LocalizedText("note")]),
|
||||
"Enable" => (MethodIds.ConditionType_Enable, []),
|
||||
"Disable" => (MethodIds.ConditionType_Disable, []),
|
||||
"OneShotShelve" => (MethodIds.AlarmConditionType_ShelvingState_OneShotShelve, []),
|
||||
"TimedShelve" => (MethodIds.AlarmConditionType_ShelvingState_TimedShelve, [60_000.0]),
|
||||
"Unshelve" => (MethodIds.AlarmConditionType_ShelvingState_Unshelve, []),
|
||||
_ => throw new ArgumentOutOfRangeException(nameof(method), method, null),
|
||||
};
|
||||
await session.CallAsync(condition, methodId, ct, args);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,373 @@
|
||||
using System.Net;
|
||||
using System.Net.Sockets;
|
||||
using System.Reflection;
|
||||
using System.Text;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Opc.Ua;
|
||||
using Opc.Ua.Client;
|
||||
using Opc.Ua.Configuration;
|
||||
using Shouldly;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
|
||||
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Authorization;
|
||||
using ZB.MOM.WW.OtOpcUa.Host.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.OpcUaServer.Security;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
|
||||
|
||||
/// <summary>
|
||||
/// A booted server for the Server-object event tests (Gitea #587 / #588): one native condition on a raw
|
||||
/// device wired to two equipment notifiers, one scripted condition under <see cref="EqA"/>, an API-key
|
||||
/// principal and anonymous, optionally an encrypted endpoint and server auditing.
|
||||
/// </summary>
|
||||
internal sealed class ServerEventRig : IAsyncDisposable
|
||||
{
|
||||
public const string Cluster = "SITE-A";
|
||||
public const string RawDeviceFolder = "Plant/Modbus/dev1";
|
||||
public const string RawAlarmPath = "Plant/Modbus/dev1/temp_hi";
|
||||
public const string ScriptedAlarmId = "scripted-eqa-hi";
|
||||
public const string EqA = "EqA";
|
||||
public const string EqB = "EqB";
|
||||
public const string KeyPrincipal = "event-probe";
|
||||
public const string KeyId = "k-server-events";
|
||||
|
||||
private readonly string _pkiRoot;
|
||||
private readonly List<Task> _pending = [];
|
||||
private OpcUaApplicationHost _host = null!;
|
||||
private OtOpcUaSdkServer _server = null!;
|
||||
private int _port;
|
||||
|
||||
private ServerEventRig(string pkiRoot) => _pkiRoot = pkiRoot;
|
||||
|
||||
public OtOpcUaNodeManager Nm => _server.NodeManager!;
|
||||
|
||||
public SdkAddressSpaceSink Sink { get; private set; } = null!;
|
||||
|
||||
public static async Task<ServerEventRig> StartAsync(string name, bool encrypted, CancellationToken ct)
|
||||
{
|
||||
var rig = new ServerEventRig(Path.Combine(Path.GetTempPath(), $"otopcua-{name}-{Guid.NewGuid():N}"));
|
||||
rig._port = AllocateFreePort();
|
||||
rig._server = new OtOpcUaSdkServer();
|
||||
rig._host = new OpcUaApplicationHost(
|
||||
new OpcUaApplicationHostOptions
|
||||
{
|
||||
ApplicationName = $"OtOpcUa.{name}",
|
||||
ApplicationUri = $"urn:OtOpcUa.{name}.{Guid.NewGuid():N}",
|
||||
OpcUaPort = rig._port,
|
||||
PublicHostname = "127.0.0.1",
|
||||
PkiStoreRoot = rig._pkiRoot,
|
||||
EnabledSecurityProfiles = encrypted
|
||||
? [OpcUaSecurityProfile.None, OpcUaSecurityProfile.Basic256Sha256SignAndEncrypt]
|
||||
: [OpcUaSecurityProfile.None],
|
||||
AutoAcceptUntrustedClientCertificates = true,
|
||||
},
|
||||
NullLogger<OpcUaApplicationHost>.Instance,
|
||||
new KeyPresentingAuthenticator());
|
||||
|
||||
await rig._host.StartAsync(rig._server, ct);
|
||||
rig.Sink = new SdkAddressSpaceSink(rig.Nm);
|
||||
return rig;
|
||||
}
|
||||
|
||||
/// <summary>Wires the authorizer over the given keys and grants, enforcing or not.</summary>
|
||||
public DataPlaneAuthorityState Enforce(bool enabled, ApiKey[] keys, params NodeAcl[] grants)
|
||||
{
|
||||
var state = new DataPlaneAuthorityState();
|
||||
state.Apply(Cluster, "rev-1", keys, grants, ScopePaths());
|
||||
Nm.DataPlaneAuthority = state;
|
||||
Nm.Authorizer = new DataPlaneAuthorizer(
|
||||
state, new TriePermissionEvaluator(state.Tries), enabled, NullLogger.Instance, TimeProvider.System);
|
||||
return state;
|
||||
}
|
||||
|
||||
public static ApiKey Key() => new()
|
||||
{
|
||||
ApiKeyId = KeyId,
|
||||
Name = KeyId,
|
||||
SecretHash = "hash",
|
||||
Principal = KeyPrincipal,
|
||||
Enabled = true,
|
||||
CreatedUtc = DateTime.UtcNow.AddDays(-1),
|
||||
};
|
||||
|
||||
public static NodeAcl ClusterGrant(string principal, string id, NodePermissions permissions) => new()
|
||||
{
|
||||
NodeAclId = id,
|
||||
ClusterId = Cluster,
|
||||
Principal = principal,
|
||||
ScopeKind = NodeAclScopeKind.Cluster,
|
||||
PermissionFlags = permissions,
|
||||
};
|
||||
|
||||
public static NodeAcl EquipmentGrant(string equipment, NodePermissions permissions) => new()
|
||||
{
|
||||
NodeAclId = $"acl-{equipment}",
|
||||
ClusterId = Cluster,
|
||||
Principal = OpcUaPrincipals.Anonymous,
|
||||
ScopeKind = NodeAclScopeKind.Equipment,
|
||||
ScopeId = $"eq-{equipment}",
|
||||
PermissionFlags = permissions,
|
||||
};
|
||||
|
||||
private static IReadOnlyDictionary<AclScopeRef, IReadOnlyList<NodeAclPath>> ScopePaths() =>
|
||||
new Dictionary<AclScopeRef, IReadOnlyList<NodeAclPath>>
|
||||
{
|
||||
[new AclScopeRef(NodeAclScopeKind.Equipment, $"eq-{EqA}")] = [new NodeAclPath([NodeAclRealms.Uns, EqA])],
|
||||
[new AclScopeRef(NodeAclScopeKind.Equipment, $"eq-{EqB}")] = [new NodeAclPath([NodeAclRealms.Uns, EqB])],
|
||||
};
|
||||
|
||||
/// <summary>The native condition, wired as an event source of both equipment folders.</summary>
|
||||
public void MaterialiseNative()
|
||||
{
|
||||
Sink.EnsureFolder(RawDeviceFolder, parentNodeId: null, displayName: "dev1", realm: AddressSpaceRealm.Raw);
|
||||
Sink.MaterialiseAlarmCondition(
|
||||
RawAlarmPath, RawDeviceFolder, "temp_hi", "OffNormalAlarm", 700, AddressSpaceRealm.Raw, isNative: true);
|
||||
EnsureEquipment();
|
||||
Sink.WireAlarmNotifiers(RawAlarmPath, AddressSpaceRealm.Raw, [EqA, EqB], AddressSpaceRealm.Uns);
|
||||
}
|
||||
|
||||
/// <summary>The scripted condition under <see cref="EqA"/>.</summary>
|
||||
public void MaterialiseScripted()
|
||||
{
|
||||
EnsureEquipment();
|
||||
Sink.MaterialiseAlarmCondition(
|
||||
ScriptedAlarmId, EqA, "hi", "AlarmCondition", 500, AddressSpaceRealm.Uns, isNative: false);
|
||||
}
|
||||
|
||||
public void EnsureEquipment()
|
||||
{
|
||||
Sink.EnsureFolder(EqA, parentNodeId: null, displayName: "A", realm: AddressSpaceRealm.Uns);
|
||||
Sink.EnsureFolder(EqB, parentNodeId: null, displayName: "B", realm: AddressSpaceRealm.Uns);
|
||||
}
|
||||
|
||||
/// <summary>Projects <paramref name="state"/> onto a condition, as the engine or a driver would.</summary>
|
||||
public void Project(bool native, AlarmConditionSnapshot state) =>
|
||||
Sink.WriteAlarmCondition(
|
||||
native ? RawAlarmPath : ScriptedAlarmId,
|
||||
state,
|
||||
DateTime.UtcNow,
|
||||
native ? AddressSpaceRealm.Raw : AddressSpaceRealm.Uns);
|
||||
|
||||
/// <summary>
|
||||
/// Re-projects <paramref name="state"/> from a pool thread, the way the engine's or the driver's answer
|
||||
/// to a routed method call arrives — it contends for the node manager's lock like the real one does.
|
||||
/// </summary>
|
||||
public void ProjectLater(bool native, AlarmConditionSnapshot state)
|
||||
{
|
||||
lock (_pending) _pending.Add(Task.Run(() => Project(native, state)));
|
||||
}
|
||||
|
||||
/// <summary>Waits for every <see cref="ProjectLater"/> re-projection to have run.</summary>
|
||||
public async Task DrainProjectionsAsync()
|
||||
{
|
||||
Task[] pending;
|
||||
lock (_pending) pending = [.. _pending];
|
||||
await Task.WhenAll(pending);
|
||||
}
|
||||
|
||||
/// <summary>The condition's NodeId in the session's namespaces.</summary>
|
||||
public static NodeId ConditionNodeId(ISession session, bool native) => native
|
||||
? new NodeId(RawAlarmPath, (ushort)session.NamespaceUris.GetIndex(V3NodeIds.RawNamespaceUri))
|
||||
: new NodeId(ScriptedAlarmId, (ushort)session.NamespaceUris.GetIndex(V3NodeIds.UnsNamespaceUri));
|
||||
|
||||
public static NodeId UnsNodeId(ISession session, string id)
|
||||
{
|
||||
var ns = (ushort)session.NamespaceUris.GetIndex(V3NodeIds.UnsNamespaceUri);
|
||||
ns.ShouldBeGreaterThan((ushort)0);
|
||||
return new NodeId(id, ns);
|
||||
}
|
||||
|
||||
/// <summary>Switches the server's <c>Auditing</c> flag (no configuration seam reaches it in this repo).</summary>
|
||||
public void SetAuditing(bool on)
|
||||
{
|
||||
var server = typeof(Opc.Ua.Server.CustomNodeManager2)
|
||||
.GetProperty("Server", BindingFlags.Instance | BindingFlags.NonPublic | BindingFlags.Public)!
|
||||
.GetValue(Nm)!;
|
||||
server.GetType()
|
||||
.GetProperty("Auditing", BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic)!
|
||||
.GetSetMethod(nonPublic: true)!
|
||||
.Invoke(server, [on]);
|
||||
}
|
||||
|
||||
public async Task<ISession> ConnectAsync(bool anonymous, CancellationToken ct, bool encrypted = false)
|
||||
{
|
||||
var pki = TestClientSecurity.AllocatePkiRoot();
|
||||
var security = TestClientSecurity.Build(pki);
|
||||
security.ApplicationCertificate.SubjectName = "CN=OtOpcUa.ServerEventClient";
|
||||
var appConfig = new ApplicationConfiguration
|
||||
{
|
||||
ApplicationName = "OtOpcUa.ServerEventClient",
|
||||
ApplicationUri = $"urn:OtOpcUa.ServerEventClient.{Guid.NewGuid():N}",
|
||||
ApplicationType = ApplicationType.Client,
|
||||
SecurityConfiguration = security,
|
||||
ClientConfiguration = new ClientConfiguration { DefaultSessionTimeout = 60_000 },
|
||||
};
|
||||
await appConfig.ValidateAsync(ApplicationType.Client, ct);
|
||||
appConfig.CertificateValidator.CertificateValidation += (_, e) => e.Accept = true;
|
||||
|
||||
if (encrypted)
|
||||
{
|
||||
#pragma warning disable CS0618 // the telemetry-context overload is not needed for a test client
|
||||
var app = new ApplicationInstance { ApplicationName = appConfig.ApplicationName, ApplicationType = ApplicationType.Client };
|
||||
#pragma warning restore CS0618
|
||||
app.ApplicationConfiguration = appConfig;
|
||||
await app.CheckApplicationInstanceCertificatesAsync(silent: true, lifeTimeInMonths: null, ct);
|
||||
}
|
||||
|
||||
var endpoint = await CoreClientUtils.SelectEndpointAsync(
|
||||
appConfig, $"opc.tcp://127.0.0.1:{_port}/OtOpcUa", encrypted, DefaultTelemetry.Create(_ => { }), ct);
|
||||
if (encrypted) endpoint!.SecurityMode.ShouldBe(MessageSecurityMode.SignAndEncrypt);
|
||||
|
||||
var identity = anonymous
|
||||
? new UserIdentity(new AnonymousIdentityToken())
|
||||
: new UserIdentity(new UserNameIdentityToken
|
||||
{
|
||||
UserName = "apikey",
|
||||
DecryptedPassword = Encoding.UTF8.GetBytes($"otua_{KeyId}_secret"),
|
||||
});
|
||||
|
||||
return await new DefaultSessionFactory(DefaultTelemetry.Create(_ => { })).CreateAsync(
|
||||
appConfig,
|
||||
new ConfiguredEndpoint(null, endpoint, EndpointConfiguration.Create(appConfig)),
|
||||
updateBeforeConnect: false,
|
||||
checkDomain: false,
|
||||
sessionName: "ServerEventRig",
|
||||
sessionTimeout: 60_000,
|
||||
identity: identity,
|
||||
preferredLocales: null,
|
||||
ct: ct);
|
||||
}
|
||||
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
await _host.DisposeAsync();
|
||||
if (Directory.Exists(_pkiRoot))
|
||||
{
|
||||
try { Directory.Delete(_pkiRoot, recursive: true); }
|
||||
catch (IOException) { /* best effort */ }
|
||||
}
|
||||
}
|
||||
|
||||
private static int AllocateFreePort()
|
||||
{
|
||||
var listener = new TcpListener(IPAddress.Loopback, 0);
|
||||
listener.Start();
|
||||
var port = ((IPEndPoint)listener.LocalEndpoint).Port;
|
||||
listener.Stop();
|
||||
return port;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Admits <c>apikey</c> as <see cref="KeyPrincipal"/> carrying the two server-wide roles (used only
|
||||
/// while enforcement is off); anonymous falls through to the <c>anonymous</c> principal.
|
||||
/// </summary>
|
||||
private sealed class KeyPresentingAuthenticator : IOpcUaUserAuthenticator
|
||||
{
|
||||
/// <inheritdoc />
|
||||
public Task<OpcUaUserAuthResult> AuthenticateUserNameAsync(string username, string password, CancellationToken ct) =>
|
||||
Task.FromResult(
|
||||
string.Equals(username, "apikey", StringComparison.Ordinal)
|
||||
&& password.StartsWith($"otua_{KeyId}_", StringComparison.Ordinal)
|
||||
? OpcUaUserAuthResult.AllowKey(
|
||||
KeyPrincipal, [OpcUaDataPlaneRoles.AlarmAck, OpcUaDataPlaneRoles.WriteOperate], KeyId)
|
||||
: OpcUaUserAuthResult.Deny("Invalid API key."));
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Records every event each monitored item received, in order, with the fields the tests select.</summary>
|
||||
internal sealed class RecordedEvents
|
||||
{
|
||||
// The select clauses, in order — see AddEventItem.
|
||||
public const int EventIdField = 0;
|
||||
public const int MessageField = 1;
|
||||
public const int EventTypeField = 2;
|
||||
public const int AckedField = 3;
|
||||
public const int ConfirmedField = 4;
|
||||
public const int EnabledField = 5;
|
||||
public const int ShelvingField = 6;
|
||||
public const int CommentField = 7;
|
||||
public const int ChangesField = 8;
|
||||
public const int SourceNodeField = 9;
|
||||
|
||||
private readonly object _gate = new();
|
||||
private readonly List<(uint Handle, VariantCollection Fields)> _events = [];
|
||||
|
||||
public void OnEvents(Subscription subscription, EventNotificationList notification, IList<string> stringTable)
|
||||
{
|
||||
lock (_gate)
|
||||
{
|
||||
foreach (var e in notification.Events) _events.Add((e.ClientHandle, e.EventFields));
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Every event <paramref name="item"/> received, in arrival order.</summary>
|
||||
public IReadOnlyList<VariantCollection> Of(MonitoredItem item)
|
||||
{
|
||||
lock (_gate) return _events.Where(e => e.Handle == item.ClientHandle).Select(e => e.Fields).ToList();
|
||||
}
|
||||
|
||||
/// <summary>The condition events (they carry <c>AckedState/Id</c>; audit and refresh events do not).</summary>
|
||||
public IReadOnlyList<VariantCollection> ConditionEvents(MonitoredItem item, string message) =>
|
||||
Of(item).Where(f => f[AckedField].Value is bool && Message(f) == message).ToList();
|
||||
|
||||
public bool Has(MonitoredItem item, string message) => Of(item).Any(f => Message(f) == message);
|
||||
|
||||
public bool HasType(MonitoredItem item, NodeId type) => Of(item).Any(f => f[EventTypeField].Value as NodeId == type);
|
||||
|
||||
public byte[] LatestEventId(MonitoredItem item, string message) =>
|
||||
(byte[])Of(item).Last(f => Message(f) == message && f[AckedField].Value is bool)[EventIdField].Value;
|
||||
|
||||
public static string? Message(VariantCollection f) => (f[MessageField].Value as LocalizedText)?.Text;
|
||||
|
||||
public static async Task<Subscription> SubscribeAsync(ISession session, RecordedEvents events, CancellationToken ct)
|
||||
{
|
||||
var subscription = new Subscription(session.DefaultSubscription) { PublishingInterval = 100 };
|
||||
subscription.FastEventCallback = events.OnEvents;
|
||||
session.AddSubscription(subscription);
|
||||
await subscription.CreateAsync(ct);
|
||||
return subscription;
|
||||
}
|
||||
|
||||
public static MonitoredItem AddEventItem(Subscription subscription, NodeId source)
|
||||
{
|
||||
var filter = new EventFilter();
|
||||
filter.AddSelectClause(ObjectTypes.BaseEventType, BrowseNames.EventId);
|
||||
filter.AddSelectClause(ObjectTypes.BaseEventType, BrowseNames.Message);
|
||||
filter.AddSelectClause(ObjectTypes.BaseEventType, BrowseNames.EventType);
|
||||
filter.SelectClauses.Add(new SimpleAttributeOperand(
|
||||
ObjectTypeIds.AcknowledgeableConditionType, [BrowseNames.AckedState, BrowseNames.Id]));
|
||||
filter.SelectClauses.Add(new SimpleAttributeOperand(
|
||||
ObjectTypeIds.AcknowledgeableConditionType, [BrowseNames.ConfirmedState, BrowseNames.Id]));
|
||||
filter.SelectClauses.Add(new SimpleAttributeOperand(
|
||||
ObjectTypeIds.ConditionType, [BrowseNames.EnabledState, BrowseNames.Id]));
|
||||
filter.SelectClauses.Add(new SimpleAttributeOperand(
|
||||
ObjectTypeIds.AlarmConditionType, [BrowseNames.ShelvingState, BrowseNames.CurrentState, BrowseNames.Id]));
|
||||
filter.AddSelectClause(ObjectTypeIds.ConditionType, BrowseNames.Comment);
|
||||
filter.AddSelectClause(ObjectTypeIds.GeneralModelChangeEventType, BrowseNames.Changes);
|
||||
filter.AddSelectClause(ObjectTypes.BaseEventType, BrowseNames.SourceNode);
|
||||
|
||||
var item = new MonitoredItem(subscription.DefaultItem)
|
||||
{
|
||||
StartNodeId = source,
|
||||
AttributeId = Attributes.EventNotifier,
|
||||
Filter = filter,
|
||||
QueueSize = 1000,
|
||||
SamplingInterval = 0,
|
||||
};
|
||||
subscription.AddItem(item);
|
||||
return item;
|
||||
}
|
||||
|
||||
public static async Task WaitUntilAsync(Func<bool> condition)
|
||||
{
|
||||
var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10);
|
||||
while (DateTime.UtcNow < deadline)
|
||||
{
|
||||
if (condition()) return;
|
||||
await Task.Delay(50);
|
||||
}
|
||||
condition().ShouldBeTrue("condition not met within timeout");
|
||||
}
|
||||
}
|
||||
+125
@@ -0,0 +1,125 @@
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Opc.Ua;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
|
||||
using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Authorization;
|
||||
using ZB.MOM.WW.OtOpcUa.OpcUaServer.Security;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #588 — <see cref="OtOpcUaNodeManager.FilterModelChangeFor"/> narrows a model-change event to the
|
||||
/// changes a session may browse. Every builder in the manager emits one change, so the over-the-wire tests
|
||||
/// only ever see "whole or nothing"; this pins the multi-change copy at its seam.
|
||||
/// </summary>
|
||||
public sealed class NodeManagerModelChangeFilterTests : IDisposable
|
||||
{
|
||||
private const string Cluster = "SITE-A";
|
||||
private const string Principal = "scadabridge";
|
||||
private const string RawFolder = "Plant/Modbus/dev1";
|
||||
|
||||
private static CancellationToken Ct => TestContext.Current.CancellationToken;
|
||||
|
||||
private readonly string _pkiRoot = Path.Combine(Path.GetTempPath(), $"otopcua-mcfilter-{Guid.NewGuid():N}");
|
||||
|
||||
/// <summary>
|
||||
/// A session granted one equipment folder receives a copy carrying only that folder's change — same
|
||||
/// EventId, and a Message that no longer names the hidden node — while a fully granted session receives
|
||||
/// the original and an ungranted one nothing.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task A_multi_change_event_is_narrowed_to_the_changes_the_session_may_browse()
|
||||
{
|
||||
var (host, server) = await BootAsync();
|
||||
try
|
||||
{
|
||||
var nm = server.NodeManager!;
|
||||
var sink = new SdkAddressSpaceSink(nm);
|
||||
sink.EnsureFolder("EqA", null, "A", AddressSpaceRealm.Uns);
|
||||
sink.EnsureFolder(RawFolder, null, "dev1", AddressSpaceRealm.Raw);
|
||||
|
||||
var e = nm.BuildNodesAddedModelChange("EqA", AddressSpaceRealm.Uns);
|
||||
var eqA = e.Changes.Value[0];
|
||||
var raw = nm.BuildNodesAddedModelChange(RawFolder, AddressSpaceRealm.Raw).Changes.Value[0];
|
||||
e.Changes.Value = [eqA, raw];
|
||||
e.Message.Value = new LocalizedText($"Nodes added under {eqA.Affected} and {raw.Affected}");
|
||||
|
||||
nm.Authorizer = Authorizer(
|
||||
new NodeAcl
|
||||
{
|
||||
NodeAclId = "acl-eqa", ClusterId = Cluster, Principal = Principal,
|
||||
ScopeKind = NodeAclScopeKind.Equipment, ScopeId = "eq-EqA",
|
||||
PermissionFlags = NodePermissions.Browse,
|
||||
},
|
||||
new NodeAcl
|
||||
{
|
||||
NodeAclId = "acl-all", ClusterId = Cluster, Principal = "everything",
|
||||
ScopeKind = NodeAclScopeKind.Cluster, PermissionFlags = NodePermissions.Browse,
|
||||
});
|
||||
|
||||
var narrowed = nm.FilterModelChangeFor(Identity(Principal), e).ShouldNotBeNull();
|
||||
narrowed.ShouldNotBeSameAs(e);
|
||||
narrowed.Changes.Value.Select(c => c.Affected).ShouldBe([eqA.Affected]);
|
||||
narrowed.EventId.Value.ShouldBe(e.EventId.Value);
|
||||
narrowed.Message.Value.Text.ShouldNotContain(RawFolder);
|
||||
|
||||
nm.FilterModelChangeFor(Identity("everything"), e).ShouldBeSameAs(e);
|
||||
nm.FilterModelChangeFor(Identity("nobody"), e).ShouldBeNull();
|
||||
}
|
||||
finally { await host.DisposeAsync(); }
|
||||
}
|
||||
|
||||
private static DataPlaneAuthorizer Authorizer(params NodeAcl[] grants)
|
||||
{
|
||||
var state = new DataPlaneAuthorityState();
|
||||
state.Apply(Cluster, "rev-1", [], grants,
|
||||
new Dictionary<AclScopeRef, IReadOnlyList<NodeAclPath>>
|
||||
{
|
||||
[new AclScopeRef(NodeAclScopeKind.Equipment, "eq-EqA")] = [new NodeAclPath([NodeAclRealms.Uns, "EqA"])],
|
||||
});
|
||||
return new DataPlaneAuthorizer(state, new TriePermissionEvaluator(state.Tries), true, NullLogger.Instance);
|
||||
}
|
||||
|
||||
private static RoleCarryingUserIdentity Identity(string principal) =>
|
||||
new(new UserNameIdentityToken { UserName = "apikey" }, [], principal, apiKeyId: null);
|
||||
|
||||
private async Task<(OpcUaApplicationHost Host, OtOpcUaSdkServer Server)> BootAsync()
|
||||
{
|
||||
var host = new OpcUaApplicationHost(
|
||||
new OpcUaApplicationHostOptions
|
||||
{
|
||||
ApplicationName = "OtOpcUa.ModelChangeFilterTest",
|
||||
ApplicationUri = $"urn:OtOpcUa.ModelChangeFilterTest:{Guid.NewGuid():N}",
|
||||
OpcUaPort = AllocateFreePort(),
|
||||
PublicHostname = "localhost",
|
||||
PkiStoreRoot = _pkiRoot,
|
||||
},
|
||||
NullLogger<OpcUaApplicationHost>.Instance);
|
||||
|
||||
var server = new OtOpcUaSdkServer();
|
||||
await host.StartAsync(server, Ct);
|
||||
return (host, server);
|
||||
}
|
||||
|
||||
private static int AllocateFreePort()
|
||||
{
|
||||
using var listener = new System.Net.Sockets.TcpListener(System.Net.IPAddress.Loopback, 0);
|
||||
listener.Start();
|
||||
var port = ((System.Net.IPEndPoint)listener.LocalEndpoint).Port;
|
||||
listener.Stop();
|
||||
return port;
|
||||
}
|
||||
|
||||
/// <summary>Cleans up the PKI root directory.</summary>
|
||||
public void Dispose()
|
||||
{
|
||||
if (Directory.Exists(_pkiRoot))
|
||||
{
|
||||
try { Directory.Delete(_pkiRoot, recursive: true); }
|
||||
catch (IOException) { /* best effort */ }
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user