testutil

Documentation

Index

• func CheckReceived(t *testing.T, consumer *kafka.Consumer, sent []model.ProduceMessage)
• func CopyFromContainer(ctx context.Context, t *testing.T, container testcontainers.Container, ...) *os.File
• func FormatCfg(cfg string) string
• func GetReceived(t *testing.T, consumer *kafka.Consumer, sent []model.ProduceMessage, ...) []kafka.Message
• func NewConsumer(ctx context.Context, t *testing.T, topic, port string) *kafka.Consumer
• func NewKafkaContainer(ctx context.Context, name, port, network string, ...) (testcontainers.Container, error)
• func NewKafkaSASLPlainContainer(ctx context.Context, name, port, network string) (testcontainers.Container, error)
• func NewKafkaSSLContainer(ctx context.Context, name, port, network string) (testcontainers.Container, error)
• func NewKrpContainer(ctx context.Context, network string, cfg string, ...) (testcontainers.Container, error)
• func NewOtelCollectorContainer(ctx context.Context, network string) (testcontainers.Container, error)
• func NewPrometheusContainer(ctx context.Context, network string) (testcontainers.Container, error)
• func NewSchemaRegistryContainer(ctx context.Context, network string) (testcontainers.Container, error)
• func ProduceAsync(ctx context.Context, t *testing.T, krp testcontainers.Container, path string, ...)
• func ProduceError(ctx context.Context, t *testing.T, krp testcontainers.Container, path string, ...) (model.ProduceErrorResponse, int)
• func ProduceSync(ctx context.Context, t *testing.T, krp testcontainers.Container, path string, ...)
• func ProjectRootDir() string
• func Ptr[T any](value T) *T
• type ConsumeOption
  • func WithReadTimeout(readTimeout time.Duration) ConsumeOption
  • func WithVerifyReceived(verifyReceived bool) ConsumeOption
• type ProduceOption
  • func WithHeaders(headers map[string]string) ProduceOption
  • func WithSuccess(success bool) ProduceOption
• type StdoutLogConsumer
  • func (lc *StdoutLogConsumer) Accept(l testcontainers.Log)

Functions

func CheckReceived

func CheckReceived(t *testing.T, consumer *kafka.Consumer, sent []model.ProduceMessage)

func CopyFromContainer

func CopyFromContainer(ctx context.Context, t *testing.T, container testcontainers.Container, path, name string) *os.File

func FormatCfg

func FormatCfg(cfg string) string

func GetReceived

func GetReceived(t *testing.T, consumer *kafka.Consumer, sent []model.ProduceMessage,
	options ...ConsumeOption) []kafka.Message

func NewConsumer

func NewConsumer(ctx context.Context, t *testing.T, topic, port string) *kafka.Consumer

func NewKafkaContainer

func NewKafkaContainer(ctx context.Context, name, port, network string, topics ...kafka.TopicSpecification) (testcontainers.Container, error)

func NewKafkaSASLPlainContainer

func NewKafkaSASLPlainContainer(ctx context.Context, name, port, network string) (testcontainers.Container, error)

func NewKafkaSSLContainer

func NewKafkaSSLContainer(ctx context.Context, name, port, network string) (testcontainers.Container, error)

func NewKrpContainer

func NewKrpContainer(ctx context.Context, network string, cfg string, files ...testcontainers.ContainerFile) (testcontainers.Container, error)

func NewOtelCollectorContainer

func NewOtelCollectorContainer(ctx context.Context, network string) (testcontainers.Container, error)

func NewPrometheusContainer

func NewPrometheusContainer(ctx context.Context, network string) (testcontainers.Container, error)

func NewSchemaRegistryContainer

func NewSchemaRegistryContainer(ctx context.Context, network string) (testcontainers.Container, error)

func ProduceAsync

func ProduceAsync(ctx context.Context, t *testing.T, krp testcontainers.Container,
	path string, req model.ProduceRequest, options ...ProduceOption)

func ProduceError

func ProduceError(ctx context.Context, t *testing.T, krp testcontainers.Container,
	path string, req any, options ...ProduceOption) (model.ProduceErrorResponse, int)

func ProduceSync

func ProduceSync(ctx context.Context, t *testing.T, krp testcontainers.Container,
	path string, req model.ProduceRequest, options ...ProduceOption)

func ProjectRootDir

func ProjectRootDir() string

func Ptr

func Ptr[T any](value T) *T

Types

type ConsumeOption

type ConsumeOption interface {
	// contains filtered or unexported methods
}

func WithReadTimeout

func WithReadTimeout(readTimeout time.Duration) ConsumeOption

func WithVerifyReceived

func WithVerifyReceived(verifyReceived bool) ConsumeOption

type ProduceOption

type ProduceOption interface {
	// contains filtered or unexported methods
}

func WithHeaders

func WithHeaders(headers map[string]string) ProduceOption

func WithSuccess

func WithSuccess(success bool) ProduceOption

type StdoutLogConsumer

type StdoutLogConsumer struct{}

func (*StdoutLogConsumer) Accept

func (lc *StdoutLogConsumer) Accept(l testcontainers.Log)