Joseph Doherty
909d75357b
Closes two test-completeness gaps flagged in review of the seam-tap commit — the ScriptedAlarmHostActor and VirtualTagActor hub emits had no test proving the hub actually receives them. - ScriptedAlarm_activation_emits_one_Alarm_item_with_matching_payload: spawns the host with a capturing fake hub, drives a real engine Inactive->Active transition through the existing ScriptedAlarms harness, and asserts exactly one TelemetryItem.Alarm carrying the Activated transition. - VirtualTag_script_log_emits_one_Script_item_with_same_payload: passes a fake hub via Props + a publisherFactory, drives an evaluator failure, and asserts a TelemetryItem.Script carrying the SAME ScriptLogEntry instance handed to the publisher. To make the VirtualTagActor tap observable, its hub emit moved to the TOP of PublishLog, before the test-only publisherFactory early-return branch, so the emit fires on BOTH the production DPS path and the factory path. Production behavior is unchanged: production passes publisherFactory: null, so it still reaches the DPS Tell; the emit just moved a couple lines earlier (both are fire-and-forget). Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
318 lines
15 KiB
C#
318 lines
15 KiB
C#
using System.Text.Json;
|
|
using Akka.Actor;
|
|
using Akka.TestKit;
|
|
using Microsoft.EntityFrameworkCore;
|
|
using Serilog;
|
|
using Shouldly;
|
|
using Xunit;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Engines;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Alerts;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Logging;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Types;
|
|
using ZB.MOM.WW.OtOpcUa.Configuration;
|
|
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.ScriptedAlarms;
|
|
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
|
|
using ZB.MOM.WW.OtOpcUa.OpcUaServer;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.Scripting;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.Telemetry;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
|
|
|
|
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Telemetry;
|
|
|
|
/// <summary>
|
|
/// Per-cluster mesh Phase 5 — the four telemetry publish seams each additionally fan their DPS payload
|
|
/// into the node-local <see cref="ITelemetryLocalHub"/>. Each test drives a producer with a capturing
|
|
/// fake hub and asserts exactly one <see cref="TelemetryItem"/> of the correct subtype carrying the
|
|
/// SAME payload that goes onto the DistributedPubSub topic. The DPS publish itself is unchanged and is
|
|
/// covered by the existing per-producer tests — these only assert the additive tap.
|
|
/// </summary>
|
|
public sealed class PublishSeamEmitsToHubTests : RuntimeActorTestBase
|
|
{
|
|
private static readonly NodeId TestNode = NodeId.Parse("telemetry-seam-test");
|
|
private static readonly RevisionHash RevA = RevisionHash.Parse(new string('a', 64));
|
|
private static readonly DateTime Ts = new(2026, 7, 23, 10, 0, 0, DateTimeKind.Utc);
|
|
private const string AlarmRawPath = "Plant/Modbus/dev1/temp_hi";
|
|
|
|
// --- driver-health seam (AkkaDriverHealthPublisher) → TelemetryItem.Health --------------------------
|
|
|
|
/// <summary>The driver-health publisher fans the SAME <see cref="DriverHealthChanged"/> it publishes to
|
|
/// the <c>driver-health</c> DPS topic into the hub as a <see cref="TelemetryItem.Health"/>.</summary>
|
|
[Fact]
|
|
public void Driver_health_publish_emits_one_Health_item_with_same_payload()
|
|
{
|
|
var hub = new CapturingHub();
|
|
var publisher = new AkkaDriverHealthPublisher(Sys, hub);
|
|
|
|
var health = new DriverHealth(DriverState.Degraded, Ts, "some error");
|
|
publisher.Publish("c1", "drv-1", health, errorCount5Min: 4);
|
|
|
|
hub.Items.Count.ShouldBe(1);
|
|
var item = hub.Items[0].ShouldBeOfType<TelemetryItem.Health>();
|
|
item.E.ClusterId.ShouldBe("c1");
|
|
item.E.DriverInstanceId.ShouldBe("drv-1");
|
|
item.E.State.ShouldBe(DriverState.Degraded.ToString());
|
|
item.E.LastError.ShouldBe("some error");
|
|
item.E.ErrorCount5Min.ShouldBe(4);
|
|
}
|
|
|
|
// --- script-logs seam (DpsScriptLogPublisher) → TelemetryItem.Script -------------------------------
|
|
|
|
/// <summary>The DPS script-log publisher fans the SAME <see cref="ScriptLogEntry"/> instance it publishes
|
|
/// to the <c>script-logs</c> topic into the hub as a <see cref="TelemetryItem.Script"/>.</summary>
|
|
[Fact]
|
|
public void Script_log_publish_emits_one_Script_item_with_same_payload()
|
|
{
|
|
var hub = new CapturingHub();
|
|
var publisher = new DpsScriptLogPublisher(() => Sys, hub);
|
|
|
|
var entry = new ScriptLogEntry(
|
|
ScriptId: "S1",
|
|
Level: "Information",
|
|
Message: "hello",
|
|
TimestampUtc: Ts,
|
|
VirtualTagId: "V1",
|
|
AlarmId: null,
|
|
EquipmentId: "EQ1");
|
|
|
|
publisher.Publish(entry);
|
|
|
|
hub.Items.Count.ShouldBe(1);
|
|
var item = hub.Items[0].ShouldBeOfType<TelemetryItem.Script>();
|
|
item.E.ShouldBeSameAs(entry); // the exact instance published to DPS
|
|
}
|
|
|
|
/// <summary>A null hub is rejected at construction (fail-fast, DI provides a real hub on driver nodes).</summary>
|
|
[Fact]
|
|
public void Script_log_publisher_rejects_null_hub()
|
|
{
|
|
Should.Throw<ArgumentNullException>(() => new DpsScriptLogPublisher(() => Sys, null!));
|
|
}
|
|
|
|
// --- alerts seam (DriverHostActor native alarm) → TelemetryItem.Alarm ------------------------------
|
|
|
|
/// <summary>A Primary-gated native-alarm transition fans the SAME <see cref="AlarmTransitionEvent"/> it
|
|
/// publishes to the <c>alerts</c> topic into the hub as a <see cref="TelemetryItem.Alarm"/>. Proves the
|
|
/// hub was threaded through <see cref="DriverHostActor"/>'s Props into the emit seam.</summary>
|
|
[Fact]
|
|
public void Native_alarm_publish_emits_one_Alarm_item_with_matching_payload()
|
|
{
|
|
var db = NewInMemoryDbFactory();
|
|
var deploymentId = SeedV3AlarmDeployment(db, RevA);
|
|
|
|
var hub = new CapturingHub();
|
|
var (actor, publish) = SpawnHostAndApply(db, deploymentId, hub);
|
|
|
|
actor.Tell(new DriverInstanceActor.AttributeAlarmPublished("drv-1", new AlarmEventArgs(
|
|
new StubAlarmHandle(),
|
|
SourceNodeId: "Temp",
|
|
ConditionId: AlarmRawPath,
|
|
AlarmType: "OffNormalAlarm",
|
|
Message: "temperature high",
|
|
Severity: AlarmSeverity.High,
|
|
SourceTimestampUtc: Ts,
|
|
Kind: AlarmTransitionKind.Raise)));
|
|
|
|
// The OPC UA condition update confirms the transition was processed (ordered before the alerts fan-out).
|
|
publish.ExpectMsg<OpcUaPublishActor.AlarmStateUpdate>(TimeSpan.FromSeconds(5));
|
|
|
|
// Exactly one Alarm item, carrying the same transition the alerts topic received.
|
|
AwaitAssert(() => hub.Items.Count.ShouldBe(1), TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(100));
|
|
var item = hub.Items[0].ShouldBeOfType<TelemetryItem.Alarm>();
|
|
item.E.AlarmId.ShouldBe(AlarmRawPath);
|
|
item.E.AlarmName.ShouldBe("temp_hi");
|
|
item.E.TransitionKind.ShouldBe("Activated");
|
|
item.E.Message.ShouldBe("temperature high");
|
|
}
|
|
|
|
/// <summary>The scripted-alarm host's engine-emission seam fans the SAME <see cref="AlarmTransitionEvent"/>
|
|
/// it publishes to the <c>alerts</c> topic into the hub as a <see cref="TelemetryItem.Alarm"/>. Drives a
|
|
/// real engine Inactive→Active transition through the same harness the ScriptedAlarmHostActor tests use.</summary>
|
|
[Fact]
|
|
public void ScriptedAlarm_activation_emits_one_Alarm_item_with_matching_payload()
|
|
{
|
|
var publish = CreateTestProbe();
|
|
var mux = CreateTestProbe();
|
|
var hub = new CapturingHub();
|
|
|
|
var upstream = new DependencyMuxTagUpstreamSource();
|
|
var logger = new LoggerConfiguration().CreateLogger();
|
|
var engine = new ScriptedAlarmEngine(upstream, new InMemoryAlarmStateStore(), new ScriptLoggerFactory(logger), logger);
|
|
var host = Sys.ActorOf(ScriptedAlarmHostActor.Props(
|
|
publish.Ref, mux.Ref, upstream, engine, localNode: null, driverMemberCountProvider: null, telemetryHub: hub));
|
|
|
|
// One enabled alarm firing when M.T > 90; wait for load to complete (RegisterInterest lands).
|
|
host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { ScriptedPlan() }));
|
|
mux.ExpectMsg<DependencyMuxActor.RegisterInterest>(TimeSpan.FromSeconds(8));
|
|
|
|
host.Tell(new VirtualTagActor.DependencyValueChanged("M.T", 99, DateTime.UtcNow));
|
|
|
|
// The OPC UA condition update confirms the transition was processed.
|
|
publish.FishForMessage<OpcUaPublishActor.AlarmStateUpdate>(m => m.State.Active, TimeSpan.FromSeconds(8));
|
|
|
|
// Exactly one Alarm item, carrying the Activated transition the alerts topic received.
|
|
AwaitAssert(() => hub.Items.Count.ShouldBe(1), TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(100));
|
|
var item = hub.Items[0].ShouldBeOfType<TelemetryItem.Alarm>();
|
|
item.E.AlarmId.ShouldBe("alm-1");
|
|
item.E.TransitionKind.ShouldBe("Activated");
|
|
item.E.Severity.ShouldBe(1000); // 800 → Critical bucket → 1000
|
|
}
|
|
|
|
// --- script-logs seam (VirtualTagActor.PublishLog) → TelemetryItem.Script --------------------------
|
|
|
|
/// <summary>The virtual-tag actor's <c>PublishLog</c> seam emits a <see cref="TelemetryItem.Script"/> into
|
|
/// the hub carrying the SAME <see cref="ScriptLogEntry"/> instance it hands to the publisher — proven via
|
|
/// the test-only <c>publisherFactory</c> path (production uses the DPS Tell; the emit fires on both because
|
|
/// it precedes the early-return branch).</summary>
|
|
[Fact]
|
|
public void VirtualTag_script_log_emits_one_Script_item_with_same_payload()
|
|
{
|
|
var hub = new CapturingHub();
|
|
var published = new List<ScriptLogEntry>();
|
|
var parent = CreateTestProbe();
|
|
|
|
var actor = parent.ChildActorOf(VirtualTagActor.Props(
|
|
"vt-1", "broken",
|
|
evaluator: new FailingEvaluator("syntax error"),
|
|
scriptId: "script-7",
|
|
publisherFactory: () => new DPSPublisher((_, payload) => published.Add((ScriptLogEntry)payload)),
|
|
telemetryHub: hub));
|
|
|
|
actor.Tell(new VirtualTagActor.DependencyValueChanged("a", 1, DateTime.UtcNow));
|
|
|
|
AwaitAssert(() =>
|
|
{
|
|
published.Count.ShouldBe(1);
|
|
hub.Items.Count.ShouldBe(1);
|
|
}, TimeSpan.FromSeconds(2), TimeSpan.FromMilliseconds(100));
|
|
|
|
var item = hub.Items[0].ShouldBeOfType<TelemetryItem.Script>();
|
|
item.E.ShouldBeSameAs(published[0]); // the exact instance handed to the publisher
|
|
item.E.ScriptId.ShouldBe("script-7");
|
|
item.E.VirtualTagId.ShouldBe("vt-1");
|
|
item.E.Level.ShouldBe("Warning");
|
|
}
|
|
|
|
/// <summary>The <c>M.T > 90</c> scripted-alarm plan shared by the activation test (mirrors the
|
|
/// ScriptedAlarmHostActor test harness).</summary>
|
|
private static EquipmentScriptedAlarmPlan ScriptedPlan() =>
|
|
new(
|
|
ScriptedAlarmId: "alm-1",
|
|
EquipmentId: "Plant/Line1/M",
|
|
Name: "HighTemp",
|
|
AlarmType: "AlarmCondition",
|
|
Severity: 800,
|
|
MessageTemplate: "condition",
|
|
PredicateScriptId: "alm-1-script",
|
|
PredicateSource: "return (int)ctx.GetTag(\"M.T\").Value > 90;",
|
|
DependencyRefs: new[] { "M.T" },
|
|
HistorizeToAveva: true,
|
|
Retain: true,
|
|
Enabled: true);
|
|
|
|
/// <summary>Test evaluator that always fails, to drive <c>VirtualTagActor.PublishLog</c>.</summary>
|
|
private sealed class FailingEvaluator : IVirtualTagEvaluator
|
|
{
|
|
private readonly string _reason;
|
|
public FailingEvaluator(string reason) => _reason = reason;
|
|
public VirtualTagEvalResult Evaluate(string id, string expr, IReadOnlyDictionary<string, object?> deps)
|
|
=> VirtualTagEvalResult.Failure(_reason);
|
|
}
|
|
|
|
/// <summary>Spawns the host wired with the capturing hub, dispatches the deployment, and waits for the
|
|
/// Applied ack + the RebuildAddressSpace so the alarm map is built before the test raises an alarm.</summary>
|
|
private (IActorRef Actor, Akka.TestKit.TestProbe Publish) SpawnHostAndApply(
|
|
IDbContextFactory<OtOpcUaConfigDbContext> db, DeploymentId deploymentId, ITelemetryLocalHub hub)
|
|
{
|
|
var coordinator = CreateTestProbe();
|
|
var publish = CreateTestProbe();
|
|
var mux = CreateTestProbe();
|
|
var vtHost = CreateTestProbe();
|
|
|
|
var actor = Sys.ActorOf(DriverHostActor.Props(
|
|
db, TestNode, coordinator.Ref,
|
|
localRoles: new HashSet<string> { "driver" },
|
|
dependencyMux: mux.Ref,
|
|
opcUaPublishActor: publish.Ref,
|
|
virtualTagEvaluator: NullVirtualTagEvaluator.Instance,
|
|
virtualTagHostOverride: vtHost.Ref,
|
|
telemetryHub: hub));
|
|
|
|
actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));
|
|
coordinator.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied);
|
|
publish.ExpectMsg<OpcUaPublishActor.RebuildAddressSpace>(TimeSpan.FromSeconds(5));
|
|
|
|
return (actor, publish);
|
|
}
|
|
|
|
/// <summary>Seeds a Sealed v3 deployment with a single Boolean raw tag carrying an <c>alarm</c> object so
|
|
/// the composer projects it as a Part 9 condition at its RawPath (mirrors DriverHostActorNativeAlarmTests).</summary>
|
|
private static DeploymentId SeedV3AlarmDeployment(IDbContextFactory<OtOpcUaConfigDbContext> db, RevisionHash rev)
|
|
{
|
|
var artifact = JsonSerializer.SerializeToUtf8Bytes(new
|
|
{
|
|
RawFolders = new[] { new { RawFolderId = "rf-plant", ParentRawFolderId = (string?)null, Name = "Plant", ClusterId = "c1" } },
|
|
DriverInstances = new[]
|
|
{
|
|
new { DriverInstanceId = "drv-1", RawFolderId = "rf-plant", Name = "Modbus", DriverType = "Modbus", DriverConfig = "{}", ClusterId = "c1", Enabled = false },
|
|
},
|
|
Devices = new[]
|
|
{
|
|
new { DeviceId = "drv-1:dev1", DriverInstanceId = "drv-1", Name = "dev1", DeviceConfig = "{}" },
|
|
},
|
|
TagGroups = Array.Empty<object>(),
|
|
Tags = new[]
|
|
{
|
|
new
|
|
{
|
|
TagId = "tag-0",
|
|
DeviceId = "drv-1:dev1",
|
|
TagGroupId = (string?)null,
|
|
Name = "temp_hi",
|
|
DataType = "Boolean",
|
|
AccessLevel = 0,
|
|
TagConfig = JsonSerializer.Serialize(new { alarm = new { alarmType = "OffNormalAlarm", severity = 700 } }),
|
|
},
|
|
},
|
|
});
|
|
|
|
var id = DeploymentId.NewId();
|
|
using var ctx = db.CreateDbContext();
|
|
ctx.Deployments.Add(new Deployment
|
|
{
|
|
DeploymentId = id.Value,
|
|
RevisionHash = rev.Value,
|
|
Status = DeploymentStatus.Sealed,
|
|
CreatedBy = "test",
|
|
SealedAtUtc = DateTime.UtcNow,
|
|
ArtifactBlob = artifact,
|
|
});
|
|
ctx.SaveChanges();
|
|
return id;
|
|
}
|
|
|
|
/// <summary>Minimal alarm-subscription handle for building <see cref="AlarmEventArgs"/>.</summary>
|
|
private sealed class StubAlarmHandle : IAlarmSubscriptionHandle
|
|
{
|
|
public string DiagnosticId => "stub-alarm-sub";
|
|
}
|
|
|
|
/// <summary>Fake hub that records every <see cref="ITelemetryLocalHub.Emit"/>. <see cref="Subscribe"/> is
|
|
/// unused by the producer seams under test.</summary>
|
|
private sealed class CapturingHub : ITelemetryLocalHub
|
|
{
|
|
public List<TelemetryItem> Items { get; } = new();
|
|
|
|
public void Emit(TelemetryItem item) => Items.Add(item);
|
|
|
|
public ITelemetrySubscription Subscribe(int boundedCapacity) => throw new NotSupportedException();
|
|
}
|
|
}
|