mxaccessgw/src/MxGateway.Worker.Tests/Ipc/WorkerFrameProtocolTests.cs

using System;
using System.IO;
using System.Linq;
using System.Threading.Tasks;
using Google.Protobuf;
using MxGateway.Contracts;
using MxGateway.Contracts.Proto;
using MxGateway.Worker.Ipc;

namespace MxGateway.Worker.Tests.Ipc;

public sealed class WorkerFrameProtocolTests
{
    private const string SessionId = "session-1";
    private const string Nonce = "nonce-secret";

    [Fact]
    public async Task WriteAndReadAsync_WithValidEnvelope_RoundTripsFrame()
    {
        WorkerFrameProtocolOptions options = CreateOptions();
        MemoryStream stream = new();
        WorkerEnvelope original = CreateGatewayHelloEnvelope();

        WorkerFrameWriter writer = new(stream, options);
        await writer.WriteAsync(original);
        stream.Position = 0;

        WorkerFrameReader reader = new(stream, options);
        WorkerEnvelope parsed = await reader.ReadAsync();

        Assert.Equal(original, parsed);
    }

    [Fact]
    public async Task ReadAsync_WithWrongProtocolVersion_ThrowsProtocolVersionMismatch()
    {
        WorkerFrameProtocolOptions options = CreateOptions();
        WorkerEnvelope envelope = CreateGatewayHelloEnvelope();
        envelope.ProtocolVersion++;
        MemoryStream stream = new(CreateFrame(envelope));

        WorkerFrameReader reader = new(stream, options);
        WorkerFrameProtocolException exception =
            await Assert.ThrowsAsync<WorkerFrameProtocolException>(
                async () => await reader.ReadAsync());

        Assert.Equal(WorkerFrameProtocolErrorCode.ProtocolVersionMismatch, exception.ErrorCode);
    }

    [Fact]
    public async Task ReadAsync_WithWrongSessionId_ThrowsSessionMismatch()
    {
        WorkerFrameProtocolOptions options = CreateOptions();
        WorkerEnvelope envelope = CreateGatewayHelloEnvelope();
        envelope.SessionId = "different-session";
        MemoryStream stream = new(CreateFrame(envelope));

        WorkerFrameReader reader = new(stream, options);
        WorkerFrameProtocolException exception =
            await Assert.ThrowsAsync<WorkerFrameProtocolException>(
                async () => await reader.ReadAsync());

        Assert.Equal(WorkerFrameProtocolErrorCode.SessionMismatch, exception.ErrorCode);
    }

    [Fact]
    public async Task ReadAsync_WithMalformedLength_ThrowsMalformedLength()
    {
        WorkerFrameProtocolOptions options = CreateOptions();
        MemoryStream stream = new(new byte[sizeof(uint)]);

        WorkerFrameReader reader = new(stream, options);
        WorkerFrameProtocolException exception =
            await Assert.ThrowsAsync<WorkerFrameProtocolException>(
                async () => await reader.ReadAsync());

        Assert.Equal(WorkerFrameProtocolErrorCode.MalformedLength, exception.ErrorCode);
    }

    [Fact]
    public async Task ReadAsync_WithMalformedPayload_ThrowsInvalidEnvelope()
    {
        WorkerFrameProtocolOptions options = CreateOptions();
        MemoryStream stream = new(CreateFrame(new byte[] { 0x80 }));

        WorkerFrameReader reader = new(stream, options);
        WorkerFrameProtocolException exception =
            await Assert.ThrowsAsync<WorkerFrameProtocolException>(
                async () => await reader.ReadAsync());

        Assert.Equal(WorkerFrameProtocolErrorCode.InvalidEnvelope, exception.ErrorCode);
    }

    [Fact]
    public async Task WriteAsync_WithConcurrentCalls_SerializesCompleteFrames()
    {
        WorkerFrameProtocolOptions options = CreateOptions();
        MemoryStream stream = new();
        WorkerFrameWriter writer = new(stream, options);

        await Task.WhenAll(
            writer.WriteAsync(CreateGatewayHelloEnvelope(sequence: 1)),
            writer.WriteAsync(CreateGatewayHelloEnvelope(sequence: 2)),
            writer.WriteAsync(CreateGatewayHelloEnvelope(sequence: 3)));

        stream.Position = 0;
        WorkerFrameReader reader = new(stream, options);

        WorkerEnvelope first = await reader.ReadAsync();
        WorkerEnvelope second = await reader.ReadAsync();
        WorkerEnvelope third = await reader.ReadAsync();

        Assert.Equal(new ulong[] { 1, 2, 3 }, new[] { first.Sequence, second.Sequence, third.Sequence }.OrderBy(sequence => sequence));
    }

    private static WorkerFrameProtocolOptions CreateOptions()
    {
        return new WorkerFrameProtocolOptions(
            SessionId,
            GatewayContractInfo.WorkerProtocolVersion,
            Nonce);
    }

    private static WorkerEnvelope CreateGatewayHelloEnvelope(ulong sequence = 1)
    {
        return new WorkerEnvelope
        {
            ProtocolVersion = GatewayContractInfo.WorkerProtocolVersion,
            SessionId = SessionId,
            Sequence = sequence,
            GatewayHello = new GatewayHello
            {
                SupportedProtocolVersion = GatewayContractInfo.WorkerProtocolVersion,
                Nonce = Nonce,
                GatewayVersion = "test-gateway",
            },
        };
    }

    private static byte[] CreateFrame(IMessage message)
    {
        return CreateFrame(message.ToByteArray());
    }

    private static byte[] CreateFrame(byte[] payload)
    {
        byte[] frame = new byte[sizeof(uint) + payload.Length];
        WriteUInt32LittleEndian(frame, (uint)payload.Length);
        payload.CopyTo(frame, sizeof(uint));

        return frame;
    }

    private static void WriteUInt32LittleEndian(
        byte[] buffer,
        uint value)
    {
        buffer[0] = (byte)value;
        buffer[1] = (byte)(value >> 8);
        buffer[2] = (byte)(value >> 16);
        buffer[3] = (byte)(value >> 24);
    }
}