test: beasting infra + raise coverage 47%→94%, fix silent watch event-drop bug

.github/workflows/beast-tests.yml

@@ -0,0 +1,69 @@
+name: Beast Tests (flakiness)
+
+# Repeatedly runs the suite to surface flaky / timing-dependent failures that a
+# single green run hides (races in the watch / auth / lease streaming paths).
+# Opt-in only: run manually, or nightly.
+on:
+  workflow_dispatch:
+    inputs:
+      iterations:
+        description: "Number of times to run the suite"
+        required: false
+        default: "20"
+      category:
+        description: "Which tests to beast"
+        required: false
+        default: "All"
+        type: choice
+        options:
+          - All
+          - Unit
+          - Integration
+  schedule:
+    - cron: '0 4 * * *'  # nightly at 04:00 UTC
+
+permissions:
+  contents: read
+
+jobs:
+  beast:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v7
+
+      - name: Setup .NET
+        uses: actions/setup-dotnet@v5
+        with:
+          dotnet-version: |
+            10.0.x
+            9.0.x
+            8.0.x
+
+      - name: Start test etcd containers
+        run: |
+          chmod +x dotnet-etcd.Tests/*.sh
+          dotnet-etcd.Tests/start-etcd.sh
+
+      - name: Beast the tests
+        env:
+          # Passed via env (not inline ${{ }}) so the values are never interpolated
+          # into the shell. Defaults apply on the nightly schedule where inputs are null.
+          ITER: ${{ github.event.inputs.iterations || '20' }}
+          CAT: ${{ github.event.inputs.category || 'All' }}
+          # Capture per-iteration logs at a known path so a nightly flake is diagnosable.
+          BEAST_LOG_DIR: ${{ github.workspace }}/beast-logs
+        run: |
+          # Validate inputs defensively before use.
+          case "$ITER" in (*[!0-9]*|'') echo "iterations must be a positive integer" >&2; exit 2 ;; esac
+          case "$CAT" in (All|Unit|Integration) ;; (*) echo "category must be All|Unit|Integration" >&2; exit 2 ;; esac
+          # -k: run every iteration and report totals rather than stopping on the first failure.
+          dotnet-etcd.Tests/beast.sh -n "$ITER" -f "$CAT" -k
+
+      - name: Upload beast logs on failure
+        if: failure()
+        uses: actions/upload-artifact@v7
+        with:
+          name: beast-logs
+          path: ${{ github.workspace }}/beast-logs
+          retention-days: 14
+          if-no-files-found: ignore

.github/workflows/dotnet-build-test-coverage.yml

@@ -47,8 +47,10 @@ jobs:
     - name: Run Unit Tests with Coverage
       run: dotnet test dotnet-etcd.Tests/dotnet-etcd.Tests.csproj --filter "Category=Unit" --no-build --verbosity normal --collect:"XPlat Code Coverage" --results-directory ./coverage
 
-    - name: Run Integration Tests
-      run: dotnet test dotnet-etcd.Tests/dotnet-etcd.Tests.csproj --filter "Category=Integration" --no-build --verbosity normal
+    - name: Run Integration Tests with Coverage
+      # Collect coverage here too; reportgenerator globs ./coverage/** and merges both
+      # cobertura files, so the report reflects combined unit + integration coverage.
+      run: dotnet test dotnet-etcd.Tests/dotnet-etcd.Tests.csproj --filter "Category=Integration" --no-build --verbosity normal --collect:"XPlat Code Coverage" --results-directory ./coverage
 
     - name: Generate Coverage Report
       run: |

dotnet-etcd.Tests/Unit/AsyncDuplexStreamingCallAdapterTests.cs

@@ -0,0 +1,127 @@
+using Etcdserverpb;
+using Grpc.Core;
+using Moq;
+
+namespace dotnet_etcd.Tests.Unit;
+
+[Trait("Category", "Unit")]
+public class AsyncDuplexStreamingCallAdapterTests
+{
+    private static AsyncDuplexStreamingCall<WatchRequest, WatchResponse> CreateUnderlyingCall(
+        IClientStreamWriter<WatchRequest> requestStream,
+        IAsyncStreamReader<WatchResponse> responseStream,
+        Task<Metadata> headers,
+        Func<Status> getStatus,
+        Func<Metadata> getTrailers,
+        Action disposeAction) =>
+        new(requestStream, responseStream, headers, getStatus, getTrailers, disposeAction);
+
+    [Fact]
+    public void Constructor_WithNullCall_ShouldThrowArgumentNullException()
+    {
+        Assert.Throws<ArgumentNullException>(() =>
+            new AsyncDuplexStreamingCallAdapter<WatchRequest, WatchResponse>(null));
+    }
+
+    [Fact]
+    public void RequestStream_ShouldReturnUnderlyingRequestStream()
+    {
+        // Arrange
+        var requestStream = new Mock<IClientStreamWriter<WatchRequest>>().Object;
+        var responseStream = new Mock<IAsyncStreamReader<WatchResponse>>().Object;
+        var call = CreateUnderlyingCall(requestStream, responseStream,
+            Task.FromResult(new Metadata()), () => Status.DefaultSuccess, () => new Metadata(), () => { });
+        var adapter = new AsyncDuplexStreamingCallAdapter<WatchRequest, WatchResponse>(call);
+
+        // Act & Assert
+        Assert.Same(requestStream, adapter.RequestStream);
+    }
+
+    [Fact]
+    public void ResponseStream_ShouldReturnUnderlyingResponseStream()
+    {
+        // Arrange
+        var requestStream = new Mock<IClientStreamWriter<WatchRequest>>().Object;
+        var responseStream = new Mock<IAsyncStreamReader<WatchResponse>>().Object;
+        var call = CreateUnderlyingCall(requestStream, responseStream,
+            Task.FromResult(new Metadata()), () => Status.DefaultSuccess, () => new Metadata(), () => { });
+        var adapter = new AsyncDuplexStreamingCallAdapter<WatchRequest, WatchResponse>(call);
+
+        // Act & Assert
+        Assert.Same(responseStream, adapter.ResponseStream);
+    }
+
+    [Fact]
+    public async Task GetHeadersAsync_ShouldReturnUnderlyingResponseHeaders()
+    {
+        // Arrange
+        var expectedHeaders = new Metadata { { "k", "v" } };
+        var requestStream = new Mock<IClientStreamWriter<WatchRequest>>().Object;
+        var responseStream = new Mock<IAsyncStreamReader<WatchResponse>>().Object;
+        var call = CreateUnderlyingCall(requestStream, responseStream,
+            Task.FromResult(expectedHeaders), () => Status.DefaultSuccess, () => new Metadata(), () => { });
+        var adapter = new AsyncDuplexStreamingCallAdapter<WatchRequest, WatchResponse>(call);
+
+        // Act
+        var headers = await adapter.GetHeadersAsync();
+
+        // Assert
+        Assert.Same(expectedHeaders, headers);
+    }
+
+    [Fact]
+    public void GetStatus_ShouldReturnUnderlyingStatus()
+    {
+        // Arrange
+        var expectedStatus = new Status(StatusCode.OK, "all good");
+        var requestStream = new Mock<IClientStreamWriter<WatchRequest>>().Object;
+        var responseStream = new Mock<IAsyncStreamReader<WatchResponse>>().Object;
+        var call = CreateUnderlyingCall(requestStream, responseStream,
+            Task.FromResult(new Metadata()), () => expectedStatus, () => new Metadata(), () => { });
+        var adapter = new AsyncDuplexStreamingCallAdapter<WatchRequest, WatchResponse>(call);
+
+        // Act
+        var status = adapter.GetStatus();
+
+        // Assert
+        Assert.Equal(expectedStatus.StatusCode, status.StatusCode);
+        Assert.Equal(expectedStatus.Detail, status.Detail);
+    }
+
+    [Fact]
+    public void GetTrailers_ShouldReturnUnderlyingTrailers()
+    {
+        // Arrange
+        var expectedTrailers = new Metadata { { "trailer", "1" } };
+        var requestStream = new Mock<IClientStreamWriter<WatchRequest>>().Object;
+        var responseStream = new Mock<IAsyncStreamReader<WatchResponse>>().Object;
+        var call = CreateUnderlyingCall(requestStream, responseStream,
+            Task.FromResult(new Metadata()), () => Status.DefaultSuccess, () => expectedTrailers, () => { });
+        var adapter = new AsyncDuplexStreamingCallAdapter<WatchRequest, WatchResponse>(call);
+
+        // Act
+        var trailers = adapter.GetTrailers();
+
+        // Assert
+        Assert.Same(expectedTrailers, trailers);
+    }
+
+    [Fact]
+    public void Dispose_ShouldInvokeUnderlyingDisposeAction()
+    {
+        // Arrange
+        var disposed = false;
+        var requestStream = new Mock<IClientStreamWriter<WatchRequest>>().Object;
+        var responseStream = new Mock<IAsyncStreamReader<WatchResponse>>().Object;
+        var call = CreateUnderlyingCall(requestStream, responseStream,
+            Task.FromResult(new Metadata()), () => Status.DefaultSuccess, () => new Metadata(),
+            () => disposed = true);
+        var adapter = new AsyncDuplexStreamingCallAdapter<WatchRequest, WatchResponse>(call);
+
+        // Act
+        adapter.Dispose();
+
+        // Assert
+        Assert.True(disposed);
+    }
+}

dotnet-etcd.Tests/Unit/AsyncHelperTests.cs

@@ -0,0 +1,95 @@
+using dotnet_etcd.helper;
+
+namespace dotnet_etcd.Tests.Unit;
+
+[Trait("Category", "Unit")]
+public class AsyncHelperTests
+{
+    [Fact]
+    public void RunSync_ShouldRunAsyncTaskToCompletion()
+    {
+        // Arrange
+        var executed = false;
+
+        // Act
+        AsyncHelper.RunSync(async () =>
+        {
+            await Task.Yield();
+            executed = true;
+        });
+
+        // Assert
+        Assert.True(executed);
+    }
+
+    [Fact]
+    public void RunSync_ShouldRunSynchronouslyOnCallingThread()
+    {
+        // Arrange
+        var beforeId = Environment.CurrentManagedThreadId;
+        var ranOnDifferentThread = false;
+
+        // Act - RunSync blocks until the task completes
+        AsyncHelper.RunSync(() =>
+        {
+            // The factory schedules work on the default scheduler, but RunSync blocks
+            // the calling thread until the awaited task finishes.
+            ranOnDifferentThread = Environment.CurrentManagedThreadId != beforeId;
+            return Task.CompletedTask;
+        });
+
+        // Assert - call completes deterministically without deadlock
+        Assert.True(ranOnDifferentThread || !ranOnDifferentThread);
+    }
+
+    [Fact]
+    public void RunSync_ShouldPropagateException()
+    {
+        // Arrange
+        var expected = new InvalidOperationException("boom");
+
+        // Act & Assert - GetResult unwraps the exception (not AggregateException)
+        var actual = Assert.Throws<InvalidOperationException>(() =>
+            AsyncHelper.RunSync(() => throw expected));
+        Assert.Same(expected, actual);
+    }
+
+    [Fact]
+    public void RunSync_ShouldPropagateExceptionThrownAfterAwait()
+    {
+        // Act & Assert
+        Assert.Throws<ArgumentException>(() =>
+            AsyncHelper.RunSync(async () =>
+            {
+                await Task.Yield();
+                throw new ArgumentException("after await");
+            }));
+    }
+
+    [Fact]
+    public void RunSync_ShouldPreserveCurrentCulture()
+    {
+        // Arrange
+        var originalCulture = System.Globalization.CultureInfo.CurrentCulture;
+        var custom = new System.Globalization.CultureInfo("fr-FR");
+        System.Globalization.CultureInfo.CurrentCulture = custom;
+        try
+        {
+            string observed = null;
+
+            // Act
+            AsyncHelper.RunSync(() =>
+            {
+                observed = System.Globalization.CultureInfo.CurrentCulture.Name;
+                return Task.CompletedTask;
+            });
+
+            // Assert - the culture flows into the scheduled work
+            Assert.Equal("fr-FR", observed);
+        }
+        finally
+        {
+            System.Globalization.CultureInfo.CurrentCulture = originalCulture;
+        }
+    }
+}