riojoy/tests/RioJoy.Core.Tests/Serial/NamedPipeTransportTests.cs

using System.IO.Pipes;
using RioJoy.Core.Serial;
using Xunit;

namespace RioJoy.Core.Tests.Serial;

public class PipeFramingTests
{
    [Fact]
    public void EncodeData_WrapsInOneFrame()
    {
        byte[] framed = PipeFraming.EncodeData(new byte[] { 0x81, 0x01, 0xFC });
        Assert.Equal(new byte[] { 0x00, 0x03, 0x81, 0x01, 0xFC }, framed);
    }

    [Fact]
    public void EncodeData_ChunksPayloadsOver255()
    {
        var data = new byte[300];
        for (int i = 0; i < data.Length; i++) data[i] = (byte)i;

        byte[] framed = PipeFraming.EncodeData(data);

        // 255-byte frame + 45-byte frame, payloads contiguous.
        Assert.Equal(300 + 4, framed.Length);
        Assert.Equal(PipeFraming.DataType, framed[0]);
        Assert.Equal(255, framed[1]);
        Assert.Equal(PipeFraming.DataType, framed[2 + 255]);
        Assert.Equal(45, framed[2 + 255 + 1]);
        Assert.Equal(data.Take(255), framed.Skip(2).Take(255));
        Assert.Equal(data.Skip(255), framed.Skip(2 + 255 + 2));
    }

    [Fact]
    public void EncodeData_EmptyYieldsNoFrames()
    {
        Assert.Empty(PipeFraming.EncodeData(new byte[0]));
    }

    [Fact]
    public void Decoder_SurvivesAnySplitAcrossReads()
    {
        var decoder = new PipeFrameDecoder();
        var data = new List<byte>();
        var lines = new List<byte>();
        decoder.Data += (buf, count) => data.AddRange(buf.Take(count));
        decoder.Lines += lines.Add;

        // A lines frame, a 2-byte data frame, a 1-byte data frame — fed one byte at a time.
        byte[] stream = { 0x01, 0x03, 0x00, 0x02, 0x81, 0x01, 0x00, 0x01, 0xFC };
        foreach (byte b in stream)
            Assert.True(decoder.Feed(new[] { b }, 1));

        Assert.Equal(new byte[] { 0x03 }, lines);
        Assert.Equal(new byte[] { 0x81, 0x01, 0xFC }, data);
    }

    [Fact]
    public void Decoder_UnknownFrameType_Poisons()
    {
        var decoder = new PipeFrameDecoder();

        Assert.False(decoder.Feed(new byte[] { 0xFF }, 1));
        Assert.Contains("0xFF", decoder.Violation);
        Assert.False(decoder.Feed(new byte[] { 0x00, 0x01, 0x42 }, 3)); // stays poisoned

        decoder.Reset();
        Assert.True(decoder.Feed(new byte[] { 0x00, 0x01, 0x42 }, 3));
        Assert.Null(decoder.Violation);
    }

    [Fact]
    public void Decoder_ZeroLengthDataFrame_Poisons()
    {
        var decoder = new PipeFrameDecoder();
        Assert.False(decoder.Feed(new byte[] { 0x00, 0x00 }, 2));
        Assert.Contains("zero-length", decoder.Violation);
    }
}

public class RioTransportFactoryTests
{
    [Theory]
    [InlineData("pipe:vrio", true)]
    [InlineData("PIPE:vrio", true)]
    [InlineData("COM3", false)]
    [InlineData("com1", false)]
    public void IsPipe_RecognizesTheScheme(string endpoint, bool expected)
    {
        Assert.Equal(expected, RioTransportFactory.IsPipe(endpoint));
    }
}

/// <summary>
/// Integration tests against a real in-process <see cref="NamedPipeServerStream"/>
/// standing in for vRIO (which serves \\.\pipe\vrio the same way).
/// </summary>
public class NamedPipeTransportTests : IDisposable
{
    private readonly string _pipeName = $"riojoy-test-{Guid.NewGuid():N}";
    private readonly NamedPipeServerStream _server;

    public NamedPipeTransportTests()
    {
        _server = new NamedPipeServerStream(_pipeName, PipeDirection.InOut, 1,
            PipeTransmissionMode.Byte, PipeOptions.Asynchronous);
    }

    public void Dispose()
    {
        try { _server.Dispose(); }
        catch (IOException) { }
    }

    /// <summary>Accept the client and construct the transport concurrently (the ctor blocks through the DTR pulse).</summary>
    private async Task<NamedPipeTransport> ConnectAsync()
    {
        Task accept = _server.WaitForConnectionAsync();
        Task<NamedPipeTransport> client = Task.Run(() => new NamedPipeTransport(_pipeName));
        await accept.WithTimeout();
        return await client.WithTimeout();
    }

    private async Task<byte[]> ServerReadAsync(int count)
    {
        var buffer = new byte[count];
        int fill = 0;
        while (fill < count)
        {
            int n = await _server.ReadAsync(buffer, fill, count - fill).WithTimeout();
            Assert.True(n > 0, "server: pipe closed before the expected bytes arrived");
            fill += n;
        }
        return buffer;
    }

    [Fact]
    public async Task Connect_SendsTheDtrResetPulse()
    {
        using NamedPipeTransport transport = await ConnectAsync();

        // Assert (DTR high), hold, release — the in-band SETDTR/CLRDTR port.
        Assert.Equal(new byte[] { 0x01, PipeFraming.LineDtr }, await ServerReadAsync(2));
        Assert.Equal(new byte[] { 0x01, 0x00 }, await ServerReadAsync(2));
    }

    [Fact]
    public async Task ReadAsync_UnwrapsDataFrames_AndSwallowsLinesFrames()
    {
        using NamedPipeTransport transport = await ConnectAsync();

        // vRIO's on-connect lines frame (board present), then a data frame —
        // split at an awkward boundary to exercise the incremental decoder.
        // Issued unawaited: the pipe's 0-byte buffers make a write complete
        // only when the peer reads it (see the transport's ctor comment).
        Task w1 = _server.WriteAsync(new byte[] { 0x01, 0x03, 0x00, 0x03, 0x81 }, 0, 5);
        Task w2 = _server.WriteAsync(new byte[] { 0x01, 0xFC }, 0, 2);

        var buffer = new byte[16];
        var got = new List<byte>();
        while (got.Count < 3)
        {
            int n = await transport.ReadAsync(buffer, CancellationToken.None).WithTimeout();
            Assert.True(n > 0);
            got.AddRange(buffer.Take(n));
        }

        Assert.Equal(new byte[] { 0x81, 0x01, 0xFC }, got);
        Assert.Equal((byte)(PipeFraming.LineDtr | PipeFraming.LineRts), transport.PeerLines);
        await w1.WithTimeout();
        await w2.WithTimeout();
    }

    [Fact]
    public async Task ReadAsync_SmallBuffer_DrainsAcrossCalls()
    {
        using NamedPipeTransport transport = await ConnectAsync();

        Task write = _server.WriteAsync(new byte[] { 0x00, 0x04, 0x10, 0x20, 0x30, 0x40 }, 0, 6);

        var buffer = new byte[3];
        Assert.Equal(3, await transport.ReadAsync(buffer, CancellationToken.None).WithTimeout());
        Assert.Equal(new byte[] { 0x10, 0x20, 0x30 }, buffer);
        Assert.Equal(1, await transport.ReadAsync(buffer, CancellationToken.None).WithTimeout());
        Assert.Equal(0x40, buffer[0]);
        await write.WithTimeout();
    }

    [Fact]
    public async Task WriteAsync_WrapsInADataFrame()
    {
        using NamedPipeTransport transport = await ConnectAsync();
        await ServerReadAsync(4); // discard the DTR pulse frames

        // Unawaited until the server drains it (0-byte pipe buffers).
        Task write = transport.WriteAsync(new byte[] { 0x81, 0x01 }, CancellationToken.None);

        Assert.Equal(new byte[] { 0x00, 0x02, 0x81, 0x01 }, await ServerReadAsync(4));
        await write.WithTimeout();
    }

    [Fact]
    public async Task ServerGone_ReadReturnsZero()
    {
        using NamedPipeTransport transport = await ConnectAsync();

        _server.Dispose();

        var buffer = new byte[16];
        Assert.Equal(0, await transport.ReadAsync(buffer, CancellationToken.None).WithTimeout());
        // And it keeps saying closed rather than reading a dead pipe.
        Assert.Equal(0, await transport.ReadAsync(buffer, CancellationToken.None).WithTimeout());
    }

    [Fact]
    public async Task ProtocolViolation_ReadReturnsZero()
    {
        using NamedPipeTransport transport = await ConnectAsync();

        Task write = _server.WriteAsync(new byte[] { 0xFF }, 0, 1);

        var buffer = new byte[16];
        Assert.Equal(0, await transport.ReadAsync(buffer, CancellationToken.None).WithTimeout());
        Assert.Contains("0xFF", transport.Violation);
        await write.WithTimeout();
    }

    [Fact]
    public async Task Dispose_UnblocksAPendingRead()
    {
        NamedPipeTransport transport = await ConnectAsync();

        Task<int> pending = transport.ReadAsync(new byte[16], CancellationToken.None);
        Assert.False(pending.IsCompleted);

        transport.Dispose();

        Assert.Equal(0, await pending.WithTimeout());
    }

    [Fact]
    public void NoServer_ConstructorTimesOutWithAClearMessage()
    {
        var ex = Assert.Throws<TimeoutException>(() =>
            new NamedPipeTransport($"riojoy-nobody-{Guid.NewGuid():N}", TimeSpan.FromMilliseconds(200)));
        Assert.Contains("vRIO", ex.Message);
    }

    [Fact]
    public async Task Factory_OpensPipeEndpoints()
    {
        Task accept = _server.WaitForConnectionAsync();
        Task<IRioTransport> client = Task.Run(() => RioTransportFactory.Open($"pipe:{_pipeName}"));
        await accept.WithTimeout();

        using IRioTransport transport = await client.WithTimeout();
        Assert.Equal($@"\\.\pipe\{_pipeName}", transport.Description);
    }
}

internal static class TaskTimeoutExtensions
{
    /// <summary>Await with a test-failure deadline, so a hung pipe fails fast instead of stalling the run.</summary>
    public static async Task<T> WithTimeout<T>(this Task<T> task, int seconds = 5)
    {
        Assert.Same(task, await Task.WhenAny(task, Task.Delay(TimeSpan.FromSeconds(seconds))));
        return await task;
    }

    public static async Task WithTimeout(this Task task, int seconds = 5)
    {
        Assert.Same(task, await Task.WhenAny(task, Task.Delay(TimeSpan.FromSeconds(seconds))));
        await task;
    }
}