Using spy endpoints for testing for messages to ServiceControl

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

@@ -1,27 +1,27 @@
 namespace ServiceControl.Audit.AcceptanceTests.Monitoring
 {
+    using System;
+    using System.Collections.Concurrent;
     using System.Linq;
     using System.Threading.Tasks;
     using AcceptanceTesting.EndpointTemplates;
     using NServiceBus;
     using NServiceBus.AcceptanceTesting;
+    using NServiceBus.Pipeline;
     using NUnit.Framework;
-    using TestSupport;
+    using Contracts.EndpointControl;
     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 = s => s.ServiceControlQueueAddress = Conventions.EndpointNamingConvention(typeof(SpyEndpoint));
 
-            var context = await Define<InterceptedMessagesScenarioContext>()
+            var context = await Define<MyContext>()
                 .WithEndpoint<Receiver>(b => b.When((bus, c) => bus.SendLocal(new MyMessage())))
+                .WithEndpoint<SpyEndpoint>()
                 .Done(c => c.SentRegisterEndpointCommands.Any())
                 .Run();
 
@@ -40,6 +40,43 @@ public class MyMessageHandler : IHandleMessages<MyMessage>
             }
         }
 
+        public class SpyEndpoint : EndpointConfigurationBuilder
+        {
+            public SpyEndpoint() => EndpointSetup<DefaultServerWithoutAudit>(c =>
+                c.Pipeline.Register(new IgnoreCustomCheckResultsBehavior(), "Ignores custom check results reported to the spied queue"));
+
+            class IgnoreCustomCheckResultsBehavior : Behavior<ITransportReceiveContext>
+            {
+                public override Task Invoke(ITransportReceiveContext context, Func<Task> next)
+                {
+                    if (context.Message.Headers.TryGetValue(Headers.EnclosedMessageTypes, out var messageTypes)
+                        && messageTypes.StartsWith(CustomCheckResultType, StringComparison.Ordinal))
+                    {
+                        return Task.CompletedTask;
+                    }
+
+                    return next();
+                }
+
+                const string CustomCheckResultType = "ServiceControl.Plugin.CustomChecks.Messages.ReportCustomCheckResult";
+            }
+
+            [Handler]
+            public class RegisterNewEndpointHandler(MyContext testContext) : IHandleMessages<RegisterNewEndpoint>
+            {
+                public Task Handle(RegisterNewEndpoint message, IMessageHandlerContext context)
+                {
+                    testContext.SentRegisterEndpointCommands.Add(message);
+                    return Task.CompletedTask;
+                }
+            }
+        }
+
         public class MyMessage : ICommand;
+
+        public class MyContext : ScenarioContext
+        {
+            public ConcurrentBag<RegisterNewEndpoint> SentRegisterEndpointCommands { get; } = [];
+        }
     }
-}
+}

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

@@ -1,31 +1,31 @@
 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 NServiceBus;
     using NServiceBus.AcceptanceTesting;
+    using NServiceBus.Pipeline;
     using NUnit.Framework;
-    using TestSupport;
+    using Contracts.EndpointControl;
+    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 = s => s.ServiceControlQueueAddress = Conventions.EndpointNamingConvention(typeof(SpyEndpoint));
 
             MessagesView auditedMessage = null;
 
             var messageId = Guid.NewGuid().ToString();
-            await Define<InterceptedMessagesScenarioContext>()
+            var context = await Define<MyContext>()
                 .WithEndpoint<Receiver>(b => b.When(s =>
                 {
                     var options = new SendOptions();
@@ -35,17 +35,19 @@ await Define<InterceptedMessagesScenarioContext>()
                     options.SetMessageId(messageId);
                     return s.Send(new MyMessage(), options);
                 }))
+                .WithEndpoint<SpyEndpoint>()
                 .Done(async c =>
                 {
                     var result = await this.TryGetSingle<MessagesView>("/api/messages", m => m.MessageId == messageId);
 
                     auditedMessage = result;
 
-                    return result;
+                    return result.HasResult && c.ReceivedRegisterEndpointCommands.Any();
                 })
                 .Run();
 
             Assert.That(auditedMessage.Status, Is.EqualTo(MessageStatus.ResolvedSuccessfully));
+            Assert.That(context.ReceivedRegisterEndpointCommands, Is.Not.Empty);
         }
 
         public class Receiver : EndpointConfigurationBuilder
@@ -59,6 +61,43 @@ public class MyMessageHandler : IHandleMessages<MyMessage>
             }
         }
 
+        public class SpyEndpoint : EndpointConfigurationBuilder
+        {
+            public SpyEndpoint() => EndpointSetup<DefaultServerWithoutAudit>(c =>
+                c.Pipeline.Register(new IgnoreCustomCheckResultsBehavior(), "Ignores custom check results reported to the spied queue"));
+
+            class IgnoreCustomCheckResultsBehavior : Behavior<ITransportReceiveContext>
+            {
+                public override Task Invoke(ITransportReceiveContext context, Func<Task> next)
+                {
+                    if (context.Message.Headers.TryGetValue(Headers.EnclosedMessageTypes, out var messageTypes)
+                        && messageTypes.StartsWith(CustomCheckResultType, StringComparison.Ordinal))
+                    {
+                        return Task.CompletedTask;
+                    }
+
+                    return next();
+                }
+
+                const string CustomCheckResultType = "ServiceControl.Plugin.CustomChecks.Messages.ReportCustomCheckResult";
+            }
+
+            [Handler]
+            public class RegisterNewEndpointHandler(MyContext testContext) : IHandleMessages<RegisterNewEndpoint>
+            {
+                public Task Handle(RegisterNewEndpoint message, IMessageHandlerContext context)
+                {
+                    testContext.ReceivedRegisterEndpointCommands.Add(message);
+                    return Task.CompletedTask;
+                }
+            }
+        }
+
         public class MyMessage : ICommand;
+
+        public class MyContext : ScenarioContext
+        {
+            public ConcurrentBag<RegisterNewEndpoint> ReceivedRegisterEndpointCommands { get; } = [];
+        }
     }
-}
+}