test(xunit): CurrentErrorItemCount default + no-over-read (#49) contract tests

CHANGELOG.md

@@ -41,6 +41,15 @@ public API only — no breaking change.
 - `DisposableStageContractTests<TSut>` — an opt-in xUnit contract-test base verifying a stage
   throws `ObjectDisposedException` after `Dispose()`/`DisposeAsync()` (the 0.17 use-after-dispose
   guard) and that disposing twice is a harmless no-op.
+- Counter contract tests on `ExtractorBaseContractTests`, `LoaderBaseContractTests`, and
+  `TransformerBaseContractTests`: `CurrentItemCount` / `CurrentSkippedItemCount` /
+  `CurrentErrorItemCount` default-to-zero and skip-count-tracking assertions, inherited free by
+  every downstream contract-test class (#248).
+- "No over-read" contract tests (#49) on all three base classes: a stage must stop pulling from
+  its source once `MaximumItemCount` is reached (≤ M+1 reads) or the run is cancelled, and a
+  pre-cancelled token must read nothing. Extractors opt in by overriding the new
+  `CreateSutOverSource` factory (a no-op by default for extractors whose source is not an
+  injectable sequence).
 
 ### Changed
 

src/Wolfgang.Etl.TestKit.Xunit/ExtractorBaseContractTests.cs

@@ -813,4 +813,115 @@ public async Task ExtractAsync_CurrentSkippedItemCount_reflects_the_number_of_it
 
         Assert.Equal(2, sut.CurrentSkippedItemCount);
     }
+
+    /// <summary>
+    /// Verifies that <c>CurrentErrorItemCount</c> is zero on a freshly created extractor, before
+    /// any item has failed (Abstractions 0.18.0 error hook).
+    /// </summary>
+    [Fact]
+    public void CurrentErrorItemCount_defaults_to_zero()
+    {
+        var sut = CreateSut();
+
+        Assert.Equal(0, sut.CurrentErrorItemCount);
+    }
+
+
+
+    // ------------------------------------------------------------------
+    // No over-read (issue #49)
+    // ------------------------------------------------------------------
+
+    /// <summary>
+    /// Override to enable the "no over-read" tests for an extractor that reads from an
+    /// injectable in-memory sequence. Return an extractor that draws its items from
+    /// <paramref name="source"/>, or <see langword="null"/> (the default) to skip those tests —
+    /// appropriate for an extractor whose source is a connection or handle that cannot be a
+    /// caller-supplied sequence.
+    /// </summary>
+    /// <param name="source">The sequence the returned extractor must read from.</param>
+    protected virtual TSut? CreateSutOverSource(IEnumerable<TItem> source) => default;
+
+    /// <summary>
+    /// Verifies that once <c>MaximumItemCount</c> is reached the extractor stops pulling from its
+    /// source rather than draining it — at most M+1 reads (the +1 discovers the limit). Skipped
+    /// unless <see cref="CreateSutOverSource"/> is overridden.
+    /// </summary>
+    [Fact]
+    public async Task ExtractAsync_does_not_over_read_past_MaximumItemCount_Async()
+    {
+        var counter = new PullCounter();
+        var sut = CreateSutOverSource(counter.CountSync(CreateExpectedItems()));
+        if (sut is null)
+        {
+            return;
+        }
+
+        sut.MaximumItemCount = 3;
+
+        await sut.ExtractAsync().ToListAsync().ConfigureAwait(false);
+
+        Assert.True(counter.Count <= 4, $"Expected at most 4 upstream reads, saw {counter.Count}.");
+    }
+
+    // Cancel() runs synchronously to cancel mid-enumeration; CancelAsync is net8.0+ only and
+    // this base targets net462+.
+#pragma warning disable CA1849, VSTHRD103
+    /// <summary>
+    /// Verifies that cancelling mid-enumeration stops the extractor pulling from its source at
+    /// the next check. Skipped unless <see cref="CreateSutOverSource"/> is overridden.
+    /// </summary>
+    [Fact]
+    public async Task ExtractAsync_stops_reading_on_cancellation_Async()
+    {
+        using var cts = new CancellationTokenSource();
+        var counter = new PullCounter();
+        var sut = CreateSutOverSource(counter.CountSync(CreateExpectedItems()));
+        if (sut is null)
+        {
+            return;
+        }
+
+        var seen = 0;
+
+        try
+        {
+            await foreach (var _ in sut.ExtractAsync(cts.Token).ConfigureAwait(false))
+            {
+                if (++seen == 3)
+                {
+                    cts.Cancel();
+                }
+            }
+        }
+        catch (OperationCanceledException)
+        {
+        }
+
+        Assert.True(counter.Count <= 4, $"Expected at most 4 upstream reads, saw {counter.Count}.");
+    }
+#pragma warning restore CA1849, VSTHRD103
+
+    /// <summary>
+    /// Verifies that a pre-cancelled token short-circuits the extractor before it pulls any item
+    /// from its source. Skipped unless <see cref="CreateSutOverSource"/> is overridden.
+    /// </summary>
+    [Fact]
+    public async Task ExtractAsync_with_a_pre_cancelled_token_reads_nothing_Async()
+    {
+        var token = new CancellationToken(canceled: true);
+        var counter = new PullCounter();
+        var sut = CreateSutOverSource(counter.CountSync(CreateExpectedItems()));
+        if (sut is null)
+        {
+            return;
+        }
+
+        await Assert.ThrowsAnyAsync<OperationCanceledException>
+        (
+            () => sut.ExtractAsync(token).ToListAsync(token).AsTask()
+        ).ConfigureAwait(false);
+
+        Assert.Equal(0, counter.Count);
+    }
 }

src/Wolfgang.Etl.TestKit.Xunit/LoaderBaseContractTests.cs

@@ -877,4 +877,89 @@ public async Task LoadAsync_CurrentSkippedItemCount_reflects_the_number_of_items
 
         Assert.Equal(2, sut.CurrentSkippedItemCount);
     }
+
+    /// <summary>
+    /// Verifies that <c>CurrentErrorItemCount</c> is zero on a freshly created loader, before any
+    /// item has failed (Abstractions 0.18.0 error hook).
+    /// </summary>
+    [Fact]
+    public void CurrentErrorItemCount_defaults_to_zero()
+    {
+        var sut = CreateSut();
+
+        Assert.Equal(0, sut.CurrentErrorItemCount);
+    }
+
+
+
+    // ------------------------------------------------------------------
+    // No over-read (issue #49)
+    // ------------------------------------------------------------------
+
+    /// <summary>
+    /// Verifies that once <c>MaximumItemCount</c> is reached the loader stops pulling from its
+    /// source rather than draining it — at most M+1 reads (the +1 discovers the limit).
+    /// </summary>
+    [Fact]
+    public async Task LoadAsync_does_not_over_read_past_MaximumItemCount_Async()
+    {
+        var sut = CreateSut();
+        sut.MaximumItemCount = 3;
+        var counter = new PullCounter();
+
+        await sut.LoadAsync(counter.CountAsync(CreateInputItemsAsync())).ConfigureAwait(false);
+
+        Assert.True(counter.Count <= 4, $"Expected at most 4 upstream reads, saw {counter.Count}.");
+    }
+
+    // Cancel() runs synchronously to cancel mid-enumeration; CancelAsync is net8.0+ only and
+    // this base targets net462+.
+#pragma warning disable CA1849, VSTHRD103
+    /// <summary>
+    /// Verifies that cancelling mid-run stops the loader pulling from its source at the next
+    /// check, rather than draining the already-available items.
+    /// </summary>
+    [Fact]
+    public async Task LoadAsync_stops_reading_on_cancellation_Async()
+    {
+        var sut = CreateSut();
+        using var cts = new CancellationTokenSource();
+        var counter = new PullCounter();
+        var source = counter.CountAsync
+        (
+            CreateInputItemsAsync(),
+            onPull: () => { if (counter.Count == 3) { cts.Cancel(); } },
+            token: cts.Token
+        );
+
+        try
+        {
+            await sut.LoadAsync(source, cts.Token).ConfigureAwait(false);
+        }
+        catch (OperationCanceledException)
+        {
+        }
+
+        Assert.True(counter.Count <= 4, $"Expected at most 4 upstream reads, saw {counter.Count}.");
+    }
+#pragma warning restore CA1849, VSTHRD103
+
+    /// <summary>
+    /// Verifies that a pre-cancelled token short-circuits the loader before it pulls any item
+    /// from its source.
+    /// </summary>
+    [Fact]
+    public async Task LoadAsync_with_a_pre_cancelled_token_reads_nothing_Async()
+    {
+        var sut = CreateSut();
+        var token = new CancellationToken(canceled: true);
+        var counter = new PullCounter();
+
+        await Assert.ThrowsAnyAsync<OperationCanceledException>
+        (
+            () => sut.LoadAsync(counter.CountAsync(CreateInputItemsAsync(), token: token), token)
+        ).ConfigureAwait(false);
+
+        Assert.Equal(0, counter.Count);
+    }
 }

src/Wolfgang.Etl.TestKit.Xunit/PublicAPI.Unshipped.txt

@@ -32,3 +32,16 @@ Wolfgang.Etl.TestKit.Xunit.LoaderBaseContractTests<TSut, TItem, TProgress>.LoadA
 Wolfgang.Etl.TestKit.Xunit.TransformerBaseContractTests<TSut, TItem, TProgress>.CurrentItemCount_defaults_to_zero() -> void
 Wolfgang.Etl.TestKit.Xunit.TransformerBaseContractTests<TSut, TItem, TProgress>.CurrentSkippedItemCount_defaults_to_zero() -> void
 Wolfgang.Etl.TestKit.Xunit.TransformerBaseContractTests<TSut, TItem, TProgress>.TransformAsync_CurrentSkippedItemCount_reflects_the_number_of_items_skipped_Async() -> System.Threading.Tasks.Task!
+Wolfgang.Etl.TestKit.Xunit.ExtractorBaseContractTests<TSut, TItem, TProgress>.CurrentErrorItemCount_defaults_to_zero() -> void
+virtual Wolfgang.Etl.TestKit.Xunit.ExtractorBaseContractTests<TSut, TItem, TProgress>.CreateSutOverSource(System.Collections.Generic.IEnumerable<TItem>! source) -> TSut?
+Wolfgang.Etl.TestKit.Xunit.ExtractorBaseContractTests<TSut, TItem, TProgress>.ExtractAsync_does_not_over_read_past_MaximumItemCount_Async() -> System.Threading.Tasks.Task!
+Wolfgang.Etl.TestKit.Xunit.ExtractorBaseContractTests<TSut, TItem, TProgress>.ExtractAsync_stops_reading_on_cancellation_Async() -> System.Threading.Tasks.Task!
+Wolfgang.Etl.TestKit.Xunit.ExtractorBaseContractTests<TSut, TItem, TProgress>.ExtractAsync_with_a_pre_cancelled_token_reads_nothing_Async() -> System.Threading.Tasks.Task!
+Wolfgang.Etl.TestKit.Xunit.LoaderBaseContractTests<TSut, TItem, TProgress>.CurrentErrorItemCount_defaults_to_zero() -> void
+Wolfgang.Etl.TestKit.Xunit.LoaderBaseContractTests<TSut, TItem, TProgress>.LoadAsync_does_not_over_read_past_MaximumItemCount_Async() -> System.Threading.Tasks.Task!
+Wolfgang.Etl.TestKit.Xunit.LoaderBaseContractTests<TSut, TItem, TProgress>.LoadAsync_stops_reading_on_cancellation_Async() -> System.Threading.Tasks.Task!
+Wolfgang.Etl.TestKit.Xunit.LoaderBaseContractTests<TSut, TItem, TProgress>.LoadAsync_with_a_pre_cancelled_token_reads_nothing_Async() -> System.Threading.Tasks.Task!
+Wolfgang.Etl.TestKit.Xunit.TransformerBaseContractTests<TSut, TItem, TProgress>.CurrentErrorItemCount_defaults_to_zero() -> void
+Wolfgang.Etl.TestKit.Xunit.TransformerBaseContractTests<TSut, TItem, TProgress>.TransformAsync_does_not_over_read_past_MaximumItemCount_Async() -> System.Threading.Tasks.Task!
+Wolfgang.Etl.TestKit.Xunit.TransformerBaseContractTests<TSut, TItem, TProgress>.TransformAsync_stops_reading_on_cancellation_Async() -> System.Threading.Tasks.Task!
+Wolfgang.Etl.TestKit.Xunit.TransformerBaseContractTests<TSut, TItem, TProgress>.TransformAsync_with_a_pre_cancelled_token_reads_nothing_Async() -> System.Threading.Tasks.Task!

src/Wolfgang.Etl.TestKit.Xunit/PullCounter.cs

@@ -0,0 +1,54 @@
+using System;
+using System.Collections.Generic;
+using System.Runtime.CompilerServices;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Wolfgang.Etl.TestKit.Xunit;
+
+/// <summary>
+/// Counts how many items a stage pulls from a wrapped source, so the "no over-read" contract
+/// tests (issue #49) can assert that a stage stops reading upstream once
+/// <c>MaximumItemCount</c> is reached or the run is cancelled, rather than draining the whole
+/// source. Internal test-harness helper.
+/// </summary>
+internal sealed class PullCounter
+{
+    private int _count;
+
+    /// <summary>The number of items pulled from the wrapped source so far.</summary>
+    public int Count => Volatile.Read(ref _count);
+
+    /// <summary>
+    /// Wraps an async source, incrementing <see cref="Count"/> as each item is pulled and
+    /// awaiting <paramref name="onPull"/> after each increment (used to trigger cancellation
+    /// after a chosen number of pulls).
+    /// </summary>
+    public async IAsyncEnumerable<T> CountAsync<T>
+    (
+        IAsyncEnumerable<T> source,
+        Action? onPull = null,
+        [EnumeratorCancellation] CancellationToken token = default
+    )
+    {
+        await foreach (var item in source.WithCancellation(token).ConfigureAwait(false))
+        {
+            _ = Interlocked.Increment(ref _count);
+            onPull?.Invoke();
+            yield return item;
+        }
+    }
+
+    /// <summary>
+    /// Wraps a synchronous source (an extractor's in-memory sequence), incrementing
+    /// <see cref="Count"/> as each item is pulled.
+    /// </summary>
+    public IEnumerable<T> CountSync<T>(IEnumerable<T> source)
+    {
+        foreach (var item in source)
+        {
+            _ = Interlocked.Increment(ref _count);
+            yield return item;
+        }
+    }
+}

src/Wolfgang.Etl.TestKit.Xunit/TransformerBaseContractTests.cs

@@ -912,4 +912,92 @@ public async Task TransformAsync_CurrentSkippedItemCount_reflects_the_number_of_
 
         Assert.Equal(2, sut.CurrentSkippedItemCount);
     }
+
+    /// <summary>
+    /// Verifies that <c>CurrentErrorItemCount</c> is zero on a freshly created transformer, before
+    /// any item has failed (Abstractions 0.18.0 error hook).
+    /// </summary>
+    [Fact]
+    public void CurrentErrorItemCount_defaults_to_zero()
+    {
+        var sut = CreateSut();
+
+        Assert.Equal(0, sut.CurrentErrorItemCount);
+    }
+
+
+
+    // ------------------------------------------------------------------
+    // No over-read (issue #49)
+    // ------------------------------------------------------------------
+
+    /// <summary>
+    /// Verifies that once <c>MaximumItemCount</c> is reached the transformer stops pulling from
+    /// its source rather than draining it — at most M+1 reads (the +1 discovers the limit).
+    /// </summary>
+    [Fact]
+    public async Task TransformAsync_does_not_over_read_past_MaximumItemCount_Async()
+    {
+        var sut = CreateSut();
+        sut.MaximumItemCount = 3;
+        var counter = new PullCounter();
+
+        await sut.TransformAsync(counter.CountAsync(CreateInputItemsAsync())).ToListAsync().ConfigureAwait(false);
+
+        Assert.True(counter.Count <= 4, $"Expected at most 4 upstream reads, saw {counter.Count}.");
+    }
+
+    // Cancel() runs synchronously to cancel mid-enumeration; CancelAsync is net8.0+ only and
+    // this base targets net462+.
+#pragma warning disable CA1849, VSTHRD103
+    /// <summary>
+    /// Verifies that cancelling mid-run stops the transformer pulling from its source at the
+    /// next check, rather than draining the already-available items.
+    /// </summary>
+    [Fact]
+    public async Task TransformAsync_stops_reading_on_cancellation_Async()
+    {
+        var sut = CreateSut();
+        using var cts = new CancellationTokenSource();
+        var counter = new PullCounter();
+        var seen = 0;
+
+        try
+        {
+            var source = counter.CountAsync(CreateInputItemsAsync(), token: cts.Token);
+
+            await foreach (var _ in sut.TransformAsync(source, cts.Token).ConfigureAwait(false))
+            {
+                if (++seen == 3)
+                {
+                    cts.Cancel();
+                }
+            }
+        }
+        catch (OperationCanceledException)
+        {
+        }
+
+        Assert.True(counter.Count <= 4, $"Expected at most 4 upstream reads, saw {counter.Count}.");
+    }
+#pragma warning restore CA1849, VSTHRD103
+
+    /// <summary>
+    /// Verifies that a pre-cancelled token short-circuits the transformer before it pulls any
+    /// item from its source.
+    /// </summary>
+    [Fact]
+    public async Task TransformAsync_with_a_pre_cancelled_token_reads_nothing_Async()
+    {
+        var sut = CreateSut();
+        var token = new CancellationToken(canceled: true);
+        var counter = new PullCounter();
+
+        await Assert.ThrowsAnyAsync<OperationCanceledException>
+        (
+            () => sut.TransformAsync(counter.CountAsync(CreateInputItemsAsync(), token: token), token).ToListAsync(token).AsTask()
+        ).ConfigureAwait(false);
+
+        Assert.Equal(0, counter.Count);
+    }
 }

tests/Wolfgang.Etl.TestKit.Xunit.Tests.Unit/TestExtractorContractTests.cs

@@ -27,6 +27,10 @@ protected override IReadOnlyList<int> CreateExpectedItems() =>
     protected override TestExtractor<int> CreateSutWithTimer(IProgressTimer timer) =>
         new TestExtractorWithTimer(Enumerable.Range(1, 5).ToList(), timer);
 
+    /// <inheritdoc/>
+    protected override TestExtractor<int> CreateSutOverSource(IEnumerable<int> source) =>
+        new TestExtractor<int>(source);
+
 
 
     // Exposes the protected timer constructor of TestExtractor<T> for contract testing.