Mock gRPC Response Stream for unit testing - C#

Question

I have a function that calls a gRPC endpoint, converts the objects into POCO objects and returns them as a list.

public class ActionPlanConnectionsService : IConnectionService
    {
            #region Fields
    
            /// <summary>
            /// Grpc client
            /// </summary>
            private readonly ConnectionDb.ConnectionDbClient _client;
            #endregion
    
            public ActionPlanConnectionsService(ConnectionDb.ConnectionDbClient channel)
            {
                _client = channel;
            }
    
            public async Task<IEnumerable<Connection>> Get(int actionPlanId, int implementation)
            {
               List<Connection> diagramConnections = new List<Connection>();
               GetConnectionsByIdAndImplementationMessage message = new GetConnectionsByIdAndImplementationMessage
               {
                   ActionPlanId = actionPlanId,
                   Implementation = implementation
               };
    
               using var call = _client.GetAllConnections(message);
               await foreach (ConnectionServiceModel connection in call.ResponseStream.ReadAllAsync())
               {
                   // Never enters here as ResponseStream has no elements when unit testing!!
                   diagramConnections.Add(new Connection(
                       connection.FromActionPlanStepId, connection.ToActionPlanStepId, connection.ActionPlanId,
                       connection.Qualifier, connection.Implementation, connection.Path));
               }
    
               return diagramConnections;
           }
   }

I have been developing a unit test for this function but the list returned always has a count of zero. This is because the ResponseStream has no elements inside of it.

How can I mock the ResponseStream?

My unit test so far:

[Test]
        public async Task GetConnectionsTest()
        {
            // Arrange
            Mock<ConnectionDb.ConnectionDbClient> mockClient = new Mock<ConnectionDb.ConnectionDbClient>();
            Mock<IAsyncStreamReader<ConnectionServiceModel>> mockResponseStream
                = new Mock<IAsyncStreamReader<ConnectionServiceModel>>();

            List<ConnectionServiceModel> connectionServiceModels =
                new List<ConnectionServiceModel>
                {
                    new ConnectionServiceModel
                    {
                        ActionPlanId = 1,
                        FromActionPlanStepId = 1,
                        ToActionPlanStepId = 1,
                        Implementation = 0,
                        Qualifier = 1,
                        Path = " 1;2;3;4;5;6;7;8;9;10;11;12;13;14"
                    }
                };

            var fakeCall = TestCalls.AsyncServerStreamingCall
                (mockResponseStream.Object,
                Task.FromResult(new Metadata()), () => Status.DefaultSuccess,
                () => new Metadata(), () => { });

            mockClient.Setup(m => m.GetAllConnections(
                It.IsAny<GetConnectionsByIdAndImplementationMessage>(),
                null, null, CancellationToken.None)).Returns(fakeCall);

            // Act
            ActionPlanConnectionsService service = new ActionPlanConnectionsService(mockClient.Object);
            IEnumerable<Connection> connections = await service.Get(1, 1);

            // Assert
            
            // CONNECTIONS WILL ALWAYS HAVE 0 Elements as the response isn't setup for it.
        }
    }

Answer 1

Expanding on what @Jan Tattermusch recommended, you probably want to just implement the IAsyncStreamReader instead of trying to mock it. Here's a simple implementation on top of an enumerator.

internal class MyAsyncStreamReader<T> : IAsyncStreamReader<T>
{
    private readonly IEnumerator<T> enumerator;

    public MyAsyncStreamReader(IEnumerable<T> results)
    {
        enumerator = results.GetEnumerator();
    }

    public T Current => enumerator.Current;

    public Task<bool> MoveNext(CancellationToken cancellationToken) =>
        Task.Run(() => enumerator.MoveNext());
}

Then something like this should work:

        [Test]
        public async Task GetConnectionsTest()
        {
            // Arrange
            Mock<ConnectionDb.ConnectionDbClient> mockClient = new Mock<ConnectionDb.ConnectionDbClient>();

            List<ConnectionServiceModel> connectionServiceModels =
                new List<ConnectionServiceModel>
                {
                    new ConnectionServiceModel
                    {
                        ActionPlanId = 1,
                        FromActionPlanStepId = 1,
                        ToActionPlanStepId = 1,
                        Implementation = 0,
                        Qualifier = 1,
                        Path = " 1;2;3;4;5;6;7;8;9;10;11;12;13;14"
                    }
                };

            // Create your async stream reader
            var reader = new MyAsyncStreamReader<ConnectionServiceModel>(connectionServiceModels);

            var fakeCall = TestCalls.AsyncServerStreamingCall(
                reader, // Pass the stream reader into the gRPC call
                Task.FromResult(new Metadata()),
                () => Status.DefaultSuccess,
                () => new Metadata(),
                () => { });

            mockClient.Setup(m => m.GetAllConnections(
                It.IsAny<GetConnectionsByIdAndImplementationMessage>(),
                null, null, CancellationToken.None)).Returns(fakeCall);

            // Act
            ActionPlanConnectionsService service = new ActionPlanConnectionsService(mockClient.Object);
            IEnumerable<Connection> connections = await service.Get(1, 1);

            // Assert
            Assert.NotEmpty(connections); // Should pass
        }
    }

Answer 2

The easiest way is to mock the IAsyncStreamReader enumerator and supply it to the AsyncServerStreamingCall:

var asyncStreamReader = new Mock<IAsyncStreamReader<MyResponse>>();
var asyncServerStreamingCall = new AsyncServerStreamingCall<MyResponse> 
(asyncStreamReader.Object, null, null, null, null, null);

var l = new List<MyResponse>
{
    new MyResponse{},
    new MyResponse(),
    new MyResponse(),
};

var enumerator = l.GetEnumerator();

asyncStreamReader.Setup(x => x.MoveNext(It.IsAny<CancellationToken> 
())).ReturnsAsync(() => enumerator.MoveNext());
asyncStreamReader.Setup(x => x.Current).Returns(() => enumerator.Current);

myGrpcService.Setup(s => s.CallService(It.IsAny<MyRequest>(), null, null,It.IsAny<CancellationToken>())).Returns(asyncServerStreamingCall);