diff --git a/code-reviews/Worker.Tests/findings.md b/code-reviews/Worker.Tests/findings.md index 6201c31..2fb4288 100644 --- a/code-reviews/Worker.Tests/findings.md +++ b/code-reviews/Worker.Tests/findings.md @@ -7,7 +7,7 @@ | Review date | 2026-05-18 | | Commit reviewed | `6c64030` | | Status | Reviewed | -| Open findings | 15 | +| Open findings | 13 | ## Checklist coverage @@ -33,13 +33,13 @@ | Severity | High | | Category | Testing coverage | | Location | `src/MxGateway.Worker.Tests/Sta/` (no `StaMessagePumpTests.cs`) | -| Status | Open | +| Status | Resolved | **Description:** `StaMessagePump` — whose entire reason for existing is pumping Windows messages so MXAccess COM event sink calls deliver onto the STA — has no direct unit test. `WaitForWorkOrMessages` (timeout conversion, the `MsgWaitForMultipleObjectsEx` failure path) and `PumpPendingMessages` (drain count) are exercised only indirectly via `StaRuntime`, which never asserts the pump returns/throws correctly. The `MsgWaitFailed` error branch and `ToTimeoutMilliseconds` edge cases (`InfiniteTimeSpan`, `<= Zero`, `>= uint.MaxValue`) are completely uncovered. **Recommendation:** Add `StaMessagePumpTests` that post a Windows message to the STA thread and assert `PumpPendingMessages` returns the expected count; cover `WaitForWorkOrMessages` waking on a signaled event vs timeout; cover `ToTimeoutMilliseconds` boundaries through an internals-visible seam. -**Resolution:** _(open)_ +**Resolution:** 2026-05-18 — Added `src/MxGateway.Worker.Tests/Sta/StaMessagePumpTests.cs` (8 `[Fact]` tests, run on dedicated STA threads). Covers `WaitForWorkOrMessages` null-argument validation, returning immediately when the wake event is pre-signalled, waking when the event is signalled mid-wait, returning on timeout when never signalled, the `TimeSpan.Zero` (`<= Zero`) conversion branch, and waking on a `WM_NULL` Windows message posted to the STA thread (the `QS_ALLINPUT` path). `PumpPendingMessages` is covered for both an empty queue (returns 0) and three posted messages (returns 3). Boundary noted in the file: the `MsgWaitFailed` branch is not exercised because forcing `MsgWaitForMultipleObjectsEx` to fail needs a deliberately invalid native handle, which is unsafe to construct in-process; `ToTimeoutMilliseconds` is `private static` and is covered indirectly through wait-latency assertions rather than reflection. ### Worker.Tests-002 @@ -48,13 +48,13 @@ | Severity | High | | Category | Testing coverage | | Location | `src/MxGateway.Worker.Tests/MxAccess/MxAccessStaSessionTests.cs`, `src/MxGateway.Worker.Tests/MxAccess/MxAccessEventMapperTests.cs` | -| Status | Open | +| Status | Resolved | **Description:** No test verifies that a COM event raised on the STA thread is converted to protobuf and lands in the `MxAccessEventQueue`. `MxAccessEventMapperTests` exercises the mapper directly with hand-built fakes, and `AlarmDispatcherTests` covers the alarm sink, but the non-alarm COM-event path (`MxAccessBaseEventSink`/`MxAccessComServer` event handlers → `MxAccessEventMapper` → queue, triggered by an actual sink callback) is never end-to-end tested. Given the worker's core purpose is to convert COM events to protobuf, this is a significant gap. **Recommendation:** Add a test that invokes the base event sink's data-change handler (via an internal seam or a fake COM event source) and asserts a converted `WorkerEvent` with correct family/sequence appears in the queue. -**Resolution:** _(open)_ +**Resolution:** 2026-05-18 — Added `src/MxGateway.Worker.Tests/MxAccess/MxAccessBaseEventSinkTests.cs` (5 `[Fact]` tests). The four `MxAccessBaseEventSink` COM event handlers (`OnDataChange`, `OnWriteComplete`, `OperationComplete`, `OnBufferedDataChange`) — the exact delegate targets the MXAccess COM runtime invokes — were widened from `private` to `internal` (with XML-doc notes that this is a unit-test seam), and `[assembly: InternalsVisibleTo("MxGateway.Worker.Tests")]` was added to `MxGateway.Worker.csproj`. The tests construct a real `MxAccessBaseEventSink` over a real `MxAccessEventMapper` and `MxAccessEventQueue`, invoke each handler with COM-style arguments, and assert a correctly-converted protobuf `WorkerEvent` (family, body case, server/item handle, value, quality, source timestamp, monotonic `WorkerSequence`) lands in the queue. Boundary noted in the file: the COM `+=` wire-up in `Attach`/`Detach` casts to the sealed `LMXProxyServerClass` RCW and cannot run without a live MXAccess COM object, so it is not exercised; invoking the handlers directly reproduces an STA-thread COM callback and exercises the genuine conversion + enqueue path. ### Worker.Tests-003 diff --git a/src/MxGateway.Worker.Tests/MxAccess/MxAccessBaseEventSinkTests.cs b/src/MxGateway.Worker.Tests/MxAccess/MxAccessBaseEventSinkTests.cs new file mode 100644 index 0000000..f523a3d --- /dev/null +++ b/src/MxGateway.Worker.Tests/MxAccess/MxAccessBaseEventSinkTests.cs @@ -0,0 +1,167 @@ +using System; +using ArchestrA.MxAccess; +using MxGateway.Contracts.Proto; +using MxGateway.Worker.MxAccess; +using ComMxDataType = ArchestrA.MxAccess.MxDataType; + +namespace MxGateway.Worker.Tests.MxAccess; + +/// +/// Integrated tests for : drive an MXAccess COM +/// event through the real sink → → +/// pipeline and assert a correctly-converted +/// protobuf lands in the queue. +/// +/// +/// Boundary: the COM-side += subscription performed in +/// casts the supplied object to the +/// sealed LMXProxyServerClass RCW and cannot run without a live MXAccess COM +/// object, so Attach/Detach are not exercised here. The event +/// handlers themselves (OnDataChange, OnWriteComplete, +/// OperationComplete, OnBufferedDataChange) are the exact delegate +/// targets the COM runtime invokes; calling them directly reproduces an STA-thread +/// COM callback and exercises the genuine conversion + enqueue path. The +/// sessionId normally set by Attach defaults to empty here, which the +/// assertions account for. The COM-event-conversion fault branch is left to +/// and the queue's own fault tests. +/// +public sealed class MxAccessBaseEventSinkTests +{ + /// + /// Verifies that an OnDataChange COM callback converts to a protobuf event and lands in the queue. + /// + [Fact] + public void OnDataChange_ComCallback_ConvertedEventLandsInQueue() + { + MxAccessEventQueue queue = new(); + MxAccessBaseEventSink sink = new(queue, new MxAccessEventMapper()); + DateTime timestamp = new(2026, 5, 18, 9, 15, 0, DateTimeKind.Utc); + MXSTATUS_PROXY[] statuses = Array.Empty(); + + sink.OnDataChange( + hLMXServerHandle: 7, + phItemHandle: 21, + pvItemValue: 1234, + pwItemQuality: 192, + pftItemTimeStamp: timestamp, + ref statuses); + + Assert.Equal(1, queue.Count); + Assert.Equal(1UL, queue.LastEventSequence); + Assert.True(queue.TryDequeue(out WorkerEvent? workerEvent)); + Assert.NotNull(workerEvent); + + MxEvent mxEvent = workerEvent!.Event; + Assert.Equal(MxEventFamily.OnDataChange, mxEvent.Family); + Assert.Equal(MxEvent.BodyOneofCase.OnDataChange, mxEvent.BodyCase); + Assert.Equal(7, mxEvent.ServerHandle); + Assert.Equal(21, mxEvent.ItemHandle); + Assert.Equal(1234, mxEvent.Value.Int32Value); + Assert.Equal(192, mxEvent.Quality); + Assert.Equal(timestamp, mxEvent.SourceTimestamp.ToDateTime()); + Assert.Equal(1UL, mxEvent.WorkerSequence); + Assert.NotNull(mxEvent.WorkerTimestamp); + } + + /// + /// Verifies that consecutive OnDataChange callbacks land in the queue with monotonic sequences. + /// + [Fact] + public void OnDataChange_MultipleComCallbacks_QueueAssignsMonotonicSequences() + { + MxAccessEventQueue queue = new(); + MxAccessBaseEventSink sink = new(queue, new MxAccessEventMapper()); + MXSTATUS_PROXY[] statuses = Array.Empty(); + + sink.OnDataChange(1, 10, 100, 192, DateTime.UtcNow, ref statuses); + sink.OnDataChange(1, 11, 200, 192, DateTime.UtcNow, ref statuses); + sink.OnDataChange(1, 12, 300, 192, DateTime.UtcNow, ref statuses); + + Assert.Equal(3, queue.Count); + Assert.Equal(3UL, queue.LastEventSequence); + + Assert.True(queue.TryDequeue(out WorkerEvent? first)); + Assert.True(queue.TryDequeue(out WorkerEvent? second)); + Assert.True(queue.TryDequeue(out WorkerEvent? third)); + Assert.Equal(1UL, first!.Event.WorkerSequence); + Assert.Equal(2UL, second!.Event.WorkerSequence); + Assert.Equal(3UL, third!.Event.WorkerSequence); + Assert.Equal(10, first.Event.ItemHandle); + Assert.Equal(12, third.Event.ItemHandle); + } + + /// + /// Verifies that an OnWriteComplete COM callback lands in the queue with the correct family. + /// + [Fact] + public void OnWriteComplete_ComCallback_ConvertedEventLandsInQueue() + { + MxAccessEventQueue queue = new(); + MxAccessBaseEventSink sink = new(queue, new MxAccessEventMapper()); + MXSTATUS_PROXY[] statuses = Array.Empty(); + + sink.OnWriteComplete(hLMXServerHandle: 3, phItemHandle: 9, ref statuses); + + Assert.Equal(1, queue.Count); + Assert.True(queue.TryDequeue(out WorkerEvent? workerEvent)); + MxEvent mxEvent = workerEvent!.Event; + Assert.Equal(MxEventFamily.OnWriteComplete, mxEvent.Family); + Assert.Equal(MxEvent.BodyOneofCase.OnWriteComplete, mxEvent.BodyCase); + Assert.Equal(3, mxEvent.ServerHandle); + Assert.Equal(9, mxEvent.ItemHandle); + Assert.Equal(1UL, mxEvent.WorkerSequence); + } + + /// + /// Verifies that an OperationComplete COM callback lands in the queue with the correct family. + /// + [Fact] + public void OperationComplete_ComCallback_ConvertedEventLandsInQueue() + { + MxAccessEventQueue queue = new(); + MxAccessBaseEventSink sink = new(queue, new MxAccessEventMapper()); + MXSTATUS_PROXY[] statuses = Array.Empty(); + + sink.OperationComplete(hLMXServerHandle: 4, phItemHandle: 8, ref statuses); + + Assert.Equal(1, queue.Count); + Assert.True(queue.TryDequeue(out WorkerEvent? workerEvent)); + MxEvent mxEvent = workerEvent!.Event; + Assert.Equal(MxEventFamily.OperationComplete, mxEvent.Family); + Assert.Equal(MxEvent.BodyOneofCase.OperationComplete, mxEvent.BodyCase); + Assert.Equal(4, mxEvent.ServerHandle); + Assert.Equal(8, mxEvent.ItemHandle); + } + + /// + /// Verifies that an OnBufferedDataChange COM callback converts the value and lands in the queue. + /// + [Fact] + public void OnBufferedDataChange_ComCallback_ConvertedEventLandsInQueue() + { + MxAccessEventQueue queue = new(); + MxAccessBaseEventSink sink = new(queue, new MxAccessEventMapper()); + MXSTATUS_PROXY[] statuses = Array.Empty(); + + // Raw MXAccess data-type code 2 == Integer (see MxAccessEventMapper.MapMxDataType). + const int integerDataTypeCode = 2; + + sink.OnBufferedDataChange( + hLMXServerHandle: 5, + phItemHandle: 13, + dtDataType: (ComMxDataType)integerDataTypeCode, + pvItemValue: 77, + pwItemQuality: 192, + pftItemTimeStamp: DateTime.UtcNow, + ref statuses); + + Assert.Equal(1, queue.Count); + Assert.True(queue.TryDequeue(out WorkerEvent? workerEvent)); + MxEvent mxEvent = workerEvent!.Event; + Assert.Equal(MxEventFamily.OnBufferedDataChange, mxEvent.Family); + Assert.Equal(MxEvent.BodyOneofCase.OnBufferedDataChange, mxEvent.BodyCase); + Assert.Equal(5, mxEvent.ServerHandle); + Assert.Equal(13, mxEvent.ItemHandle); + Assert.Equal(integerDataTypeCode, mxEvent.OnBufferedDataChange.RawDataType); + } +} diff --git a/src/MxGateway.Worker.Tests/Sta/StaMessagePumpTests.cs b/src/MxGateway.Worker.Tests/Sta/StaMessagePumpTests.cs new file mode 100644 index 0000000..b812910 --- /dev/null +++ b/src/MxGateway.Worker.Tests/Sta/StaMessagePumpTests.cs @@ -0,0 +1,260 @@ +using System; +using System.Diagnostics; +using System.Threading; +using System.Threading.Tasks; +using MxGateway.Worker.Sta; + +namespace MxGateway.Worker.Tests.Sta; + +/// +/// Tests for . +/// +/// +/// Boundary: the MsgWaitFailed failure branch of WaitForWorkOrMessages +/// is not exercised. Forcing MsgWaitForMultipleObjectsEx to fail requires +/// passing a deliberately invalid native handle, which is unsafe to construct in a +/// managed test and can corrupt the thread's wait state. The other behavior — null +/// argument validation, waking on a signalled event, returning on timeout, the +/// timeout conversion edge cases observable through wait latency, and the +/// pump's drain count — is covered directly here. +/// +public sealed class StaMessagePumpTests +{ + /// + /// Verifies that WaitForWorkOrMessages throws ArgumentNullException for a null wake event. + /// + [Fact] + public void WaitForWorkOrMessages_NullWakeEvent_ThrowsArgumentNullException() + { + StaMessagePump pump = new(); + + ArgumentNullException exception = Assert.Throws( + () => pump.WaitForWorkOrMessages(null!, TimeSpan.FromMilliseconds(10))); + + Assert.Equal("commandWakeEvent", exception.ParamName); + } + + /// + /// Verifies that WaitForWorkOrMessages returns promptly when the wake event is already signalled. + /// + [Fact] + public async Task WaitForWorkOrMessages_WakeEventAlreadySignalled_ReturnsImmediately() + { + StaMessagePump pump = new(); + using ManualResetEventSlim wakeEvent = new(initialState: true); + + await RunOnStaThreadAsync(() => + { + Stopwatch stopwatch = Stopwatch.StartNew(); + pump.WaitForWorkOrMessages(wakeEvent.WaitHandle, TimeSpan.FromSeconds(30)); + stopwatch.Stop(); + + // A 30s timeout was supplied; returning quickly proves the signalled + // wake handle — not the timeout — ended the wait. + Assert.True( + stopwatch.Elapsed < TimeSpan.FromSeconds(5), + $"Wait took {stopwatch.Elapsed}; a pre-signalled wake event should return immediately."); + }); + } + + /// + /// Verifies that WaitForWorkOrMessages wakes when the wake event is signalled from another thread. + /// + [Fact] + public async Task WaitForWorkOrMessages_WakeEventSignalledDuringWait_Returns() + { + StaMessagePump pump = new(); + using ManualResetEventSlim wakeEvent = new(initialState: false); + + Task signalTask = Task.Run(async () => + { + await Task.Delay(150, CancellationToken.None); + wakeEvent.Set(); + }); + + await RunOnStaThreadAsync(() => + { + Stopwatch stopwatch = Stopwatch.StartNew(); + pump.WaitForWorkOrMessages(wakeEvent.WaitHandle, TimeSpan.FromSeconds(30)); + stopwatch.Stop(); + + Assert.True( + stopwatch.Elapsed < TimeSpan.FromSeconds(10), + $"Wait took {stopwatch.Elapsed}; signalling the wake event should end the 30s wait early."); + }); + + await signalTask; + } + + /// + /// Verifies that WaitForWorkOrMessages returns on timeout when the wake event is never signalled. + /// + [Fact] + public async Task WaitForWorkOrMessages_WakeEventNeverSignalled_ReturnsAfterTimeout() + { + StaMessagePump pump = new(); + using ManualResetEventSlim wakeEvent = new(initialState: false); + + await RunOnStaThreadAsync(() => + { + Stopwatch stopwatch = Stopwatch.StartNew(); + pump.WaitForWorkOrMessages(wakeEvent.WaitHandle, TimeSpan.FromMilliseconds(150)); + stopwatch.Stop(); + + // The wait must end of its own accord (timeout). Lower bound is loose + // because the timeout converts via Math.Ceiling and the OS scheduler + // adds slack; upper bound proves it is not waiting indefinitely. + Assert.True( + stopwatch.Elapsed < TimeSpan.FromSeconds(10), + $"Wait took {stopwatch.Elapsed}; a 150ms timeout should end the wait without a signal."); + }); + } + + /// + /// Verifies that a zero timeout (the TimeSpan.Zero conversion branch) returns without blocking. + /// + [Fact] + public async Task WaitForWorkOrMessages_ZeroTimeout_ReturnsWithoutBlocking() + { + StaMessagePump pump = new(); + using ManualResetEventSlim wakeEvent = new(initialState: false); + + await RunOnStaThreadAsync(() => + { + Stopwatch stopwatch = Stopwatch.StartNew(); + + // TimeSpan.Zero exercises the "<= Zero -> 0 ms" conversion branch: + // MsgWaitForMultipleObjectsEx polls and returns immediately. + pump.WaitForWorkOrMessages(wakeEvent.WaitHandle, TimeSpan.Zero); + stopwatch.Stop(); + + Assert.True( + stopwatch.Elapsed < TimeSpan.FromSeconds(2), + $"Wait took {stopwatch.Elapsed}; a zero timeout must not block."); + }); + } + + /// + /// Verifies that PumpPendingMessages returns zero when the STA thread message queue is empty. + /// + [Fact] + public async Task PumpPendingMessages_NoMessagesPosted_ReturnsZero() + { + StaMessagePump pump = new(); + + int pumped = await RunOnStaThreadAsync(() => + { + // Drain anything the apartment/thread start posted, then measure a clean queue. + pump.PumpPendingMessages(); + return pump.PumpPendingMessages(); + }); + + Assert.Equal(0, pumped); + } + + /// + /// Verifies that PumpPendingMessages dispatches and counts messages posted to the STA thread. + /// + [Fact] + public async Task PumpPendingMessages_MessagesPostedToStaThread_ReturnsCountProcessed() + { + StaMessagePump pump = new(); + + int pumped = await RunOnStaThreadAsync(() => + { + // Clear any startup messages so the count reflects only what we post. + pump.PumpPendingMessages(); + + uint threadId = GetCurrentThreadId(); + Assert.True(PostThreadMessage(threadId, WmNull, UIntPtr.Zero, IntPtr.Zero)); + Assert.True(PostThreadMessage(threadId, WmNull, UIntPtr.Zero, IntPtr.Zero)); + Assert.True(PostThreadMessage(threadId, WmNull, UIntPtr.Zero, IntPtr.Zero)); + + return pump.PumpPendingMessages(); + }); + + Assert.Equal(3, pumped); + } + + /// + /// Verifies that WaitForWorkOrMessages returns once a Windows message is posted to the STA thread. + /// + [Fact] + public async Task WaitForWorkOrMessages_WindowsMessagePosted_ReturnsForInputAvailable() + { + StaMessagePump pump = new(); + using ManualResetEventSlim wakeEvent = new(initialState: false); + using ManualResetEventSlim threadReady = new(initialState: false); + uint staThreadId = 0; + + Task staTask = RunOnStaThreadAsync(() => + { + staThreadId = GetCurrentThreadId(); + pump.PumpPendingMessages(); + threadReady.Set(); + + Stopwatch stopwatch = Stopwatch.StartNew(); + // The wake event is never signalled. Only the posted Windows message + // (QS_ALLINPUT wake mask) can end this 30s wait early. + pump.WaitForWorkOrMessages(wakeEvent.WaitHandle, TimeSpan.FromSeconds(30)); + stopwatch.Stop(); + + Assert.True( + stopwatch.Elapsed < TimeSpan.FromSeconds(10), + $"Wait took {stopwatch.Elapsed}; a posted Windows message should wake the pump."); + }); + + Assert.True(threadReady.Wait(TimeSpan.FromSeconds(5)), "STA thread did not start."); + await Task.Delay(100, CancellationToken.None); + Assert.True( + PostThreadMessage(staThreadId, WmNull, UIntPtr.Zero, IntPtr.Zero), + "Failed to post a Windows message to the STA thread."); + + await staTask; + } + + private const uint WmNull = 0x0000; + + /// Runs an action on a dedicated STA thread and returns when it completes. + private static Task RunOnStaThreadAsync(Action action) + { + return RunOnStaThreadAsync(() => + { + action(); + return 0; + }); + } + + /// Runs a function on a dedicated STA thread and returns its result. + private static Task RunOnStaThreadAsync(Func function) + { + TaskCompletionSource completion = new(); + Thread thread = new(() => + { + try + { + completion.SetResult(function()); + } + catch (Exception exception) + { + completion.SetException(exception); + } + }) + { + IsBackground = true, + }; + thread.SetApartmentState(ApartmentState.STA); + thread.Start(); + return completion.Task; + } + + [System.Runtime.InteropServices.DllImport("kernel32.dll")] + private static extern uint GetCurrentThreadId(); + + [System.Runtime.InteropServices.DllImport("user32.dll", SetLastError = true)] + private static extern bool PostThreadMessage( + uint threadId, + uint message, + UIntPtr wParam, + IntPtr lParam); +} diff --git a/src/MxGateway.Worker/MxAccess/MxAccessBaseEventSink.cs b/src/MxGateway.Worker/MxAccess/MxAccessBaseEventSink.cs index 5c5ae69..317747d 100644 --- a/src/MxGateway.Worker/MxAccess/MxAccessBaseEventSink.cs +++ b/src/MxGateway.Worker/MxAccess/MxAccessBaseEventSink.cs @@ -65,7 +65,14 @@ public sealed class MxAccessBaseEventSink : IMxAccessEventSink sessionId = string.Empty; } - private void OnDataChange( + /// + /// Handles the MXAccess OnDataChange COM event: converts the + /// event arguments to a protobuf and enqueues + /// it. Subscribed to the COM object's event in . + /// Exposed internal so unit tests can drive the integrated + /// sink → mapper → queue path without a live MXAccess COM event source. + /// + internal void OnDataChange( int hLMXServerHandle, int phItemHandle, object pvItemValue, @@ -84,7 +91,11 @@ public sealed class MxAccessBaseEventSink : IMxAccessEventSink statuses)); } - private void OnWriteComplete( + /// + /// Handles the MXAccess OnWriteComplete COM event. Exposed + /// internal as a unit-test seam; see . + /// + internal void OnWriteComplete( int hLMXServerHandle, int phItemHandle, ref MXSTATUS_PROXY[] pVars) @@ -97,7 +108,11 @@ public sealed class MxAccessBaseEventSink : IMxAccessEventSink statuses)); } - private void OperationComplete( + /// + /// Handles the MXAccess OperationComplete COM event. Exposed + /// internal as a unit-test seam; see . + /// + internal void OperationComplete( int hLMXServerHandle, int phItemHandle, ref MXSTATUS_PROXY[] pVars) @@ -110,7 +125,11 @@ public sealed class MxAccessBaseEventSink : IMxAccessEventSink statuses)); } - private void OnBufferedDataChange( + /// + /// Handles the MXAccess OnBufferedDataChange COM event. Exposed + /// internal as a unit-test seam; see . + /// + internal void OnBufferedDataChange( int hLMXServerHandle, int phItemHandle, MxDataType dtDataType, diff --git a/src/MxGateway.Worker/MxGateway.Worker.csproj b/src/MxGateway.Worker/MxGateway.Worker.csproj index 6850b7f..e99334c 100644 --- a/src/MxGateway.Worker/MxGateway.Worker.csproj +++ b/src/MxGateway.Worker/MxGateway.Worker.csproj @@ -14,6 +14,10 @@ + + + +