lmxopcua/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Wonderware.Client.Tests/FakeSidecarServer.cs

using System.IO.Pipes;
using MessagePack;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Wonderware.Client.Ipc;

namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Wonderware.Client.Tests;

/// <summary>
///     In-process fake of the Wonderware historian sidecar. Reuses the client-side framing
///     code (which is byte-identical to the real sidecar) so the wire bytes round-trip
///     correctly without requiring the .NET 4.8 sidecar binary at test time.
/// </summary>
internal sealed class FakeSidecarServer : IAsyncDisposable
{
    private readonly string _pipeName;
    private readonly string _expectedSecret;
    private readonly CancellationTokenSource _cts = new();
    private Task? _loop;

    /// <summary>Gets or sets the handler for ReadRaw requests.</summary>
    public Func<ReadRawRequest, ReadRawReply> OnReadRaw { get; set; } = _ => new ReadRawReply { Success = true };

    /// <summary>Gets or sets the handler for ReadProcessed requests.</summary>
    public Func<ReadProcessedRequest, ReadProcessedReply> OnReadProcessed { get; set; } = _ => new ReadProcessedReply { Success = true };

    /// <summary>Gets or sets the handler for ReadAtTime requests.</summary>
    public Func<ReadAtTimeRequest, ReadAtTimeReply> OnReadAtTime { get; set; } = _ => new ReadAtTimeReply { Success = true };

    /// <summary>Gets or sets the handler for ReadEvents requests.</summary>
    public Func<ReadEventsRequest, ReadEventsReply> OnReadEvents { get; set; } = _ => new ReadEventsReply { Success = true };

    /// <summary>Gets or sets the handler for WriteAlarmEvents requests.</summary>
    public Func<WriteAlarmEventsRequest, WriteAlarmEventsReply> OnWriteAlarmEvents { get; set; } = req
        => new WriteAlarmEventsReply { Success = true, PerEventOk = Enumerable.Repeat(true, req.Events.Length).ToArray() };

    /// <summary>Force-disconnect the next accepted client mid-call to exercise reconnect.</summary>
    public bool DisconnectAfterHandshake { get; set; }

    /// <summary>
    ///     Drop the connection after the handshake but before replying to any non-Hello request.
    ///     Armed for every connection until reset. Used to exercise the WriteBatchAsync catch
    ///     path and the second-attempt-also-fails propagation path.
    /// </summary>
    public bool DisconnectBeforeReply { get; set; }

    /// <summary>
    ///     Reply to the first non-Hello request with this kind instead of the expected kind,
    ///     to exercise <see cref="System.IO.InvalidDataException"/> detection in ExchangeAsync.
    ///     Reset to null after the first mis-routed reply.
    /// </summary>
    public MessageKind? ReplyWithWrongKind { get; set; }

    /// <summary>
    ///     Stall indefinitely after receiving a request before sending any reply, so the client's
    ///     call-timeout token fires. Used to test the CallTimeout path.
    /// </summary>
    public bool StallAfterRequest { get; set; }

    /// <summary>Initializes a new instance of FakeSidecarServer with the specified pipe name and expected secret.</summary>
    /// <param name="pipeName">The name of the named pipe to use for communication.</param>
    /// <param name="expectedSecret">The expected shared secret for handshake validation.</param>
    public FakeSidecarServer(string pipeName, string expectedSecret)
    {
        _pipeName = pipeName;
        _expectedSecret = expectedSecret;
    }

    /// <summary>Gets the named pipe name used for communication.</summary>
    public string PipeName => _pipeName;

    /// <summary>Starts the fake sidecar server asynchronously.</summary>
    public Task StartAsync()
    {
        _loop = Task.Run(() => RunAsync(_cts.Token));
        // Give the listener a moment to start so client connect doesn't race.
        return Task.Delay(50);
    }

    private async Task RunAsync(CancellationToken ct)
    {
        while (!ct.IsCancellationRequested)
        {
            await using var pipe = new NamedPipeServerStream(
                _pipeName, PipeDirection.InOut, maxNumberOfServerInstances: 1,
                PipeTransmissionMode.Byte, PipeOptions.Asynchronous,
                inBufferSize: 64 * 1024, outBufferSize: 64 * 1024);

            try { await pipe.WaitForConnectionAsync(ct).ConfigureAwait(false); }
            catch (OperationCanceledException) { break; }

            try
            {
                using var reader = new FrameReader(pipe, leaveOpen: true);
                using var writer = new FrameWriter(pipe, leaveOpen: true);

                // Hello handshake.
                var first = await reader.ReadFrameAsync(ct).ConfigureAwait(false);
                if (first is null || first.Value.Kind != MessageKind.Hello) continue;
                var hello = MessagePackSerializer.Deserialize<Hello>(first.Value.Body);

                if (!string.Equals(hello.SharedSecret, _expectedSecret, StringComparison.Ordinal))
                {
                    await writer.WriteAsync(MessageKind.HelloAck, new HelloAck { Accepted = false, RejectReason = "shared-secret-mismatch" }, ct);
                    continue;
                }

                await writer.WriteAsync(MessageKind.HelloAck, new HelloAck { Accepted = true, HostName = "fake-sidecar" }, ct);

                if (DisconnectAfterHandshake)
                {
                    DisconnectAfterHandshake = false; // arm once
                    pipe.Disconnect();
                    continue;
                }

                while (!ct.IsCancellationRequested)
                {
                    var frame = await reader.ReadFrameAsync(ct).ConfigureAwait(false);
                    if (frame is null) break;

                    // Drop before sending any reply — lets the client fall into its catch /
                    // retry path or propagate on second failure.
                    if (DisconnectBeforeReply)
                    {
                        pipe.Disconnect();
                        break;
                    }

                    // Stall indefinitely to let the client's call-timeout token fire.
                    if (StallAfterRequest)
                    {
                        await Task.Delay(Timeout.Infinite, ct).ConfigureAwait(false);
                        break;
                    }

                    // Optionally send a deliberately wrong kind back to exercise
                    // InvalidDataException detection in the client's ExchangeAsync.
                    if (ReplyWithWrongKind.HasValue)
                    {
                        var wrongKind = ReplyWithWrongKind.Value;
                        ReplyWithWrongKind = null; // arm once
                        // Send an empty body with the wrong kind so the client can parse it.
                        await writer.WriteAsync(wrongKind, new ReadRawReply { Success = false }, ct);
                        continue;
                    }

                    switch (frame.Value.Kind)
                    {
                        case MessageKind.ReadRawRequest:
                        {
                            var req = MessagePackSerializer.Deserialize<ReadRawRequest>(frame.Value.Body);
                            var reply = OnReadRaw(req);
                            reply.CorrelationId = req.CorrelationId;
                            await writer.WriteAsync(MessageKind.ReadRawReply, reply, ct);
                            break;
                        }
                        case MessageKind.ReadProcessedRequest:
                        {
                            var req = MessagePackSerializer.Deserialize<ReadProcessedRequest>(frame.Value.Body);
                            var reply = OnReadProcessed(req);
                            reply.CorrelationId = req.CorrelationId;
                            await writer.WriteAsync(MessageKind.ReadProcessedReply, reply, ct);
                            break;
                        }
                        case MessageKind.ReadAtTimeRequest:
                        {
                            var req = MessagePackSerializer.Deserialize<ReadAtTimeRequest>(frame.Value.Body);
                            var reply = OnReadAtTime(req);
                            reply.CorrelationId = req.CorrelationId;
                            await writer.WriteAsync(MessageKind.ReadAtTimeReply, reply, ct);
                            break;
                        }
                        case MessageKind.ReadEventsRequest:
                        {
                            var req = MessagePackSerializer.Deserialize<ReadEventsRequest>(frame.Value.Body);
                            var reply = OnReadEvents(req);
                            reply.CorrelationId = req.CorrelationId;
                            await writer.WriteAsync(MessageKind.ReadEventsReply, reply, ct);
                            break;
                        }
                        case MessageKind.WriteAlarmEventsRequest:
                        {
                            var req = MessagePackSerializer.Deserialize<WriteAlarmEventsRequest>(frame.Value.Body);
                            var reply = OnWriteAlarmEvents(req);
                            reply.CorrelationId = req.CorrelationId;
                            await writer.WriteAsync(MessageKind.WriteAlarmEventsReply, reply, ct);
                            break;
                        }
                    }
                }
            }
            catch (OperationCanceledException) { break; }
            catch (IOException) { /* peer dropped — accept next */ }
        }
    }

    /// <summary>Releases all resources used by the fake sidecar server.</summary>
    public async ValueTask DisposeAsync()
    {
        _cts.Cancel();
        if (_loop is not null)
        {
            try { await _loop.ConfigureAwait(false); } catch { /* ignore shutdown errors */ }
        }
        _cts.Dispose();
    }
}