33b15f10a4
Introduce ICentralTransport as the site->central choke point inside SiteCommunicationActor. The seven site->central sends (notification submit/ status, audit + cached-telemetry ingest, reconcile, health, heartbeat) now delegate to an injected transport instead of owning ClusterClient.Send inline. - AkkaCentralTransport: verbatim extraction of today's ClusterClient.Send path, including the exact sender-forwarding that routes central's reply straight back to the waiting Ask. Default when no transport is injected -> behaviour unchanged, existing SiteCommunicationActorTests pass as-is. - GrpcCentralTransport + CentralChannelProvider: dial CentralControlService with sticky failover + background failback (1s-doubling-cap-60s), PSK + x-scadabridge-site via ControlPlaneCredentials, per-call deadlines mirroring today's Ask timeouts. Cross-node retry ONLY on provably-unsent connect failures; never on DeadlineExceeded. Heartbeat stays fire-and-forget. - StaticSitePskProvider: site's single own-key provider (fail-closed). - CommunicationOptions: CentralTransport flag (default Akka) + CentralGrpcEndpoints (validator: required when transport=Grpc). Host selects the impl; the ClusterClient is created only on the Akka path. Tests: actor-with-fake-transport (7 delegations + fault routing + heartbeat no-fault), AkkaCentralTransport sender-forwarding, GrpcCentralTransport over in-process TestServer (failover flip, sticky, failback, PSK+header, deadline, no-retry-on-deadline), validator. Communication.Tests 371 green, Host.Tests 391 green; the three above-seam suites pass unmodified.
161 lines
5.9 KiB
C#
161 lines
5.9 KiB
C#
using Microsoft.Extensions.Options;
|
|
|
|
namespace ZB.MOM.WW.ScadaBridge.Communication.Tests;
|
|
|
|
/// <summary>
|
|
/// Regression: <see cref="CommunicationOptions"/> timeouts feed per-pattern Ask
|
|
/// deadlines and gRPC keepalive/stream-lifetime settings; a zero/negative value
|
|
/// (or a non-positive <c>GrpcMaxConcurrentStreams</c>) must be rejected at startup
|
|
/// by an <see cref="IValidateOptions{TOptions}"/> with a clear, key-naming message
|
|
/// rather than surfacing at first Ask/gRPC use.
|
|
/// </summary>
|
|
public class CommunicationOptionsValidatorTests
|
|
{
|
|
private static ValidateOptionsResult Validate(CommunicationOptions options) =>
|
|
new CommunicationOptionsValidator().Validate(Options.DefaultName, options);
|
|
|
|
[Fact]
|
|
public void DefaultOptions_AreValid()
|
|
{
|
|
var result = Validate(new CommunicationOptions());
|
|
Assert.True(result.Succeeded, result.FailureMessage);
|
|
}
|
|
|
|
[Fact]
|
|
public void ZeroDeploymentTimeout_IsRejected()
|
|
{
|
|
var result = Validate(new CommunicationOptions { DeploymentTimeout = TimeSpan.Zero });
|
|
Assert.True(result.Failed);
|
|
Assert.Contains("DeploymentTimeout", result.FailureMessage);
|
|
}
|
|
|
|
[Fact]
|
|
public void NonPositiveGrpcMaxConcurrentStreams_IsRejected()
|
|
{
|
|
var result = Validate(new CommunicationOptions { GrpcMaxConcurrentStreams = 0 });
|
|
Assert.True(result.Failed);
|
|
Assert.Contains("GrpcMaxConcurrentStreams", result.FailureMessage);
|
|
}
|
|
|
|
// ── Aggregated live alarm cache options (plan #10, Task 6) ───────────────────
|
|
|
|
[Fact]
|
|
public void ZeroLiveAlarmCacheLinger_IsValid()
|
|
{
|
|
// Zero linger = stop the aggregator immediately when the last viewer leaves.
|
|
var result = Validate(new CommunicationOptions { LiveAlarmCacheLinger = TimeSpan.Zero });
|
|
Assert.True(result.Succeeded, result.FailureMessage);
|
|
}
|
|
|
|
[Fact]
|
|
public void NegativeLiveAlarmCacheLinger_IsRejected()
|
|
{
|
|
var result = Validate(new CommunicationOptions { LiveAlarmCacheLinger = TimeSpan.FromSeconds(-1) });
|
|
Assert.True(result.Failed);
|
|
Assert.Contains("LiveAlarmCacheLinger", result.FailureMessage);
|
|
}
|
|
|
|
[Fact]
|
|
public void ZeroLiveAlarmCacheReconcileInterval_IsRejected()
|
|
{
|
|
var result = Validate(new CommunicationOptions { LiveAlarmCacheReconcileInterval = TimeSpan.Zero });
|
|
Assert.True(result.Failed);
|
|
Assert.Contains("LiveAlarmCacheReconcileInterval", result.FailureMessage);
|
|
}
|
|
|
|
[Fact]
|
|
public void ZeroLiveAlarmCacheSeedConcurrency_IsRejected()
|
|
{
|
|
var result = Validate(new CommunicationOptions { LiveAlarmCacheSeedConcurrency = 0 });
|
|
Assert.True(result.Failed);
|
|
Assert.Contains("LiveAlarmCacheSeedConcurrency", result.FailureMessage);
|
|
}
|
|
|
|
[Fact]
|
|
public void ExcessiveLiveAlarmCacheSeedConcurrency_IsRejected()
|
|
{
|
|
var result = Validate(new CommunicationOptions { LiveAlarmCacheSeedConcurrency = 65 });
|
|
Assert.True(result.Failed);
|
|
Assert.Contains("LiveAlarmCacheSeedConcurrency", result.FailureMessage);
|
|
}
|
|
|
|
[Fact]
|
|
public void ZeroLiveAlarmCacheMaxSubscribersPerSite_IsRejected()
|
|
{
|
|
var result = Validate(new CommunicationOptions { LiveAlarmCacheMaxSubscribersPerSite = 0 });
|
|
Assert.True(result.Failed);
|
|
Assert.Contains("LiveAlarmCacheMaxSubscribersPerSite", result.FailureMessage);
|
|
}
|
|
|
|
// ── R2 T10: live-delta publish-coalescing window (N6) ────────────────────────
|
|
|
|
[Fact]
|
|
public void ZeroLiveAlarmCachePublishCoalesce_IsValid()
|
|
{
|
|
// Zero = publish per delta (legacy behavior).
|
|
var result = Validate(new CommunicationOptions { LiveAlarmCachePublishCoalesce = TimeSpan.Zero });
|
|
Assert.True(result.Succeeded, result.FailureMessage);
|
|
}
|
|
|
|
[Fact]
|
|
public void NegativeLiveAlarmCachePublishCoalesce_IsRejected()
|
|
{
|
|
var result = Validate(new CommunicationOptions { LiveAlarmCachePublishCoalesce = TimeSpan.FromMilliseconds(-1) });
|
|
Assert.True(result.Failed);
|
|
Assert.Contains("LiveAlarmCachePublishCoalesce", result.FailureMessage);
|
|
}
|
|
|
|
// ── T1A.3: gRPC central transport endpoints (required only when selected) ────
|
|
|
|
[Fact]
|
|
public void GrpcTransport_WithNoEndpoints_IsRejected()
|
|
{
|
|
var result = Validate(new CommunicationOptions
|
|
{
|
|
CentralTransport = CentralTransportMode.Grpc,
|
|
CentralGrpcEndpoints = new List<string>(),
|
|
});
|
|
Assert.True(result.Failed);
|
|
Assert.Contains("CentralGrpcEndpoints", result.FailureMessage);
|
|
}
|
|
|
|
[Fact]
|
|
public void GrpcTransport_WithBlankEndpoint_IsRejected()
|
|
{
|
|
var result = Validate(new CommunicationOptions
|
|
{
|
|
CentralTransport = CentralTransportMode.Grpc,
|
|
CentralGrpcEndpoints = new List<string> { " " },
|
|
});
|
|
Assert.True(result.Failed);
|
|
Assert.Contains("CentralGrpcEndpoints", result.FailureMessage);
|
|
}
|
|
|
|
[Fact]
|
|
public void GrpcTransport_WithEndpoints_IsValid()
|
|
{
|
|
var result = Validate(new CommunicationOptions
|
|
{
|
|
CentralTransport = CentralTransportMode.Grpc,
|
|
CentralGrpcEndpoints = new List<string>
|
|
{
|
|
"http://scadabridge-central-a:8083",
|
|
"http://scadabridge-central-b:8083",
|
|
},
|
|
});
|
|
Assert.True(result.Succeeded, result.FailureMessage);
|
|
}
|
|
|
|
[Fact]
|
|
public void AkkaTransport_IgnoresMissingGrpcEndpoints()
|
|
{
|
|
// The default transport must not be forced to declare gRPC endpoints it never dials.
|
|
var result = Validate(new CommunicationOptions
|
|
{
|
|
CentralTransport = CentralTransportMode.Akka,
|
|
CentralGrpcEndpoints = new List<string>(),
|
|
});
|
|
Assert.True(result.Succeeded, result.FailureMessage);
|
|
}
|
|
}
|