7. Тестирование async систем

package service

import (
    "context"
    "errors"
    "fmt"
)

// PaymentMessage представляет входящее событие из очереди
type PaymentMessage struct {
    EventID string  `json:"event_id"`
    UserID  string  `json:"user_id"`
    Amount  float64 `json:"amount"`
}

// OrderRepository — интерфейс хранилища, легко подменяемый моком
type OrderRepository interface {
    UpdateBalance(ctx context.Context, userID string, delta float64) error
    RecordEvent(ctx context.Context, eventID string) error
    HasEvent(ctx context.Context, eventID string) (bool, error)
}

type PaymentHandler struct {
    repo OrderRepository
}

func NewPaymentHandler(repo OrderRepository) *PaymentHandler {
    return &PaymentHandler{repo: repo}
}

func (h *PaymentHandler) HandlePayment(ctx context.Context, msg PaymentMessage) error {
    if msg.EventID == "" {
        return errors.New("missing event_id")
    }
    if msg.Amount <= 0 {
        return fmt.Errorf("invalid payment amount: %.2f", msg.Amount)
    }

    // Проверка идемпотентности
    alreadyProcessed, err := h.repo.HasEvent(ctx, msg.EventID)
    if err != nil {
        return fmt.Errorf("idempotency check failed: %w", err)
    }
    if alreadyProcessed {
        // Повторное сообщение — подтверждаем без выполнения побочных эффектов
        return nil
    }

    // Списание и фиксация события
    if err := h.repo.UpdateBalance(ctx, msg.UserID, -msg.Amount); err != nil {
        return fmt.Errorf("balance update failed: %w", err)
    }
    if err := h.repo.RecordEvent(ctx, msg.EventID); err != nil {
        return fmt.Errorf("record event failed: %w", err)
    }

    return nil
}

package service_test

import (
    "context"
    "errors"
    "strings"
    "sync"
    "testing"

    "github.com/stretchr/testify/assert"
    "github.com/stretchr/testify/require"
)

type mockOrderRepo struct {
    mu           sync.Mutex
    events       map[string]bool
    balances     map[string]float64
    updateErr    error
    hasEventErr  error
}

func newMockOrderRepo() *mockOrderRepo {
    return &mockOrderRepo{
        events:   make(map[string]bool),
        balances: make(map[string]float64),
    }
}

func (m *mockOrderRepo) UpdateBalance(ctx context.Context, userID string, delta float64) error {
    m.mu.Lock()
    defer m.mu.Unlock()
    if m.updateErr != nil {
        return m.updateErr
    }
    m.balances[userID] += delta
    return nil
}

func (m *mockOrderRepo) RecordEvent(ctx context.Context, eventID string) error {
    m.mu.Lock()
    defer m.mu.Unlock()
    m.events[eventID] = true
    return nil
}

func (m *mockOrderRepo) HasEvent(ctx context.Context, eventID string) (bool, error) {
    m.mu.Lock()
    defer m.mu.Unlock()
    if m.hasEventErr != nil {
        return false, m.hasEventErr
    }
    return m.events[eventID], nil
}

func TestPaymentHandler_Table(t *testing.T) {
    t.Parallel()

    tests := []struct {
        name        string
        msg         PaymentMessage
        setupMock   func(m *mockOrderRepo)
        wantErr     string
        checkExpect func(t *testing.T, m *mockOrderRepo)
    }{
        {
            name: "Success: New payment processed",
            msg:  PaymentMessage{EventID: "evt-1", UserID: "usr-100", Amount: 50.0},
            setupMock: func(m *mockOrderRepo) {
                m.balances["usr-100"] = 200.0
            },
            wantErr: "",
            checkExpect: func(t *testing.T, m *mockOrderRepo) {
                assert.Equal(t, 150.0, m.balances["usr-100"])
                assert.True(t, m.events["evt-1"])
            },
        },
        {
            name:      "Error: Zero or negative amount",
            msg:       PaymentMessage{EventID: "evt-2", UserID: "usr-100", Amount: -10.0},
            setupMock: func(m *mockOrderRepo) {},
            wantErr:   "invalid payment amount",
        },
        {
            name: "Error: Database update balance failure",
            msg:  PaymentMessage{EventID: "evt-3", UserID: "usr-100", Amount: 30.0},
            setupMock: func(m *mockOrderRepo) {
                m.updateErr = errors.New("connection reset by peer")
            },
            wantErr: "balance update failed",
        },
    }

    for _, tt := range tests {
        tt := tt
        t.Run(tt.name, func(t *testing.T) {
            t.Parallel()
            mockRepo := newMockOrderRepo()
            tt.setupMock(mockRepo)

            h := NewPaymentHandler(mockRepo)
            err := h.HandlePayment(context.Background(), tt.msg)

            if tt.wantErr != "" {
                require.Error(t, err)
                assert.Contains(t, err.Error(), tt.wantErr)
            } else {
                require.NoError(t, err)
                if tt.checkExpect != nil {
                    tt.checkExpect(t, mockRepo)
                }
            }
        })
    }
}

func TestPaymentHandler_Idempotency_Concurrent(t *testing.T) {
    mockRepo := newMockOrderRepo()
    mockRepo.balances["usr-vip"] = 1000.0

    h := NewPaymentHandler(mockRepo)
    msg := PaymentMessage{
        EventID: "tx-duplicate-999",
        UserID:  "usr-vip",
        Amount:  100.0,
    }

    const goroutines = 10
    var wg sync.WaitGroup
    wg.Add(goroutines)

    // Запускаем 10 параллельных горутин с абсолютно одинаковым сообщением
    for i := 0; i < goroutines; i++ {
        go func() {
            defer wg.Done()
            _ = h.HandlePayment(context.Background(), msg)
        }()
    }

    wg.Wait()

    // Инвариант: списание 100.0 должно произойти строго 1 раз! Баланс обязан стать 900.0, а не 0.0
    assert.Equal(t, 900.0, mockRepo.balances["usr-vip"], "Баланс должен быть списан ровно один раз")
    assert.True(t, mockRepo.events["tx-duplicate-999"])
}

import "go.uber.org/goleak"

func TestMain(m *testing.M) {
    // Проверяет, что после завершения пакета тестов в рантайме не осталось забытых горутин
    goleak.VerifyTestMain(m)
}

func TestHandlerLifecycle(t *testing.T) {
    defer goleak.VerifyNone(t)
    // Тело теста с запуском и остановкой воркеров
}

package service_test

import (
    "context"
    "errors"
    "sync"
    "sync/atomic"
    "testing"
    "time"

    "github.com/stretchr/testify/assert"
    "github.com/stretchr/testify/require"
)

type mockBroker struct {
    mu           sync.Mutex
    attempts     int
    failuresLeft int
}

func (m *mockBroker) Publish(ctx context.Context, topic string, payload []byte) error {
    m.mu.Lock()
    defer m.mu.Unlock()
    m.attempts++
    if m.failuresLeft > 0 {
        m.failuresLeft--
        return errors.New("network io timeout")
    }
    return nil
}

type mockDLQ struct {
    mu       sync.Mutex
    messages [][]byte
}

func (d *mockDLQ) SendToDLQ(ctx context.Context, payload []byte, reason error) error {
    d.mu.Lock()
    defer d.mu.Unlock()
    d.messages = append(d.messages, payload)
    return nil
}

type RetryConfig struct {
    MaxRetries int
    BaseDelay  time.Duration
}

type RetryPublisher struct {
    broker *mockBroker
    dlq    *mockDLQ
    cfg    RetryConfig
}

func (p *RetryPublisher) PublishWithRetry(ctx context.Context, topic string, payload []byte) error {
    delay := p.cfg.BaseDelay
    for attempt := 0; attempt <= p.cfg.MaxRetries; attempt++ {
        err := p.broker.Publish(ctx, topic, payload)
        if err == nil {
            return nil
        }
        if attempt == p.cfg.MaxRetries {
            // Попытки исчерпаны — отправляем в DLQ
            _ = p.dlq.SendToDLQ(ctx, payload, err)
            return fmt.Errorf("retries exhausted: %w", err)
        }
        select {
        case <-ctx.Done():
            return ctx.Err()
        case <-time.After(delay):
            delay *= 2 // exponential backoff
        }
    }
    return nil
}

func TestRetryPublisher_SuccessAfterRetries(t *testing.T) {
    broker := &mockBroker{failuresLeft: 2} // Первые 2 вызова упадут, 3-й успешен
    dlq := &mockDLQ{}
    pub := &RetryPublisher{
        broker: broker,
        dlq:    dlq,
        cfg:    RetryConfig{MaxRetries: 3, BaseDelay: 2 * time.Millisecond},
    }

    err := pub.PublishWithRetry(context.Background(), "orders", []byte(`{"id": 10}`))
    require.NoError(t, err)

    broker.mu.Lock()
    assert.Equal(t, 3, broker.attempts, "Ожидалось ровно 3 попытки (2 сбоя + 1 успех)")
    broker.mu.Unlock()

    dlq.mu.Lock()
    assert.Empty(t, dlq.messages, "DLQ не должна содержать сообщений при успехе")
    dlq.mu.Unlock()
}

func TestRetryPublisher_ExhaustedToDLQ(t *testing.T) {
    broker := &mockBroker{failuresLeft: 10} // Все попытки завершатся сбоем
    dlq := &mockDLQ{}
    pub := &RetryPublisher{
        broker: broker,
        dlq:    dlq,
        cfg:    RetryConfig{MaxRetries: 3, BaseDelay: 2 * time.Millisecond},
    }

    err := pub.PublishWithRetry(context.Background(), "orders", []byte(`{"id": 99}`))
    require.Error(t, err)

    broker.mu.Lock()
    // 1 исходная попытка + 3 ретрая = 4 вызова
    assert.Equal(t, 4, broker.attempts)
    broker.mu.Unlock()

    dlq.mu.Lock()
    require.Len(t, dlq.messages, 1, "Сообщение обязано поступить в DLQ")
    assert.JSONEq(t, `{"id": 99}`, string(dlq.messages[0]))
    dlq.mu.Unlock()
}

//go:build integration
// +build integration

package integration_test

import (
    "context"
    "testing"
    "time"

    "github.com/stretchr/testify/require"
    "github.com/testcontainers/testcontainers-go"
    tckafka "github.com/testcontainers/testcontainers-go/modules/kafka"
    "github.com/twmb/franz-go/pkg/kgo"
)

func TestKafkaProduceConsume(t *testing.T) {
    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
    defer cancel()

    // 1. Поднимаем легковесный одиночный контейнер Kafka в Docker
    kafkaContainer, err := tckafka.RunContainer(ctx,
        tckafka.WithClusterID("test-cluster"),
        testcontainers.WithImage("confluentinc/confluent-local:7.5.0"),
    )
    require.NoError(t, err)
    defer func() {
        _ = kafkaContainer.Terminate(context.Background())
    }()

    // 2. Извлекаем реальные хост и порт, выставленные наружу Docker-демоном
    brokers, err := kafkaContainer.Brokers(ctx)
    require.NoError(t, err)
    require.NotEmpty(t, brokers)

    topicName := "test-topic"

    // 3. Создаем топик через Kafka Admin API
    adminCl, err := kgo.NewClient(kgo.SeedBrokers(brokers...))
    require.NoError(t, err)
    defer adminCl.Close()

    _, err = adminCl.CreateTopics(ctx, &kgo.CreateTopicsRequest{
        Topics: []kgo.CreateTopicsRequestTopic{{
            Topic:             topicName,
            NumPartitions:     1,
            ReplicationFactor: 1,
        }},
    })
    require.NoError(t, err)

    // 4. Инициализируем продюсер и отправляем контрольное сообщение
    produceCl, err := kgo.NewClient(
        kgo.SeedBrokers(brokers...),
        kgo.DefaultProduceTopic(topicName),
    )
    require.NoError(t, err)

    record := &kgo.Record{Value: []byte("hello-async-world")}
    results := produceCl.ProduceSync(ctx, record)
    require.NoError(t, results.FirstErr())
    produceCl.Close()

    // 5. Запускаем консьюмер и проверяем получение
    consumeCl, err := kgo.NewClient(
        kgo.SeedBrokers(brokers...),
        kgo.ConsumeTopics(topicName),
        kgo.ConsumerGroup("test-group"),
    )
    require.NoError(t, err)
    defer consumeCl.Close()

    pollCtx, pollCancel := context.WithTimeout(ctx, 15*time.Second)
    defer pollCancel()

    var received []string
    for {
        fetches := consumeCl.PollFetches(pollCtx)
        if fetches.IsClientClosed() || pollCtx.Err() != nil {
            break
        }
        fetches.EachRecord(func(r *kgo.Record) {
            received = append(received, string(r.Value))
        })
        if len(received) >= 1 {
            break
        }
    }

    require.Equal(t, []string{"hello-async-world"}, received)
}

func TestGracefulShutdown(t *testing.T) {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    client, err := kgo.NewClient(kgo.SeedBrokers(brokers...), kgo.ConsumeTopics("shutdown-topic"))
    require.NoError(t, err)
    defer client.Close()

    processed := make(chan string, 100)
    consumerDone := make(chan struct{})

    // Фоновый цикл вычитки
    go func() {
        defer close(consumerDone)
        for {
            fetches := client.PollFetches(ctx)
            if fetches.IsClientClosed() || ctx.Err() != nil {
                // Контекст отменен — выходим из цикла вычитки
                break
            }
            fetches.EachRecord(func(r *kgo.Record) {
                processed <- string(r.Value)
            })
        }
    }()

    // Ждем, пока консьюмер гарантированно начнет работу и обработает хотя бы одно сообщение
    require.Eventually(t, func() bool {
        return len(processed) > 0
    }, 5*time.Second, 50*time.Millisecond)

    // Имитируем приход сигнала SIGTERM в микросервис
    cancel()

    // Убеждаемся, что горутина консьюмера штатно завершилась без зависания
    select {
    case <-consumerDone:
        // Успешный выход
    case <-time.After(3 * time.Second):
        t.Fatal("Consumer failed to shutdown gracefully within 3 seconds")
    }
}

func TestConcurrentProcessing_OrderAndRaces(t *testing.T) {
    const totalMessages = 500
    var processedCount int64

    var mu sync.Mutex
    orderPerKey := make(map[string][]int)

    // Симулируем конкурентный воркер-пул
    jobs := make(chan struct {
        key string
        seq int
    }, totalMessages)

    var wg sync.WaitGroup
    const workers = 8
    for w := 0; w < workers; w++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for j := range jobs {
                // Имитируем работу
                atomic.AddInt64(&processedCount, 1)

                mu.Lock()
                orderPerKey[j.key] = append(orderPerKey[j.key], j.seq)
                mu.muUnlock := func() { mu.Unlock() }
                mu.muUnlock()
            }
        }()
    }

    // Отправляем сообщения с тремя разными ключами
    for i := 0; i < totalMessages; i++ {
        key := fmt.Sprintf("partition-%d", i%3)
        jobs <- struct {
            key string
            seq int
        }{key: key, seq: i}
    }
    close(jobs)
    wg.Wait()

    assert.Equal(t, int64(totalMessages), atomic.LoadInt64(&processedCount))
}

func FuzzPaymentMessageHandler(f *testing.F) {
    // Добавляем начальный корпус валидных семян (Seed Corpus)
    f.Add([]byte(`{"event_id":"evt-1","user_id":"u-10","amount":100.5}`))
    f.Add([]byte(`{"event_id":"","user_id":"","amount":0}`))
    f.Add([]byte(`{}`))
    f.Add([]byte(`null`))

    repo := newMockOrderRepo()
    handler := NewPaymentHandler(repo)

    f.Fuzz(func(t *testing.T, data []byte) {
        var msg PaymentMessage
        // 1. Проверяем парсинг — он не должен вызывать панику
        if err := json.Unmarshal(data, &msg); err != nil {
            // Невалидный JSON — штатная ситуация, игнорируем
            return
        }

        // 2. Проверяем доменную валидацию — любые крайние значения float64 (NaN, +Inf, -Inf, Overflow)
        // не должны ломать внутренние инварианты
        _ = handler.HandlePayment(context.Background(), msg)
    })
}

go test -fuzz=FuzzPaymentMessageHandler -fuzztime=30s ./service/...