Перевести FakePipelineStore на NSubstitute

src/core/tests/Deal.Tests.Unit/Api/IngressRateLimitInterceptorTests.cs

@@ -49,7 +49,7 @@ public sealed class IngressRateLimitInterceptorTests
 
         await RunAsync(
             registry,
-            services => services.AddScoped<IPipelineStore>(_ => new FakePipelineStore()),
+            services => services.AddScoped<IPipelineStore>(_ => new TestPipelineStore().Store),
             async channel =>
             {
                 PushSourceReply first = await PushAsync(channel, TenantA);
@@ -77,7 +77,7 @@ public sealed class IngressRateLimitInterceptorTests
 
         await RunAsync(
             registry,
-            services => services.AddScoped<IPipelineStore>(_ => new FakePipelineStore()),
+            services => services.AddScoped<IPipelineStore>(_ => new TestPipelineStore().Store),
             async channel =>
             {
                 // Окно ингресса — 1/мин: единственный разрешённый вызов исчерпывает лимит.

src/core/tests/Deal.Tests.Unit/Api/RuntimeDepthsCollectorTests.cs

@@ -29,10 +29,10 @@ public sealed class RuntimeDepthsCollectorTests
     [Fact]
     public async Task CollectAsync_SumsQueueAndOutboxAcrossTenants()
     {
-        var pipelineByTenant = new Dictionary<string, FakePipelineStore>
+        var pipelineByTenant = new Dictionary<string, TestPipelineStore>
         {
-            [TenantA.ToString("N")] = new FakePipelineStore(),
+            [TenantA.ToString("N")] = new TestPipelineStore(),
-            [TenantB.ToString("N")] = new FakePipelineStore(),
+            [TenantB.ToString("N")] = new TestPipelineStore(),
         };
         pipelineByTenant[TenantA.ToString("N")].SeedQueue(QueueItem("p_a1"));
         pipelineByTenant[TenantA.ToString("N")].SeedQueue(QueueItem("p_a2"));
@@ -105,7 +105,7 @@ public sealed class RuntimeDepthsCollectorTests
     public async Task CollectAsync_NoTenants_ReturnsZeros()
     {
         RuntimeDepthsCollector collector = Build(
-            new Dictionary<string, FakePipelineStore>(),
+            new Dictionary<string, TestPipelineStore>(),
             new Dictionary<string, TestMlLearningStore>(),
             tenants: Array.Empty<TenantRecordDto>());
 
@@ -123,8 +123,8 @@ public sealed class RuntimeDepthsCollectorTests
         Status = PipelineQueueStatuses.New,
     };
 
-    private static Dictionary<string, FakePipelineStore> PipelineStores(params Guid[] tenants)
+    private static Dictionary<string, TestPipelineStore> PipelineStores(params Guid[] tenants)
-        => tenants.ToDictionary(tenant => tenant.ToString("N"), _ => new FakePipelineStore());
+        => tenants.ToDictionary(tenant => tenant.ToString("N"), _ => new TestPipelineStore());
 
     private static Dictionary<string, TestMlLearningStore> OutboxStores(params Guid[] tenants)
         => tenants.ToDictionary(tenant => tenant.ToString("N"), _ => new TestMlLearningStore());
@@ -139,7 +139,7 @@ public sealed class RuntimeDepthsCollectorTests
 
     // Собирает коллектор поверх tenant-scoped фейков (как реальные адаптеры по ITenantContext).
     private static RuntimeDepthsCollector Build(
-        IReadOnlyDictionary<string, FakePipelineStore> pipelineByTenant,
+        IReadOnlyDictionary<string, TestPipelineStore> pipelineByTenant,
         IReadOnlyDictionary<string, TestMlLearningStore> outboxByTenant,
         IReadOnlyList<TenantRecordDto>? tenants = null,
         ITenantLimitStore? limitStore = null)
@@ -153,7 +153,7 @@ public sealed class RuntimeDepthsCollectorTests
                 new TenantRecordDto(TenantB, "B", "active", DateTimeOffset.UtcNow),
             }).ToArray()).Repository);
 
-        services.AddScoped<IPipelineStore>(provider => pipelineByTenant[CurrentTenant(provider)]);
+        services.AddScoped<IPipelineStore>(provider => pipelineByTenant[CurrentTenant(provider)].Store);
         services.AddScoped<IMlClient>(_ => new TestMlClient().Client);
         services.AddScoped<PipelineIngestService>();
         services.AddScoped<PipelineProcessingService>();

src/core/tests/Deal.Tests.Unit/Contracts/PipelineCardWriterTests.cs

@@ -18,9 +18,9 @@ public sealed class PipelineCardWriterTests
     [Fact]
     public async Task CreateCard_CreatesInboxCardAndLinksDedupClaim()
     {
-        (PipelineCardWriter writer, FakeKanjStore store, FakePipelineStore pipelineStore) = Create();
+        (PipelineCardWriter writer, FakeKanjStore store, TestPipelineStore pipelineStore) = Create();
         const string hash = "abcdef0123456789abcdef0123456789abcdef01";
-        await pipelineStore.ClaimAsync(hash, CancellationToken.None); // заявка воркера (LeadId=null, Ruling 8)
+        await pipelineStore.Store.ClaimAsync(hash, CancellationToken.None); // заявка воркера (LeadId=null, Ruling 8)
         AiParsedCardDto parsed = Parsed(
             title: "Python-разработчик на бота",
             budget: new AiBudgetDto(2000, 2000, "USD"),
@@ -38,16 +38,16 @@ public sealed class PipelineCardWriterTests
         Assert.Equal(new CardBudgetDto(2000, 2000, "USD"), card.Budget);
 
         Assert.Equal(card.Id, pipelineStore.DedupLeadId(hash));
-        Assert.True(await pipelineStore.ExistsAsync(hash, CancellationToken.None));
+        Assert.True(await pipelineStore.Store.ExistsAsync(hash, CancellationToken.None));
     }
 
     [Fact]
     public async Task CreateCard_AssignedBoardAccepted_WritesCardIntoBoardColumn()
     {
-        (PipelineCardWriter writer, FakeKanjStore store, FakePipelineStore pipelineStore) = Create();
+        (PipelineCardWriter writer, FakeKanjStore store, TestPipelineStore pipelineStore) = Create();
         store.SeedBoard(new ContainerDto { Id = "b_py" });
         const string hash = "abcdef0123456789abcdef0123456789abcdef01";
-        await pipelineStore.ClaimAsync(hash, CancellationToken.None);
+        await pipelineStore.Store.ClaimAsync(hash, CancellationToken.None);
 
         CardDto card = await writer.CreateCardAsync(
             Parsed(title: "Python-разработчик", board: "b_py"),
@@ -66,10 +66,10 @@ public sealed class PipelineCardWriterTests
     [Fact]
     public async Task CreateCard_AddCardFails_ThrowsAndDoesNotLinkDedup()
     {
-        (PipelineCardWriter writer, FakeKanjStore store, FakePipelineStore pipelineStore) = Create();
+        (PipelineCardWriter writer, FakeKanjStore store, TestPipelineStore pipelineStore) = Create();
         store.FailAddCard = true; // сбой адаптера записи (например, недоступна БД тенанта)
         const string hash = "abcdef0123456789abcdef0123456789abcdef01";
-        await pipelineStore.ClaimAsync(hash, CancellationToken.None);
+        await pipelineStore.Store.ClaimAsync(hash, CancellationToken.None);
 
         await Assert.ThrowsAsync<InvalidOperationException>(
             () => writer.CreateCardAsync(Parsed(title: "Python-разработчик"), Message("Текст"), hash, CancellationToken.None));
@@ -77,19 +77,19 @@ public sealed class PipelineCardWriterTests
         // Карточка не создана → связывать нечего: заявка дедупа осталась несвязанной (LeadId=null).
         Assert.Empty(store.CardDtos);
         Assert.Null(pipelineStore.DedupLeadId(hash));
-        Assert.True(await pipelineStore.ExistsAsync(hash, CancellationToken.None)); // заявка на месте
+        Assert.True(await pipelineStore.Store.ExistsAsync(hash, CancellationToken.None)); // заявка на месте
     }
 
     // ─── Хелперы ──────────────────────────────────────────────────────────────────────────
 
-    // Создаёт контекст теста: фейки канбана, пайплайна и композитор поверх них.
+    // Создаёт контекст теста: подставки канбана, пайплайна и композитор поверх них.
     // Возвращает: Кортеж (обёртка, канбан-хранилище, хранилище пайплайна).
-    private static (PipelineCardWriter Writer, FakeKanjStore Store, FakePipelineStore PipelineStore) Create()
+    private static (PipelineCardWriter Writer, FakeKanjStore Store, TestPipelineStore PipelineStore) Create()
     {
         var store = new FakeKanjStore();
-        var pipelineStore = new FakePipelineStore();
+        var pipelineStore = new TestPipelineStore();
         var composer = new CardComposer(store, new TestSettingsStore().Store);
-        return (new PipelineCardWriter(store, pipelineStore, composer), store, pipelineStore);
+        return (new PipelineCardWriter(store, pipelineStore.Store, composer), store, pipelineStore);
     }
 
     // Разбор карточки со значениями по умолчанию (сценарий теста перекрывает нужные поля).

src/core/tests/Deal.Tests.Unit/Modules/Kanban/FakePipelineStore.cs

@@ -1,315 +0,0 @@
-using Deal.Modules.Cards.Application.Sources;
-using Deal.Modules.Kanban.Application.Models;
-using Deal.Modules.Pipeline.Application.Abstractions;
-using Deal.Modules.Pipeline.Application.Models;
-
-namespace Deal.Tests.Unit.Modules.Kanban;
-
-/// <summary>
-/// In-memory реализация <see cref="IPipelineStore"/> для unit-тестов PipelineIngestService/PipelineProcessingService.
-/// </summary>
-public class FakePipelineStore : IPipelineStore
-{
-    private readonly List<QueueItemDto> _queue = [];
-    private readonly List<RejectedItemDto> _rejected = [];
-    private readonly Dictionary<string, string?> _dedup = new(StringComparer.Ordinal);
-
-    /// <summary>
-    /// Строки очереди фейка
-    /// </summary>
-    public IReadOnlyList<QueueItemDto> Queue => _queue.ToList();
-
-    /// <summary>
-    /// Записи отсева фейка
-    /// </summary>
-    public IReadOnlyList<RejectedItemDto> Rejected => _rejected.ToList();
-
-    /// <summary>
-    /// Кладёт строку очереди напрямую
-    /// </summary>
-    /// <param name="item">Строка как если бы была сохранена в БД.</param>
-    public void SeedQueue(QueueItemDto item)
-    {
-        _queue.Add(item);
-    }
-
-    /// <summary>
-    /// Кладёт запись отсева напрямую
-    /// </summary>
-    /// <param name="item">Запись как если бы была сохранена в БД.</param>
-    public void SeedRejected(RejectedItemDto item)
-    {
-        _rejected.Add(item);
-    }
-
-    // ── Очередь (QueueItems) ───────────────────────────────────────────────
-
-    /// <inheritdoc />
-    public Task<bool> ExistsDuplicateAsync(
-        SourceRef source,
-        CancellationToken ct)
-    {
-        string key = source.DedupeKey();
-        bool found = _queue.Any(item => item.Source.DedupeKey() == key);
-        return Task.FromResult(found);
-    }
-
-    /// <inheritdoc />
-    public Task AddAsync(QueueItemDto item, CancellationToken ct)
-    {
-        _queue.Add(item);
-        return Task.CompletedTask;
-    }
-
-    /// <inheritdoc />
-    public virtual Task<IReadOnlyList<QueueItemDto>> ListAsync(
-        string? status,
-        int limit,
-        CancellationToken ct)
-    {
-        IReadOnlyList<QueueItemDto> items = _queue
-            .Where(item => status is null || item.Status == status)
-            .OrderBy(item => item.QueuedAtMs)
-            .Take(limit)
-            .ToList();
-        return Task.FromResult(items);
-    }
-
-    /// <inheritdoc />
-    public Task<int> CountByStatusAsync(string status, CancellationToken ct)
-    {
-        return Task.FromResult(_queue.Count(item => item.Status == status));
-    }
-
-    /// <inheritdoc />
-    public Task SetStatusAsync(
-        string id,
-        string status,
-        CancellationToken ct)
-    {
-        int index = _queue.FindIndex(item => item.Id == id);
-        if (index >= 0)
-        {
-            _queue[index] = _queue[index] with { Status = status };
-        }
-
-        return Task.CompletedTask;
-    }
-
-    /// <inheritdoc />
-    public Task RemoveAsync(string id, CancellationToken ct)
-    {
-        _queue.RemoveAll(item => item.Id == id);
-        return Task.CompletedTask;
-    }
-
-    // ── Отсев (RejectedItems) ──────────────────────────────────────────────
-
-    /// <inheritdoc />
-    public Task UpsertAsync(RejectRecord record, CancellationToken ct)
-    {
-        if (string.IsNullOrWhiteSpace(record.Text))
-        {
-            return Task.CompletedTask;
-        }
-
-        string id = record.DeterministicId ?? PrefixId.New(PipelineIdPrefixes.Rejected);
-        long nowMs = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
-
-        int index = _rejected.FindIndex(item => item.Id == id);
-        RejectedItemDto row = ToItem(record, id, nowMs);
-        if (index >= 0)
-        {
-            RejectedItemDto existing = _rejected[index];
-            row = row with
-            {
-                Returned = existing.Returned,
-                ReturnedAtMs = existing.ReturnedAtMs,
-                ReturnReason = existing.ReturnReason,
-            };
-            _rejected[index] = row;
-        }
-        else
-        {
-            _rejected.Add(row);
-        }
-
-        return Task.CompletedTask;
-    }
-
-    /// <inheritdoc />
-    public Task<IReadOnlyList<RejectedItemDto>> ListPageAsync(
-        int offset,
-        int limit,
-        CancellationToken ct)
-    {
-        IReadOnlyList<RejectedItemDto> page = _rejected
-            .OrderByDescending(item => item.RejectedAtMs)
-            .Skip(offset)
-            .Take(limit)
-            .ToList();
-        return Task.FromResult(page);
-    }
-
-    /// <inheritdoc />
-    public Task<IReadOnlyList<RejectedItemDto>> SearchAsync(
-        string q,
-        int limitFts,
-        int limitLike,
-        CancellationToken ct)
-    {
-        string query = q.Trim();
-        IReadOnlyList<RejectedItemDto> rows = _rejected
-            .Where(item => LikeField(item.Text, query)
-                || LikeField(item.Reason, query)
-                || LikeField(item.Kw, query)
-                || LikeField(item.Source.DisplayName ?? string.Empty, query))
-            .OrderByDescending(item => item.RejectedAtMs)
-            .Take(limitLike)
-            .ToList();
-        return Task.FromResult(rows);
-    }
-
-    /// <inheritdoc />
-    public Task<int> CountAsync(CancellationToken ct)
-    {
-        return Task.FromResult(_rejected.Count);
-    }
-
-    /// <inheritdoc />
-    public Task<RejectedItemDto?> GetAsync(string id, CancellationToken ct)
-    {
-        return Task.FromResult(_rejected.FirstOrDefault(item => item.Id == id));
-    }
-
-    /// <inheritdoc />
-    public Task DeleteAsync(string id, CancellationToken ct)
-    {
-        _rejected.RemoveAll(item => item.Id == id);
-        return Task.CompletedTask;
-    }
-
-    /// <inheritdoc />
-    public Task<int> ClearAsync(CancellationToken ct)
-    {
-        int cleared = _rejected.Count;
-        _rejected.Clear();
-        return Task.FromResult(cleared);
-    }
-
-    /// <inheritdoc />
-    public Task<int> PurgeExpiredAsync(DateTimeOffset olderThan, CancellationToken ct)
-    {
-        long cutMs = olderThan.ToUnixTimeMilliseconds();
-        int removed = _rejected.RemoveAll(item => item.RejectedAtMs < cutMs);
-        return Task.FromResult(removed);
-    }
-
-    /// <inheritdoc />
-    public Task MarkReturnedAsync(
-        string id,
-        string reason,
-        DateTimeOffset returnedAt,
-        CancellationToken ct)
-    {
-        int index = _rejected.FindIndex(item => item.Id == id);
-        if (index >= 0)
-        {
-            _rejected[index] = _rejected[index] with
-            {
-                Returned = true,
-                ReturnedAtMs = returnedAt.ToUnixTimeMilliseconds(),
-                ReturnReason = reason,
-            };
-        }
-
-        return Task.CompletedTask;
-    }
-
-    // ── Дедуп (DedupEntries) ───────────────────────────────────────────────
-
-    /// <summary>
-    /// Id карточки, связанной с хэшем дедупа
-    /// </summary>
-    /// <param name="hash">SHA1-hex нормализованного текста.</param>
-    public string? DedupLeadId(string hash)
-    {
-        return _dedup.TryGetValue(hash, out string? leadId) ? leadId : null;
-    }
-
-    /// <inheritdoc />
-    public virtual Task<bool> ExistsAsync(string hash, CancellationToken ct)
-    {
-        return Task.FromResult(_dedup.ContainsKey(hash));
-    }
-
-    /// <inheritdoc />
-    public virtual Task<bool> ClaimAsync(string hash, CancellationToken ct)
-    {
-        // Атомарность claim'а (как PipelineStore.ClaimAsync, ON CONFLICT DO NOTHING): true — заявка занята
-        // этим вызовом, false — хэш уже заявлен (TryAdd) другим проходом pump.
-        return Task.FromResult(_dedup.TryAdd(hash, null));
-    }
-
-    /// <inheritdoc />
-    public virtual Task DeleteClaimAsync(string hash, CancellationToken ct)
-    {
-        if (_dedup.TryGetValue(hash, out string? leadId) && leadId is null)
-        {
-            _dedup.Remove(hash);
-        }
-
-        return Task.CompletedTask;
-    }
-
-    /// <inheritdoc />
-    public Task LinkAsync(
-        string hash,
-        string cardId,
-        CancellationToken ct)
-    {
-        _dedup[hash] = cardId;
-        return Task.CompletedTask;
-    }
-
-    /// <inheritdoc />
-    public Task DeleteByCardAsync(string cardId, CancellationToken ct)
-    {
-        foreach (string hash in _dedup.Where(pair => pair.Value == cardId).Select(pair => pair.Key).ToList())
-        {
-            _dedup.Remove(hash);
-        }
-
-        return Task.CompletedTask;
-    }
-
-    // ── Маппинг (как адаптер: подписи — словари PipelineRejectConstants) ────
-
-    // Подстрока q в lower(поле) — LIKE-половина адаптера (поле колонки SQL приводится к нижнему регистру).
-    // field: Поле записи отсева (text/reason/kw/ch_name).
-    // query: Поисковый запрос, нормализованный сервисом (trim+lowercase).
-    // Возвращает: True — поле содержит запрос без учёта регистра поля.
-    private static bool LikeField(string field, string query)
-    {
-        return field.ToLowerInvariant().Contains(query, StringComparison.Ordinal);
-    }
-
-    private static RejectedItemDto ToItem(
-        RejectRecord record,
-        string id,
-        long rejectedAtMs) => new()
-    {
-        Id = id,
-        Source = record.Source,
-        Content = record.Content,
-        Text = record.Text,
-        Stage = record.Stage,
-        StageLabel = PipelineRejectConstants.StageLabel(record.Stage),
-        Reason = record.Reason,
-        Kw = record.Kw,
-        DecidedBy = record.DecidedBy,
-        DecidedByLabel = PipelineRejectConstants.SourceLabel(record.DecidedBy),
-        MsgAtMs = record.MsgAtMs,
-        RejectedAtMs = rejectedAtMs,
-    };
-}

src/core/tests/Deal.Tests.Unit/Modules/Pipeline/PipelineIngestServiceTests.cs

@@ -1,7 +1,6 @@
 using Deal.Modules.Cards.Application.Sources;
 using Deal.Modules.Pipeline.Application.Models;
 using Deal.Modules.Pipeline.Application.Services;
-using Deal.Tests.Unit.Modules.Kanban;
 using Deal.Tests.Unit.Support;
 
 namespace Deal.Tests.Unit.Modules.Pipeline;
@@ -16,8 +15,8 @@ public sealed class PipelineIngestServiceTests
     [Fact]
     public async Task EnqueueAsync_TrimsTextAndWritesNewRowWithPId()
     {
-        FakePipelineStore store = new();
+        TestPipelineStore store = new();
-        PipelineIngestService service = new(store);
+        PipelineIngestService service = new(store.Store);
 
         PipelineIngestResultDto result = await service.EnqueueAsync(
             Message(
@@ -48,8 +47,8 @@ public sealed class PipelineIngestServiceTests
     [Fact]
     public async Task EnqueueAsync_MsgAtAbsent_ZeroMsgAtAndQueuedNow()
     {
-        FakePipelineStore store = new();
+        TestPipelineStore store = new();
-        PipelineIngestService service = new(store);
+        PipelineIngestService service = new(store.Store);
 
         await service.EnqueueAsync(
             Message("Сообщение без времени", receivedAt: default(DateTimeOffset)),
@@ -63,8 +62,8 @@ public sealed class PipelineIngestServiceTests
     [Fact]
     public async Task EnqueueAsync_ForceFlag_PropagatedToQueueRow()
     {
-        FakePipelineStore store = new();
+        TestPipelineStore store = new();
-        PipelineIngestService service = new(store);
+        PipelineIngestService service = new(store.Store);
 
         await service.EnqueueAsync(Message("Возврат из отсева", force: true), CancellationToken.None);
 
@@ -76,8 +75,8 @@ public sealed class PipelineIngestServiceTests
     [Fact]
     public async Task EnqueueAsync_EmptyOrWhitespaceText_Noop()
     {
-        FakePipelineStore store = new();
+        TestPipelineStore store = new();
-        PipelineIngestService service = new(store);
+        PipelineIngestService service = new(store.Store);
 
         PipelineIngestResultDto empty = await service.EnqueueAsync(Message(text: string.Empty), CancellationToken.None);
         PipelineIngestResultDto spaces = await service.EnqueueAsync(Message(text: "   "), CancellationToken.None);
@@ -93,8 +92,8 @@ public sealed class PipelineIngestServiceTests
     [Fact]
     public async Task EnqueueAsync_SameSourceTwice_SecondIsDuplicate()
     {
-        FakePipelineStore store = new();
+        TestPipelineStore store = new();
-        PipelineIngestService service = new(store);
+        PipelineIngestService service = new(store.Store);
         QueuedMessage first = Message("Сообщение", origin: "d_1", externalId: "7");
         QueuedMessage second = Message("Сообщение снова", origin: "d_1", externalId: "7");
 
@@ -111,8 +110,8 @@ public sealed class PipelineIngestServiceTests
     [Fact]
     public async Task EnqueueAsync_SameSourceDifferentExternalIds_BothAccepted()
     {
-        FakePipelineStore store = new();
+        TestPipelineStore store = new();
-        PipelineIngestService service = new(store);
+        PipelineIngestService service = new(store.Store);
 
         await service.EnqueueAsync(Message("Первое", origin: "d_1", externalId: "1"), CancellationToken.None);
         PipelineIngestResultDto second = await service.EnqueueAsync(Message("Второе", origin: "d_1", externalId: "2"), CancellationToken.None);
@@ -124,8 +123,8 @@ public sealed class PipelineIngestServiceTests
     [Fact]
     public async Task EnqueueAsync_SameOriginWithoutExternalId_SecondIsDuplicate()
     {
-        FakePipelineStore store = new();
+        TestPipelineStore store = new();
-        PipelineIngestService service = new(store);
+        PipelineIngestService service = new(store.Store);
 
         PipelineIngestResultDto first = await service.EnqueueAsync(
             Message("Первое", origin: "d_1", externalId: null), CancellationToken.None);
@@ -142,8 +141,8 @@ public sealed class PipelineIngestServiceTests
     [Fact]
     public async Task EnqueueAsync_LongText_TruncatedTo6000CodePoints()
     {
-        FakePipelineStore store = new();
+        TestPipelineStore store = new();
-        PipelineIngestService service = new(store);
+        PipelineIngestService service = new(store.Store);
         string longText = new string('а', 7_000);
 
         await service.EnqueueAsync(Message(longText), CancellationToken.None);

src/core/tests/Deal.Tests.Unit/Modules/Pipeline/PipelineProcessingServiceTests.cs

@@ -3,7 +3,6 @@ using Deal.Modules.Cards.Application.Sources;
 using Deal.Modules.Pipeline.Application.Models;
 using Deal.Modules.Pipeline.Application.Services;
 using Deal.Tests.Unit.Contracts;
-using Deal.Tests.Unit.Modules.Kanban;
 using Deal.Tests.Unit.Support;
 
 namespace Deal.Tests.Unit.Modules.Pipeline;
@@ -17,7 +16,7 @@ public sealed class PipelineProcessingServiceTests
     [Fact]
     public async Task RejectAsync_WhitespaceText_Noop()
     {
-        (PipelineProcessingService service, FakePipelineStore store, _) = Create();
+        (PipelineProcessingService service, TestPipelineStore store, _) = Create();
 
         await service.RejectAsync(new RejectRecord { Text = "   ", DecidedBy = "stop", Stage = "length" }, CancellationToken.None);
 
@@ -27,7 +26,7 @@ public sealed class PipelineProcessingServiceTests
     [Fact]
     public async Task RejectAsync_TruncatesFieldsDefaultsHueAndUsesDeterministicId()
     {
-        (PipelineProcessingService service, FakePipelineStore store, _) = Create();
+        (PipelineProcessingService service, TestPipelineStore store, _) = Create();
 
         await service.RejectAsync(new RejectRecord
         {
@@ -53,7 +52,7 @@ public sealed class PipelineProcessingServiceTests
     [Fact]
     public async Task RejectAsync_SameDialogAndMsgId_UpsertsSingleRow()
     {
-        (PipelineProcessingService service, FakePipelineStore store, _) = Create();
+        (PipelineProcessingService service, TestPipelineStore store, _) = Create();
 
         await service.RejectAsync(Reject("d_1", 7, stage: "length", reason: "короткое"), CancellationToken.None);
         await service.RejectAsync(Reject("d_1", 7, stage: "stop", reason: "стоп-фраза «реклама»"), CancellationToken.None);
@@ -68,7 +67,7 @@ public sealed class PipelineProcessingServiceTests
     [Fact]
     public async Task QueueCountsAsync_CountsNewAndFiltered()
     {
-        (PipelineProcessingService service, FakePipelineStore store, _) = Create();
+        (PipelineProcessingService service, TestPipelineStore store, _) = Create();
         store.SeedQueue(QueueItem("p_1", PipelineQueueStatuses.New, queuedAtMs: 1));
         store.SeedQueue(QueueItem("p_2", PipelineQueueStatuses.New, queuedAtMs: 2));
         store.SeedQueue(QueueItem("p_3", PipelineQueueStatuses.Filtered, queuedAtMs: 3));
@@ -83,7 +82,7 @@ public sealed class PipelineProcessingServiceTests
     [Fact]
     public async Task StatsAsync_ReturnsQueueCountsAndRejectedCount()
     {
-        (PipelineProcessingService service, FakePipelineStore store, _) = Create();
+        (PipelineProcessingService service, TestPipelineStore store, _) = Create();
         store.SeedQueue(QueueItem("p_1", PipelineQueueStatuses.New, queuedAtMs: 1));
         store.SeedRejected(Rejected("r_1"));
         store.SeedRejected(Rejected("r_2"));
@@ -99,7 +98,7 @@ public sealed class PipelineProcessingServiceTests
     [Fact]
     public async Task ListQueueAsync_OrdersByQueuedAtAndClampsLimitToMinimumOne()
     {
-        (PipelineProcessingService service, FakePipelineStore store, _) = Create();
+        (PipelineProcessingService service, TestPipelineStore store, _) = Create();
         store.SeedQueue(QueueItem("p_1", PipelineQueueStatuses.New, queuedAtMs: 30));
         store.SeedQueue(QueueItem("p_2", PipelineQueueStatuses.Filtered, queuedAtMs: 10));
         store.SeedQueue(QueueItem("p_3", PipelineQueueStatuses.New, queuedAtMs: 20));
@@ -116,7 +115,7 @@ public sealed class PipelineProcessingServiceTests
     [Fact]
     public async Task ListRejectedAsync_NoQuery_NewestFirstWithOffsetAndLimit()
     {
-        (PipelineProcessingService service, FakePipelineStore store, _) = Create();
+        (PipelineProcessingService service, TestPipelineStore store, _) = Create();
         store.SeedRejected(Rejected("r_1", rejectedAtMs: 1_000));
         store.SeedRejected(Rejected("r_2", rejectedAtMs: 2_000));
         store.SeedRejected(Rejected("r_3", rejectedAtMs: 3_000));
@@ -132,7 +131,7 @@ public sealed class PipelineProcessingServiceTests
     [Fact]
     public async Task ListRejectedAsync_QueryMatchesTextReasonKwAndSourceName()
     {
-        (PipelineProcessingService service, FakePipelineStore store, _) = Create();
+        (PipelineProcessingService service, TestPipelineStore store, _) = Create();
         store.SeedRejected(Rejected("r_text", text: "Middle Python разработчик", rejectedAtMs: 1_000));
         store.SeedRejected(Rejected("r_reason", reason: "стоп-фраза «реклама»", rejectedAtMs: 2_000));
         store.SeedRejected(Rejected("r_kw", kw: "спам", rejectedAtMs: 3_000));
@@ -152,7 +151,7 @@ public sealed class PipelineProcessingServiceTests
     [Fact]
     public async Task ListRejectedAsync_QueryWithOffsetPagesOverCandidatesAndEchoesTotal()
     {
-        (PipelineProcessingService service, FakePipelineStore store, _) = Create();
+        (PipelineProcessingService service, TestPipelineStore store, _) = Create();
         store.SeedRejected(Rejected("r_1", text: "middle a", rejectedAtMs: 1_000));
         store.SeedRejected(Rejected("r_2", text: "middle b", rejectedAtMs: 2_000));
         store.SeedRejected(Rejected("r_3", text: "middle c", rejectedAtMs: 3_000));
@@ -168,7 +167,7 @@ public sealed class PipelineProcessingServiceTests
     [Fact]
     public async Task ReturnAsync_RecordNotFound_ReturnsNull()
     {
-        (PipelineProcessingService service, FakePipelineStore store, TestMlClient ml) = Create();
+        (PipelineProcessingService service, TestPipelineStore store, TestMlClient ml) = Create();
 
         RejectReturnResultDto? result = await service.ReturnAsync("r_missing", string.Empty, CancellationToken.None);
 
@@ -180,7 +179,7 @@ public sealed class PipelineProcessingServiceTests
     [Fact]
     public async Task ReturnAsync_AlreadyReturned_Returns400()
     {
-        (PipelineProcessingService service, FakePipelineStore store, TestMlClient ml) = Create();
+        (PipelineProcessingService service, TestPipelineStore store, TestMlClient ml) = Create();
         store.SeedRejected(Rejected("r_1", returned: true, returnReason: "было"));
 
         RejectReturnResultDto? result = await service.ReturnAsync("r_1", "ещё раз", CancellationToken.None);
@@ -195,7 +194,7 @@ public sealed class PipelineProcessingServiceTests
     [Fact]
     public async Task ReturnAsync_DuplicateSource_Returns400()
     {
-        (PipelineProcessingService service, FakePipelineStore store, TestMlClient ml) = Create();
+        (PipelineProcessingService service, TestPipelineStore store, TestMlClient ml) = Create();
         store.SeedRejected(Rejected("r_1", decidedBy: "dup", stage: "dup"));
 
         RejectReturnResultDto? result = await service.ReturnAsync("r_1", string.Empty, CancellationToken.None);
@@ -209,7 +208,7 @@ public sealed class PipelineProcessingServiceTests
     [Fact]
     public async Task ReturnAsync_EmptyText_Returns400()
     {
-        (PipelineProcessingService service, FakePipelineStore store, TestMlClient ml) = Create();
+        (PipelineProcessingService service, TestPipelineStore store, TestMlClient ml) = Create();
         store.SeedRejected(Rejected("r_1", text: "   "));
 
         RejectReturnResultDto? result = await service.ReturnAsync("r_1", string.Empty, CancellationToken.None);
@@ -223,7 +222,7 @@ public sealed class PipelineProcessingServiceTests
     [Fact]
     public async Task ReturnAsync_SpamStage_UnlearnsSpamMarksRecordAndEnqueuesForceRow()
     {
-        (PipelineProcessingService service, FakePipelineStore store, TestMlClient ml) = Create();
+        (PipelineProcessingService service, TestPipelineStore store, TestMlClient ml) = Create();
         store.SeedRejected(Rejected(
             "r_1",
             text: "  Продвижение в каналах  ",
@@ -270,7 +269,7 @@ public sealed class PipelineProcessingServiceTests
     [Fact]
     public async Task ReturnAsync_NonSpamStage_NoLearningPushButEnqueuesForceRow()
     {
-        (PipelineProcessingService service, FakePipelineStore store, TestMlClient ml) = Create();
+        (PipelineProcessingService service, TestPipelineStore store, TestMlClient ml) = Create();
         store.SeedRejected(Rejected(
             "r_1",
             text: "Короткое сообщение",
@@ -296,7 +295,7 @@ public sealed class PipelineProcessingServiceTests
     [Fact]
     public async Task ReturnAsync_LongReason_TruncatedTo500CodePointsOnRecord()
     {
-        (PipelineProcessingService service, FakePipelineStore store, _) = Create();
+        (PipelineProcessingService service, TestPipelineStore store, _) = Create();
         store.SeedRejected(Rejected("r_1", text: "Текст", stage: "stale", decidedBy: "stale", origin: "d_1"));
 
         await service.ReturnAsync("r_1", new string('п', 700), CancellationToken.None);
@@ -309,7 +308,7 @@ public sealed class PipelineProcessingServiceTests
     [Fact]
     public async Task ReturnAsync_NoSourceInfo_EnqueuesDirectlyWithDefaultHue()
     {
-        (PipelineProcessingService service, FakePipelineStore store, TestMlClient ml) = Create();
+        (PipelineProcessingService service, TestPipelineStore store, TestMlClient ml) = Create();
         store.SeedRejected(Rejected(
             "r_old",
             text: "Старая запись без ссылки на исходное",
@@ -340,7 +339,7 @@ public sealed class PipelineProcessingServiceTests
     [Fact]
     public async Task DeleteAsync_RemovesSingleRecord()
     {
-        (PipelineProcessingService service, FakePipelineStore store, _) = Create();
+        (PipelineProcessingService service, TestPipelineStore store, _) = Create();
         store.SeedRejected(Rejected("r_1"));
         store.SeedRejected(Rejected("r_2"));
 
@@ -352,7 +351,7 @@ public sealed class PipelineProcessingServiceTests
     [Fact]
     public async Task ClearAsync_ReturnsCountAndEmptiesRejects()
     {
-        (PipelineProcessingService service, FakePipelineStore store, _) = Create();
+        (PipelineProcessingService service, TestPipelineStore store, _) = Create();
         store.SeedRejected(Rejected("r_1"));
         store.SeedRejected(Rejected("r_2"));
         store.SeedRejected(Rejected("r_3"));
@@ -368,7 +367,7 @@ public sealed class PipelineProcessingServiceTests
     [Fact]
     public async Task PurgeExpiredAsync_RemovesOnlyRowsOlderThanRetention()
     {
-        (PipelineProcessingService service, FakePipelineStore store, _) = Create();
+        (PipelineProcessingService service, TestPipelineStore store, _) = Create();
         long nowMs = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
         store.SeedRejected(Rejected("r_old", rejectedAtMs: (long)(nowMs - TimeSpan.FromDays(4).TotalMilliseconds)));
         store.SeedRejected(Rejected("r_fresh", rejectedAtMs: (long)(nowMs - TimeSpan.FromDays(1).TotalMilliseconds)));
@@ -381,12 +380,12 @@ public sealed class PipelineProcessingServiceTests
 
     // ─── Помощники ──────────────────────────────────────────────────────────
 
-    // Сервис на общих фейках: очередь/отсев и ML-клиент одного сценария.
+    // Сервис на общих подставках: очередь/отсев и ML-клиент одного сценария.
-    private static (PipelineProcessingService Service, FakePipelineStore Store, TestMlClient Ml) Create()
+    private static (PipelineProcessingService Service, TestPipelineStore Store, TestMlClient Ml) Create()
     {
-        FakePipelineStore store = new();
+        TestPipelineStore store = new();
         TestMlClient ml = new();
-        PipelineProcessingService service = new(store, ml.Client, new PipelineIngestService(store));
+        PipelineProcessingService service = new(store.Store, ml.Client, new PipelineIngestService(store.Store));
         return (service, store, ml);
     }
 

src/core/tests/Deal.Tests.Unit/Support/PipelineWorkerSchedulerTests.cs

@@ -41,10 +41,10 @@ public sealed class PipelineWorkerSchedulerTests
     // Контекст теста: планировщик на общих фейках + каналы подписок тенантов.
     private sealed record Context(
         PipelineWorkerScheduler Scheduler,
-        FakePipelineStore PipelineA,
+        TestPipelineStore PipelineA,
         FakeKanjStore KanjA,
         SseSubscription SubscriptionA,
-        FakePipelineStore PipelineB,
+        TestPipelineStore PipelineB,
         FakeKanjStore KanjB,
         SseSubscription SubscriptionB,
         PipelinePumpGate PumpGate,
@@ -151,8 +151,8 @@ public sealed class PipelineWorkerSchedulerTests
     {
         var tenants = new TestTenantRepository(Tenant(TenantA), Tenant(TenantB));
         var tenantContext = new TenantContext();
-        FakePipelineStore pipelineA = withThrowingQueueReadA ? new ThrowingQueueReadPipelineStore() : new FakePipelineStore();
+        var pipelineA = new TestPipelineStore(throwOnList: withThrowingQueueReadA);
-        var pipelineB = new FakePipelineStore();
+        var pipelineB = new TestPipelineStore();
         var kanjA = new FakeKanjStore();
         var kanjB = new FakeKanjStore();
         var settingsA = new TestSettingsStore();
@@ -174,7 +174,7 @@ public sealed class PipelineWorkerSchedulerTests
         // (эталон StorageTickSchedulerTests/ConnectionStringProvider.ForTenant).
         services.AddScoped<ICardStore>(provider => TenantOf(provider) == TenantA ? kanjA : kanjB);
         services.AddScoped<ISettingsStore>(provider => TenantOf(provider) == TenantA ? settingsA.Store : settingsB.Store);
-        services.AddScoped<IPipelineStore>(provider => TenantOf(provider) == TenantA ? pipelineA : pipelineB);
+        services.AddScoped<IPipelineStore>(provider => TenantOf(provider) == TenantA ? pipelineA.Store : pipelineB.Store);
         // Реальные сервисы модуля Pipeline — как AddPipelineModule в Program.cs: цикл резолвит их в tenant-scope.
         services.AddScoped<IncomingRules>();
         services.AddScoped<LocalFieldsParser>();
@@ -267,18 +267,7 @@ public sealed class PipelineWorkerSchedulerTests
         return cardIds;
     }
 
-    // Хранилище со сбоем чтения очереди: ListAsync бросает (сценарий «БД/схема недоступны» на pump).
+    // Хранилище со сбоем чтения очереди задаётся флагом throwOnList в TestPipelineStore.
-    private sealed class ThrowingQueueReadPipelineStore : FakePipelineStore
-    {
-        /// <inheritdoc />
-        public override Task<IReadOnlyList<QueueItemDto>> ListAsync(
-            string? status,
-            int limit,
-            CancellationToken ct)
-        {
-            throw new InvalidOperationException("Тестовый сбой чтения очереди (ListAsync).");
-        }
-    }
 
     // Логгер-коллектор: копит сообщения планировщика (диагностика в тестах сбоя pump).
     private sealed class ListLogger : ILogger<PipelineWorkerScheduler>

src/core/tests/Deal.Tests.Unit/Support/SourceIngressGrpcServiceTests.cs

@@ -46,12 +46,12 @@ public sealed class SourceIngressGrpcServiceTests
     [Fact]
     public async Task PushSource_ValidTenant_EnqueuesQueueRowAndAccepts()
     {
-        var store = new FakePipelineStore();
+        var store = new TestPipelineStore();
         var registry = TestRegistry.With(Tenant(TenantA));
 
         await RunAsync(
             registry,
-            services => services.AddScoped<IPipelineStore>(_ => store),
+            services => services.AddScoped<IPipelineStore>(_ => store.Store),
             async channel =>
             {
                 PushSourceReply reply = await PushAsync(
@@ -80,12 +80,12 @@ public sealed class SourceIngressGrpcServiceTests
     [Fact]
     public async Task PushSource_SameSourceKeyTwice_SecondIsDuplicateAndQueueNotGrown()
     {
-        var store = new FakePipelineStore();
+        var store = new TestPipelineStore();
         var registry = TestRegistry.With(Tenant(TenantA));
 
         await RunAsync(
             registry,
-            services => services.AddScoped<IPipelineStore>(_ => store),
+            services => services.AddScoped<IPipelineStore>(_ => store.Store),
             async channel =>
             {
                 PushSourceReply first = await PushAsync(channel, Request("Сообщение канала", externalId: "42"), TenantA, ValidToken);
@@ -104,12 +104,12 @@ public sealed class SourceIngressGrpcServiceTests
     [Fact]
     public async Task PushSource_EmptyText_NotAcceptedAndQueueNotGrown()
     {
-        var store = new FakePipelineStore();
+        var store = new TestPipelineStore();
         var registry = TestRegistry.With(Tenant(TenantA));
 
         await RunAsync(
             registry,
-            services => services.AddScoped<IPipelineStore>(_ => store),
+            services => services.AddScoped<IPipelineStore>(_ => store.Store),
             async channel =>
             {
                 PushSourceReply reply = await PushAsync(channel, Request("   ", externalId: "9"), TenantA, ValidToken);
@@ -126,12 +126,12 @@ public sealed class SourceIngressGrpcServiceTests
     [Fact]
     public async Task PushSource_UnknownTenant_NotAcceptedWithoutRpcError()
     {
-        var store = new FakePipelineStore();
+        var store = new TestPipelineStore();
         var registry = TestRegistry.With(Tenant(TenantA));
 
         await RunAsync(
             registry,
-            services => services.AddScoped<IPipelineStore>(_ => store),
+            services => services.AddScoped<IPipelineStore>(_ => store.Store),
             async channel =>
             {
                 PushSourceReply reply = await PushAsync(channel, Request("Сообщение чужого тенанта", externalId: "5"), Guid.NewGuid(), ValidToken);
@@ -150,10 +150,10 @@ public sealed class SourceIngressGrpcServiceTests
     [Fact]
     public async Task PushSource_WithoutToken_IsUnauthenticated()
     {
-        var store = new FakePipelineStore();
+        var store = new TestPipelineStore();
         var registry = TestRegistry.With(Tenant(TenantA));
 
-        await RunAsync(registry, services => services.AddScoped<IPipelineStore>(_ => store), async channel =>
+        await RunAsync(registry, services => services.AddScoped<IPipelineStore>(_ => store.Store), async channel =>
         {
             RpcException exception = await Assert.ThrowsAsync<RpcException>(
                 () => PushAsync(channel, Request("текст", externalId: "1"), TenantA, null));
@@ -167,10 +167,10 @@ public sealed class SourceIngressGrpcServiceTests
     [Fact]
     public async Task PushSource_WithWrongToken_IsUnauthenticated()
     {
-        var store = new FakePipelineStore();
+        var store = new TestPipelineStore();
         var registry = TestRegistry.With(Tenant(TenantA));
 
-        await RunAsync(registry, services => services.AddScoped<IPipelineStore>(_ => store), async channel =>
+        await RunAsync(registry, services => services.AddScoped<IPipelineStore>(_ => store.Store), async channel =>
         {
             RpcException exception = await Assert.ThrowsAsync<RpcException>(
                 () => PushAsync(channel, Request("текст", externalId: "2"), TenantA, "wrong-token"));
@@ -184,7 +184,7 @@ public sealed class SourceIngressGrpcServiceTests
     [Fact]
     public async Task PushSource_UnsetEnvToken_FailsClosed()
     {
-        var store = new FakePipelineStore();
+        var store = new TestPipelineStore();
         var registry = TestRegistry.With(Tenant(TenantA));
 
         await TelegramIngressTestHost.RunAsync(
@@ -192,7 +192,7 @@ public sealed class SourceIngressGrpcServiceTests
             configureServices: services =>
             {
                 services.AddSingleton<ITenantRepository>(registry);
-                services.AddScoped<IPipelineStore>(_ => store);
+                services.AddScoped<IPipelineStore>(_ => store.Store);
             },
             scenario: async channel =>
             {
@@ -208,10 +208,10 @@ public sealed class SourceIngressGrpcServiceTests
     [Fact]
     public async Task PushSource_WithoutTenantIdMetadata_IsUnauthenticated()
     {
-        var store = new FakePipelineStore();
+        var store = new TestPipelineStore();
         var registry = TestRegistry.With(Tenant(TenantA));
 
-        await RunAsync(registry, services => services.AddScoped<IPipelineStore>(_ => store), async channel =>
+        await RunAsync(registry, services => services.AddScoped<IPipelineStore>(_ => store.Store), async channel =>
         {
             RpcException exception = await Assert.ThrowsAsync<RpcException>(
                 () => PushAsync(channel, Request("текст", externalId: "4"), null, ValidToken));

src/core/tests/Deal.Tests.Unit/Support/TestPipelineStore.cs

@@ -34,6 +34,7 @@ public sealed class TestPipelineStore
 
     private readonly bool _claimRace;
     private readonly bool _claimResult;
+    private readonly bool _throwOnList;
 
     /// <summary>
     /// Сколько раз вызван ClaimAsync
@@ -50,17 +51,31 @@ public sealed class TestPipelineStore
     /// </summary>
     /// <param name="claimRace">Сценарий claim-гонки: ExistsAsync=false, ClaimAsync возвращает claimResult.</param>
     /// <param name="claimResult">Результат ClaimAsync в сценарии гонки (false — хэш уже заявлен другим проходом).</param>
-    public TestPipelineStore(bool claimRace = false, bool claimResult = false)
+    /// <param name="throwOnList">Сценарий сбоя чтения очереди: ListAsync бросает (имитация недоступной схемы/БД).</param>
+    public TestPipelineStore(
+        bool claimRace = false,
+        bool claimResult = false,
+        bool throwOnList = false)
     {
         _claimRace = claimRace;
         _claimResult = claimResult;
+        _throwOnList = throwOnList;
         Store = Substitute.For<IPipelineStore>();
         Store.ExistsDuplicateAsync(Arg.Any<SourceRef>(), Arg.Any<CancellationToken>())
             .Returns(ci => ExistsDuplicate(ci.Arg<SourceRef>()));
         Store.When(s => s.AddAsync(Arg.Any<QueueItemDto>(), Arg.Any<CancellationToken>()))
             .Do(ci => _queue.Add(ci.Arg<QueueItemDto>()));
+        if (_throwOnList)
+        {
+            Store.ListAsync(Arg.Any<string?>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
+                .Returns<Task<IReadOnlyList<QueueItemDto>>>(_ =>
+                    throw new InvalidOperationException("Тестовый сбой чтения очереди (ListAsync)."));
+        }
+        else
+        {
             Store.ListAsync(Arg.Any<string?>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
                 .Returns(ci => List(ci.ArgAt<string?>(0), ci.ArgAt<int>(1)));
+        }
         Store.CountByStatusAsync(Arg.Any<string>(), Arg.Any<CancellationToken>())
             .Returns(ci => CountByStatus(ci.ArgAt<string>(0)));
         Store.When(s => s.SetStatusAsync(Arg.Any<string>(), Arg.Any<string>(), Arg.Any<CancellationToken>()))
@@ -73,7 +88,7 @@ public sealed class TestPipelineStore
             .Returns(ci => ListPage(ci.ArgAt<int>(0), ci.ArgAt<int>(1)));
         Store.SearchAsync(Arg.Any<string>(), Arg.Any<int>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
             .Returns(ci => Search(ci.ArgAt<string>(0), ci.ArgAt<int>(2)));
-        Store.CountAsync(Arg.Any<CancellationToken>()).Returns(_rejected.Count);
+        Store.CountAsync(Arg.Any<CancellationToken>()).Returns(_ => _rejected.Count);
         Store.GetAsync(Arg.Any<string>(), Arg.Any<CancellationToken>())
             .Returns(ci => _rejected.FirstOrDefault(item => item.Id == ci.ArgAt<string>(0)));
         Store.When(s => s.DeleteAsync(Arg.Any<string>(), Arg.Any<CancellationToken>()))