test(communication): pin wire round-trip + type identity of cross-cluster message contracts
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,191 @@
|
||||
using Akka.TestKit.Xunit2;
|
||||
using ZB.MOM.WW.Audit;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Health;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Notification;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.Communication.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// Characterization pins for the cross-cluster wire contracts. The Akka default
|
||||
/// reflective-JSON wire format couples compatibility to <b>type and namespace
|
||||
/// identity</b> plus round-trip fidelity, and nothing else tested that
|
||||
/// (<c>MessageContractTests</c> asserts correlation-ids only). These tests pin
|
||||
/// (a) round-trip fidelity and (b) the fully-qualified type identity of the
|
||||
/// highest-traffic central↔site contracts, so a rename/move that would break a
|
||||
/// rolling central↔site upgrade fails a test here instead of in production.
|
||||
///
|
||||
/// The serializer swap (proto / explicit bindings) is deliberately deferred — it
|
||||
/// changes the wire format and needs a coordinated rolling-upgrade plan; until
|
||||
/// then these pins are the guardrail.
|
||||
/// </summary>
|
||||
public class WireSerializationPinTests : TestKit
|
||||
{
|
||||
/// <summary>
|
||||
/// Serializes with the same serializer Akka would pick on the wire, then
|
||||
/// deserializes back through the type manifest — exactly the path a peer node
|
||||
/// takes when it receives the bytes.
|
||||
/// </summary>
|
||||
private T RoundTrip<T>(T message)
|
||||
{
|
||||
var serialization = Sys.Serialization;
|
||||
var serializer = serialization.FindSerializerFor(message);
|
||||
var bytes = serializer.ToBinary(message);
|
||||
return (T)serialization.Deserialize(bytes, serializer.Identifier, message!.GetType());
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void HeartbeatMessage_RoundTripsOnTheWire()
|
||||
{
|
||||
var original = new HeartbeatMessage("site-1", "host-a", true, DateTimeOffset.UtcNow);
|
||||
var back = RoundTrip(original);
|
||||
Assert.Equal(original, back); // all-scalar record → structural equality holds
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void NotificationSubmit_RoundTripsOnTheWire()
|
||||
{
|
||||
var original = new NotificationSubmit(
|
||||
NotificationId: Guid.NewGuid().ToString("N"),
|
||||
ListName: "Operators",
|
||||
Subject: "Pump alarm",
|
||||
Body: "Pump 3 tripped",
|
||||
SourceSiteId: "site-1",
|
||||
SourceInstanceId: "Plant.Pump3",
|
||||
SourceScript: "ScriptActor:MonitorSpeed",
|
||||
SiteEnqueuedAt: DateTimeOffset.UtcNow,
|
||||
OriginExecutionId: Guid.NewGuid(),
|
||||
OriginParentExecutionId: Guid.NewGuid(),
|
||||
SourceNode: "node-a");
|
||||
|
||||
var back = RoundTrip(original);
|
||||
|
||||
// All-scalar record; assert every field so a dropped ctor arg on either
|
||||
// side (a wire-fidelity bug) is caught field-by-field, not just by ==.
|
||||
Assert.Equal(original.NotificationId, back.NotificationId);
|
||||
Assert.Equal(original.ListName, back.ListName);
|
||||
Assert.Equal(original.Subject, back.Subject);
|
||||
Assert.Equal(original.Body, back.Body);
|
||||
Assert.Equal(original.SourceSiteId, back.SourceSiteId);
|
||||
Assert.Equal(original.SourceInstanceId, back.SourceInstanceId);
|
||||
Assert.Equal(original.SourceScript, back.SourceScript);
|
||||
Assert.Equal(original.SiteEnqueuedAt, back.SiteEnqueuedAt);
|
||||
Assert.Equal(original.OriginExecutionId, back.OriginExecutionId);
|
||||
Assert.Equal(original.OriginParentExecutionId, back.OriginParentExecutionId);
|
||||
Assert.Equal(original.SourceNode, back.SourceNode);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void IngestAuditEventsCommand_RoundTripsOnTheWire()
|
||||
{
|
||||
var occurredAt = new DateTime(2026, 5, 20, 10, 15, 30, 123, DateTimeKind.Utc);
|
||||
var evt = ScadaBridgeAuditEventFactory.Create(
|
||||
channel: AuditChannel.ApiOutbound,
|
||||
kind: AuditKind.ApiCallCached,
|
||||
status: AuditStatus.Forwarded,
|
||||
eventId: Guid.NewGuid(),
|
||||
occurredAtUtc: occurredAt,
|
||||
actor: "design-key",
|
||||
target: "weather-api",
|
||||
sourceNode: "node-a",
|
||||
correlationId: Guid.NewGuid(),
|
||||
executionId: Guid.NewGuid(),
|
||||
parentExecutionId: Guid.NewGuid(),
|
||||
sourceSiteId: "site-1",
|
||||
sourceInstanceId: "Pump01",
|
||||
sourceScript: "OnDemand",
|
||||
httpStatus: 200,
|
||||
durationMs: 42,
|
||||
errorMessage: "transient timeout",
|
||||
errorDetail: "stack-trace",
|
||||
requestSummary: "GET /weather",
|
||||
responseSummary: "{ \"ok\": true }",
|
||||
payloadTruncated: true,
|
||||
extra: "{ \"retryCount\": 1 }");
|
||||
|
||||
var original = new IngestAuditEventsCommand(new[] { evt });
|
||||
var back = RoundTrip(original);
|
||||
|
||||
Assert.Single(back.Events);
|
||||
var o = original.Events[0].AsRow();
|
||||
var rt = back.Events[0].AsRow();
|
||||
Assert.Equal(o.EventId, rt.EventId);
|
||||
Assert.Equal(o.OccurredAtUtc, rt.OccurredAtUtc);
|
||||
Assert.Equal(o.Channel, rt.Channel);
|
||||
Assert.Equal(o.Kind, rt.Kind);
|
||||
Assert.Equal(o.Status, rt.Status);
|
||||
Assert.Equal(o.CorrelationId, rt.CorrelationId);
|
||||
Assert.Equal(o.ExecutionId, rt.ExecutionId);
|
||||
Assert.Equal(o.ParentExecutionId, rt.ParentExecutionId);
|
||||
Assert.Equal(o.SourceSiteId, rt.SourceSiteId);
|
||||
Assert.Equal(o.SourceNode, rt.SourceNode);
|
||||
Assert.Equal(o.SourceInstanceId, rt.SourceInstanceId);
|
||||
Assert.Equal(o.SourceScript, rt.SourceScript);
|
||||
Assert.Equal(o.Actor, rt.Actor);
|
||||
Assert.Equal(o.Target, rt.Target);
|
||||
Assert.Equal(o.HttpStatus, rt.HttpStatus);
|
||||
Assert.Equal(o.DurationMs, rt.DurationMs);
|
||||
Assert.Equal(o.ErrorMessage, rt.ErrorMessage);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void SiteHealthReport_RoundTripsOnTheWire()
|
||||
{
|
||||
var original = new SiteHealthReport(
|
||||
SiteId: "site-1",
|
||||
SequenceNumber: 42,
|
||||
ReportTimestamp: DateTimeOffset.UtcNow,
|
||||
DataConnectionStatuses: new Dictionary<string, ConnectionHealth>
|
||||
{
|
||||
["opc-1"] = ConnectionHealth.Connected,
|
||||
},
|
||||
TagResolutionCounts: new Dictionary<string, TagResolutionStatus>
|
||||
{
|
||||
["opc-1"] = new TagResolutionStatus(10, 9),
|
||||
},
|
||||
ScriptErrorCount: 1,
|
||||
AlarmEvaluationErrorCount: 2,
|
||||
StoreAndForwardBufferDepths: new Dictionary<string, int>
|
||||
{
|
||||
["Notification"] = 3,
|
||||
},
|
||||
DeadLetterCount: 4,
|
||||
DeployedInstanceCount: 5,
|
||||
EnabledInstanceCount: 6,
|
||||
DisabledInstanceCount: 7,
|
||||
NodeRole: "Active",
|
||||
NodeHostname: "host-a",
|
||||
ParkedMessageCount: 8);
|
||||
|
||||
var back = RoundTrip(original);
|
||||
|
||||
// Scalars + one entry per primary collection (record equality does NOT
|
||||
// hold — dictionaries compare by reference — so assert field/entry-wise).
|
||||
Assert.Equal(original.SiteId, back.SiteId);
|
||||
Assert.Equal(original.SequenceNumber, back.SequenceNumber);
|
||||
Assert.Equal(original.ReportTimestamp, back.ReportTimestamp);
|
||||
Assert.Equal(ConnectionHealth.Connected, back.DataConnectionStatuses["opc-1"]);
|
||||
Assert.Equal(new TagResolutionStatus(10, 9), back.TagResolutionCounts["opc-1"]);
|
||||
Assert.Equal(3, back.StoreAndForwardBufferDepths["Notification"]);
|
||||
Assert.Equal(original.DeadLetterCount, back.DeadLetterCount);
|
||||
Assert.Equal(original.DeployedInstanceCount, back.DeployedInstanceCount);
|
||||
Assert.Equal(original.NodeRole, back.NodeRole);
|
||||
Assert.Equal(original.NodeHostname, back.NodeHostname);
|
||||
Assert.Equal(original.ParkedMessageCount, back.ParkedMessageCount);
|
||||
}
|
||||
|
||||
// Type-identity pins: the reflective-JSON wire embeds CLR type manifests, so a
|
||||
// rename/move of any of these types silently breaks rolling cross-version
|
||||
// central↔site communication. If one of these assertions fails, you are making
|
||||
// a wire-breaking change — coordinate a full-fleet upgrade or add an explicit
|
||||
// serialization binding before renaming/moving the type.
|
||||
[Theory]
|
||||
[InlineData(typeof(HeartbeatMessage), "ZB.MOM.WW.ScadaBridge.Commons.Messages.Health.HeartbeatMessage")]
|
||||
[InlineData(typeof(NotificationSubmit), "ZB.MOM.WW.ScadaBridge.Commons.Messages.Notification.NotificationSubmit")]
|
||||
[InlineData(typeof(IngestAuditEventsCommand), "ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit.IngestAuditEventsCommand")]
|
||||
[InlineData(typeof(SiteHealthReport), "ZB.MOM.WW.ScadaBridge.Commons.Messages.Health.SiteHealthReport")]
|
||||
public void CrossClusterContract_TypeIdentity_IsPinned(Type type, string expectedFullName) =>
|
||||
Assert.Equal(expectedFullName, type.FullName);
|
||||
}
|
||||
+82
@@ -0,0 +1,82 @@
|
||||
using Akka.TestKit.Xunit2;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// Characterization pin for the intra-cluster S&F replication contract. A
|
||||
/// <see cref="ReplicationOperation"/> (carrying a full <see cref="StoreAndForwardMessage"/>)
|
||||
/// is Told from the active node to the standby node's replication actor and rides
|
||||
/// the Akka default reflective-JSON wire format. Nothing tested that it round-trips
|
||||
/// or that its type identity is stable — a rename/move, a dropped setter, or a
|
||||
/// non-default-constructible message would silently break standby buffer sync and
|
||||
/// only surface as a divergent buffer after a failover.
|
||||
///
|
||||
/// The serializer swap (proto / explicit bindings) is deferred; until then this pin
|
||||
/// is the guardrail.
|
||||
/// </summary>
|
||||
public class ReplicationWireSerializationPinTests : TestKit
|
||||
{
|
||||
private T RoundTrip<T>(T message)
|
||||
{
|
||||
var serialization = Sys.Serialization;
|
||||
var serializer = serialization.FindSerializerFor(message);
|
||||
var bytes = serializer.ToBinary(message);
|
||||
return (T)serialization.Deserialize(bytes, serializer.Identifier, message!.GetType());
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void ReplicationOperation_WithFullMessage_RoundTripsOnTheWire()
|
||||
{
|
||||
var message = new StoreAndForwardMessage
|
||||
{
|
||||
Id = Guid.NewGuid().ToString("N"),
|
||||
Category = StoreAndForwardCategory.Notification,
|
||||
Target = "Operators",
|
||||
PayloadJson = "{\"notificationId\":\"abc\"}",
|
||||
RetryCount = 4,
|
||||
MaxRetries = 0,
|
||||
RetryIntervalMs = 30000,
|
||||
CreatedAt = DateTimeOffset.UtcNow,
|
||||
LastAttemptAt = DateTimeOffset.UtcNow,
|
||||
Status = StoreAndForwardMessageStatus.Parked,
|
||||
LastError = "central rejected",
|
||||
OriginInstanceName = "Plant.Pump3",
|
||||
ExecutionId = Guid.NewGuid(),
|
||||
SourceScript = "ScriptActor:MonitorSpeed",
|
||||
ParentExecutionId = Guid.NewGuid(),
|
||||
};
|
||||
var original = new ReplicationOperation(ReplicationOperationType.Park, message.Id, message);
|
||||
|
||||
var back = RoundTrip(original);
|
||||
|
||||
Assert.Equal(original.OperationType, back.OperationType);
|
||||
Assert.Equal(original.MessageId, back.MessageId);
|
||||
Assert.NotNull(back.Message);
|
||||
var m = back.Message!;
|
||||
Assert.Equal(message.Id, m.Id);
|
||||
Assert.Equal(message.Category, m.Category);
|
||||
Assert.Equal(message.Target, m.Target);
|
||||
Assert.Equal(message.PayloadJson, m.PayloadJson);
|
||||
Assert.Equal(message.RetryCount, m.RetryCount);
|
||||
Assert.Equal(message.MaxRetries, m.MaxRetries);
|
||||
Assert.Equal(message.RetryIntervalMs, m.RetryIntervalMs);
|
||||
Assert.Equal(message.CreatedAt, m.CreatedAt);
|
||||
Assert.Equal(message.LastAttemptAt, m.LastAttemptAt);
|
||||
Assert.Equal(message.Status, m.Status);
|
||||
Assert.Equal(message.LastError, m.LastError);
|
||||
Assert.Equal(message.OriginInstanceName, m.OriginInstanceName);
|
||||
Assert.Equal(message.ExecutionId, m.ExecutionId);
|
||||
Assert.Equal(message.SourceScript, m.SourceScript);
|
||||
Assert.Equal(message.ParentExecutionId, m.ParentExecutionId);
|
||||
}
|
||||
|
||||
// Type-identity pins: the reflective-JSON wire embeds CLR type manifests, so a
|
||||
// rename/move of either type silently breaks standby replication across a
|
||||
// rolling upgrade. If one fails, you are making a wire-breaking change.
|
||||
[Theory]
|
||||
[InlineData(typeof(ReplicationOperation), "ZB.MOM.WW.ScadaBridge.StoreAndForward.ReplicationOperation")]
|
||||
[InlineData(typeof(StoreAndForwardMessage), "ZB.MOM.WW.ScadaBridge.StoreAndForward.StoreAndForwardMessage")]
|
||||
public void ReplicationContract_TypeIdentity_IsPinned(Type type, string expectedFullName) =>
|
||||
Assert.Equal(expectedFullName, type.FullName);
|
||||
}
|
||||
Reference in New Issue
Block a user