158 lines
6.3 KiB
C#
158 lines
6.3 KiB
C#
using FluentAssertions;
|
|
using Fake.Application;
|
|
using Fake.EventBus.Distributed;
|
|
using Fake.ObjectStorage;
|
|
using Microsoft.Extensions.Logging.Abstractions;
|
|
using Microsoft.Extensions.Options;
|
|
using Moq;
|
|
using Oms.Application.Dtos.Orders;
|
|
using Oms.Application.Messaging;
|
|
using Oms.Application.Services.Orders.Batch;
|
|
using Oms.Domain.ChannelsAggregate.Repositories;
|
|
using Oms.Domain.OrdersAggregate.Consts;
|
|
using Oms.Domain.OrdersAggregate.Entities;
|
|
using Oms.Domain.OrdersAggregate.Repositories;
|
|
using Oms.Domain.SystemAggregate.Repositories;
|
|
|
|
namespace Oms.Tests;
|
|
|
|
public class OrderBatchJobCoordinatorGovernanceTests
|
|
{
|
|
private readonly Mock<IOrderBatchJobRepository> _batchJobRepository = new(MockBehavior.Strict);
|
|
private readonly Mock<IOrderRepository> _orderRepository = new(MockBehavior.Loose);
|
|
private readonly Mock<ICountryRepository> _countryRepository = new(MockBehavior.Loose);
|
|
private readonly Mock<IServiceChannelRepository> _serviceChannelRepository = new(MockBehavior.Loose);
|
|
private readonly Mock<IObjectStorage> _fileStorage = new(MockBehavior.Loose);
|
|
private readonly Mock<IDistributedEventBus> _eventBus = new(MockBehavior.Strict);
|
|
private readonly OrderBatchJobCoordinator _coordinator;
|
|
|
|
public OrderBatchJobCoordinatorGovernanceTests()
|
|
{
|
|
var batchOptions = Options.Create(new OrderBatchRabbitMqOptions { Enabled = true });
|
|
var actionExecutor = new Mock<IOrderBatchActionExecutor>(MockBehavior.Loose);
|
|
var processor = new OrderBatchJobProcessor(
|
|
_batchJobRepository.Object,
|
|
_orderRepository.Object,
|
|
actionExecutor.Object,
|
|
batchOptions,
|
|
NullLogger<OrderBatchJobProcessor>.Instance);
|
|
_coordinator = new OrderBatchJobCoordinator(
|
|
_batchJobRepository.Object,
|
|
_orderRepository.Object,
|
|
_countryRepository.Object,
|
|
_serviceChannelRepository.Object,
|
|
_fileStorage.Object,
|
|
_eventBus.Object,
|
|
processor,
|
|
batchOptions,
|
|
NullLogger<OrderBatchJobCoordinator>.Instance);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RetryFailedAsync_ShouldResetFailedItemsAndPublish()
|
|
{
|
|
var job = CreateJob();
|
|
var failedItem = CreateItem(1, OrderBatchJobItemStatusCodes.Failed, "网络错误");
|
|
var successItem = CreateItem(2, OrderBatchJobItemStatusCodes.Success);
|
|
|
|
_batchJobRepository
|
|
.Setup(x => x.GetByIdAsync(job.Id, It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(job);
|
|
_batchJobRepository
|
|
.Setup(x => x.GetItemsByJobIdAsync(job.Id, It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync([failedItem, successItem]);
|
|
_batchJobRepository
|
|
.Setup(x => x.UpdateItemAsync(failedItem, It.IsAny<CancellationToken>()))
|
|
.Returns(Task.CompletedTask);
|
|
_batchJobRepository
|
|
.Setup(x => x.UpdateJobAsync(job, It.IsAny<CancellationToken>()))
|
|
.Returns(Task.CompletedTask);
|
|
_eventBus
|
|
.Setup(x => x.PublishAsync(It.IsAny<OrderBatchJobIntegrationEvent>(), It.IsAny<CancellationToken>()))
|
|
.Returns(Task.CompletedTask);
|
|
|
|
var output = await _coordinator.RetryFailedAsync(job.Id, null, 99);
|
|
|
|
output.JobId.Should().Be(job.Id);
|
|
output.RetriedCount.Should().Be(1);
|
|
failedItem.Status.Should().Be(OrderBatchJobItemStatusCodes.Pending);
|
|
failedItem.RetryCount.Should().Be(0);
|
|
failedItem.FailReason.Should().BeNull();
|
|
job.Status.Should().Be(OrderBatchJobStatusCodes.Running);
|
|
_eventBus.Verify(
|
|
x => x.PublishAsync(
|
|
It.Is<OrderBatchJobIntegrationEvent>(m => m.JobItemId == failedItem.Id),
|
|
It.IsAny<CancellationToken>()),
|
|
Times.Once);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task CancelPendingAsync_ShouldMarkPendingAsCancelled()
|
|
{
|
|
var job = CreateJob();
|
|
var pendingItem = CreateItem(3, OrderBatchJobItemStatusCodes.Pending);
|
|
var successItem = CreateItem(4, OrderBatchJobItemStatusCodes.Success);
|
|
|
|
_batchJobRepository
|
|
.Setup(x => x.GetByIdAsync(job.Id, It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(job);
|
|
_batchJobRepository
|
|
.Setup(x => x.GetItemsByJobIdAsync(job.Id, It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync([pendingItem, successItem]);
|
|
_batchJobRepository
|
|
.Setup(x => x.UpdateItemAsync(pendingItem, It.IsAny<CancellationToken>()))
|
|
.Returns(Task.CompletedTask);
|
|
_batchJobRepository
|
|
.Setup(x => x.UpdateJobAsync(job, It.IsAny<CancellationToken>()))
|
|
.Returns(Task.CompletedTask);
|
|
|
|
var output = await _coordinator.CancelPendingAsync(job.Id, 99);
|
|
|
|
output.JobId.Should().Be(job.Id);
|
|
output.CancelledCount.Should().Be(1);
|
|
pendingItem.Status.Should().Be(OrderBatchJobItemStatusCodes.Cancelled);
|
|
pendingItem.FailReason.Should().Be("任务已取消");
|
|
job.Status.Should().Be(OrderBatchJobStatusCodes.Completed);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RetryFailedAsync_WhenNoFailedItems_ShouldThrow()
|
|
{
|
|
var job = CreateJob();
|
|
_batchJobRepository
|
|
.Setup(x => x.GetByIdAsync(job.Id, It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync(job);
|
|
_batchJobRepository
|
|
.Setup(x => x.GetItemsByJobIdAsync(job.Id, It.IsAny<CancellationToken>()))
|
|
.ReturnsAsync([CreateItem(5, OrderBatchJobItemStatusCodes.Success)]);
|
|
|
|
var act = () => _coordinator.RetryFailedAsync(job.Id, null, 99);
|
|
|
|
await act.Should().ThrowAsync<BusinessException>()
|
|
.WithMessage("没有可重试的失败子项");
|
|
}
|
|
|
|
private static OrderBatchJob CreateJob() =>
|
|
new()
|
|
{
|
|
Id = 501,
|
|
Action = OrderBatchActionCodes.Pay,
|
|
Status = OrderBatchJobStatusCodes.Completed,
|
|
Total = 2,
|
|
SuccessCount = 1,
|
|
FailCount = 1,
|
|
};
|
|
|
|
private static OrderBatchJobItem CreateItem(int id, string status, string? failReason = null) =>
|
|
new()
|
|
{
|
|
Id = id,
|
|
JobId = 501,
|
|
OrderId = 9000 + id,
|
|
Action = OrderBatchActionCodes.Pay,
|
|
Status = status,
|
|
FailReason = failReason,
|
|
MessageId = $"501:{id}:pay",
|
|
};
|
|
}
|