lmxopcua/tests/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Host.Tests/EndToEndIpcTests.cs

using System;
using System.IO;
using System.IO.Pipes;
using System.Security.Principal;
using System.Threading;
using System.Threading.Tasks;
using MessagePack;
using Serilog;
using Serilog.Core;
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Host.Backend;
using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Host.Ipc;
using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Shared;
using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Shared.Contracts;

namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Host.Tests
{
    /// <summary>
    ///     Drives every <see cref="MessageKind"/> the Phase 2 plan exposes through the full
    ///     Host-side stack (<see cref="PipeServer"/> + <see cref="GalaxyFrameHandler"/> +
    ///     <see cref="StubGalaxyBackend"/>) using a hand-rolled IPC client built on Shared's
    ///     <see cref="FrameReader"/>/<see cref="FrameWriter"/>. The Proxy's <c>GalaxyIpcClient</c>
    ///     is net10-only and cannot load in this net48 x86 test process, so we exercise the same
    ///     wire protocol through the framing primitives directly. The dispatcher/backend response
    ///     shapes are the production code path verbatim.
    /// </summary>
    [Trait("Category", "Integration")]
    public sealed class EndToEndIpcTests
    {
        private sealed class TestStack : IDisposable
        {
            public PipeServer Server = null!;
            public NamedPipeClientStream Stream = null!;
            public FrameReader Reader = null!;
            public FrameWriter Writer = null!;
            public Task ServerTask = null!;
            public CancellationTokenSource Cts = null!;

            public void Dispose()
            {
                Cts.Cancel();
                try { ServerTask.GetAwaiter().GetResult(); } catch { /* shutdown */ }
                Server.Dispose();
                Stream.Dispose();
                Reader.Dispose();
                Writer.Dispose();
                Cts.Dispose();
            }
        }

        private static async Task<TestStack> StartAsync()
        {
            using var identity = WindowsIdentity.GetCurrent();
            var sid = identity.User!;
            var pipe = $"OtOpcUaGalaxyE2E-{Guid.NewGuid():N}";
            const string secret = "e2e-secret";
            Logger log = new LoggerConfiguration().CreateLogger();
            var cts = new CancellationTokenSource(TimeSpan.FromSeconds(15));

            var server = new PipeServer(pipe, sid, secret, log);
            var serverTask = Task.Run(() => server.RunAsync(
                new GalaxyFrameHandler(new StubGalaxyBackend(), log), cts.Token));

            var stream = new NamedPipeClientStream(".", pipe, PipeDirection.InOut, PipeOptions.Asynchronous);
            await stream.ConnectAsync(5_000, cts.Token);
            var reader = new FrameReader(stream, leaveOpen: true);
            var writer = new FrameWriter(stream, leaveOpen: true);
            await writer.WriteAsync(MessageKind.Hello,
                new Hello { PeerName = "e2e", SharedSecret = secret }, cts.Token);
            var ack = await reader.ReadFrameAsync(cts.Token);
            if (ack is null || ack.Value.Kind != MessageKind.HelloAck)
                throw new InvalidOperationException("Hello handshake failed");

            return new TestStack
            {
                Server = server,
                Stream = stream,
                Reader = reader,
                Writer = writer,
                ServerTask = serverTask,
                Cts = cts,
            };
        }

        private static async Task<TResp> RoundTripAsync<TReq, TResp>(
            TestStack s, MessageKind reqKind, TReq req, MessageKind respKind)
        {
            await s.Writer.WriteAsync(reqKind, req, s.Cts.Token);
            var frame = await s.Reader.ReadFrameAsync(s.Cts.Token);
            frame.HasValue.ShouldBeTrue();
            frame!.Value.Kind.ShouldBe(respKind);
            return MessagePackSerializer.Deserialize<TResp>(frame.Value.Body);
        }

        [Fact]
        public async Task OpenSession_succeeds_with_an_assigned_session_id()
        {
            using var s = await StartAsync();

            var resp = await RoundTripAsync<OpenSessionRequest, OpenSessionResponse>(
                s, MessageKind.OpenSessionRequest,
                new OpenSessionRequest { DriverInstanceId = "gal-e2e", DriverConfigJson = "{}" },
                MessageKind.OpenSessionResponse);

            resp.Success.ShouldBeTrue();
            resp.SessionId.ShouldBeGreaterThan(0L);
        }

        [Fact]
        public async Task Discover_against_stub_returns_an_error_response()
        {
            using var s = await StartAsync();

            var resp = await RoundTripAsync<DiscoverHierarchyRequest, DiscoverHierarchyResponse>(
                s, MessageKind.DiscoverHierarchyRequest,
                new DiscoverHierarchyRequest { SessionId = 1 },
                MessageKind.DiscoverHierarchyResponse);

            resp.Success.ShouldBeFalse();
            resp.Error.ShouldContain("MXAccess code lift pending");
        }

        [Fact]
        public async Task WriteValues_returns_per_tag_BadInternalError_status()
        {
            using var s = await StartAsync();

            var resp = await RoundTripAsync<WriteValuesRequest, WriteValuesResponse>(
                s, MessageKind.WriteValuesRequest,
                new WriteValuesRequest
                {
                    SessionId = 1,
                    Writes = new[] { new GalaxyDataValue { TagReference = "TagA" } },
                },
                MessageKind.WriteValuesResponse);

            resp.Results.Length.ShouldBe(1);
            resp.Results[0].StatusCode.ShouldBe(0x80020000u);
        }

        [Fact]
        public async Task Subscribe_returns_a_subscription_id()
        {
            using var s = await StartAsync();

            var sub = await RoundTripAsync<SubscribeRequest, SubscribeResponse>(
                s, MessageKind.SubscribeRequest,
                new SubscribeRequest { SessionId = 1, TagReferences = new[] { "TagA" }, RequestedIntervalMs = 500 },
                MessageKind.SubscribeResponse);

            sub.Success.ShouldBeTrue();
            sub.SubscriptionId.ShouldBeGreaterThan(0L);
        }

        [Fact]
        public async Task Recycle_returns_the_grace_window_from_the_backend()
        {
            using var s = await StartAsync();

            var resp = await RoundTripAsync<RecycleHostRequest, RecycleStatusResponse>(
                s, MessageKind.RecycleHostRequest,
                new RecycleHostRequest { Kind = "Soft", Reason = "test" },
                MessageKind.RecycleStatusResponse);

            resp.Accepted.ShouldBeTrue();
            resp.GraceSeconds.ShouldBe(15);
        }
    }
}