Consume the lifted JasperFx.Events ProjectionScenario

Closes #404. Deletes Polecat's seven-file copy of Marten's pre-lift harness and
replaces it with one subclass of
JasperFx.Events.TestSupport.ProjectionScenario<TOperations, TQuerySession>
(jasperfx#616, shipped in JasperFx.Events 2.38.0), closed over Polecat's
IDocumentSession / IQuerySession. Marten's equivalent adoption is marten#5133.

Nearly free, as #404 predicted: all seven abstract seam members already existed
as PolecatComplianceFixture bodies, because that seam was deliberately shaped to
match EventStoreComplianceFixture -- including the object-id load dispatch.

What Polecat gets that its own copy never had:

- The 15 missing overloads exist, inherited rather than written. The one that
  actually bit was IEnumerable<object>: a caller holding a List<object> had to
  spread it at every call site.
- An arrange-only scenario is no longer a silent no-op. Appends only flushed when
  the next step was an assertion, so a trailing append was disposed uncommitted
  (marten#5126).
- A failed action stops the scenario instead of letting later steps run against
  state nobody intended; failed assertions still accumulate and report together.
- A scenario can only execute once, Timeout is configurable, and the
  CancellationToken is honored.
- DoNotDeleteExistingData -> DeleteExistingData (default true), Execute ->
  ExecuteAsync, and DocumentShouldExist/NotExist take object ids.

Two things this change does beyond the swap:

AdvancedOperations.CleanAsync gains a non-generic Type overload and the generic
one now delegates to it. DeleteExistingDataAsync has to reset each projection's
Options.StorageTypes, which are Types, and the alternative -- CleanAllDocumentsAsync
-- would take out documents a scenario deliberately seeded first. It also removes
the last ad-hoc SqlConnection in this file: the old harness cleaned projected
documents on a hand-opened connection, outside StoreOptions.ResiliencePipeline.

Six tests added beside the five that already existed, covering the seam rather
than re-testing scripting behavior jasperfx#616 already unit-tests against a fake
store: the IEnumerable overload, a string stream key through the object-id load
dispatch, the trailing-append commit rule, the re-execution guard,
DeleteExistingData = false, and an async-lifecycle projection that can only pass
if BuildDaemonAsync really stood up a daemon and the scenario waited on it.

projection_scenario_tests 11/11.

Author: jeremydmiller

src/Polecat.Tests/Events/projection_scenario_tests.cs

@@ -1,5 +1,6 @@
 using JasperFx.Events;
 using JasperFx.Events.Projections;
+using Polecat.Events.TestSupport;
 using Polecat.Projections;
 using Polecat.Tests.Harness;
 
@@ -16,6 +17,18 @@ public partial class ScenarioQuestParty
     public void Apply(MembersDeparted e) => Members.RemoveAll(m => e.Members.Contains(m));
 }
 
+/// <summary>
+///     String-keyed twin of <see cref="ScenarioQuestParty" />, so the scenario's object-id load
+///     dispatch gets exercised with something other than a Guid.
+/// </summary>
+public partial class ScenarioStringQuestParty
+{
+    public string Id { get; set; } = string.Empty;
+    public string Name { get; set; } = string.Empty;
+
+    public void Apply(QuestStarted e) => Name = e.Name;
+}
+
 [Collection("integration")]
 public class projection_scenario_tests : IntegrationContext
 {
@@ -140,7 +153,7 @@ await StoreOptions(opts =>
 
         var missingId = Guid.NewGuid();
 
-        await Should.ThrowAsync<Polecat.Events.TestSupport.ProjectionScenarioException>(async () =>
+        await Should.ThrowAsync<JasperFx.Events.TestSupport.ProjectionScenarioException>(async () =>
         {
             await theStore.Advanced.EventProjectionScenario(scenario =>
             {
@@ -149,4 +162,166 @@ await theStore.Advanced.EventProjectionScenario(scenario =>
             });
         });
     }
+
+    // The tests below cover what #404 bought: the harness is now
+    // JasperFx.Events.TestSupport.ProjectionScenario<,> and Polecat supplies only the seam. They
+    // exercise the seam through surface Polecat's own copy never had, rather than re-testing
+    // scripting behavior that jasperfx#616 already unit-tests against a fake store.
+
+    [Fact]
+    public async Task append_accepts_an_enumerable_of_events()
+    {
+        await StoreOptions(opts =>
+        {
+            opts.DatabaseSchemaName = "scenario_enumerable";
+            opts.Projections.Add<SingleStreamProjection<ScenarioQuestParty, Guid>>(ProjectionLifecycle.Inline);
+        });
+
+        var questId = Guid.NewGuid();
+
+        // The headline gap in the original #404: a caller holding a List<object> had to spread it
+        // at every call site, because Polecat's copy was params-only.
+        var events = new List<object>
+        {
+            new QuestStarted("Enumerable Quest"),
+            new MembersJoined(1, "Bree", ["Barliman"])
+        };
+
+        await theStore.Advanced.EventProjectionScenario(scenario =>
+        {
+            scenario.StartStream(questId, events);
+
+            scenario.DocumentShouldExist<ScenarioQuestParty>(questId, doc =>
+            {
+                doc.Name.ShouldBe("Enumerable Quest");
+                doc.Members.ShouldContain("Barliman");
+            });
+        }, TestContext.Current.CancellationToken);
+    }
+
+    [Fact]
+    public async Task string_stream_key_flows_through_the_object_id_load_dispatch()
+    {
+        await StoreOptions(opts =>
+        {
+            opts.DatabaseSchemaName = "scenario_stringkey";
+            opts.Events.StreamIdentity = StreamIdentity.AsString;
+            opts.Projections.Add<SingleStreamProjection<ScenarioStringQuestParty, string>>(
+                ProjectionLifecycle.Inline);
+        });
+
+        var key = "quest-" + Guid.NewGuid();
+
+        // DocumentShouldExist takes object now rather than one overload per identity type, so the
+        // seam's LoadDocumentAsync has to dispatch on the runtime type. That is the part only a
+        // real store can prove.
+        await theStore.Advanced.EventProjectionScenario(scenario =>
+        {
+            scenario.StartStream(key, new QuestStarted("Keyed Quest"));
+            scenario.DocumentShouldExist<ScenarioStringQuestParty>(key, doc => doc.Name.ShouldBe("Keyed Quest"));
+        }, TestContext.Current.CancellationToken);
+    }
+
+    [Fact]
+    public async Task a_trailing_append_with_no_assertion_after_it_is_still_committed()
+    {
+        await StoreOptions(opts =>
+        {
+            opts.DatabaseSchemaName = "scenario_trailing";
+            opts.Projections.Add<SingleStreamProjection<ScenarioQuestParty, Guid>>(ProjectionLifecycle.Inline);
+        });
+
+        var questId = Guid.NewGuid();
+
+        // An arrange-only scenario used to be a silent no-op that passed: appends only flushed when
+        // the next step was an assertion, and the trailing one was disposed uncommitted (marten#5126).
+        await theStore.Advanced.EventProjectionScenario(scenario =>
+        {
+            scenario.Append(questId, new QuestStarted("Trailing Quest"));
+        }, TestContext.Current.CancellationToken);
+
+        await using var query = theStore.QuerySession();
+        var party = await query.LoadAsync<ScenarioQuestParty>(questId, TestContext.Current.CancellationToken);
+
+        party.ShouldNotBeNull();
+        party.Name.ShouldBe("Trailing Quest");
+    }
+
+    [Fact]
+    public async Task a_scenario_cannot_be_executed_twice()
+    {
+        await StoreOptions(opts =>
+        {
+            opts.DatabaseSchemaName = "scenario_once";
+            opts.Projections.Add<SingleStreamProjection<ScenarioQuestParty, Guid>>(ProjectionLifecycle.Inline);
+        });
+
+        var scenario = new ProjectionScenario(theStore);
+        scenario.Append(Guid.NewGuid(), new QuestStarted("Once"));
+
+        await scenario.ExecuteAsync(TestContext.Current.CancellationToken);
+
+        // The steps were consumed by the first run, so a second run would be a silent no-op. It
+        // should be a loud failure instead.
+        await Should.ThrowAsync<InvalidOperationException>(async () =>
+        {
+            await scenario.ExecuteAsync(TestContext.Current.CancellationToken);
+        });
+    }
+
+    [Fact]
+    public async Task delete_existing_data_can_be_turned_off()
+    {
+        await StoreOptions(opts =>
+        {
+            opts.DatabaseSchemaName = "scenario_keepdata";
+            opts.Projections.Add<SingleStreamProjection<ScenarioQuestParty, Guid>>(ProjectionLifecycle.Inline);
+        });
+
+        var first = Guid.NewGuid();
+        var second = Guid.NewGuid();
+
+        await theStore.Advanced.EventProjectionScenario(scenario =>
+        {
+            scenario.Append(first, new QuestStarted("Survivor"));
+        }, TestContext.Current.CancellationToken);
+
+        // DeleteExistingData replaced the double-negative DoNotDeleteExistingData, and defaults to
+        // true -- so without this the first quest would be wiped by the second run.
+        await theStore.Advanced.EventProjectionScenario(scenario =>
+        {
+            scenario.DeleteExistingData = false;
+            scenario.Append(second, new QuestStarted("Newcomer"));
+
+            scenario.DocumentShouldExist<ScenarioQuestParty>(first, doc => doc.Name.ShouldBe("Survivor"));
+            scenario.DocumentShouldExist<ScenarioQuestParty>(second, doc => doc.Name.ShouldBe("Newcomer"));
+        }, TestContext.Current.CancellationToken);
+    }
+
+    [Fact]
+    public async Task scenario_stands_up_a_daemon_for_an_async_projection()
+    {
+        await StoreOptions(opts =>
+        {
+            opts.DatabaseSchemaName = "scenario_async";
+            opts.Projections.Add<SingleStreamProjection<ScenarioQuestParty, Guid>>(ProjectionLifecycle.Async);
+        });
+
+        var questId = Guid.NewGuid();
+
+        // Exercises the BuildDaemonAsync seam and the non-stale wait: with an async lifecycle the
+        // assertion can only pass if the scenario actually ran a daemon and waited for it.
+        await theStore.Advanced.EventProjectionScenario(scenario =>
+        {
+            scenario.Append(questId,
+                new QuestStarted("Async Quest"),
+                new MembersJoined(1, "Rivendell", ["Elrond"]));
+
+            scenario.DocumentShouldExist<ScenarioQuestParty>(questId, doc =>
+            {
+                doc.Name.ShouldBe("Async Quest");
+                doc.Members.ShouldContain("Elrond");
+            });
+        }, TestContext.Current.CancellationToken);
+    }
 }

src/Polecat/AdvancedOperations.cs

@@ -492,9 +492,20 @@ private static async Task DeleteFlatTablesAsync(SqlConnection conn, string[] qua
     /// <summary>
     ///     Delete all rows from the document table for type T.
     /// </summary>
-    public async Task CleanAsync<T>(CancellationToken token = default)
+    public Task CleanAsync<T>(CancellationToken token = default) => CleanAsync(typeof(T), token);
+
+    /// <summary>
+    ///     Delete all rows from the document table for <paramref name="documentType" />.
+    /// </summary>
+    /// <remarks>
+    ///     The non-generic twin of <see cref="CleanAsync{T}" />, for callers holding a
+    ///     <see cref="Type" /> rather than a generic parameter — a projection's
+    ///     <c>Options.StorageTypes</c>, for one, which is how <see cref="ProjectionScenario" />
+    ///     resets exactly the documents its projections own instead of every table in the schema.
+    /// </remarks>
+    public async Task CleanAsync(Type documentType, CancellationToken token = default)
     {
-        var provider = _store.GetProvider(typeof(T));
+        var provider = _store.GetProvider(documentType);
         var tableName = provider.Mapping.QualifiedTableName;
         var connStr = _store.Options.ConnectionString;
         await _resilience.ExecuteAsync(static async (state, ct) =>
@@ -683,7 +694,7 @@ public Task EventProjectionScenario(Action<ProjectionScenario> configuration, Ca
     {
         var scenario = new ProjectionScenario(_store);
         configuration(scenario);
-        return scenario.Execute(ct);
+        return scenario.ExecuteAsync(ct);
     }
 
     // ---- runtime tenant onboarding under managed per-tenant partitioning (#335) ----