diff --git a/src/ZB.MOM.WW.MxGateway.IntegrationTests/WorkerLiveMxAccessSmokeTests.cs b/src/ZB.MOM.WW.MxGateway.IntegrationTests/WorkerLiveMxAccessSmokeTests.cs index 8d92057..d2af758 100644 --- a/src/ZB.MOM.WW.MxGateway.IntegrationTests/WorkerLiveMxAccessSmokeTests.cs +++ b/src/ZB.MOM.WW.MxGateway.IntegrationTests/WorkerLiveMxAccessSmokeTests.cs @@ -562,6 +562,329 @@ public sealed class WorkerLiveMxAccessSmokeTests(ITestOutputHelper output) Assert.DoesNotContain(verifyPassword, recordedOutput.Captured, StringComparison.Ordinal); } + /// + /// B8 live verification of the COM commands the B-bundle added against a fake + /// IMxAccessServer: AuthenticateUser, ArchestrAUserToId, Suspend, + /// and Activate. The contract being proven is that each command round-trips + /// to the worker and back carrying a real MXAccess outcome (Ok / an MxStatusProxy / + /// a non-zero HResult) and is NOT short-circuited to INVALID_REQUEST the way an + /// unimplemented command would be. MXAccess-level rejections (a wrong item class for + /// Suspend/Activate commonly returns 0x80070057) are parity, not test failures — we + /// assert the reply kind plus a non-INVALID_REQUEST protocol status, and log the + /// HResult for the record. + /// + [LiveMxAccessFact] + public async Task GatewaySession_WithLiveWorker_NewComCommands_RoundTripWithRealReplies() + { + string workerExecutablePath = IntegrationTestEnvironment.ResolveLiveMxAccessWorkerExecutablePath(); + Assert.True( + File.Exists(workerExecutablePath), + $"Live MXAccess worker executable was not found at {workerExecutablePath}. Build the worker or set {IntegrationTestEnvironment.LiveMxAccessWorkerExecutableVariableName}."); + + // Credential-redaction: AuthenticateUser carries a password. Route every test + // log surface through the buffering helper so the post-run assertion proves the + // password never reached the gateway logger, worker stdout/stderr, or any + // WriteLine the test body issued (same pattern as the WriteSecured parity test). + RecordingTestOutputHelper recordedOutput = new(output); + TestWorkerProcessFactory processFactory = new(recordedOutput); + await using GatewayServiceFixture fixture = new(workerExecutablePath, processFactory, recordedOutput); + + string? sessionId = null; + (string verifyUser, string verifyPassword) = ResolveLiveMxAccessSecuredCredentials(); + + try + { + OpenSessionReply openReply = await fixture.Service.OpenSession( + new OpenSessionRequest + { + ClientSessionName = "live-mxaccess-new-com-commands", + ClientCorrelationId = "live-open-new-com", + CommandTimeout = Duration.FromTimeSpan(CommandTimeout), + }, + new TestServerCallContext()).ConfigureAwait(false); + + sessionId = openReply.SessionId; + Assert.Equal(ProtocolStatusCode.Ok, openReply.ProtocolStatus.Code); + + MxCommandReply registerReply = await fixture.Service.Invoke( + CreateRegisterRequest(sessionId), + new TestServerCallContext()).ConfigureAwait(false); + LogReplyTo(recordedOutput, "Register", registerReply); + Assert.Equal(ProtocolStatusCode.Ok, registerReply.ProtocolStatus.Code); + int serverHandle = registerReply.Register.ServerHandle; + + MxCommandReply addItemReply = await fixture.Service.Invoke( + CreateAddItemRequest(sessionId, serverHandle), + new TestServerCallContext()).ConfigureAwait(false); + LogReplyTo(recordedOutput, "AddItem", addItemReply); + Assert.Equal(ProtocolStatusCode.Ok, addItemReply.ProtocolStatus.Code); + int itemHandle = addItemReply.AddItem.ItemHandle; + + // AuthenticateUser — the B-bundle command under live verification. Before the + // B-bundle this command was unimplemented and the worker short-circuited it to + // INVALID_REQUEST. It must now produce a real reply (Ok with a user id when the + // provider accepts the credential, or a real MXAccess HResult when it does not). + MxCommandReply authReply = await fixture.Service.Invoke( + CreateAuthenticateUserRequest(sessionId, serverHandle, verifyUser, verifyPassword), + new TestServerCallContext()).ConfigureAwait(false); + recordedOutput.WriteLine( + $"AuthenticateUser status={authReply.ProtocolStatus.Code} hresult={authReply.Hresult} user_id={authReply.AuthenticateUser?.UserId}"); + Assert.Equal(MxCommandKind.AuthenticateUser, authReply.Kind); + Assert.NotEqual(ProtocolStatusCode.InvalidRequest, authReply.ProtocolStatus.Code); + Assert.True( + authReply.ProtocolStatus.Code is ProtocolStatusCode.Ok or ProtocolStatusCode.MxaccessFailure, + $"AuthenticateUser must surface a real MXAccess outcome, got {authReply.ProtocolStatus.Code}."); + int authenticatedUserId = + authReply.ProtocolStatus.Code == ProtocolStatusCode.Ok && authReply.AuthenticateUser is not null + ? authReply.AuthenticateUser.UserId + : 0; + if (authReply.ProtocolStatus.Code == ProtocolStatusCode.Ok) + { + // On the dev rig AuthenticateUser("Administrator","") resolves to user id 1. + // Don't pin the exact value (provider/user-store dependent) — just prove a + // success carried a usable, non-zero ArchestrA user id through the reply. + Assert.NotEqual(0, authenticatedUserId); + } + + // ArchestrAUserToId — resolves an ArchestrA user GUID to an integer user id. + // We feed an empty/placeholder GUID: the value is provider-dependent, so the + // assertion is the parity one (real reply, never INVALID_REQUEST). A non-zero + // HResult here is the expected MXAccess rejection of an unknown GUID. + MxCommandReply userToIdReply = await fixture.Service.Invoke( + CreateArchestrAUserToIdRequest(sessionId, serverHandle, userIdGuid: string.Empty), + new TestServerCallContext()).ConfigureAwait(false); + LogReplyTo(recordedOutput, "ArchestrAUserToId", userToIdReply); + recordedOutput.WriteLine($"ArchestrAUserToId user_id={userToIdReply.ArchestraUserToId?.UserId}"); + Assert.Equal(MxCommandKind.ArchestraUserToId, userToIdReply.Kind); + Assert.NotEqual(ProtocolStatusCode.InvalidRequest, userToIdReply.ProtocolStatus.Code); + Assert.True( + userToIdReply.ProtocolStatus.Code is ProtocolStatusCode.Ok or ProtocolStatusCode.MxaccessFailure, + $"ArchestrAUserToId must surface a real MXAccess outcome, got {userToIdReply.ProtocolStatus.Code}."); + + // Suspend / Activate against the advised item. The dev-rig TestInt item class + // may not be suspendable (MXAccess returns 0x80070057 / E_INVALIDARG for a + // wrong item class — see B8 notes). That is MXAccess parity: assert the reply + // kind and a non-INVALID_REQUEST status, surface the HResult and MxStatusProxy + // for the record, and do NOT treat a provider-side rejection as a test failure. + MxCommandReply suspendReply = await fixture.Service.Invoke( + CreateSuspendRequest(sessionId, serverHandle, itemHandle), + new TestServerCallContext()).ConfigureAwait(false); + LogReplyTo(recordedOutput, "Suspend", suspendReply); + recordedOutput.WriteLine( + $"Suspend status_proxy success={suspendReply.Suspend?.Status?.Success} hresult=0x{(uint)suspendReply.Hresult:X8}"); + Assert.Equal(MxCommandKind.Suspend, suspendReply.Kind); + Assert.NotEqual(ProtocolStatusCode.InvalidRequest, suspendReply.ProtocolStatus.Code); + Assert.True( + suspendReply.ProtocolStatus.Code is ProtocolStatusCode.Ok or ProtocolStatusCode.MxaccessFailure, + $"Suspend must surface a real MXAccess outcome, got {suspendReply.ProtocolStatus.Code}."); + + MxCommandReply activateReply = await fixture.Service.Invoke( + CreateActivateRequest(sessionId, serverHandle, itemHandle), + new TestServerCallContext()).ConfigureAwait(false); + LogReplyTo(recordedOutput, "Activate", activateReply); + recordedOutput.WriteLine( + $"Activate status_proxy success={activateReply.Activate?.Status?.Success} hresult=0x{(uint)activateReply.Hresult:X8}"); + Assert.Equal(MxCommandKind.Activate, activateReply.Kind); + Assert.NotEqual(ProtocolStatusCode.InvalidRequest, activateReply.ProtocolStatus.Code); + Assert.True( + activateReply.ProtocolStatus.Code is ProtocolStatusCode.Ok or ProtocolStatusCode.MxaccessFailure, + $"Activate must surface a real MXAccess outcome, got {activateReply.ProtocolStatus.Code}."); + } + finally + { + await ShutDownAsync(fixture, processFactory, sessionId, streamTask: null).ConfigureAwait(false); + } + + // Credential contract: the AuthenticateUser password must never reach any log + // surface (gateway logger, worker stdout/stderr, or test WriteLine). + Assert.DoesNotContain(verifyPassword, recordedOutput.Captured, StringComparison.Ordinal); + } + + /// + /// B8 §3.2 buffered-data path: adds a BUFFERED item (AddBufferedItem), sets the + /// buffered update interval (SetBufferedUpdateInterval), advises it, then attempts + /// to observe an event carrying multiple + /// samples so the worker's multi-sample conversion (VariantConverter → + /// OnBufferedDataChangeEvent quality/timestamp arrays) can be validated live. + /// + /// The AddBufferedItem + SetBufferedUpdateInterval round-trips are asserted unconditionally + /// (they are the B-bundle commands under verification). The buffered EVENT capture is + /// best-effort: if the rig's object logic does not drive a buffered batch within the live + /// event timeout (the same environmental limitation seen with the externally-undrivable + /// alarm rig), the test records the buffered conversion as an unverified residual rather + /// than failing — the command path is proven, the live multi-sample conversion is not. + /// When a batch IS captured, the converted value and quality/timestamp arrays are asserted + /// to be non-empty and internally consistent (no crash, no dropped payload). + /// + /// + [LiveMxAccessFact] + public async Task GatewaySession_WithLiveWorker_BufferedItem_AddsSetsIntervalAndAttemptsCapture() + { + string workerExecutablePath = IntegrationTestEnvironment.ResolveLiveMxAccessWorkerExecutablePath(); + Assert.True( + File.Exists(workerExecutablePath), + $"Live MXAccess worker executable was not found at {workerExecutablePath}. Build the worker or set {IntegrationTestEnvironment.LiveMxAccessWorkerExecutableVariableName}."); + + TestWorkerProcessFactory processFactory = new(output); + await using GatewayServiceFixture fixture = new(workerExecutablePath, processFactory, output); + using RecordingServerStreamWriter eventWriter = new(); + + string? sessionId = null; + Task? streamTask = null; + using CancellationTokenSource streamCancellation = new(); + + // AddBufferedItem takes (item_definition, item_context) like AddItem2. The dev rig + // exposes TestChildObject.TestInt; the buffered form is item="TestInt", + // context="TestChildObject" (per B8 notes). Split the configured live item so a + // custom MXGATEWAY_LIVE_MXACCESS_ITEM override still works. + (string bufferedItem, string bufferedContext) = SplitLiveItemForBuffered(IntegrationTestEnvironment.LiveMxAccessItem); + + try + { + OpenSessionReply openReply = await fixture.Service.OpenSession( + new OpenSessionRequest + { + ClientSessionName = "live-mxaccess-buffered", + ClientCorrelationId = "live-open-buffered", + CommandTimeout = Duration.FromTimeSpan(CommandTimeout), + }, + new TestServerCallContext()).ConfigureAwait(false); + + sessionId = openReply.SessionId; + Assert.Equal(ProtocolStatusCode.Ok, openReply.ProtocolStatus.Code); + + streamTask = fixture.Service.StreamEvents( + new StreamEventsRequest { SessionId = sessionId }, + eventWriter, + new TestServerCallContext(streamCancellation.Token)); + + MxCommandReply registerReply = await fixture.Service.Invoke( + CreateRegisterRequest(sessionId), + new TestServerCallContext()).ConfigureAwait(false); + LogReply("Register", registerReply); + Assert.Equal(ProtocolStatusCode.Ok, registerReply.ProtocolStatus.Code); + int serverHandle = registerReply.Register.ServerHandle; + + // SetBufferedUpdateInterval first so the buffered cadence is established before + // the item is added/advised. MXAccess rounds to 100ms units and rejects < 1. + MxCommandReply intervalReply = await fixture.Service.Invoke( + CreateSetBufferedUpdateIntervalRequest(sessionId, serverHandle, updateIntervalMilliseconds: 1000), + new TestServerCallContext()).ConfigureAwait(false); + LogReply("SetBufferedUpdateInterval", intervalReply); + Assert.Equal(MxCommandKind.SetBufferedUpdateInterval, intervalReply.Kind); + Assert.NotEqual(ProtocolStatusCode.InvalidRequest, intervalReply.ProtocolStatus.Code); + Assert.Equal(ProtocolStatusCode.Ok, intervalReply.ProtocolStatus.Code); + + // AddBufferedItem — must return a real item handle (the dev rig yields handle 1). + MxCommandReply addBufferedReply = await fixture.Service.Invoke( + CreateAddBufferedItemRequest(sessionId, serverHandle, bufferedItem, bufferedContext), + new TestServerCallContext()).ConfigureAwait(false); + LogReply("AddBufferedItem", addBufferedReply); + Assert.Equal(MxCommandKind.AddBufferedItem, addBufferedReply.Kind); + Assert.NotEqual(ProtocolStatusCode.InvalidRequest, addBufferedReply.ProtocolStatus.Code); + Assert.Equal(ProtocolStatusCode.Ok, addBufferedReply.ProtocolStatus.Code); + Assert.NotNull(addBufferedReply.AddBufferedItem); + int bufferedItemHandle = addBufferedReply.AddBufferedItem.ItemHandle; + Assert.True(bufferedItemHandle > 0, "AddBufferedItem must yield a usable item handle."); + + MxCommandReply adviseReply = await fixture.Service.Invoke( + CreateAdviseRequest(sessionId, serverHandle, bufferedItemHandle), + new TestServerCallContext()).ConfigureAwait(false); + LogReply("Advise(buffered)", adviseReply); + Assert.Equal(ProtocolStatusCode.Ok, adviseReply.ProtocolStatus.Code); + + // Best-effort capture of a SAMPLE-BEARING buffered batch. + // + // Live observation (B8): immediately after Advise the provider delivers an + // initial OnBufferedDataChange with data_type=NoData / raw_data_type=0 and zero + // quality+timestamp samples — the buffered analogue of the bad-quality/ + // registration-state bootstrap event the OnDataChange tests skip with their + // family-match predicate. That empty bootstrap is parity, NOT a dropped payload: + // the converter ran without crashing and there were simply no samples to carry. + // We therefore match only a batch that actually carries samples, so a real + // multi-sample conversion can be validated and the empty bootstrap is skipped + // rather than mistaken for a defect. + MxEvent? bufferedBatch = null; + try + { + bufferedBatch = await eventWriter + .WaitForMessageAsync( + candidate => candidate.Family == MxEventFamily.OnBufferedDataChange + && candidate.ServerHandle == serverHandle + && candidate.ItemHandle == bufferedItemHandle + && candidate.OnBufferedDataChange is not null + && (CountArrayElements(candidate.OnBufferedDataChange.QualityValues) > 0 + || CountArrayElements(candidate.OnBufferedDataChange.TimestampValues) > 0), + IntegrationTestEnvironment.LiveMxAccessEventTimeout, + streamCancellation.Token) + .ConfigureAwait(false); + } + catch (TimeoutException) + { + bufferedBatch = null; + } + + // Whether or not a sample-bearing batch arrived, record what buffered events the + // rig DID deliver (typically just the empty NoData bootstrap) for the record. + int bootstrapBufferedEvents = CountMatchingEvents( + eventWriter, + e => e.Family == MxEventFamily.OnBufferedDataChange + && e.ServerHandle == serverHandle + && e.ItemHandle == bufferedItemHandle); + + if (bufferedBatch is null) + { + // RESIDUAL (documented): the command path (AddBufferedItem + + // SetBufferedUpdateInterval + Advise) is proven and the buffered EVENT plumbing + // is live (the empty NoData bootstrap arrives and converts without crashing), + // but the rig did not drive a sample-bearing buffered batch within the timeout + // — the same environmental limitation as the externally-undrivable alarm rig. + // The §3.2 OnBufferedDataChange MULTI-SAMPLE conversion therefore remains + // unverified live. This is environmental, not a defect — let the test pass. + output.WriteLine( + "B8 RESIDUAL: AddBufferedItem/SetBufferedUpdateInterval/Advise round-tripped and " + + $"{bootstrapBufferedEvents} OnBufferedDataChange event(s) arrived (empty NoData " + + "bootstrap, converted without crash/drop), but no sample-bearing buffered batch " + + $"was observed within {IntegrationTestEnvironment.LiveMxAccessEventTimeout}. Live " + + "§3.2 multi-sample conversion remains unverified (rig object logic may not drive " + + "buffered samples on demand)."); + return; + } + + // A SAMPLE-BEARING buffered batch was captured — validate the §3.2 conversion. + LogEvent(bufferedBatch); + OnBufferedDataChangeEvent body = bufferedBatch.OnBufferedDataChange; + Assert.NotNull(body); + + int qualityCount = CountArrayElements(body.QualityValues); + int timestampCount = CountArrayElements(body.TimestampValues); + output.WriteLine( + $"B8 CAPTURED buffered batch: data_type={body.DataType} raw_data_type={body.RawDataType} " + + $"quality_samples={qualityCount} timestamp_samples={timestampCount} " + + $"value_kind={bufferedBatch.Value?.KindCase}"); + + // The predicate guaranteed at least one sample; the converted aggregate value + // must also exist (no crash, no dropped payload). + Assert.True( + qualityCount > 0 || timestampCount > 0, + "Sample-bearing OnBufferedDataChange lost its samples after the predicate matched."); + Assert.NotNull(bufferedBatch.Value); + + // When MXAccess delivers parallel quality + timestamp arrays the converted + // arrays must agree in length; a mismatch is a real conversion defect (a sample + // was dropped on one side). + if (qualityCount > 0 && timestampCount > 0) + { + Assert.Equal(qualityCount, timestampCount); + } + } + finally + { + streamCancellation.Cancel(); + await ShutDownAsync(fixture, processFactory, sessionId, streamTask).ConfigureAwait(false); + } + } + /// /// Verifies that killing the worker process marks the session /// with a clean fault classification — the gateway @@ -939,6 +1262,113 @@ public sealed class WorkerLiveMxAccessSmokeTests(ITestOutputHelper output) }; } + private static MxCommandRequest CreateArchestrAUserToIdRequest( + string sessionId, + int serverHandle, + string userIdGuid) + { + return new MxCommandRequest + { + SessionId = sessionId, + ClientCorrelationId = "live-archestra-user-to-id", + Command = new MxCommand + { + Kind = MxCommandKind.ArchestraUserToId, + ArchestraUserToId = new ArchestrAUserToIdCommand + { + ServerHandle = serverHandle, + UserIdGuid = userIdGuid, + }, + }, + }; + } + + private static MxCommandRequest CreateAddBufferedItemRequest( + string sessionId, + int serverHandle, + string itemDefinition, + string itemContext) + { + return new MxCommandRequest + { + SessionId = sessionId, + ClientCorrelationId = "live-add-buffered-item", + Command = new MxCommand + { + Kind = MxCommandKind.AddBufferedItem, + AddBufferedItem = new AddBufferedItemCommand + { + ServerHandle = serverHandle, + ItemDefinition = itemDefinition, + ItemContext = itemContext, + }, + }, + }; + } + + private static MxCommandRequest CreateSetBufferedUpdateIntervalRequest( + string sessionId, + int serverHandle, + int updateIntervalMilliseconds) + { + return new MxCommandRequest + { + SessionId = sessionId, + ClientCorrelationId = "live-set-buffered-update-interval", + Command = new MxCommand + { + Kind = MxCommandKind.SetBufferedUpdateInterval, + SetBufferedUpdateInterval = new SetBufferedUpdateIntervalCommand + { + ServerHandle = serverHandle, + UpdateIntervalMilliseconds = updateIntervalMilliseconds, + }, + }, + }; + } + + private static MxCommandRequest CreateSuspendRequest( + string sessionId, + int serverHandle, + int itemHandle) + { + return new MxCommandRequest + { + SessionId = sessionId, + ClientCorrelationId = "live-suspend", + Command = new MxCommand + { + Kind = MxCommandKind.Suspend, + Suspend = new SuspendCommand + { + ServerHandle = serverHandle, + ItemHandle = itemHandle, + }, + }, + }; + } + + private static MxCommandRequest CreateActivateRequest( + string sessionId, + int serverHandle, + int itemHandle) + { + return new MxCommandRequest + { + SessionId = sessionId, + ClientCorrelationId = "live-activate", + Command = new MxCommand + { + Kind = MxCommandKind.Activate, + Activate = new ActivateCommand + { + ServerHandle = serverHandle, + ItemHandle = itemHandle, + }, + }, + }; + } + private static MxCommandRequest CreateWriteSecuredRequest( string sessionId, int serverHandle, @@ -978,6 +1408,50 @@ public sealed class WorkerLiveMxAccessSmokeTests(ITestOutputHelper output) return (verifyUser, verifyPassword); } + /// + /// Splits a dotted MXAccess reference (e.g. "TestChildObject.TestInt") into the + /// (item_definition, item_context) pair AddBufferedItem expects — attribute name and + /// owning object. An undotted reference is passed through with an empty context. + /// + private static (string Item, string Context) SplitLiveItemForBuffered(string liveItem) + { + int lastDot = liveItem.LastIndexOf('.'); + if (lastDot <= 0 || lastDot >= liveItem.Length - 1) + { + return (liveItem, string.Empty); + } + + string context = liveItem[..lastDot]; + string item = liveItem[(lastDot + 1)..]; + return (item, context); + } + + /// + /// Counts the elements in a converted buffered across whichever + /// typed-array oneof case the VariantConverter populated, so the buffered-capture + /// assertions are independent of the rig item's element type. + /// + private static int CountArrayElements(MxArray? array) + { + if (array is null) + { + return 0; + } + + return array.ValuesCase switch + { + MxArray.ValuesOneofCase.BoolValues => array.BoolValues.Values.Count, + MxArray.ValuesOneofCase.Int32Values => array.Int32Values.Values.Count, + MxArray.ValuesOneofCase.Int64Values => array.Int64Values.Values.Count, + MxArray.ValuesOneofCase.FloatValues => array.FloatValues.Values.Count, + MxArray.ValuesOneofCase.DoubleValues => array.DoubleValues.Values.Count, + MxArray.ValuesOneofCase.StringValues => array.StringValues.Values.Count, + MxArray.ValuesOneofCase.TimestampValues => array.TimestampValues.Values.Count, + MxArray.ValuesOneofCase.RawValues => array.RawValues.Values.Count, + _ => 0, + }; + } + private static int CountMatchingEvents( RecordingServerStreamWriter writer, Func predicate)