zszq-trs/UnitTestProject/Modules/EodModule/TrsContractKafkaPushServiceTest.cs

using YLErp.Abstract;
using YLErp.Helpers;
using YLErp.Modules.EodModule;

namespace YLErp.Modules.EodModuleTests
{
    [TestClass]
    public class TrsContractKafkaPushServiceTest
    {
        [TestMethod]
        public void Push_空日快照_发送一条空消息并使用业务日期作为Key()
        {
            var valueDate = new DateTime(2026, 8, 24);
            var producer = new RecordingKafkaProducer();
            var service = CreateService(producer, valueDate);

            service.Push(valueDate);

            Assert.AreEqual(1, producer.Messages.Count);
            Assert.AreEqual("onederiv.trs.contract.v1", producer.Messages[0].Topic);
            Assert.AreEqual("2026-08-24", producer.Messages[0].Key);
            var payload = JsonHelper.Deserialize<TrsContractSnapshot>(producer.Messages[0].Message);
            Assert.AreEqual("2026-08-24", payload.ValueDate);
            Assert.AreEqual(0, payload.ContractCount);
            Assert.AreEqual(0, payload.Contracts.Count);
        }

        [TestMethod]
        public void Push_区间内每天分别调用_每个日期各发送一条快照()
        {
            var valueDates = new[]
            {
                new DateTime(2026, 8, 20),
                new DateTime(2026, 8, 21),
                new DateTime(2026, 8, 24)
            };
            var producer = new RecordingKafkaProducer();
            var service = new TestableTrsContractKafkaPushService(producer, valueDates.ToDictionary(x => x, CreateEmptySnapshot));

            foreach (var valueDate in valueDates)
            {
                service.Push(valueDate);
            }

            CollectionAssert.AreEqual(
                new[] { "2026-08-20", "2026-08-21", "2026-08-24" },
                producer.Messages.Select(x => x.Key).ToArray());
            CollectionAssert.AreEqual(
                new[] { "2026-08-20", "2026-08-21", "2026-08-24" },
                producer.Messages.Select(x => JsonHelper.Deserialize<TrsContractSnapshot>(x.Message).ValueDate).ToArray());
        }

        [TestMethod]
        public void Push_首次失败后成功_停止重试且不记录最终失败()
        {
            var valueDate = new DateTime(2026, 8, 24);
            var producer = new RecordingKafkaProducer { FailuresBeforeSuccess = 1 };
            var service = CreateService(producer, valueDate);

            service.Push(valueDate);

            Assert.AreEqual(2, producer.AttemptCount);
            Assert.AreEqual(1, producer.Messages.Count);
            Assert.AreEqual(0, service.FailureRecords.Count);
        }

        [TestMethod]
        public void Push_连续失败三次_记录最终失败和三次尝试()
        {
            var valueDate = new DateTime(2026, 8, 24);
            var producer = new RecordingKafkaProducer { FailuresBeforeSuccess = int.MaxValue };
            var service = CreateService(producer, valueDate);

            service.Push(valueDate);

            Assert.AreEqual(3, producer.AttemptCount);
            Assert.AreEqual(0, producer.Messages.Count);
            Assert.AreEqual(1, service.FailureRecords.Count);
            Assert.AreEqual(valueDate, service.FailureRecords[0].ValueDate);
            Assert.AreEqual(3, service.FailureRecords[0].RetryCount);
            Assert.IsInstanceOfType(service.FailureRecords[0].Exception, typeof(InvalidOperationException));
        }

        [TestMethod]
        public void BuildContract_字段使用日终快照和约定来源()
        {
            var valueDate = new DateTime(2026, 8, 24);
            var eodSwap = new eod_swap
            {
                id = 10,
                ValueDate = valueDate,
                SwapTradeId = 7,
                SwapTradeNo = "TRS-001",
                BookId = 3,
                ClientId = 8,
                NotionalValue = 1000000m,
                dv01 = 12.34m,
                InitMarginGain = 100m,
                InitMarginLoss = 0m
            };
            var trade = new trade
            {
                id = 7,
                UnderlyingCode = "600000.SH",
                UnderlyingAssetName = "浦发银行",
                UnderlyingInstrumentType = "Stock",
                StartDate = new DateTime(2026, 8, 1),
                ExerciseDate = new DateTime(2027, 8, 1)
            };
            var positions = new List<eod_swap_position>
            {
                new() { SwapTradeId = 7, PositionId = 101, UnderlyingCode = "600000.SH", PositionType = 1 },
                new() { SwapTradeId = 7, PositionId = 102, InterestMode = (int)InterestModeEnum.固定值, InterestRateDefault = 0.0123m, InterestDirection = 2 }
            };
            var swapPositions = new Dictionary<long, swap_position>
            {
                [101] = new() { id = 101, category_tag = "互换利率" },
                [102] = new() { id = 102, category_tag = "互换利率" }
            };

            var item = TrsContractKafkaPushService.BuildContract(
                eodSwap,
                new Dictionary<int, trade> { [7] = trade },
                positions,
                swapPositions);

            Assert.AreEqual("2026-08-24", item.TradeDate);
            Assert.AreEqual(3, item.BookId);
            Assert.AreEqual("TRS-001", item.SwapTradeNo);
            Assert.AreEqual(8, item.ClientId);
            Assert.AreEqual("600000.SH", item.UnderlyingCode);
            Assert.AreEqual("浦发银行", item.UnderlyingName);
            Assert.AreEqual("Stock", item.UnderlyingInstrumentType);
            Assert.AreEqual(1000000m, item.NotionalValue);
            Assert.AreEqual("2026-08-01", item.StartDate);
            Assert.AreEqual("2027-08-01", item.MaturityDate);
            Assert.AreEqual(12.34m, item.Dv01);
            Assert.AreEqual(0.0123m, item.FixedRate);
            Assert.AreEqual(2, item.InterestDirection);
            Assert.AreEqual(1, item.FloatingDirection);
            Assert.AreEqual(100m, item.InitMarginGain);
            Assert.AreEqual(0m, item.InitMarginLoss);
        }

        private static TestableTrsContractKafkaPushService CreateService(RecordingKafkaProducer producer, DateTime valueDate)
        {
            return new TestableTrsContractKafkaPushService(
                producer,
                new Dictionary<DateTime, TrsContractSnapshot> { [valueDate] = CreateEmptySnapshot(valueDate) });
        }

        private static TrsContractSnapshot CreateEmptySnapshot(DateTime valueDate)
        {
            return new TrsContractSnapshot
            {
                SchemaVersion = "v1",
                ValueDate = valueDate.ToString("yyyy-MM-dd"),
                PushTime = "2026-08-24 12:00:00",
                ContractCount = 0,
                Contracts = new List<TrsContractSnapshotItem>()
            };
        }

        private sealed class TestableTrsContractKafkaPushService : TrsContractKafkaPushService
        {
            private readonly IReadOnlyDictionary<DateTime, TrsContractSnapshot> _snapshots;

            public List<(DateTime ValueDate, int RetryCount, Exception Exception)> FailureRecords { get; } = new();

            public TestableTrsContractKafkaPushService(IKafkaProduce producer, IReadOnlyDictionary<DateTime, TrsContractSnapshot> snapshots)
                : base(new YLContext(), producer, "onederiv.trs.contract.v1")
            {
                _snapshots = snapshots;
            }

            protected override TrsContractSnapshot BuildSnapshot(DateTime valueDate)
            {
                return _snapshots[valueDate];
            }

            protected override void RecordFailures(DateTime valueDate, int retryCount, Exception exception)
            {
                FailureRecords.Add((valueDate, retryCount, exception));
            }
        }

        private sealed class RecordingKafkaProducer : IKafkaProduce
        {
            public int FailuresBeforeSuccess { get; set; }
            public int AttemptCount { get; private set; }
            public List<(string Topic, string Key, string Message)> Messages { get; } = new();

            public void Produce(string topic, string message)
            {
                throw new NotSupportedException();
            }

            public void Produce(string topic, string key, string message)
            {
                AttemptCount++;
                if (AttemptCount <= FailuresBeforeSuccess)
                {
                    throw new InvalidOperationException("Kafka unavailable");
                }

                Messages.Add((topic, key, message));
            }
        }
    }
}