grpc tests WIP

.gitignore

@@ -195,4 +195,5 @@ samples/apps/autogen-studio/autogenstudio/models/test/
 notebook/coding
 
 # dotnet artifacts
-artifacts
+artifacts
+/dotnet/.NCrunch_AutoGen/StoredText

dotnet/src/Microsoft.AutoGen/Microsoft.AutoGen.Runtime.Grpc/GrpcGatewayService.cs

@@ -7,7 +7,7 @@
 namespace Microsoft.AutoGen.Runtime.Grpc;
 
 // gRPC service which handles communication between the agent worker and the cluster.
-internal sealed class GrpcGatewayService : AgentRpc.AgentRpcBase
+public sealed class GrpcGatewayService : AgentRpc.AgentRpcBase
 {
     private readonly GrpcGateway Gateway;
     public GrpcGatewayService(GrpcGateway gateway)

dotnet/test/Microsoft.AutoGen.Runtime.Grpc.Tests/GrpcGatewayServiceTests.cs

@@ -0,0 +1,69 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// GrpcGatewayServiceTests.cs
+
+using FluentAssertions;
+using Microsoft.AutoGen.Abstractions;
+using Microsoft.AutoGen.Runtime.Grpc.Tests.Helpers.Grpc;
+using Microsoft.AutoGen.Runtime.Grpc.Tests.Helpers.Orleans;
+using Microsoft.Extensions.Logging;
+using Moq;
+
+namespace Microsoft.AutoGen.Runtime.Grpc.Tests;
+[Collection(ClusterCollection.Name)]
+public class GrpcGatewayServiceTests
+{
+    private readonly ClusterFixture _fixture;
+
+    public GrpcGatewayServiceTests(ClusterFixture fixture)
+    {
+        _fixture = fixture;
+    }
+    // Test broadcast Event
+    [Fact]
+    public async Task Test_OpenChannel()
+    {
+        var logger = Mock.Of<ILogger<GrpcGateway>>();
+        var gateway = new GrpcGateway(_fixture.Cluster.Client, logger);
+        var service = new GrpcGatewayService(gateway);
+        var callContext = TestServerCallContext.Create();
+        var requestStream = new TestAsyncStreamReader<Message>(callContext);
+        var responseStream = new TestServerStreamWriter<Message>(callContext);
+
+        await service.OpenChannel(requestStream, responseStream, callContext);
+
+        requestStream.AddMessage(new Message {  });
+
+        requestStream.Complete();
+
+        responseStream.Complete();
+
+        var responseMessage = await responseStream.ReadNextAsync();
+        responseMessage.Should().NotBeNull();
+    }
+
+    [Fact]
+    public async Task Test_SaveState()
+    {
+        var logger = Mock.Of<ILogger<GrpcGateway>>();
+        var gateway = new GrpcGateway(_fixture.Cluster.Client, logger);
+        var service = new GrpcGatewayService(gateway);
+        var callContext = TestServerCallContext.Create();
+
+        var response = await service.SaveState(new AgentState { }, callContext);
+
+        response.Should().NotBeNull();
+    }
+
+    [Fact]
+    public async Task Test_GetState()
+    {
+        var logger = Mock.Of<ILogger<GrpcGateway>>();
+        var gateway = new GrpcGateway(_fixture.Cluster.Client, logger);
+        var service = new GrpcGatewayService(gateway);
+        var callContext = TestServerCallContext.Create();
+
+        var response = await service.GetState(new AgentId { }, callContext);
+
+        response.Should().NotBeNull();
+    }
+}

dotnet/test/Microsoft.AutoGen.Runtime.Grpc.Tests/GrpcGatewayTests.cs

@@ -2,7 +2,7 @@
 // GrpcGatewayTests.cs
 
 using Microsoft.AutoGen.Abstractions;
-using Microsoft.AutoGen.Runtime.Grpc.Tests.Helpers;
+using Microsoft.AutoGen.Runtime.Grpc.Tests.Helpers.Orleans;
 using Microsoft.Extensions.Logging;
 using Moq;
 

dotnet/test/Microsoft.AutoGen.Runtime.Grpc.Tests/Helpers/Grpc/TestAsyncStreamReader.cs

@@ -0,0 +1,63 @@
+#pragma warning disable IDE0073
+// Copyright 2019 The gRPC Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using System.Threading.Channels;
+using Grpc.Core;
+
+namespace Microsoft.AutoGen.Runtime.Grpc.Tests.Helpers.Grpc;
+
+public class TestAsyncStreamReader<T> : IAsyncStreamReader<T> where T : class
+{
+    private readonly Channel<T> _channel;
+    private readonly ServerCallContext _serverCallContext;
+
+    public T Current { get; private set; } = null!;
+
+    public TestAsyncStreamReader(ServerCallContext serverCallContext)
+    {
+        _channel = Channel.CreateUnbounded<T>();
+        _serverCallContext = serverCallContext;
+    }
+
+    public void AddMessage(T message)
+    {
+        if (!_channel.Writer.TryWrite(message))
+        {
+            throw new InvalidOperationException("Unable to write message.");
+        }
+    }
+
+    public void Complete()
+    {
+        _channel.Writer.Complete();
+    }
+
+    public async Task<bool> MoveNext(CancellationToken cancellationToken)
+    {
+        _serverCallContext.CancellationToken.ThrowIfCancellationRequested();
+
+        if (await _channel.Reader.WaitToReadAsync(cancellationToken) &&
+            _channel.Reader.TryRead(out var message))
+        {
+            Current = message;
+            return true;
+        }
+        else
+        {
+            Current = null!;
+            return false;
+        }
+    }
+}

dotnet/test/Microsoft.AutoGen.Runtime.Grpc.Tests/Helpers/Grpc/TestServerCallContext.cs

@@ -0,0 +1,73 @@
+#pragma warning disable IDE0073
+// Copyright 2019 The gRPC Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using Grpc.Core;
+
+namespace Microsoft.AutoGen.Runtime.Grpc.Tests.Helpers.Grpc;
+
+public class TestServerCallContext : ServerCallContext
+{
+    private readonly Metadata _requestHeaders;
+    private readonly CancellationToken _cancellationToken;
+    private readonly Metadata _responseTrailers;
+    private readonly AuthContext _authContext;
+    private readonly Dictionary<object, object> _userState;
+    private WriteOptions? _writeOptions;
+
+    public Metadata? ResponseHeaders { get; private set; }
+
+    private TestServerCallContext(Metadata requestHeaders, CancellationToken cancellationToken)
+    {
+        _requestHeaders = requestHeaders;
+        _cancellationToken = cancellationToken;
+        _responseTrailers = new Metadata();
+        _authContext = new AuthContext(string.Empty, new Dictionary<string, List<AuthProperty>>());
+        _userState = new Dictionary<object, object>();
+    }
+
+    protected override string MethodCore => "MethodName";
+    protected override string HostCore => "HostName";
+    protected override string PeerCore => "PeerName";
+    protected override DateTime DeadlineCore { get; }
+    protected override Metadata RequestHeadersCore => _requestHeaders;
+    protected override CancellationToken CancellationTokenCore => _cancellationToken;
+    protected override Metadata ResponseTrailersCore => _responseTrailers;
+    protected override Status StatusCore { get; set; }
+    protected override WriteOptions? WriteOptionsCore { get => _writeOptions; set { _writeOptions = value; } }
+    protected override AuthContext AuthContextCore => _authContext;
+
+    protected override ContextPropagationToken CreatePropagationTokenCore(ContextPropagationOptions? options)
+    {
+        throw new NotImplementedException();
+    }
+
+    protected override Task WriteResponseHeadersAsyncCore(Metadata responseHeaders)
+    {
+        if (ResponseHeaders != null)
+        {
+            throw new InvalidOperationException("Response headers have already been written.");
+        }
+
+        ResponseHeaders = responseHeaders;
+        return Task.CompletedTask;
+    }
+
+    protected override IDictionary<object, object> UserStateCore => _userState;
+
+    public static TestServerCallContext Create(Metadata? requestHeaders = null, CancellationToken cancellationToken = default)
+    {
+        return new TestServerCallContext(requestHeaders ?? new Metadata(), cancellationToken);
+    }
+}

dotnet/test/Microsoft.AutoGen.Runtime.Grpc.Tests/Helpers/Grpc/TestServerStreamWriter.cs

@@ -0,0 +1,81 @@
+#pragma warning disable IDE0073
+// Copyright 2019 The gRPC Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using System.Threading.Channels;
+using Grpc.Core;
+
+namespace Microsoft.AutoGen.Runtime.Grpc.Tests.Helpers.Grpc;
+
+public class TestServerStreamWriter<T> : IServerStreamWriter<T> where T : class
+{
+    private readonly ServerCallContext _serverCallContext;
+    private readonly Channel<T> _channel;
+
+    public WriteOptions? WriteOptions { get; set; }
+
+    public TestServerStreamWriter(ServerCallContext serverCallContext)
+    {
+        _channel = Channel.CreateUnbounded<T>();
+
+        _serverCallContext = serverCallContext;
+    }
+
+    public void Complete()
+    {
+        _channel.Writer.Complete();
+    }
+
+    public IAsyncEnumerable<T> ReadAllAsync()
+    {
+        return _channel.Reader.ReadAllAsync();
+    }
+
+    public async Task<T?> ReadNextAsync()
+    {
+        if (await _channel.Reader.WaitToReadAsync())
+        {
+            _channel.Reader.TryRead(out var message);
+            return message;
+        }
+        else
+        {
+            return null;
+        }
+    }
+
+    public Task WriteAsync(T message, CancellationToken cancellationToken)
+    {
+        if (cancellationToken.IsCancellationRequested)
+        {
+            return Task.FromCanceled(cancellationToken);
+        }
+        if (_serverCallContext.CancellationToken.IsCancellationRequested)
+        {
+            return Task.FromCanceled(_serverCallContext.CancellationToken);
+        }
+
+        if (!_channel.Writer.TryWrite(message))
+        {
+            throw new InvalidOperationException("Unable to write message.");
+        }
+
+        return Task.CompletedTask;
+    }
+
+    public Task WriteAsync(T message)
+    {
+        return WriteAsync(message, CancellationToken.None);
+    }
+}

dotnet/test/Microsoft.AutoGen.Runtime.Grpc.Tests/Helpers/ClusterCollection.cs → dotnet/test/Microsoft.AutoGen.Runtime.Grpc.Tests/Helpers/Orleans/ClusterCollection.cs

@@ -1,7 +1,7 @@
 // Copyright (c) Microsoft Corporation. All rights reserved.
 // ClusterCollection.cs
 
-namespace Microsoft.AutoGen.Runtime.Grpc.Tests.Helpers;
+namespace Microsoft.AutoGen.Runtime.Grpc.Tests.Helpers.Orleans;
 
 [CollectionDefinition(Name)]
 public sealed class ClusterCollection : ICollectionFixture<ClusterFixture>

dotnet/test/Microsoft.AutoGen.Runtime.Grpc.Tests/Helpers/ClusterFixture.cs → dotnet/test/Microsoft.AutoGen.Runtime.Grpc.Tests/Helpers/Orleans/ClusterFixture.cs

@@ -3,7 +3,7 @@
 
 using Orleans.TestingHost;
 
-namespace Microsoft.AutoGen.Runtime.Grpc.Tests.Helpers;
+namespace Microsoft.AutoGen.Runtime.Grpc.Tests.Helpers.Orleans;
 
 public sealed class ClusterFixture : IDisposable
 {