using System.Net.Http.Json; using System.Text.Json; using FluentAssertions; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Options; using Minio.DataModel.Args; using Minio.Exceptions; using StackExchange.Redis; [Collection("Integration")] public class NotificationPipelineTests : IAsyncLifetime { private readonly ApiFixture _fixture; private readonly HttpClient _http; public NotificationPipelineTests(ApiFixture fixture) { _fixture = fixture; _http = fixture.CreateClient(); } public async Task InitializeAsync() { using var scope = _fixture.Services.CreateScope(); var db = scope.ServiceProvider.GetRequiredService(); await DbResetHelper.ResetAsync(db); db.AlertThresholds.Add(new AlertThreshold { Id = Guid.NewGuid(), ObservationCode = "POTASSIUM_MEQ_L", DisplayName = "Serum Potassium", Unit = "mEq/L", CriticalLow = 2.5m, WarningLow = 3.5m, WarningHigh = 5.0m, CriticalHigh = 6.5m, CreatedAt = DateTimeOffset.UtcNow, }); await db.SaveChangesAsync(); var redis = scope.ServiceProvider.GetRequiredService(); await redis.GetDatabase(1).StringSetAsync( "threshold:POTASSIUM_MEQ_L", """{"ObservationCode":"POTASSIUM_MEQ_L","CriticalLow":2.5,"WarningLow":3.5,"WarningHigh":5.0,"CriticalHigh":6.5}"""); } public Task DisposeAsync() => Task.CompletedTask; [Fact] public async Task DischargeEncounter_UploadsSummaryToMinIO() { var encounterId = await CreateActiveEncounterAsync(); var resp = await _http.PatchAsJsonAsync( $"/api/v1/encounters/{encounterId}/status", new { status = "Discharged" }); resp.EnsureSuccessStatusCode(); // outbox relay → Kafka → notification-publisher → RabbitMQ → discharge worker → MinIO await Task.Delay(TimeSpan.FromSeconds(8)); var mo = _fixture.Services.GetRequiredService>().Value; var client = MinioClientFactory.Build(mo); var objectKey = $"discharge-summaries/{encounterId}/summary.pdf"; var exists = false; try { await client.StatObjectAsync(new StatObjectArgs() .WithBucket(mo.BucketName) .WithObject(objectKey)); exists = true; } catch (ObjectNotFoundException) { } exists.Should().BeTrue($"MinIO object not found: {objectKey}"); } [Fact] public async Task CriticalAlert_Unacknowledged_EscalatesAfterTimeout() { var encounterId = await CreateActiveEncounterAsync(); var alertId = await IngestCriticalPotassiumAsync(encounterId); // paging worker (≤6s) → DLQ (5s TTL) → escalation worker. // Poll instead of a fixed sleep: earlier tests may leave paging jobs queued // (prefetch=1), so wall-clock time varies across the full integration suite. var deadline = DateTimeOffset.UtcNow.AddSeconds(35); AlertStatus status; do { await Task.Delay(500); using var scope = _fixture.Services.CreateScope(); var db = scope.ServiceProvider.GetRequiredService(); status = (await db.ClinicalAlerts.FindAsync(alertId))!.Status; } while (status == AlertStatus.Open && DateTimeOffset.UtcNow < deadline); status.Should().Be(AlertStatus.Escalated); } [Fact] public async Task CriticalAlert_AcknowledgedBeforeTimeout_DoesNotEscalate() { var encounterId = await CreateActiveEncounterAsync(); var alertId = await IngestCriticalPotassiumAsync(encounterId); await Task.Delay(TimeSpan.FromSeconds(3)); var ackResp = await _http.PostAsJsonAsync( $"/api/v1/alerts/{alertId}/acknowledge", new AcknowledgeAlertRequest("Dr. Kwame Mensah", "Reviewed — will adjust potassium replacement.")); ackResp.EnsureSuccessStatusCode(); await Task.Delay(TimeSpan.FromSeconds(8)); using var scope = _fixture.Services.CreateScope(); var db = scope.ServiceProvider.GetRequiredService(); var alert = await db.ClinicalAlerts.FindAsync(alertId); alert.Should().NotBeNull(); alert!.Status.Should().Be(AlertStatus.Acknowledged); } private async Task CreateActiveEncounterAsync() { var patientId = await CreatePatientAsync(); var resp = await _http.PostAsJsonAsync( $"/api/v1/patients/{patientId}/encounters", new OpenEncounterRequest(EncounterType.Inpatient, "ICU", "Dr. Osei")); resp.EnsureSuccessStatusCode(); var body = await resp.Content.ReadFromJsonAsync(); return body!.RootElement.GetProperty("data").GetProperty("id").GetGuid(); } private async Task CreatePatientAsync() { var resp = await _http.PostAsJsonAsync( "/api/v1/patients", new RegisterPatientRequest("Eleanor", "Vance", new DateOnly(1962, 9, 14), "Female")); resp.EnsureSuccessStatusCode(); var body = await resp.Content.ReadFromJsonAsync(); return body!.RootElement.GetProperty("data").GetProperty("id").GetGuid(); } private async Task IngestCriticalPotassiumAsync(Guid encounterId) { var ingestResp = await _http.PostAsJsonAsync( $"/api/v1/encounters/{encounterId}/observations", new BatchIngestRequest(new List { new("POTASSIUM_MEQ_L", 2.1m, "mEq/L", ObservationSource.Device, DateTimeOffset.UtcNow, null), })); ingestResp.EnsureSuccessStatusCode(); var body = await ingestResp.Content.ReadFromJsonAsync(); var alertId = body!.RootElement.GetProperty("data").GetProperty("alertId").GetGuid(); alertId.Should().NotBe(Guid.Empty); return alertId; } }