run test on phase 24 and 33

VigilCareClinicalAPI.Tests/Helpers/RabbitMqTestHelper.cs

@@ -1,21 +1,56 @@
+using FluentAssertions;
 using Microsoft.Extensions.Options;
 using RabbitMQ.Client;
 using RabbitMQ.Client.Exceptions;
+using System.Net.Http.Headers;
+using System.Text;
+using System.Text.Json;
 
 public static class RabbitMqTestHelper
 {
+    private static readonly string[] NotificationQueues =
+    [
+        "alerts.paging.queue",
+        "alerts.paging.dlq",
+        "alerts.escalation.queue",
+    ];
+
+    /// <summary>
+    /// Creates the configured virtual host and grants the test user full access.
+    /// Integration tests use a dedicated vhost so a concurrently running dev API
+    /// on the default "/" vhost cannot steal paging messages or recreate the DLQ TTL.
+    /// </summary>
+    public static async Task EnsureVirtualHostAsync(RabbitMqOptions o)
+    {
+        if (string.IsNullOrEmpty(o.VirtualHost) || o.VirtualHost == "/")
+            return;
+
+        using var http = new HttpClient();
+        var credentials = Convert.ToBase64String(Encoding.UTF8.GetBytes($"{o.Username}:{o.Password}"));
+        http.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Basic", credentials);
+
+        var mgmtBase = $"http://{o.Host}:{o.Port + 10000}";
+        var vhost = Uri.EscapeDataString(o.VirtualHost);
+
+        using var create = await http.PutAsync($"{mgmtBase}/api/vhosts/{vhost}", null);
+        create.EnsureSuccessStatusCode();
+
+        var permissions = JsonSerializer.Serialize(new
+        {
+            configure = ".*",
+            write = ".*",
+            read = ".*",
+        });
+        using var grant = await http.PutAsync(
+            $"{mgmtBase}/api/permissions/{vhost}/{Uri.EscapeDataString(o.Username)}",
+            new StringContent(permissions, Encoding.UTF8, "application/json"));
+        grant.EnsureSuccessStatusCode();
+    }
+
     public static void PurgeNotificationQueues(IOptions<RabbitMqOptions> opts)
     {
         var o = opts.Value;
-        var factory = new ConnectionFactory
-        {
-            HostName = o.Host,
-            Port = o.Port,
-            UserName = o.Username,
-            Password = o.Password,
-        };
-
-        using var connection = factory.CreateConnection("test-queue-purge");
+        using var connection = CreateConnection(o, "test-queue-purge");
         using var channel = connection.CreateModel();
 
         // x-message-ttl is immutable once the queue exists. A concurrently running dev API
@@ -36,6 +71,67 @@ public static class RabbitMqTestHelper
         }
     }
 
+    /// <summary>
+    /// Waits for stale alert.generated Kafka messages to finish the paging → DLQ → escalation
+    /// cycle. Purging alone is not enough: NotificationPublisher keeps republishing backlog
+    /// after the purge, and prefetch=1 leaves one in-flight page ahead of the test alert.
+    /// </summary>
+    public static async Task WaitForNotificationPipelineIdleAsync(
+        IOptions<RabbitMqOptions> opts,
+        TimeSpan maxWait)
+    {
+        var o = opts.Value;
+        // One full cycle: paging timeout + DLQ TTL + worker margin.
+        var settle = TimeSpan.FromMilliseconds(o.PagingAckTimeoutMs * 2 + 10_000);
+        var deadline = DateTimeOffset.UtcNow.Add(maxWait);
+        DateTimeOffset? lastBusy = null;
+
+        while (DateTimeOffset.UtcNow < deadline)
+        {
+            if (!AreReadyQueuesEmpty(o))
+            {
+                lastBusy = DateTimeOffset.UtcNow;
+                await Task.Delay(500);
+                continue;
+            }
+
+            lastBusy ??= DateTimeOffset.UtcNow;
+
+            if (DateTimeOffset.UtcNow - lastBusy.Value >= settle)
+                return;
+
+            await Task.Delay(500);
+        }
+    }
+
+    private static bool AreReadyQueuesEmpty(RabbitMqOptions o)
+    {
+        using var connection = CreateConnection(o, "test-queue-idle-check");
+        using var channel = connection.CreateModel();
+
+        foreach (var queue in NotificationQueues)
+        {
+            try
+            {
+                var stats = channel.QueueDeclarePassive(queue);
+                if (stats.MessageCount > 0)
+                    return false;
+            }
+            catch (OperationInterruptedException)
+            {
+                // Queue may not exist yet on a cold broker.
+            }
+        }
+
+        return true;
+    }
+
+    private static ConnectionFactory CreateFactory(RabbitMqOptions o) =>
+        RabbitMqConnectionFactory.Create(o);
+
+    private static IConnection CreateConnection(RabbitMqOptions o, string name) =>
+        CreateFactory(o).CreateConnection(name);
+
     private static void RecreatePagingDlq(IModel channel, int pagingAckTimeoutMs)
     {
         const string dlq = "alerts.paging.dlq";