Joseph Doherty
472 lines
18 KiB
C#
472 lines
18 KiB
C#
using System.Collections.Concurrent;
|
|
using Akka.Actor;
|
|
using Akka.TestKit.Xunit2;
|
|
using Microsoft.EntityFrameworkCore;
|
|
using Microsoft.Extensions.Logging.Abstractions;
|
|
using Shouldly;
|
|
using Xunit;
|
|
using ZB.MOM.WW.OtOpcUa.AdminUI.Hubs;
|
|
using ZB.MOM.WW.OtOpcUa.AdminUI.Telemetry;
|
|
using ZB.MOM.WW.OtOpcUa.Cluster;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Alerts;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Logging;
|
|
using ZB.MOM.WW.OtOpcUa.Configuration;
|
|
using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
|
|
|
|
namespace ZB.MOM.WW.OtOpcUa.AdminUI.Tests.Telemetry;
|
|
|
|
/// <summary>
|
|
/// Unit tests for <see cref="TelemetryDialSupervisor"/> — the central-side dial supervisor that
|
|
/// discovers driver nodes, keeps one reconnecting dialer each, and feeds the four AdminUI
|
|
/// in-process sinks. All seams are faked: no real DB, no real gRPC.
|
|
/// </summary>
|
|
public sealed class TelemetryDialSupervisorTests : TestKit
|
|
{
|
|
private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(5);
|
|
|
|
[Fact]
|
|
public void Discovery_starts_one_dialer_per_target_with_distinct_correlation_ids()
|
|
{
|
|
var dial = new FakeDialLoop();
|
|
var targets = new[]
|
|
{
|
|
new TelemetryDialTarget("node-1", "http://h1:5300"),
|
|
new TelemetryDialTarget("node-2", "http://h2:5300"),
|
|
};
|
|
|
|
Spawn(dial, () => targets);
|
|
|
|
var invocations = dial.WaitForCount(2, Timeout);
|
|
invocations.Select(i => i.Target.NodeId).ShouldBe(new[] { "node-1", "node-2" }, ignoreOrder: true);
|
|
invocations.ShouldContain(i => i.CorrelationId == "central-node-1-1");
|
|
invocations.ShouldContain(i => i.CorrelationId == "central-node-2-1");
|
|
invocations.ShouldAllBe(i => i.Ct.CanBeCanceled);
|
|
}
|
|
|
|
[Fact]
|
|
public void Refresh_removes_departed_nodes_and_adds_new_ones()
|
|
{
|
|
var dial = new FakeDialLoop();
|
|
var current = new List<TelemetryDialTarget>
|
|
{
|
|
new("node-1", "http://h1:5300"),
|
|
new("node-2", "http://h2:5300"),
|
|
};
|
|
|
|
var actor = Spawn(dial, () => current.ToArray());
|
|
dial.WaitForCount(2, Timeout);
|
|
|
|
var node1 = dial.Invocations.Single(i => i.Target.NodeId == "node-1");
|
|
|
|
// node-1 leaves, node-3 joins.
|
|
current.Clear();
|
|
current.Add(new TelemetryDialTarget("node-2", "http://h2:5300"));
|
|
current.Add(new TelemetryDialTarget("node-3", "http://h3:5300"));
|
|
actor.Tell(new TelemetryDialSupervisor.RefreshNodes());
|
|
|
|
// node-1's dialer is cancelled and dropped.
|
|
AwaitCondition(() => node1.Ct.IsCancellationRequested, Timeout);
|
|
// node-3 gets a fresh dialer.
|
|
var invocations = dial.WaitForCount(3, Timeout);
|
|
invocations.ShouldContain(i => i.Target.NodeId == "node-3" && i.CorrelationId == "central-node-3-1");
|
|
|
|
// node-2 is present in BOTH refreshes with an unchanged endpoint: it must NOT be restarted, or
|
|
// its live stream would be dropped on every refresh. Exactly one dial invocation for it.
|
|
ExpectNoMsg(TimeSpan.FromMilliseconds(200));
|
|
dial.Invocations.Count(i => i.Target.NodeId == "node-2").ShouldBe(1);
|
|
}
|
|
|
|
[Fact]
|
|
public void Refresh_redials_a_node_whose_endpoint_changed_but_not_an_unchanged_one()
|
|
{
|
|
var dial = new FakeDialLoop();
|
|
var current = new List<TelemetryDialTarget>
|
|
{
|
|
new("node-1", "http://old-host:5300"),
|
|
new("node-2", "http://h2:5300"),
|
|
};
|
|
|
|
var actor = Spawn(dial, () => current.ToArray());
|
|
dial.WaitForCount(2, Timeout);
|
|
var node1Original = dial.Invocations.Single(i => i.Target.NodeId == "node-1");
|
|
|
|
// node-1 re-provisioned to a new host; node-2 unchanged.
|
|
current.Clear();
|
|
current.Add(new TelemetryDialTarget("node-1", "http://new-host:5300"));
|
|
current.Add(new TelemetryDialTarget("node-2", "http://h2:5300"));
|
|
actor.Tell(new TelemetryDialSupervisor.RefreshNodes());
|
|
|
|
// The old node-1 stream is torn down and a fresh dialer opens at the NEW endpoint (gen 2).
|
|
AwaitCondition(() => node1Original.Ct.IsCancellationRequested, Timeout);
|
|
var invocations = dial.WaitForCount(3, Timeout);
|
|
invocations.ShouldContain(i =>
|
|
i.Target.NodeId == "node-1"
|
|
&& i.Target.Endpoint == "http://new-host:5300"
|
|
&& i.CorrelationId == "central-node-1-2");
|
|
|
|
// node-2's endpoint did not change → still exactly one dial invocation (no churn).
|
|
ExpectNoMsg(TimeSpan.FromMilliseconds(200));
|
|
dial.Invocations.Count(i => i.Target.NodeId == "node-2").ShouldBe(1);
|
|
}
|
|
|
|
[Fact]
|
|
public void Each_record_type_lands_on_its_sink()
|
|
{
|
|
var dial = new FakeDialLoop();
|
|
var sinks = new Sinks();
|
|
Spawn(dial, () => new[] { new TelemetryDialTarget("node-1", "http://h1:5300") }, sinks);
|
|
|
|
var inv = dial.WaitForCount(1, Timeout).Single();
|
|
|
|
inv.OnMapped(Alarm("a1"));
|
|
inv.OnMapped(Script("s1"));
|
|
inv.OnMapped(Health("d1"));
|
|
inv.OnMapped(Resilience("d1"));
|
|
|
|
AwaitAssert(
|
|
() =>
|
|
{
|
|
sinks.Alarms.ShouldHaveSingleItem().AlarmId.ShouldBe("a1");
|
|
sinks.Scripts.ShouldHaveSingleItem().ScriptId.ShouldBe("s1");
|
|
sinks.Health.ShouldHaveSingleItem().DriverInstanceId.ShouldBe("d1");
|
|
sinks.Resilience.ShouldHaveSingleItem().DriverInstanceId.ShouldBe("d1");
|
|
},
|
|
Timeout);
|
|
}
|
|
|
|
[Fact]
|
|
public void Stream_failure_reconnects_with_an_incremented_generation()
|
|
{
|
|
var dial = new FakeDialLoop();
|
|
Spawn(dial, () => new[] { new TelemetryDialTarget("node-1", "http://h1:5300") });
|
|
|
|
var first = dial.WaitForCount(1, Timeout).Single();
|
|
first.CorrelationId.ShouldBe("central-node-1-1");
|
|
|
|
// Transport failure → the client would end the stream after onError, so end the fake loop too.
|
|
first.OnError(new InvalidOperationException("boom"));
|
|
first.Complete();
|
|
|
|
// First retry is immediate; generation is bumped.
|
|
var invocations = dial.WaitForCount(2, Timeout);
|
|
invocations[1].CorrelationId.ShouldBe("central-node-1-2");
|
|
}
|
|
|
|
[Fact]
|
|
public void Late_event_from_a_superseded_generation_is_dropped()
|
|
{
|
|
var dial = new FakeDialLoop();
|
|
var sinks = new Sinks();
|
|
Spawn(dial, () => new[] { new TelemetryDialTarget("node-1", "http://h1:5300") }, sinks);
|
|
|
|
var first = dial.WaitForCount(1, Timeout).Single();
|
|
first.OnError(new InvalidOperationException("boom"));
|
|
first.Complete();
|
|
dial.WaitForCount(2, Timeout); // gen-2 dialer is live.
|
|
|
|
// A late event arriving on the OLD (gen-1) stream must be dropped, not routed.
|
|
first.OnMapped(Alarm("stale"));
|
|
|
|
// Give the message a chance to be (wrongly) routed, then assert it was not.
|
|
ExpectNoMsg(TimeSpan.FromMilliseconds(300));
|
|
sinks.Alarms.ShouldBeEmpty();
|
|
}
|
|
|
|
[Fact]
|
|
public void Late_failure_from_a_superseded_generation_does_not_trigger_a_second_reconnect()
|
|
{
|
|
var dial = new FakeDialLoop();
|
|
Spawn(dial, () => new[] { new TelemetryDialTarget("node-1", "http://h1:5300") });
|
|
|
|
var first = dial.WaitForCount(1, Timeout).Single();
|
|
first.OnError(new InvalidOperationException("boom"));
|
|
first.Complete();
|
|
dial.WaitForCount(2, Timeout); // gen-2 dialer.
|
|
|
|
// A duplicate/late failure from gen-1 must be ignored (no third dial).
|
|
first.OnError(new InvalidOperationException("late duplicate"));
|
|
|
|
ExpectNoMsg(TimeSpan.FromMilliseconds(400));
|
|
dial.Invocations.Count.ShouldBe(2);
|
|
}
|
|
|
|
[Fact]
|
|
public void Pill_flips_true_on_first_connection_and_false_only_when_all_down()
|
|
{
|
|
var dial = new FakeDialLoop();
|
|
var sinks = new Sinks();
|
|
Spawn(
|
|
dial,
|
|
() => new[]
|
|
{
|
|
new TelemetryDialTarget("node-1", "http://h1:5300"),
|
|
new TelemetryDialTarget("node-2", "http://h2:5300"),
|
|
},
|
|
sinks);
|
|
|
|
var inv = dial.WaitForCount(2, Timeout);
|
|
var node1 = inv.Single(i => i.Target.NodeId == "node-1");
|
|
var node2 = inv.Single(i => i.Target.NodeId == "node-2");
|
|
|
|
// First node connects → both broadcasters flip to connected.
|
|
node1.OnMapped(Alarm("a1"));
|
|
AwaitAssert(() => sinks.PillStates.ToArray().ShouldBe(new[] { true }), Timeout);
|
|
|
|
// Second node connects → still connected, no extra transition.
|
|
node2.OnMapped(Alarm("a2"));
|
|
ExpectNoMsg(TimeSpan.FromMilliseconds(200));
|
|
sinks.PillStates.ToArray().ShouldBe(new[] { true });
|
|
|
|
// node-1 drops but node-2 is still up → pill stays true.
|
|
node1.OnError(new InvalidOperationException("down"));
|
|
node1.Complete();
|
|
ExpectNoMsg(TimeSpan.FromMilliseconds(200));
|
|
sinks.PillStates.ToArray().ShouldBe(new[] { true });
|
|
|
|
// node-2 drops too → all down → pill flips false.
|
|
node2.OnError(new InvalidOperationException("down"));
|
|
node2.Complete();
|
|
AwaitAssert(() => sinks.PillStates.ToArray().ShouldBe(new[] { true, false }), Timeout);
|
|
}
|
|
|
|
[Fact]
|
|
public void PostStop_cancels_every_dialer()
|
|
{
|
|
var dial = new FakeDialLoop();
|
|
var actor = Spawn(
|
|
dial,
|
|
() => new[]
|
|
{
|
|
new TelemetryDialTarget("node-1", "http://h1:5300"),
|
|
new TelemetryDialTarget("node-2", "http://h2:5300"),
|
|
});
|
|
|
|
var inv = dial.WaitForCount(2, Timeout);
|
|
|
|
Sys.Stop(actor);
|
|
|
|
AwaitCondition(() => inv.All(i => i.Ct.IsCancellationRequested), Timeout);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Production_node_source_skips_null_grpc_port_and_maintenance_rows()
|
|
{
|
|
var factory = new InMemoryFactory(Guid.NewGuid().ToString());
|
|
await using (var db = factory.CreateDbContext())
|
|
{
|
|
db.ClusterNodes.AddRange(
|
|
Node("has-port", "h1", grpcPort: 5300, enabled: true, maintenance: false),
|
|
Node("null-port", "h2", grpcPort: null, enabled: true, maintenance: false),
|
|
Node("maintenance", "h3", grpcPort: 5300, enabled: true, maintenance: true),
|
|
Node("disabled", "h4", grpcPort: 5300, enabled: false, maintenance: false));
|
|
await db.SaveChangesAsync();
|
|
}
|
|
|
|
var source = TelemetryNodeSource.Create(factory, NullLogger.Instance);
|
|
var targets = await source();
|
|
|
|
targets.ShouldHaveSingleItem();
|
|
targets[0].NodeId.ShouldBe("has-port");
|
|
targets[0].Endpoint.ShouldBe("http://h1:5300");
|
|
}
|
|
|
|
private IActorRef Spawn(
|
|
FakeDialLoop dial,
|
|
Func<TelemetryDialTarget[]> nodeSource,
|
|
Sinks? sinks = null)
|
|
{
|
|
sinks ??= new Sinks();
|
|
var options = new TelemetryDialOptions { ContactRefreshSeconds = 3600 };
|
|
return Sys.ActorOf(TelemetryDialSupervisor.Props(
|
|
() => Task.FromResult<IReadOnlyList<TelemetryDialTarget>>(nodeSource()),
|
|
dial.Loop,
|
|
sinks.AlarmBroadcaster,
|
|
sinks.ScriptBroadcaster,
|
|
sinks.HealthStore,
|
|
sinks.ResilienceStore,
|
|
options));
|
|
}
|
|
|
|
private static AlarmTransitionEvent Alarm(string id) =>
|
|
new(id, "Area/Line/Eq", "hi", "Activated", 500, "msg", "system", DateTime.UtcNow);
|
|
|
|
private static ScriptLogEntry Script(string id) =>
|
|
new(id, "Info", "msg", DateTime.UtcNow, null, null, null);
|
|
|
|
private static DriverHealthChanged Health(string id) =>
|
|
new("C1", id, "Healthy", DateTime.UtcNow, null, 0, DateTime.UtcNow);
|
|
|
|
private static DriverResilienceStatusChanged Resilience(string id) =>
|
|
new(id, "host", false, 0, 0, null, DateTime.UtcNow, DateTime.UtcNow);
|
|
|
|
private static ClusterNode Node(string id, string host, int? grpcPort, bool enabled, bool maintenance) =>
|
|
new()
|
|
{
|
|
NodeId = id,
|
|
ClusterId = "C1",
|
|
Host = host,
|
|
GrpcPort = grpcPort,
|
|
Enabled = enabled,
|
|
MaintenanceMode = maintenance,
|
|
ApplicationUri = $"urn:{id}",
|
|
CreatedBy = "test",
|
|
};
|
|
|
|
/// <summary>A captured dial-loop invocation with the callbacks the test drives by hand.</summary>
|
|
private sealed class DialInvocation
|
|
{
|
|
private readonly TaskCompletionSource _completion = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
|
|
public required TelemetryDialTarget Target { get; init; }
|
|
public required string CorrelationId { get; init; }
|
|
public required Action<object> OnMapped { get; init; }
|
|
public required Action<Exception> OnError { get; init; }
|
|
public required CancellationToken Ct { get; init; }
|
|
|
|
public Task Completion => _completion.Task;
|
|
|
|
/// <summary>Ends the fake loop (as the client returns after a server-ended or post-onError stream).</summary>
|
|
public void Complete() => _completion.TrySetResult();
|
|
}
|
|
|
|
/// <summary>A fake <see cref="TelemetryDialLoop"/> that records each invocation and blocks until told to end.</summary>
|
|
private sealed class FakeDialLoop
|
|
{
|
|
private readonly ConcurrentQueue<DialInvocation> _invocations = new();
|
|
|
|
public IReadOnlyList<DialInvocation> Invocations => _invocations.ToArray();
|
|
|
|
public TelemetryDialLoop Loop => async (target, correlationId, onMapped, onError, ct) =>
|
|
{
|
|
var invocation = new DialInvocation
|
|
{
|
|
Target = target,
|
|
CorrelationId = correlationId,
|
|
OnMapped = onMapped,
|
|
OnError = onError,
|
|
Ct = ct,
|
|
};
|
|
_invocations.Enqueue(invocation);
|
|
|
|
// Complete when the test ends the loop OR the supervisor cancels us.
|
|
await using var reg = ct.Register(() => invocation.Complete());
|
|
await invocation.Completion;
|
|
};
|
|
|
|
public IReadOnlyList<DialInvocation> WaitForCount(int count, TimeSpan timeout)
|
|
{
|
|
var deadline = DateTime.UtcNow + timeout;
|
|
while (DateTime.UtcNow < deadline)
|
|
{
|
|
var snapshot = _invocations.ToArray();
|
|
if (snapshot.Length >= count)
|
|
{
|
|
return snapshot;
|
|
}
|
|
|
|
Thread.Sleep(10);
|
|
}
|
|
|
|
throw new Xunit.Sdk.XunitException(
|
|
$"Expected at least {count} dial invocations within {timeout}, saw {_invocations.Count}.");
|
|
}
|
|
}
|
|
|
|
/// <summary>The four fake sinks plus the pill-transition record.</summary>
|
|
private sealed class Sinks
|
|
{
|
|
public RecordingBroadcaster<AlarmTransitionEvent> AlarmBroadcaster { get; }
|
|
public RecordingBroadcaster<ScriptLogEntry> ScriptBroadcaster { get; }
|
|
public RecordingHealthStore HealthStore { get; } = new();
|
|
public RecordingResilienceStore ResilienceStore { get; } = new();
|
|
|
|
public ConcurrentQueue<AlarmTransitionEvent> Alarms { get; } = new();
|
|
public ConcurrentQueue<ScriptLogEntry> Scripts { get; } = new();
|
|
public ConcurrentQueue<bool> PillStates { get; } = new();
|
|
|
|
public ConcurrentQueue<DriverHealthChanged> Health => HealthStore.Upserts;
|
|
public ConcurrentQueue<DriverResilienceStatusChanged> Resilience => ResilienceStore.Upserts;
|
|
|
|
public Sinks()
|
|
{
|
|
// Both broadcasters share the ONE pill-transition record so a test can assert the alarm and
|
|
// script pills move together (matching the supervisor calling SetConnected on both).
|
|
AlarmBroadcaster = new RecordingBroadcaster<AlarmTransitionEvent>(Alarms, PillStates);
|
|
ScriptBroadcaster = new RecordingBroadcaster<ScriptLogEntry>(Scripts, null);
|
|
}
|
|
}
|
|
|
|
private sealed class RecordingBroadcaster<T>(ConcurrentQueue<T> published, ConcurrentQueue<bool>? pill)
|
|
: IInProcessBroadcaster<T>
|
|
{
|
|
public event Action<T>? Received;
|
|
public event Action<bool>? ConnectionStateChanged;
|
|
|
|
public bool IsConnected { get; private set; }
|
|
|
|
public void Publish(T item)
|
|
{
|
|
published.Enqueue(item);
|
|
Received?.Invoke(item);
|
|
}
|
|
|
|
public void SetConnected(bool connected)
|
|
{
|
|
IsConnected = connected;
|
|
pill?.Enqueue(connected);
|
|
ConnectionStateChanged?.Invoke(connected);
|
|
}
|
|
}
|
|
|
|
private sealed class RecordingHealthStore : IDriverStatusSnapshotStore
|
|
{
|
|
public ConcurrentQueue<DriverHealthChanged> Upserts { get; } = new();
|
|
|
|
public event Action<DriverHealthChanged>? SnapshotChanged;
|
|
|
|
public void Upsert(DriverHealthChanged snapshot)
|
|
{
|
|
Upserts.Enqueue(snapshot);
|
|
SnapshotChanged?.Invoke(snapshot);
|
|
}
|
|
|
|
public bool TryGet(string driverInstanceId, out DriverHealthChanged snapshot)
|
|
{
|
|
snapshot = null!;
|
|
return false;
|
|
}
|
|
|
|
public IReadOnlyCollection<DriverHealthChanged> GetAll() => Upserts.ToArray();
|
|
}
|
|
|
|
private sealed class RecordingResilienceStore : IDriverResilienceStatusStore
|
|
{
|
|
public ConcurrentQueue<DriverResilienceStatusChanged> Upserts { get; } = new();
|
|
|
|
public event Action<DriverResilienceStatusChanged>? SnapshotChanged;
|
|
|
|
public void Upsert(DriverResilienceStatusChanged snapshot)
|
|
{
|
|
Upserts.Enqueue(snapshot);
|
|
SnapshotChanged?.Invoke(snapshot);
|
|
}
|
|
|
|
public IReadOnlyList<DriverResilienceStatusChanged> GetForInstance(string driverInstanceId) =>
|
|
Upserts.ToArray();
|
|
|
|
public IReadOnlyCollection<DriverResilienceStatusChanged> GetAll() => Upserts.ToArray();
|
|
}
|
|
|
|
private sealed class InMemoryFactory(string name) : IDbContextFactory<OtOpcUaConfigDbContext>
|
|
{
|
|
public OtOpcUaConfigDbContext CreateDbContext() =>
|
|
new(new DbContextOptionsBuilder<OtOpcUaConfigDbContext>()
|
|
.UseInMemoryDatabase(name)
|
|
.Options);
|
|
|
|
public Task<OtOpcUaConfigDbContext> CreateDbContextAsync(CancellationToken cancellationToken = default) =>
|
|
Task.FromResult(CreateDbContext());
|
|
}
|
|
}
|