using System;
using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;
using Google.Protobuf.Collections;
using ZB.MOM.WW.MxGateway.Contracts.Proto;
using ZB.MOM.WW.MxGateway.Worker.MxAccess;
namespace ZB.MOM.WW.MxGateway.Worker.Tests.MxAccess;
///
/// Unit tests for . The cache is
/// consumed by the write command executor's bounded pump-wait so a
/// WriteSecured/WriteSecured2 reply can carry the correlated
/// OnWriteComplete outcome; its version-baseline contract is exercised in
/// isolation here before the STA / COM plumbing gets layered on top.
///
public sealed class MxAccessWriteCompletionCacheTests
{
/// Verifies that Record bumps the version per key and keys stay isolated.
[Fact]
public void Record_IncrementsVersionPerKey()
{
MxAccessWriteCompletionCache cache = new();
Assert.Equal(0UL, cache.CurrentVersion(7, 21));
cache.Record(7, 21, BuildStatuses(detail: 100));
Assert.Equal(1UL, cache.CurrentVersion(7, 21));
cache.Record(7, 21, BuildStatuses(detail: 200));
Assert.Equal(2UL, cache.CurrentVersion(7, 21));
cache.Record(7, 22, BuildStatuses(detail: 300));
Assert.Equal(1UL, cache.CurrentVersion(7, 22));
Assert.Equal(2UL, cache.CurrentVersion(7, 21));
}
/// Verifies that a completion newer than the baseline is returned with its status rows.
[Fact]
public void TryWaitForCompletion_WhenCompletionNewerThanBaseline_ReturnsStatuses()
{
MxAccessWriteCompletionCache cache = new();
cache.Record(7, 21, BuildStatuses(detail: 4321));
bool found = cache.TryWaitForCompletion(
7,
21,
sinceVersion: 0UL,
deadlineUtc: DateTime.UtcNow.AddSeconds(5),
pumpStep: static () => { },
out RepeatedField statuses);
Assert.True(found);
MxStatusProxy row = Assert.Single(statuses);
Assert.Equal(4321, row.Detail);
Assert.Equal(MxStatusCategory.Ok, row.Category);
}
///
/// Verifies that a completion recorded before the baseline was captured is
/// never misattributed to the waiting write: only a strictly newer version
/// satisfies the wait, so a stale row times the wait out.
///
[Fact]
public void TryWaitForCompletion_WhenOnlyStaleCompletion_TimesOut()
{
MxAccessWriteCompletionCache cache = new();
cache.Record(7, 21, BuildStatuses(detail: 4321));
ulong baseline = cache.CurrentVersion(7, 21);
bool found = cache.TryWaitForCompletion(
7,
21,
sinceVersion: baseline,
deadlineUtc: DateTime.UtcNow.AddMilliseconds(50),
pumpStep: static () => { },
out RepeatedField statuses);
Assert.False(found);
Assert.Empty(statuses);
}
///
/// Verifies the pump loop is what lets a completion land: the completion is
/// recorded from inside a later pump step (standing in for the STA
/// dispatching the OnWriteComplete message) and the wait then succeeds.
///
[Fact]
public void TryWaitForCompletion_InvokesPumpStepEachIteration()
{
MxAccessWriteCompletionCache cache = new();
int pumpCalls = 0;
bool found = cache.TryWaitForCompletion(
7,
21,
sinceVersion: 0UL,
deadlineUtc: DateTime.UtcNow.AddSeconds(5),
pumpStep: () =>
{
pumpCalls++;
if (pumpCalls == 2)
{
cache.Record(7, 21, BuildStatuses(detail: 55));
}
},
out RepeatedField statuses);
Assert.True(found);
Assert.True(pumpCalls >= 2);
Assert.Equal(55, Assert.Single(statuses).Detail);
}
///
/// Verifies the wait slice stays bounded no matter what poll interval the
/// caller asks for: with a 10 s interval and a 3 s deadline, a completion
/// recorded 50 ms in is still returned in time, because every slice is
/// capped at the 50 ms fallback tick. The blind Thread.Sleep this
/// replaced would have slept the full interval — on the STA, 10 s with no
/// Windows messages dispatched — and then reported a timeout.
///
///
/// Scope: this proves the cadence bound, not the signal path. The clamp
/// alone would satisfy it, so it stays green if the Record-side
/// Set() is deleted. StaWaitHelperTests owns the signal
/// path, where a five-second wait with no clamp in play makes the handle
/// the only thing that can end it early.
///
/// A task that represents the asynchronous operation.
[Fact]
public async Task TryWaitForCompletion_CompletesWithinDeadline_WhenPollIntervalExceedsTheFallbackTick()
{
MxAccessWriteCompletionCache cache = new();
using ManualResetEventSlim waitEntered = new(false);
Task recorder = Task.Run(async () =>
{
// Handshake so the completion cannot land before the wait starts —
// otherwise the first check would satisfy it and prove nothing.
waitEntered.Wait(TimeSpan.FromSeconds(5));
await Task.Delay(50, CancellationToken.None);
cache.Record(7, 21, BuildStatuses(detail: 8080));
});
Stopwatch elapsed = Stopwatch.StartNew();
bool found = cache.TryWaitForCompletion(
7,
21,
sinceVersion: 0UL,
deadlineUtc: DateTime.UtcNow.AddSeconds(3),
pumpStep: () => waitEntered.Set(),
out RepeatedField statuses,
pollIntervalMs: 10_000);
elapsed.Stop();
await recorder;
Assert.True(found);
Assert.Equal(8080, Assert.Single(statuses).Detail);
Assert.True(
elapsed.Elapsed < TimeSpan.FromSeconds(2),
$"The wait should have re-checked within the fallback tick, not slept out the poll interval; took {elapsed.ElapsedMilliseconds} ms.");
}
/// Verifies that Record stores an independent clone of the caller's rows.
[Fact]
public void Record_ClonesStatuses()
{
MxAccessWriteCompletionCache cache = new();
RepeatedField callerRows = BuildStatuses(detail: 77);
cache.Record(7, 21, callerRows);
callerRows[0].Detail = 999;
callerRows.Add(new MxStatusProxy());
Assert.True(cache.TryWaitForCompletion(
7,
21,
sinceVersion: 0UL,
deadlineUtc: DateTime.UtcNow.AddSeconds(5),
pumpStep: static () => { },
out RepeatedField statuses));
Assert.Equal(77, Assert.Single(statuses).Detail);
}
/// Verifies that Remove forgets only the given handle pair.
[Fact]
public void Remove_ForgetsOnlyThatHandlePair()
{
MxAccessWriteCompletionCache cache = new();
cache.Record(7, 21, BuildStatuses(detail: 1));
cache.Record(7, 22, BuildStatuses(detail: 2));
cache.Remove(7, 21);
cache.Remove(7, 99); // unknown pair is a no-op
Assert.Equal(1, cache.Count);
Assert.Equal(0UL, cache.CurrentVersion(7, 21));
Assert.Equal(1UL, cache.CurrentVersion(7, 22));
}
/// Verifies that RemoveServer forgets every pair of that server and no other.
[Fact]
public void RemoveServer_ForgetsOnlyThatServersPairs()
{
MxAccessWriteCompletionCache cache = new();
cache.Record(7, 21, BuildStatuses(detail: 1));
cache.Record(7, -5, BuildStatuses(detail: 2));
cache.Record(8, 21, BuildStatuses(detail: 3));
cache.Record(-7, 21, BuildStatuses(detail: 4));
cache.RemoveServer(7);
Assert.Equal(2, cache.Count);
Assert.Equal(1UL, cache.CurrentVersion(8, 21));
Assert.Equal(1UL, cache.CurrentVersion(-7, 21));
}
private static RepeatedField BuildStatuses(int detail)
{
return new RepeatedField
{
new MxStatusProxy
{
Success = 1,
Category = MxStatusCategory.Ok,
Detail = detail,
},
};
}
}