stream/options_test.go

Rama main · réplica de lectura del repositorio original.

// Copyright 2023 Harness, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package stream

import (
	"testing"
	"time"
)

func TestWithConcurrency(t *testing.T) {
	t.Run("valid concurrency", func(t *testing.T) {
		config := &ConsumerConfig{}
		opt := WithConcurrency(10)
		opt.apply(config)

		if config.Concurrency != 10 {
			t.Errorf("expected concurrency 10, got %d", config.Concurrency)
		}
	})

	t.Run("minimum valid concurrency", func(t *testing.T) {
		config := &ConsumerConfig{}
		opt := WithConcurrency(1)
		opt.apply(config)

		if config.Concurrency != 1 {
			t.Errorf("expected concurrency 1, got %d", config.Concurrency)
		}
	})

	t.Run("maximum valid concurrency", func(t *testing.T) {
		config := &ConsumerConfig{}
		opt := WithConcurrency(MaxConcurrency)
		opt.apply(config)

		if config.Concurrency != MaxConcurrency {
			t.Errorf("expected concurrency %d, got %d", MaxConcurrency, config.Concurrency)
		}
	})

	t.Run("zero concurrency panics", func(t *testing.T) {
		defer func() {
			if r := recover(); r == nil {
				t.Error("expected panic for zero concurrency")
			}
		}()
		WithConcurrency(0)
	})

	t.Run("negative concurrency panics", func(t *testing.T) {
		defer func() {
			if r := recover(); r == nil {
				t.Error("expected panic for negative concurrency")
			}
		}()
		WithConcurrency(-1)
	})

	t.Run("concurrency above max panics", func(t *testing.T) {
		defer func() {
			if r := recover(); r == nil {
				t.Error("expected panic for concurrency above max")
			}
		}()
		WithConcurrency(MaxConcurrency + 1)
	})
}

func TestWithMaxRetries(t *testing.T) {
	t.Run("valid max retries", func(t *testing.T) {
		config := &HandlerConfig{}
		opt := WithMaxRetries(5)
		opt.apply(config)

		if config.maxRetries != 5 {
			t.Errorf("expected maxRetries 5, got %d", config.maxRetries)
		}
	})

	t.Run("zero max retries", func(t *testing.T) {
		config := &HandlerConfig{}
		opt := WithMaxRetries(0)
		opt.apply(config)

		if config.maxRetries != 0 {
			t.Errorf("expected maxRetries 0, got %d", config.maxRetries)
		}
	})

	t.Run("maximum valid retries", func(t *testing.T) {
		config := &HandlerConfig{}
		opt := WithMaxRetries(MaxMaxRetries)
		opt.apply(config)

		if config.maxRetries != MaxMaxRetries {
			t.Errorf("expected maxRetries %d, got %d", MaxMaxRetries, config.maxRetries)
		}
	})

	t.Run("negative max retries panics", func(t *testing.T) {
		defer func() {
			if r := recover(); r == nil {
				t.Error("expected panic for negative max retries")
			}
		}()
		WithMaxRetries(-1)
	})

	t.Run("max retries above max panics", func(t *testing.T) {
		defer func() {
			if r := recover(); r == nil {
				t.Error("expected panic for max retries above max")
			}
		}()
		WithMaxRetries(MaxMaxRetries + 1)
	})
}

func TestWithIdleTimeout(t *testing.T) {
	t.Run("valid idle timeout", func(t *testing.T) {
		config := &HandlerConfig{}
		timeout := 10 * time.Second
		opt := WithIdleTimeout(timeout)
		opt.apply(config)

		if config.idleTimeout != timeout {
			t.Errorf("expected idleTimeout %v, got %v", timeout, config.idleTimeout)
		}
	})

	t.Run("minimum valid idle timeout", func(t *testing.T) {
		config := &HandlerConfig{}
		opt := WithIdleTimeout(MinIdleTimeout)
		opt.apply(config)

		if config.idleTimeout != MinIdleTimeout {
			t.Errorf("expected idleTimeout %v, got %v", MinIdleTimeout, config.idleTimeout)
		}
	})

	t.Run("idle timeout below minimum panics", func(t *testing.T) {
		defer func() {
			if r := recover(); r == nil {
				t.Error("expected panic for idle timeout below minimum")
			}
		}()
		WithIdleTimeout(MinIdleTimeout - 1*time.Second)
	})

	t.Run("zero idle timeout panics", func(t *testing.T) {
		defer func() {
			if r := recover(); r == nil {
				t.Error("expected panic for zero idle timeout")
			}
		}()
		WithIdleTimeout(0)
	})
}

func TestWithHandlerOptions(t *testing.T) {
	t.Run("applies single handler option", func(t *testing.T) {
		config := &ConsumerConfig{}
		opt := WithHandlerOptions(WithMaxRetries(10))
		opt.apply(config)

		if config.DefaultHandlerConfig.maxRetries != 10 {
			t.Errorf("expected maxRetries 10, got %d", config.DefaultHandlerConfig.maxRetries)
		}
	})

	t.Run("applies multiple handler options", func(t *testing.T) {
		config := &ConsumerConfig{}
		timeout := 10 * time.Second
		opt := WithHandlerOptions(
			WithMaxRetries(5),
			WithIdleTimeout(timeout),
		)
		opt.apply(config)

		if config.DefaultHandlerConfig.maxRetries != 5 {
			t.Errorf("expected maxRetries 5, got %d", config.DefaultHandlerConfig.maxRetries)
		}
		if config.DefaultHandlerConfig.idleTimeout != timeout {
			t.Errorf("expected idleTimeout %v, got %v", timeout, config.DefaultHandlerConfig.idleTimeout)
		}
	})

	t.Run("applies no handler options", func(t *testing.T) {
		config := &ConsumerConfig{}
		opt := WithHandlerOptions()
		opt.apply(config)

		// Should not panic and should leave config unchanged
		if config.DefaultHandlerConfig.maxRetries != 0 {
			t.Errorf("expected maxRetries 0, got %d", config.DefaultHandlerConfig.maxRetries)
		}
	})
}