feat: изменил логику анализа отзывов

2026-05-22 01:30:41 +03:00
parent 168d6af860
commit 8ac593d36f
36 changed files with 858 additions and 457 deletions
@@ -0,0 +1,96 @@
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Options;
+using UniVerse.Api.BackgroundServices;
+using UniVerse.Api.Options;
+using UniVerse.Application.Interfaces;
+using UniVerse.Infrastructure.Data;
+using Xunit;
+
+namespace UniVerse.Api.Tests.Reviews;
+
+public class ReviewAnalysisWorkerTests
+{
+    [Theory]
+    [InlineData(1)]
+    [InlineData(2)]
+    public async Task Worker_DoesNotExceedConfiguredConcurrency(int maxConcurrentProcessing)
+    {
+        var queue = new ReviewAnalysisQueue();
+        var analysisService = new RecordingLlmAnalysisService();
+        await using var provider = CreateServiceProvider(analysisService);
+        var worker = new ReviewAnalysisWorker(
+            provider,
+            queue,
+            Microsoft.Extensions.Options.Options.Create(
+                new ReviewAnalysisOptions { MaxConcurrentProcessing = maxConcurrentProcessing }),
+            NullLogger<ReviewAnalysisWorker>.Instance);
+
+        for (var reviewId = 1; reviewId <= 6; reviewId++)
+            await queue.EnqueueAsync(reviewId);
+
+        analysisService.ExpectProcessed(6);
+        await worker.StartAsync(CancellationToken.None);
+        await analysisService.WaitForProcessedAsync();
+        await worker.StopAsync(CancellationToken.None);
+
+        Assert.True(
+            analysisService.MaxRunning <= maxConcurrentProcessing,
+            $"Expected at most {maxConcurrentProcessing} concurrent analyses, got {analysisService.MaxRunning}.");
+    }
+
+    private static ServiceProvider CreateServiceProvider(ILlmAnalysisService analysisService)
+    {
+        var services = new ServiceCollection();
+        services.AddDbContext<AppDbContext>(options =>
+            options.UseInMemoryDatabase($"ReviewAnalysisWorkerTests_{Guid.NewGuid()}"));
+        services.AddScoped(_ => analysisService);
+        return services.BuildServiceProvider();
+    }
+
+    private sealed class RecordingLlmAnalysisService : ILlmAnalysisService
+    {
+        private readonly TaskCompletionSource _processedAll = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        private int _expectedCount;
+        private int _processedCount;
+        private int _running;
+        private int _maxRunning;
+
+        public int MaxRunning => _maxRunning;
+
+        public void ExpectProcessed(int expectedCount)
+        {
+            Volatile.Write(ref _expectedCount, expectedCount);
+        }
+
+        public async Task AnalyzeReviewAsync(int reviewId)
+        {
+            var running = Interlocked.Increment(ref _running);
+            UpdateMaxRunning(running);
+
+            await Task.Delay(50);
+
+            Interlocked.Decrement(ref _running);
+            if (Interlocked.Increment(ref _processedCount) >= Volatile.Read(ref _expectedCount))
+                _processedAll.TrySetResult();
+        }
+
+        public async Task WaitForProcessedAsync()
+        {
+            using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
+            using var registration = timeout.Token.Register(() => _processedAll.TrySetCanceled(timeout.Token));
+            await _processedAll.Task;
+        }
+
+        private void UpdateMaxRunning(int running)
+        {
+            while (true)
+            {
+                var current = Volatile.Read(ref _maxRunning);
+                if (running <= current) return;
+                if (Interlocked.CompareExchange(ref _maxRunning, running, current) == current) return;
+            }
+        }
+    }
+}