401 lines
16 KiB
C#
401 lines
16 KiB
C#
using Akka.Actor;
|
|
using Akka.TestKit.Xunit2;
|
|
using Microsoft.Extensions.DependencyInjection;
|
|
using Microsoft.Extensions.Logging.Abstractions;
|
|
using NSubstitute;
|
|
using ScadaLink.Commons.Entities.Audit;
|
|
using ScadaLink.Commons.Entities.Notifications;
|
|
using ScadaLink.Commons.Interfaces.Repositories;
|
|
using ScadaLink.Commons.Interfaces.Services;
|
|
using ScadaLink.Commons.Messages.Notification;
|
|
using ScadaLink.Commons.Types.Enums;
|
|
using ScadaLink.NotificationOutbox.Delivery;
|
|
using ScadaLink.NotificationOutbox.Messages;
|
|
|
|
namespace ScadaLink.NotificationOutbox.Tests;
|
|
|
|
/// <summary>
|
|
/// M4 Bundle B (B3) — verifies the <see cref="NotificationOutboxActor"/>
|
|
/// emits a second
|
|
/// <see cref="AuditChannel.Notification"/>/<see cref="AuditKind.NotifyDeliver"/>
|
|
/// audit row carrying the terminal status (Delivered, Parked, Discarded) on
|
|
/// every terminal-state transition. The B2 Attempted row is still emitted
|
|
/// alongside the terminal one — these tests assert ONLY the terminal row
|
|
/// presence and status.
|
|
/// </summary>
|
|
public class NotificationOutboxActorTerminalEmissionTests : TestKit
|
|
{
|
|
private readonly INotificationOutboxRepository _outboxRepository =
|
|
Substitute.For<INotificationOutboxRepository>();
|
|
|
|
private readonly INotificationRepository _notificationRepository =
|
|
Substitute.For<INotificationRepository>();
|
|
|
|
private readonly RecordingCentralAuditWriter _auditWriter = new();
|
|
|
|
private sealed class RecordingCentralAuditWriter : ICentralAuditWriter
|
|
{
|
|
public List<AuditEvent> Events { get; } = new();
|
|
public Func<AuditEvent, Task>? OnWrite { get; set; }
|
|
|
|
public Task WriteAsync(AuditEvent evt, CancellationToken ct = default)
|
|
{
|
|
lock (Events)
|
|
{
|
|
Events.Add(evt);
|
|
}
|
|
|
|
return OnWrite?.Invoke(evt) ?? Task.CompletedTask;
|
|
}
|
|
}
|
|
|
|
private IServiceProvider BuildServiceProvider(IEnumerable<INotificationDeliveryAdapter> adapters)
|
|
{
|
|
var services = new ServiceCollection();
|
|
services.AddScoped(_ => _outboxRepository);
|
|
services.AddScoped(_ => _notificationRepository);
|
|
foreach (var adapter in adapters)
|
|
{
|
|
services.AddScoped<INotificationDeliveryAdapter>(_ => adapter);
|
|
}
|
|
|
|
return services.BuildServiceProvider();
|
|
}
|
|
|
|
private sealed class StubAdapter : INotificationDeliveryAdapter
|
|
{
|
|
private readonly Func<DeliveryOutcome> _outcome;
|
|
|
|
public StubAdapter(Func<DeliveryOutcome> outcome) { _outcome = outcome; }
|
|
|
|
public NotificationType Type => NotificationType.Email;
|
|
|
|
public Task<DeliveryOutcome> DeliverAsync(
|
|
Notification notification, CancellationToken cancellationToken = default)
|
|
=> Task.FromResult(_outcome());
|
|
}
|
|
|
|
private IActorRef CreateActor(IEnumerable<INotificationDeliveryAdapter> adapters)
|
|
{
|
|
return Sys.ActorOf(Props.Create(() => new NotificationOutboxActor(
|
|
BuildServiceProvider(adapters),
|
|
new NotificationOutboxOptions { DispatchInterval = TimeSpan.FromHours(1) },
|
|
(ICentralAuditWriter)_auditWriter,
|
|
NullLogger<NotificationOutboxActor>.Instance)));
|
|
}
|
|
|
|
private static Notification MakeNotification(
|
|
NotificationStatus status = NotificationStatus.Pending,
|
|
int retryCount = 0,
|
|
Guid? notificationId = null,
|
|
Guid? originExecutionId = null,
|
|
Guid? originParentExecutionId = null)
|
|
{
|
|
return new Notification(
|
|
(notificationId ?? Guid.NewGuid()).ToString("D"),
|
|
NotificationType.Email,
|
|
"ops-team",
|
|
"Tank overflow",
|
|
"Tank 3 level critical",
|
|
"site-1")
|
|
{
|
|
Status = status,
|
|
RetryCount = retryCount,
|
|
CreatedAt = DateTimeOffset.UtcNow,
|
|
OriginExecutionId = originExecutionId,
|
|
OriginParentExecutionId = originParentExecutionId,
|
|
};
|
|
}
|
|
|
|
private void SetupSmtpRetryPolicy(int maxRetries, TimeSpan retryDelay)
|
|
{
|
|
var config = new SmtpConfiguration("smtp.example.com", "Basic", "noreply@example.com")
|
|
{
|
|
MaxRetries = maxRetries,
|
|
RetryDelay = retryDelay,
|
|
};
|
|
_notificationRepository.GetAllSmtpConfigurationsAsync(Arg.Any<CancellationToken>())
|
|
.Returns(new[] { config });
|
|
}
|
|
|
|
private List<AuditEvent> EventsByStatus(AuditStatus status)
|
|
{
|
|
lock (_auditWriter.Events)
|
|
{
|
|
return _auditWriter.Events.Where(e => e.Status == status).ToList();
|
|
}
|
|
}
|
|
|
|
[Fact]
|
|
public void Terminal_Delivered_EmitsEvent_StatusDelivered()
|
|
{
|
|
SetupSmtpRetryPolicy(maxRetries: 5, retryDelay: TimeSpan.FromMinutes(1));
|
|
var notification = MakeNotification();
|
|
_outboxRepository.GetDueAsync(Arg.Any<DateTimeOffset>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
|
|
.Returns(new[] { notification });
|
|
var adapter = new StubAdapter(() => DeliveryOutcome.Success("ops@example.com"));
|
|
var actor = CreateActor([adapter]);
|
|
|
|
actor.Tell(InternalMessages.DispatchTick.Instance);
|
|
|
|
AwaitAssert(() =>
|
|
{
|
|
var delivered = EventsByStatus(AuditStatus.Delivered);
|
|
Assert.Single(delivered);
|
|
var evt = delivered[0];
|
|
Assert.Equal(AuditChannel.Notification, evt.Channel);
|
|
Assert.Equal(AuditKind.NotifyDeliver, evt.Kind);
|
|
Assert.Equal("ops-team", evt.Target);
|
|
});
|
|
}
|
|
|
|
[Fact]
|
|
public void Terminal_Delivered_CarriesOriginExecutionId_AsExecutionId()
|
|
{
|
|
// Audit Log #23: the terminal NotifyDeliver row must echo the
|
|
// notification's OriginExecutionId so it shares the per-run id with
|
|
// the site-emitted NotifySend row.
|
|
SetupSmtpRetryPolicy(maxRetries: 5, retryDelay: TimeSpan.FromMinutes(1));
|
|
var executionId = Guid.NewGuid();
|
|
var notification = MakeNotification(originExecutionId: executionId);
|
|
_outboxRepository.GetDueAsync(Arg.Any<DateTimeOffset>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
|
|
.Returns(new[] { notification });
|
|
var adapter = new StubAdapter(() => DeliveryOutcome.Success("ops@example.com"));
|
|
var actor = CreateActor([adapter]);
|
|
|
|
actor.Tell(InternalMessages.DispatchTick.Instance);
|
|
|
|
AwaitAssert(() =>
|
|
{
|
|
var delivered = EventsByStatus(AuditStatus.Delivered);
|
|
Assert.Single(delivered);
|
|
Assert.Equal(executionId, delivered[0].ExecutionId);
|
|
});
|
|
}
|
|
|
|
[Fact]
|
|
public void Terminal_Delivered_NullOriginExecutionId_HasNullExecutionId()
|
|
{
|
|
SetupSmtpRetryPolicy(maxRetries: 5, retryDelay: TimeSpan.FromMinutes(1));
|
|
var notification = MakeNotification(originExecutionId: null);
|
|
_outboxRepository.GetDueAsync(Arg.Any<DateTimeOffset>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
|
|
.Returns(new[] { notification });
|
|
var adapter = new StubAdapter(() => DeliveryOutcome.Success("ops@example.com"));
|
|
var actor = CreateActor([adapter]);
|
|
|
|
actor.Tell(InternalMessages.DispatchTick.Instance);
|
|
|
|
AwaitAssert(() =>
|
|
{
|
|
var delivered = EventsByStatus(AuditStatus.Delivered);
|
|
Assert.Single(delivered);
|
|
Assert.Null(delivered[0].ExecutionId);
|
|
});
|
|
}
|
|
|
|
[Fact]
|
|
public void Terminal_Delivered_CarriesOriginParentExecutionId_AsParentExecutionId()
|
|
{
|
|
// Audit Log ParentExecutionId: the terminal NotifyDeliver row must echo
|
|
// the notification's OriginParentExecutionId so the central dispatcher's
|
|
// rows carry the routed run's parent id.
|
|
SetupSmtpRetryPolicy(maxRetries: 5, retryDelay: TimeSpan.FromMinutes(1));
|
|
var parentExecutionId = Guid.NewGuid();
|
|
var notification = MakeNotification(originParentExecutionId: parentExecutionId);
|
|
_outboxRepository.GetDueAsync(Arg.Any<DateTimeOffset>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
|
|
.Returns(new[] { notification });
|
|
var adapter = new StubAdapter(() => DeliveryOutcome.Success("ops@example.com"));
|
|
var actor = CreateActor([adapter]);
|
|
|
|
actor.Tell(InternalMessages.DispatchTick.Instance);
|
|
|
|
AwaitAssert(() =>
|
|
{
|
|
var delivered = EventsByStatus(AuditStatus.Delivered);
|
|
Assert.Single(delivered);
|
|
Assert.Equal(parentExecutionId, delivered[0].ParentExecutionId);
|
|
});
|
|
}
|
|
|
|
[Fact]
|
|
public void Terminal_Delivered_NullOriginParentExecutionId_HasNullParentExecutionId()
|
|
{
|
|
SetupSmtpRetryPolicy(maxRetries: 5, retryDelay: TimeSpan.FromMinutes(1));
|
|
var notification = MakeNotification(originParentExecutionId: null);
|
|
_outboxRepository.GetDueAsync(Arg.Any<DateTimeOffset>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
|
|
.Returns(new[] { notification });
|
|
var adapter = new StubAdapter(() => DeliveryOutcome.Success("ops@example.com"));
|
|
var actor = CreateActor([adapter]);
|
|
|
|
actor.Tell(InternalMessages.DispatchTick.Instance);
|
|
|
|
AwaitAssert(() =>
|
|
{
|
|
var delivered = EventsByStatus(AuditStatus.Delivered);
|
|
Assert.Single(delivered);
|
|
Assert.Null(delivered[0].ParentExecutionId);
|
|
});
|
|
}
|
|
|
|
[Fact]
|
|
public void Terminal_Parked_OnPermanentFailure_EmitsEvent_StatusParked()
|
|
{
|
|
SetupSmtpRetryPolicy(maxRetries: 5, retryDelay: TimeSpan.FromMinutes(1));
|
|
var notification = MakeNotification();
|
|
_outboxRepository.GetDueAsync(Arg.Any<DateTimeOffset>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
|
|
.Returns(new[] { notification });
|
|
var adapter = new StubAdapter(() => DeliveryOutcome.Permanent("invalid recipient address"));
|
|
var actor = CreateActor([adapter]);
|
|
|
|
actor.Tell(InternalMessages.DispatchTick.Instance);
|
|
|
|
AwaitAssert(() =>
|
|
{
|
|
var parked = EventsByStatus(AuditStatus.Parked);
|
|
Assert.Single(parked);
|
|
Assert.Equal(AuditKind.NotifyDeliver, parked[0].Kind);
|
|
Assert.Equal("invalid recipient address", parked[0].ErrorMessage);
|
|
});
|
|
}
|
|
|
|
[Fact]
|
|
public void Terminal_Parked_CarriesOriginExecutionId_AsExecutionId()
|
|
{
|
|
// Audit Log #23: the Parked terminal NotifyDeliver row flows through the
|
|
// same BuildNotifyDeliverEvent path as the Delivered row, so it must
|
|
// likewise echo the notification's OriginExecutionId — sharing the
|
|
// per-run id with the site-emitted NotifySend row.
|
|
SetupSmtpRetryPolicy(maxRetries: 5, retryDelay: TimeSpan.FromMinutes(1));
|
|
var executionId = Guid.NewGuid();
|
|
var notification = MakeNotification(originExecutionId: executionId);
|
|
_outboxRepository.GetDueAsync(Arg.Any<DateTimeOffset>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
|
|
.Returns(new[] { notification });
|
|
var adapter = new StubAdapter(() => DeliveryOutcome.Permanent("invalid recipient address"));
|
|
var actor = CreateActor([adapter]);
|
|
|
|
actor.Tell(InternalMessages.DispatchTick.Instance);
|
|
|
|
AwaitAssert(() =>
|
|
{
|
|
var parked = EventsByStatus(AuditStatus.Parked);
|
|
Assert.Single(parked);
|
|
Assert.Equal(executionId, parked[0].ExecutionId);
|
|
});
|
|
}
|
|
|
|
[Fact]
|
|
public void Terminal_Parked_OnTransientReachingMaxRetries_EmitsEvent_StatusParked()
|
|
{
|
|
SetupSmtpRetryPolicy(maxRetries: 3, retryDelay: TimeSpan.FromMinutes(1));
|
|
// RetryCount starts at max-1; the failed attempt increments it to max
|
|
// which triggers the Parked terminal transition.
|
|
var notification = MakeNotification(retryCount: 2);
|
|
_outboxRepository.GetDueAsync(Arg.Any<DateTimeOffset>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
|
|
.Returns(new[] { notification });
|
|
var adapter = new StubAdapter(() => DeliveryOutcome.Transient("smtp timeout"));
|
|
var actor = CreateActor([adapter]);
|
|
|
|
actor.Tell(InternalMessages.DispatchTick.Instance);
|
|
|
|
AwaitAssert(() =>
|
|
{
|
|
var parked = EventsByStatus(AuditStatus.Parked);
|
|
Assert.Single(parked);
|
|
Assert.Equal(AuditKind.NotifyDeliver, parked[0].Kind);
|
|
});
|
|
}
|
|
|
|
[Fact]
|
|
public void Terminal_Parked_OnMissingAdapter_EmitsEvent_StatusParked()
|
|
{
|
|
SetupSmtpRetryPolicy(maxRetries: 5, retryDelay: TimeSpan.FromMinutes(1));
|
|
var notification = MakeNotification();
|
|
_outboxRepository.GetDueAsync(Arg.Any<DateTimeOffset>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
|
|
.Returns(new[] { notification });
|
|
// No adapters registered: the missing-adapter park path runs.
|
|
var actor = CreateActor([]);
|
|
|
|
actor.Tell(InternalMessages.DispatchTick.Instance);
|
|
|
|
AwaitAssert(() =>
|
|
{
|
|
var parked = EventsByStatus(AuditStatus.Parked);
|
|
Assert.Single(parked);
|
|
Assert.Equal(AuditKind.NotifyDeliver, parked[0].Kind);
|
|
Assert.Contains("no delivery adapter", parked[0].ErrorMessage!);
|
|
});
|
|
}
|
|
|
|
[Fact]
|
|
public void Transient_BelowMaxRetries_DoesNotEmitTerminalRow()
|
|
{
|
|
// A transient failure that does not reach max-retries leaves the row
|
|
// in Retrying — non-terminal, so no terminal audit row should be
|
|
// emitted (only the Attempted row from B2).
|
|
SetupSmtpRetryPolicy(maxRetries: 5, retryDelay: TimeSpan.FromMinutes(1));
|
|
var notification = MakeNotification(retryCount: 0);
|
|
_outboxRepository.GetDueAsync(Arg.Any<DateTimeOffset>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
|
|
.Returns(new[] { notification });
|
|
var adapter = new StubAdapter(() => DeliveryOutcome.Transient("smtp timeout"));
|
|
var actor = CreateActor([adapter]);
|
|
|
|
actor.Tell(InternalMessages.DispatchTick.Instance);
|
|
|
|
// Wait for the Attempted row to land so we know dispatch has run.
|
|
AwaitAssert(() => Assert.Single(EventsByStatus(AuditStatus.Attempted)));
|
|
|
|
// No terminal rows of any kind.
|
|
Assert.Empty(EventsByStatus(AuditStatus.Delivered));
|
|
Assert.Empty(EventsByStatus(AuditStatus.Parked));
|
|
Assert.Empty(EventsByStatus(AuditStatus.Discarded));
|
|
}
|
|
|
|
[Fact]
|
|
public void Terminal_Discarded_OnManualDiscard_EmitsEvent_StatusDiscarded()
|
|
{
|
|
// Wire the actor with a parked row that GetByIdAsync returns; the
|
|
// discard handler must emit a terminal Discarded audit row.
|
|
SetupSmtpRetryPolicy(maxRetries: 5, retryDelay: TimeSpan.FromMinutes(1));
|
|
var notification = MakeNotification(status: NotificationStatus.Parked);
|
|
_outboxRepository.GetByIdAsync(notification.NotificationId, Arg.Any<CancellationToken>())
|
|
.Returns(notification);
|
|
var actor = CreateActor([]);
|
|
|
|
actor.Tell(new DiscardNotificationRequest(
|
|
CorrelationId: "test-corr", NotificationId: notification.NotificationId));
|
|
|
|
// First wait for the discard handler to reply (handshake), then assert
|
|
// the audit row landed.
|
|
ExpectMsg<DiscardNotificationResponse>(r => r.Success);
|
|
AwaitAssert(() =>
|
|
{
|
|
var discarded = EventsByStatus(AuditStatus.Discarded);
|
|
Assert.Single(discarded);
|
|
Assert.Equal(AuditKind.NotifyDeliver, discarded[0].Kind);
|
|
});
|
|
}
|
|
|
|
[Fact]
|
|
public void AuditWriter_Throws_TerminalUpdate_StillSucceeds()
|
|
{
|
|
// Audit failure NEVER aborts the user-facing action: the dispatcher
|
|
// must still persist the Delivered status via UpdateAsync.
|
|
SetupSmtpRetryPolicy(maxRetries: 5, retryDelay: TimeSpan.FromMinutes(1));
|
|
var notification = MakeNotification();
|
|
_outboxRepository.GetDueAsync(Arg.Any<DateTimeOffset>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
|
|
.Returns(new[] { notification });
|
|
var adapter = new StubAdapter(() => DeliveryOutcome.Success("ops@example.com"));
|
|
_auditWriter.OnWrite = _ => throw new InvalidOperationException("audit dead");
|
|
var actor = CreateActor([adapter]);
|
|
|
|
actor.Tell(InternalMessages.DispatchTick.Instance);
|
|
|
|
AwaitAssert(() =>
|
|
{
|
|
_outboxRepository.Received(1).UpdateAsync(
|
|
Arg.Is<Notification>(n => n.Status == NotificationStatus.Delivered),
|
|
Arg.Any<CancellationToken>());
|
|
});
|
|
}
|
|
}
|