Replace Audit interception helper with spy endpoints

src/ServiceControl.Audit.AcceptanceTests/Monitoring/When_a_new_endpoint_is_detected.cs

@@ -1,32 +1,51 @@
 namespace ServiceControl.Audit.AcceptanceTests.Monitoring
 {
+    using System.Collections.Concurrent;
     using System.Linq;
     using System.Threading.Tasks;
     using AcceptanceTesting.EndpointTemplates;
+    using Contracts.EndpointControl;
     using NServiceBus;
     using NServiceBus.AcceptanceTesting;
     using NUnit.Framework;
-    using TestSupport;
     using Conventions = NServiceBus.AcceptanceTesting.Customization.Conventions;
 
     class When_a_new_endpoint_is_detected : AcceptanceTest
     {
         [Test]
         public async Task Should_notify_service_control()
         {
-            CustomConfiguration = endpointConfiguration =>
-            {
-                endpointConfiguration.Pipeline.Register(typeof(InterceptMessagesDestinedToServiceControl),
-                    "Intercepts messages destined to ServiceControl");
-            };
+            SetSettings = settings => settings.ServiceControlQueueAddress = Conventions.EndpointNamingConvention(typeof(ServiceControlSpy));
 
-            var context = await Define<InterceptedMessagesScenarioContext>()
+            var context = await Define<Context>()
+                // The audit instance probes the primary queue on startup with an empty message the spy cannot deserialize.
+                .WithEndpoint<ServiceControlSpy>(b => b.DoNotFailOnErrorMessages())
                 .WithEndpoint<Receiver>(b => b.When((bus, c) => bus.SendLocal(new MyMessage())))
-                .Done(c => c.SentRegisterEndpointCommands.Any())
+                .Done(c => c.SentRegisterEndpointCommands.Any(command => command.Endpoint.Name == Conventions.EndpointNamingConvention(typeof(Receiver))))
                 .Run();
 
-            var command = context.SentRegisterEndpointCommands.Single();
-            Assert.That(command.Endpoint.Name, Is.EqualTo(Conventions.EndpointNamingConvention(typeof(Receiver))));
+            Assert.That(context.SentRegisterEndpointCommands.Select(command => command.Endpoint.Name),
+                Does.Contain(Conventions.EndpointNamingConvention(typeof(Receiver))));
+        }
+
+        public class Context : ScenarioContext
+        {
+            public ConcurrentBag<RegisterNewEndpoint> SentRegisterEndpointCommands { get; } = [];
+        }
+
+        public class ServiceControlSpy : EndpointConfigurationBuilder
+        {
+            public ServiceControlSpy() => EndpointSetup<DefaultServerWithoutAudit>();
+
+            [Handler]
+            public class RegisterNewEndpointHandler(Context testContext) : IHandleMessages<RegisterNewEndpoint>
+            {
+                public Task Handle(RegisterNewEndpoint message, IMessageHandlerContext context)
+                {
+                    testContext.SentRegisterEndpointCommands.Add(message);
+                    return Task.CompletedTask;
+                }
+            }
         }
 
         public class Receiver : EndpointConfigurationBuilder

src/ServiceControl.Audit.AcceptanceTests/Recoverability/When_importing_a_message_resolved_by_a_retry.cs

@@ -1,31 +1,32 @@
-namespace ServiceControl.Audit.AcceptanceTests.Recoverability
+namespace ServiceControl.Audit.AcceptanceTests.Recoverability
 {
     using System;
+    using System.Collections.Concurrent;
+    using System.Linq;
     using System.Threading.Tasks;
     using AcceptanceTesting;
     using AcceptanceTesting.EndpointTemplates;
     using Audit.Auditing.MessagesView;
     using Audit.Monitoring;
+    using Contracts.EndpointControl;
     using NServiceBus;
     using NServiceBus.AcceptanceTesting;
     using NUnit.Framework;
-    using TestSupport;
+    using Conventions = NServiceBus.AcceptanceTesting.Customization.Conventions;
 
     class When_importing_a_message_resolved_by_a_retry : AcceptanceTest
     {
         [Test]
         public async Task Should_set_status_to_resolved()
         {
-            CustomConfiguration = endpointConfiguration =>
-            {
-                endpointConfiguration.Pipeline.Register(typeof(InterceptMessagesDestinedToServiceControl),
-                    "Intercepts messages destined to ServiceControl");
-            };
+            SetSettings = settings => settings.ServiceControlQueueAddress = Conventions.EndpointNamingConvention(typeof(ServiceControlSpy));
 
             MessagesView auditedMessage = null;
 
             var messageId = Guid.NewGuid().ToString();
-            await Define<InterceptedMessagesScenarioContext>()
+            var context = await Define<Context>()
+                // The audit instance probes the primary queue on startup with an empty message the spy cannot deserialize.
+                .WithEndpoint<ServiceControlSpy>(b => b.DoNotFailOnErrorMessages())
                 .WithEndpoint<Receiver>(b => b.When(s =>
                 {
                     var options = new SendOptions();
@@ -37,15 +38,45 @@ await Define<InterceptedMessagesScenarioContext>()
                 }))
                 .Done(async c =>
                 {
+                    var receiverRegistered = c.SentRegisterEndpointCommands.Any(command => command.Endpoint.Name == Conventions.EndpointNamingConvention(typeof(Receiver)));
+                    if (!receiverRegistered)
+                    {
+                        return false;
+                    }
+
                     var result = await this.TryGetSingle<MessagesView>("/api/messages", m => m.MessageId == messageId);
 
                     auditedMessage = result;
 
-                    return result;
+                    return result && receiverRegistered;
                 })
                 .Run();
 
-            Assert.That(auditedMessage.Status, Is.EqualTo(MessageStatus.ResolvedSuccessfully));
+            using (Assert.EnterMultipleScope())
+            {
+                Assert.That(auditedMessage.Status, Is.EqualTo(MessageStatus.ResolvedSuccessfully));
+                Assert.That(context.SentRegisterEndpointCommands, Has.Some.Matches<RegisterNewEndpoint>(command => command.Endpoint.Name == Conventions.EndpointNamingConvention(typeof(Receiver))));
+            }
+        }
+
+        public class Context : ScenarioContext
+        {
+            public ConcurrentBag<RegisterNewEndpoint> SentRegisterEndpointCommands { get; } = [];
+        }
+
+        public class ServiceControlSpy : EndpointConfigurationBuilder
+        {
+            public ServiceControlSpy() => EndpointSetup<DefaultServerWithoutAudit>();
+
+            [Handler]
+            public class RegisterNewEndpointHandler(Context testContext) : IHandleMessages<RegisterNewEndpoint>
+            {
+                public Task Handle(RegisterNewEndpoint message, IMessageHandlerContext context)
+                {
+                    testContext.SentRegisterEndpointCommands.Add(message);
+                    return Task.CompletedTask;
+                }
+            }
         }
 
         public class Receiver : EndpointConfigurationBuilder