using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using NSubstitute;
using NSubstitute.ExceptionExtensions;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
using ZB.MOM.WW.ScadaBridge.Communication;
using ZB.MOM.WW.ScadaBridge.TemplateEngine.Flattening;
namespace ZB.MOM.WW.ScadaBridge.DeploymentManager.Tests;
///
/// Gitea #90: the terminal-record purge rate limiter is an injectable, clock-driven singleton
/// instead of a process-wide static, so the sweep's rate limit and its swallow-on-failure path
/// can be tested independently of every other test in the assembly.
///
public class TerminalPurgeRateLimiterTests
{
private static readonly DateTimeOffset Start = new(2026, 11, 9, 12, 0, 0, TimeSpan.Zero);
private sealed class ManualClock(DateTimeOffset start) : TimeProvider
{
public DateTimeOffset Now { get; set; } = start;
public override DateTimeOffset GetUtcNow() => Now;
}
private sealed class CapturingLogger : ILogger
{
public List<(LogLevel Level, Exception? Exception, string Message)> Entries { get; } = new();
public IDisposable? BeginScope(TState state) where TState : notnull => null;
public bool IsEnabled(LogLevel logLevel) => true;
public void Log(
LogLevel logLevel, EventId eventId, TState state, Exception? exception,
Func formatter)
{
lock (Entries) Entries.Add((logLevel, exception, formatter(state, exception)));
}
}
private static DeploymentService CreateService(
IDeploymentManagerRepository repo,
TerminalPurgeRateLimiter? limiter,
ILogger? logger = null) =>
new(
repo,
Substitute.For(),
Substitute.For(),
Substitute.For(),
new CommunicationService(
Options.Create(new CommunicationOptions()), NullLogger.Instance),
new OperationLockManager(),
Substitute.For(),
new DiffService(),
new RevisionHashService(),
new DeploymentStatusNotifier(NullLogger.Instance),
Options.Create(new DeploymentManagerOptions()),
Options.Create(new CommunicationOptions()),
logger ?? NullLogger.Instance,
limiter);
private static IDeploymentManagerRepository PurgingRepo(int terminal = 1, int artifact = 3)
{
var repo = Substitute.For();
repo.PurgeTerminalDeploymentRecordsAsync(Arg.Any(), Arg.Any(), Arg.Any())
.Returns(terminal);
repo.PurgeSystemArtifactDeploymentRecordsAsync(Arg.Any(), Arg.Any(), Arg.Any())
.Returns(artifact);
return repo;
}
/// The first claim wins; a second inside the interval loses; after it, a claim wins again.
[Fact]
public void TryClaim_OncePerInterval_OnTheInjectedClock()
{
var clock = new ManualClock(Start);
var limiter = new TerminalPurgeRateLimiter(clock);
Assert.True(limiter.TryClaim(out var first));
Assert.Equal(Start, first);
clock.Now = Start + TerminalPurgeRateLimiter.MinInterval - TimeSpan.FromSeconds(1);
Assert.False(limiter.TryClaim(out _));
clock.Now = Start + TerminalPurgeRateLimiter.MinInterval;
Assert.True(limiter.TryClaim(out var later));
Assert.Equal(clock.Now, later);
Assert.False(limiter.TryClaim(out _));
}
/// Of many concurrent claimants exactly one wins.
/// A task that represents the asynchronous operation.
[Fact]
public async Task TryClaim_ConcurrentCallers_ExactlyOneWins()
{
var limiter = new TerminalPurgeRateLimiter(new ManualClock(Start));
using var gate = new Barrier(16);
var wins = await Task.WhenAll(Enumerable.Range(0, 16).Select(_ => Task.Run(() =>
{
gate.SignalAndWait();
return limiter.TryClaim(out var _);
})));
Assert.Equal(1, wins.Count(w => w));
}
///
/// The limiter is per instance, not process-wide: two services with their own limiters both
/// sweep, and two sharing one (as the DI singleton makes them) sweep once.
///
/// A task that represents the asynchronous operation.
[Fact]
public async Task Sweep_RateLimitIsPerLimiter_NotProcessWide()
{
var clock = new ManualClock(Start);
Assert.Equal(4, await CreateService(PurgingRepo(), new TerminalPurgeRateLimiter(clock))
.TryPurgeTerminalDeploymentRecordsAsync());
Assert.Equal(4, await CreateService(PurgingRepo(), new TerminalPurgeRateLimiter(clock))
.TryPurgeTerminalDeploymentRecordsAsync());
var shared = new TerminalPurgeRateLimiter(clock);
var secondRepo = PurgingRepo();
Assert.Equal(4, await CreateService(PurgingRepo(), shared).TryPurgeTerminalDeploymentRecordsAsync());
Assert.Equal(0, await CreateService(secondRepo, shared).TryPurgeTerminalDeploymentRecordsAsync());
await secondRepo.DidNotReceiveWithAnyArgs().PurgeTerminalDeploymentRecordsAsync(default, default, default);
}
///
/// The sweep's cutoff is taken from the injected clock: now minus the retention window.
///
/// A task that represents the asynchronous operation.
[Fact]
public async Task Sweep_CutoffIsInjectedClockMinusRetention()
{
var repo = PurgingRepo();
await CreateService(repo, new TerminalPurgeRateLimiter(new ManualClock(Start)))
.TryPurgeTerminalDeploymentRecordsAsync();
var cutoff = Start - new DeploymentManagerOptions().TerminalDeploymentRecordRetention;
await repo.Received(1).PurgeTerminalDeploymentRecordsAsync(cutoff, Arg.Any(), Arg.Any());
await repo.Received(1).PurgeSystemArtifactDeploymentRecordsAsync(cutoff, Arg.Any(), Arg.Any());
}
///
/// A failing sweep never fails the caller: the exception is logged as a warning and swallowed,
/// the rows already purged are still reported, and the failed sweep still consumes the
/// interval — the next call inside it does not retry, the first call after it does.
///
/// A task that represents the asynchronous operation.
[Fact]
public async Task Sweep_Failure_IsLoggedAndSwallowed_AndWaitsOutTheInterval()
{
var clock = new ManualClock(Start);
var limiter = new TerminalPurgeRateLimiter(clock);
var logger = new CapturingLogger();
var boom = new InvalidOperationException("deadlock victim");
var repo = Substitute.For();
repo.PurgeTerminalDeploymentRecordsAsync(Arg.Any(), Arg.Any(), Arg.Any())
.Returns(2);
repo.PurgeSystemArtifactDeploymentRecordsAsync(Arg.Any(), Arg.Any(), Arg.Any())
.ThrowsAsync(boom);
var service = CreateService(repo, limiter, logger);
var purged = await service.TryPurgeTerminalDeploymentRecordsAsync();
Assert.Equal(2, purged); // the terminal batch that landed before the failure
var warning = Assert.Single(logger.Entries, e => e.Level == LogLevel.Warning);
Assert.Same(boom, warning.Exception);
Assert.Contains("retention sweep failed", warning.Message, StringComparison.Ordinal);
Assert.DoesNotContain(logger.Entries, e => e.Level == LogLevel.Information);
// Inside the interval: rate-limited, no retry.
repo.ClearReceivedCalls();
clock.Now = Start + TimeSpan.FromHours(1);
Assert.Equal(0, await service.TryPurgeTerminalDeploymentRecordsAsync());
await repo.DidNotReceiveWithAnyArgs().PurgeTerminalDeploymentRecordsAsync(default, default, default);
// After the interval: the sweep runs again (and fails again, still swallowed).
clock.Now = Start + TerminalPurgeRateLimiter.MinInterval;
Assert.Equal(2, await service.TryPurgeTerminalDeploymentRecordsAsync());
await repo.ReceivedWithAnyArgs(1).PurgeTerminalDeploymentRecordsAsync(default, default, default);
Assert.Equal(2, logger.Entries.Count(e => e.Level == LogLevel.Warning));
}
/// A failure on the very first batch reports zero and is swallowed.
/// A task that represents the asynchronous operation.
[Fact]
public async Task Sweep_FirstBatchFailure_ReturnsZero_DoesNotThrow()
{
var repo = Substitute.For();
repo.PurgeTerminalDeploymentRecordsAsync(Arg.Any(), Arg.Any(), Arg.Any())
.ThrowsAsync(new TimeoutException("SQL timeout"));
var logger = new CapturingLogger();
var purged = await CreateService(repo, new TerminalPurgeRateLimiter(new ManualClock(Start)), logger)
.TryPurgeTerminalDeploymentRecordsAsync();
Assert.Equal(0, purged);
Assert.IsType(Assert.Single(logger.Entries, e => e.Level == LogLevel.Warning).Exception);
await repo.DidNotReceiveWithAnyArgs().PurgeSystemArtifactDeploymentRecordsAsync(default, default, default);
}
///
/// AddDeploymentManager registers the limiter as a singleton, so every scoped DeploymentService
/// shares one rate limit.
///
[Fact]
public void AddDeploymentManager_RegistersLimiterAsSingleton()
{
var services = new ServiceCollection();
services.AddDeploymentManager();
var descriptor = Assert.Single(services, d => d.ServiceType == typeof(TerminalPurgeRateLimiter));
Assert.Equal(ServiceLifetime.Singleton, descriptor.Lifetime);
using var provider = services.BuildServiceProvider();
using var a = provider.CreateScope();
using var b = provider.CreateScope();
Assert.Same(
a.ServiceProvider.GetRequiredService(),
b.ServiceProvider.GetRequiredService());
}
}