api_suite_test.go

// Copyright 2023 Redpanda Data, Inc.
//
// Use of this software is governed by the Business Source License
// included in the file licenses/BSL.md
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0

//go:build integration

package integration

import (
	"context"
	"math/rand"
	"net"
	"strconv"
	"testing"
	"time"

	"github.com/cloudhut/common/rest"
	adminapi "github.com/redpanda-data/common-go/rpadmin"
	"github.com/redpanda-data/common-go/rpsr"
	"github.com/stretchr/testify/require"
	"github.com/stretchr/testify/suite"
	"github.com/testcontainers/testcontainers-go"
	"github.com/testcontainers/testcontainers-go/modules/redpanda"
	"github.com/testcontainers/testcontainers-go/network"
	"github.com/twmb/franz-go/pkg/kadm"
	"github.com/twmb/franz-go/pkg/kgo"
	"github.com/twmb/franz-go/pkg/sr"

	"github.com/redpanda-data/console/backend/pkg/api"
	"github.com/redpanda-data/console/backend/pkg/config"
	"github.com/redpanda-data/console/backend/pkg/testutil"
)

type APISuite struct {
	suite.Suite

	redpandaContainer    *redpanda.Container
	kConnectContainer    testcontainers.Container
	network              *testcontainers.DockerNetwork
	kafkaClient          *kgo.Client
	kafkaAdminClient     *kadm.Client
	redpandaAdminClient  *adminapi.AdminAPI
	schemaRegistryClient *rpsr.Client

	cfg *config.Config
	api *api.API

	testSeedBroker string
}

func TestSuite(t *testing.T) {
	suite.Run(t, &APISuite{})
}

func (s *APISuite) SetupSuite() {
	t := s.T()
	require := require.New(t)

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
	defer cancel()

	// 1. setup docker network
	ntw, err := network.New(ctx)

	require.NoError(err)

	s.network = ntw

	// 2. Start Redpanda Docker container
	container, err := redpanda.Run(ctx,
		testutil.RedpandaImage(),
		redpanda.WithEnableWasmTransform(),
		network.WithNetwork([]string{"redpanda"}, s.network),
		redpanda.WithListener("redpanda:29092"),
	)
	testutil.LogContainerLogsIfFailed(ctx, t, container, err)
	require.NoError(err)
	s.redpandaContainer = container

	// 2. Retrieve Redpanda connection details
	seedBroker, err := container.KafkaSeedBroker(ctx)
	require.NoError(err)
	schemaRegistryAddress, err := container.SchemaRegistryAddress(ctx)
	require.NoError(err)
	adminApiAddr, err := container.AdminAPIAddress(ctx)
	require.NoError(err)

	require.NoError(err)

	s.testSeedBroker = seedBroker

	// 3. Start KafkaConnect Docker container
	kConnectContainer, err := testutil.RunRedpandaConnectorsContainer(
		ctx,
		[]string{"redpanda:29092"},
		network.WithNetwork([]string{"kconnect"}, s.network),
	)
	testutil.LogContainerLogsIfFailed(ctx, t, kConnectContainer, err)
	require.NoError(err)

	s.kConnectContainer = kConnectContainer

	// 4. Create Kafka clients
	s.kafkaClient, s.kafkaAdminClient = testutil.CreateClients(t, []string{seedBroker})

	kConnectClusterURL, err := kConnectContainer.PortEndpoint(ctx, "8083/tcp", "http")
	require.NoError(err)

	// 5. Create Redpanda client
	adminApiClient, err := adminapi.NewAdminAPI([]string{adminApiAddr}, &adminapi.NopAuth{}, nil)
	require.NoError(err)
	s.redpandaAdminClient = adminApiClient

	srCl, err := sr.NewClient(sr.URLs(schemaRegistryAddress))
	require.NoError(err)
	rpsrCl, err := rpsr.NewClient(srCl)
	require.NoError(err)
	s.schemaRegistryClient = rpsrCl

	// 5. Configure & start Redpanda Console
	httpListenPort := rand.Intn(50000) + 10000
	s.cfg = &config.Config{}
	s.cfg.SetDefaults()
	s.cfg.ServeFrontend = false
	s.cfg.REST = config.Server{
		Config: rest.Config{
			HTTPListenAddress: "0.0.0.0",
			HTTPListenPort:    httpListenPort,
		},
	}
	s.cfg.Kafka.Brokers = []string{s.testSeedBroker}
	s.cfg.SchemaRegistry.Enabled = true
	s.cfg.SchemaRegistry.URLs = []string{schemaRegistryAddress}

	s.cfg.Redpanda.AdminAPI.Enabled = true
	s.cfg.Redpanda.AdminAPI.URLs = []string{adminApiAddr}
	s.cfg.Redpanda.AdminAPI.TLS.Enabled = false

	s.cfg.KafkaConnect = config.KafkaConnect{
		Enabled: true,
		Clusters: []config.KafkaConnectCluster{
			{
				Name: "connect-cluster",
				URL:  kConnectClusterURL,
			},
		},
	}
	s.api, err = api.New(s.cfg)
	require.NoError(err)

	go func() {
		if err := s.api.Start(context.Background()); err != nil {
			s.T().Errorf("failed to start API: %v", err)
		}
	}()

	// 5. Wait until Console API is up
	httpServerAddress := net.JoinHostPort("localhost", strconv.Itoa(httpListenPort))
	retries := 60
	for retries > 0 {
		ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
		if _, err := (&net.Dialer{}).DialContext(ctx, "tcp", httpServerAddress); err == nil {
			cancel()
			break
		}
		cancel()
		time.Sleep(100 * time.Millisecond)
		retries--
	}
}

func (s *APISuite) TearDownSuite() {
	t := s.T()
	assert := require.New(t)

	ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second)
	defer cancel()

	// 1. Terminate KafkaConnect container
	assert.NoError(s.kConnectContainer.Terminate(ctx))
	// 2. Terminate Redpanda container
	assert.NoError(s.redpandaContainer.Terminate(ctx))
	// 3. Remove docker network
	assert.NoError(s.network.Remove(ctx))
	// 4. Stop API
	assert.NoError(s.api.Stop(ctx))
}

func (s *APISuite) httpAddress() string {
	if s.api.Cfg.REST.TLS.Enabled {
		return "https://" + s.api.Cfg.REST.HTTPListenAddress + ":" + strconv.FormatInt(int64(s.api.Cfg.REST.HTTPSListenPort), 10)
	}
	return "http://" + s.api.Cfg.REST.HTTPListenAddress + ":" + strconv.FormatInt(int64(s.api.Cfg.REST.HTTPListenPort), 10)
}