natsnet/dotnet/tests/ZB.MOM.NatsNet.Server.Tests/JetStream/ConsumerPoliciesTests.cs

using System.Text.Json;
using Shouldly;
using ZB.MOM.NatsNet.Server;

namespace ZB.MOM.NatsNet.Server.Tests.JetStream;

public sealed class ConsumerPoliciesTests
{
    [Fact]
    public void ConsumerAction_StringAndJsonParity_ShouldMatchGo()
    {
        ConsumerAction.CreateOrUpdate.String().ShouldBe("\"create_or_update\"");
        ConsumerAction.Create.String().ShouldBe("\"create\"");
        ConsumerAction.Update.String().ShouldBe("\"update\"");

        JsonSerializer.Serialize(ConsumerAction.Create).ShouldBe("\"create\"");
        JsonSerializer.Deserialize<ConsumerAction>("\"update\"").ShouldBe(ConsumerAction.Update);
        Should.Throw<JsonException>(() => JsonSerializer.Deserialize<ConsumerAction>("\"bogus\""));
    }

    [Fact]
    public void PriorityPolicy_StringAndJsonParity_ShouldMatchGo()
    {
        PriorityPolicy.PriorityNone.String().ShouldBe("\"none\"");
        PriorityPolicy.PriorityOverflow.String().ShouldBe("\"overflow\"");
        PriorityPolicy.PriorityPinnedClient.String().ShouldBe("\"pinned_client\"");
        PriorityPolicy.PriorityPrioritized.String().ShouldBe("\"prioritized\"");

        JsonSerializer.Serialize(PriorityPolicy.PriorityPinnedClient).ShouldBe("\"pinned_client\"");
        JsonSerializer.Deserialize<PriorityPolicy>("\"prioritized\"").ShouldBe(PriorityPolicy.PriorityPrioritized);
        Should.Throw<JsonException>(() => JsonSerializer.Deserialize<PriorityPolicy>("\"none-ish\""));
    }

    [Fact]
    public void ConsumerPolicies_StringParity_ShouldMatchGo()
    {
        DeliverPolicy.DeliverByStartSequence.String().ShouldBe("by_start_sequence");
        AckPolicy.AckExplicit.String().ShouldBe("explicit");
        ReplayPolicy.ReplayInstant.String().ShouldBe("instant");
    }

    [Fact]
    public void SubjectTokens_Subjects_RemovesEmptyValues()
    {
        var subjects = SubjectTokens.Subjects(new[] { "foo.*", string.Empty, " ", "bar.>" });
        subjects.ShouldBe(["foo.*", "bar.>"]);
    }

    [Fact]
    public void SetConsumerConfigDefaults_InvalidNegativesInPedanticMode_ReturnsError()
    {
        var cfg = new ConsumerConfig { MaxDeliver = -2 };
        var streamCfg = new StreamConfig { Name = "ORDERS", Replicas = 3 };

        var err = NatsConsumer.SetConsumerConfigDefaults(cfg, streamCfg, null, pedantic: true);

        err.ShouldNotBeNull();
        cfg.MaxDeliver.ShouldBe(-2);
    }

    [Fact]
    public void SetConsumerConfigDefaults_AppliesGoDefaults_ShouldPopulateExpectedValues()
    {
        var cfg = new ConsumerConfig
        {
            Durable = "D",
            MaxDeliver = 0,
            AckPolicy = AckPolicy.AckExplicit,
            Replicas = 0,
        };
        var streamCfg = new StreamConfig { Name = "ORDERS", Replicas = 3 };
        var limits = new JetStreamAccountLimits { MaxAckPending = 2500 };

        var err = NatsConsumer.SetConsumerConfigDefaults(cfg, streamCfg, limits, pedantic: false);

        err.ShouldBeNull();
        cfg.MaxDeliver.ShouldBe(-1);
        cfg.AckWait.ShouldBe(TimeSpan.FromSeconds(30));
        cfg.MaxAckPending.ShouldBe(2500);
        cfg.Replicas.ShouldBe(3);
    }

    [Fact]
    public void CheckConsumerCfg_DurableNameMismatch_ReturnsError()
    {
        var cfg = new ConsumerConfig { Name = "A", Durable = "B", AckPolicy = AckPolicy.AckExplicit };
        var streamCfg = new StreamConfig { Name = "ORDERS", Replicas = 1 };

        var err = NatsConsumer.CheckConsumerCfg(cfg, streamCfg, null, isRecovering: false);

        err.ShouldNotBeNull();
        err.ErrCode.ShouldBe(JsApiErrors.ConsumerCreateDurableAndNameMismatch.ErrCode);
    }

    [Fact]
    public void CheckConsumerCfg_OverlappingFilterSubjects_ReturnsError()
    {
        var cfg = new ConsumerConfig
        {
            Durable = "D",
            AckPolicy = AckPolicy.AckExplicit,
            FilterSubjects = ["orders.*", "orders.created"],
        };
        var streamCfg = new StreamConfig { Name = "ORDERS", Replicas = 1 };

        var err = NatsConsumer.CheckConsumerCfg(cfg, streamCfg, null, isRecovering: false);

        err.ShouldNotBeNull();
        err.ErrCode.ShouldBe(JsApiErrors.ConsumerOverlappingSubjectFilters.ErrCode);
    }

    [Fact]
    public void CheckConsumerCfg_WithValidPullConfig_ReturnsNull()
    {
        var cfg = new ConsumerConfig
        {
            Durable = "D",
            AckPolicy = AckPolicy.AckExplicit,
            FilterSubject = "orders.created",
        };
        var streamCfg = new StreamConfig
        {
            Name = "ORDERS",
            Replicas = 1,
            Retention = RetentionPolicy.LimitsPolicy,
            Subjects = ["orders.>"],
        };

        var err = NatsConsumer.CheckConsumerCfg(cfg, streamCfg, null, isRecovering: false);

        err.ShouldBeNull();
    }
}