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()); } }