using Grpc.Core;
using ZB.MOM.WW.Secrets.Abstractions;
namespace ZB.MOM.WW.Secrets.Replicator.Grpc.Tests;
///
/// Rotation, stickiness and containment behaviour of the multi-endpoint hub reader. Exercised
/// against scripted fakes: the per-endpoint wire behaviour (headers,
/// deadlines, parse isolation) belongs to GrpcSecretsHubClient and is tested there — this
/// suite is only about which endpoint gets asked, in what order, and what escapes when they all fail.
///
public sealed class FailoverSecretsHubReaderTests
{
private const string PrimaryEndpoint = "https://central-a:5001";
private const string FallbackEndpoint = "https://central-b:5001";
private readonly ScriptedHubReader _primary = new();
private readonly ScriptedHubReader _fallback = new();
private readonly RecordingLogger _logger = new();
private FailoverSecretsHubReader CreateReader() => new(
[_primary, _fallback],
[PrimaryEndpoint, FallbackEndpoint],
_logger);
private static RpcException Unavailable() => new(new Status(
StatusCode.Unavailable, "connection refused"));
[Fact]
public async Task A_healthy_primary_is_the_only_endpoint_asked()
{
FailoverSecretsHubReader reader = CreateReader();
await reader.GetManifestAsync(CancellationToken.None);
await reader.GetManyAsync([new SecretName("db/password")], CancellationToken.None);
Assert.Equal(1, _primary.ManifestCalls);
Assert.Equal(1, _primary.ManyCalls);
Assert.Equal(0, _fallback.ManifestCalls);
Assert.Equal(0, _fallback.ManyCalls);
Assert.Equal(0, _logger.WarningCount);
}
[Fact]
public async Task A_transport_fault_on_the_primary_fails_over_to_the_fallback()
{
_primary.FaultWith = Unavailable;
_fallback.Manifest = [Entry("db/password")];
IReadOnlyList manifest =
await CreateReader().GetManifestAsync(CancellationToken.None);
Assert.Single(manifest);
Assert.Equal(1, _primary.ManifestCalls);
Assert.Equal(1, _fallback.ManifestCalls);
}
[Fact]
public async Task The_failover_warning_names_both_endpoints()
{
_primary.FaultWith = Unavailable;
await CreateReader().GetManifestAsync(CancellationToken.None);
string warning = Assert.Single(_logger.Warnings);
Assert.Contains(PrimaryEndpoint, warning, StringComparison.Ordinal);
Assert.Contains(FallbackEndpoint, warning, StringComparison.Ordinal);
}
[Fact]
public async Task GetMany_fails_over_too()
{
_primary.FaultWith = Unavailable;
_fallback.Rows = [Row("db/password")];
IReadOnlyList rows = await CreateReader()
.GetManyAsync([new SecretName("db/password")], CancellationToken.None);
Assert.Single(rows);
Assert.Equal(1, _primary.ManyCalls);
Assert.Equal(1, _fallback.ManyCalls);
}
[Fact]
public async Task The_endpoint_that_answered_is_sticky_for_the_next_call()
{
// A sweep is one GetManifest plus N GetSecrets batches. Without stickiness, every call
// after a primary outage would re-pay the primary's full CallDeadline of latency before
// reaching the fallback that is actually answering.
_primary.FaultWith = Unavailable;
FailoverSecretsHubReader reader = CreateReader();
await reader.GetManifestAsync(CancellationToken.None);
await reader.GetManyAsync([new SecretName("db/password")], CancellationToken.None);
Assert.Equal(1, _primary.ManifestCalls);
Assert.Equal(0, _primary.ManyCalls);
Assert.Equal(1, _fallback.ManyCalls);
}
[Fact]
public async Task Rotation_wraps_back_to_a_recovered_primary_when_the_fallback_fails_in_turn()
{
_primary.FaultWith = Unavailable;
FailoverSecretsHubReader reader = CreateReader();
// Sticks to the fallback…
await reader.GetManifestAsync(CancellationToken.None);
// …then the fallback dies and the primary has recovered.
_primary.FaultWith = null;
_fallback.FaultWith = Unavailable;
await reader.GetManifestAsync(CancellationToken.None);
Assert.Equal(2, _primary.ManifestCalls);
Assert.Equal(2, _fallback.ManifestCalls);
}
[Fact]
public async Task When_every_endpoint_fails_the_last_exception_is_rethrown()
{
var primaryFault = new RpcException(new Status(StatusCode.Unavailable, "primary down"));
var fallbackFault = new RpcException(new Status(StatusCode.DeadlineExceeded, "fallback down"));
_primary.FaultWith = () => primaryFault;
_fallback.FaultWith = () => fallbackFault;
RpcException thrown = await Assert.ThrowsAsync(
() => CreateReader().GetManifestAsync(CancellationToken.None));
// The LAST endpoint tried, so the sweep's catch-all logs the freshest failure rather than
// one from an endpoint a later attempt already moved past.
Assert.Same(fallbackFault, thrown);
}
[Fact]
public async Task Cancellation_propagates_immediately_without_failing_over()
{
// Cancellation is the host stopping, not an endpoint fault — dialing the fallback with an
// already-cancelled token would only delay shutdown by another deadline.
_primary.FaultWith = () => new OperationCanceledException();
await Assert.ThrowsAsync(
() => CreateReader().GetManifestAsync(CancellationToken.None));
Assert.Equal(0, _fallback.ManifestCalls);
Assert.Equal(0, _logger.WarningCount);
}
[Fact]
public async Task A_non_transport_exception_propagates_immediately_without_failing_over()
{
_primary.FaultWith = () => new InvalidOperationException("a defect, not an outage");
await Assert.ThrowsAsync(
() => CreateReader().GetManifestAsync(CancellationToken.None));
Assert.Equal(0, _fallback.ManifestCalls);
}
[Fact]
public async Task No_failover_log_message_carries_token_material()
{
// The warning's whole content is the two endpoint URIs (not secrets) and the gRPC status
// code. Even a status detail that happens to quote a credential must not reach the message.
const string leak = "s3cr3t-bearer-token";
_primary.FaultWith = () => new RpcException(new Status(
StatusCode.Unauthenticated, $"Bearer {leak} rejected"));
await CreateReader().GetManifestAsync(CancellationToken.None);
foreach (string warning in _logger.Warnings)
{
Assert.DoesNotContain(leak, warning, StringComparison.OrdinalIgnoreCase);
}
Assert.Equal(1, _logger.WarningCount);
// The exception object must not ride along either: sinks render an attached exception's
// ToString(), which for RpcException includes Status.Detail — the very text the message
// template deliberately excludes. A clean message with the exception attached would still
// leak in production, so the warning must carry NO exception at all.
Assert.All(_logger.WarningExceptions, Assert.Null);
}
[Fact]
public async Task Concurrent_calls_share_the_sticky_index_without_tearing()
{
// The sticky index is plain shared state written from every call path; this pins that
// parallel sweeps (manifest + batches overlapping) never corrupt it or deadlock.
_primary.FaultWith = Unavailable;
FailoverSecretsHubReader reader = CreateReader();
await Task.WhenAll(Enumerable.Range(0, 32).Select(async _ =>
{
IReadOnlyList manifest =
await reader.GetManifestAsync(CancellationToken.None);
Assert.NotNull(manifest);
}));
// Every one of the 32 calls ended on the answering fallback.
Assert.Equal(32, _fallback.ManifestCalls);
// And the index is now stuck there: one more call never touches the primary.
int primaryCallsBefore = _primary.ManifestCalls;
await reader.GetManifestAsync(CancellationToken.None);
Assert.Equal(primaryCallsBefore, _primary.ManifestCalls);
}
private static SecretManifestEntry Entry(string name) => new()
{
Name = new SecretName(name),
Revision = 0,
UpdatedUtc = TestWire.FixedInstant,
};
private static StoredSecret Row(string name) => new()
{
Name = new SecretName(name),
ContentType = SecretContentType.Text,
Ciphertext = [0xAA],
Nonce = [1],
Tag = [2],
WrappedDek = [3],
WrapNonce = [4],
WrapTag = [5],
KekId = "sha256:test",
Revision = 0,
CreatedUtc = TestWire.FixedInstant,
UpdatedUtc = TestWire.FixedInstant,
};
///
/// An standing in for one endpoint's client: counts calls,
/// answers from canned data, or faults with whatever a test scripts.
///
private sealed class ScriptedHubReader : ISecretsHubReader
{
/// Number of GetManifest calls received.
public int ManifestCalls;
/// Number of GetMany calls received.
public int ManyCalls;
/// When set, every call throws this instead of answering.
public Func? FaultWith { get; set; }
/// Canned manifest returned when not faulted.
public IReadOnlyList Manifest { get; set; } = [];
/// Canned rows returned when not faulted.
public IReadOnlyList Rows { get; set; } = [];
///
public Task> GetManifestAsync(CancellationToken ct)
{
Interlocked.Increment(ref ManifestCalls);
return FaultWith is { } fault
? Task.FromException>(fault())
: Task.FromResult(Manifest);
}
///
public Task> GetManyAsync(
IReadOnlyList names, CancellationToken ct)
{
Interlocked.Increment(ref ManyCalls);
return FaultWith is { } fault
? Task.FromException>(fault())
: Task.FromResult(Rows);
}
}
}