95 lines
2.9 KiB
Go
95 lines
2.9 KiB
Go
package integration
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"net"
|
|
"os"
|
|
"strconv"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
integrationbiz "kra/internal/biz/integration"
|
|
)
|
|
|
|
type kafkaConnectivityRepo struct{}
|
|
|
|
func (*kafkaConnectivityRepo) ListIntegrationConfigs(context.Context, string) ([]*integrationbiz.IntegrationConfig, error) {
|
|
return nil, nil
|
|
}
|
|
func (*kafkaConnectivityRepo) FindIntegrationConfig(context.Context, string, string) (*integrationbiz.IntegrationConfig, error) {
|
|
return nil, nil
|
|
}
|
|
func (*kafkaConnectivityRepo) SaveIntegrationConfig(context.Context, *integrationbiz.IntegrationConfig) error {
|
|
return nil
|
|
}
|
|
func (*kafkaConnectivityRepo) DeleteIntegrationConfig(context.Context, string, string) error {
|
|
return nil
|
|
}
|
|
|
|
func TestKafkaConnectivityIntegration(t *testing.T) {
|
|
rawBrokers := strings.TrimSpace(os.Getenv("KRA_KAFKA_TEST_BROKERS"))
|
|
if rawBrokers == "" {
|
|
t.Skip("KRA_KAFKA_TEST_BROKERS is not configured")
|
|
}
|
|
brokers := make([]string, 0)
|
|
for _, broker := range strings.Split(rawBrokers, ",") {
|
|
if broker = strings.TrimSpace(broker); broker != "" {
|
|
brokers = append(brokers, broker)
|
|
}
|
|
}
|
|
values := integrationbiz.DefaultIntegrationConfig(integrationbiz.IntegrationKindMQ, "kafka")
|
|
values["brokers"] = brokers
|
|
values["client_id"] = fmt.Sprintf("kra-connectivity-%d", time.Now().UnixNano())
|
|
values["group_id"] = fmt.Sprintf("kra-connectivity-%d", time.Now().UnixNano())
|
|
testMQConnectivityIntegration(t, "kafka", values)
|
|
}
|
|
|
|
func TestEMQXConnectivityIntegration(t *testing.T) {
|
|
broker := strings.TrimSpace(os.Getenv("KRA_EMQX_TEST_BROKER"))
|
|
if broker == "" {
|
|
t.Skip("KRA_EMQX_TEST_BROKER is not configured")
|
|
}
|
|
values := integrationbiz.DefaultIntegrationConfig(integrationbiz.IntegrationKindMQ, "emqx")
|
|
values["broker"] = broker
|
|
values["client_id"] = fmt.Sprintf("kra-connectivity-%d", time.Now().UnixNano())
|
|
testMQConnectivityIntegration(t, "emqx", values)
|
|
}
|
|
|
|
func TestRabbitMQConnectivityIntegration(t *testing.T) {
|
|
address := strings.TrimSpace(os.Getenv("KRA_RABBITMQ_TEST_ADDR"))
|
|
if address == "" {
|
|
t.Skip("KRA_RABBITMQ_TEST_ADDR is not configured")
|
|
}
|
|
host, portText, err := net.SplitHostPort(address)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
port, err := strconv.Atoi(portText)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
values := integrationbiz.DefaultIntegrationConfig(integrationbiz.IntegrationKindMQ, "rabbitmq")
|
|
values["host"] = host
|
|
values["port"] = port
|
|
testMQConnectivityIntegration(t, "rabbitmq", values)
|
|
}
|
|
|
|
func testMQConnectivityIntegration(t *testing.T, provider string, values map[string]any) {
|
|
t.Helper()
|
|
raw, err := json.Marshal(values)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
usecase := integrationbiz.NewIntegrationConfigUsecase(&kafkaConnectivityRepo{}, NewConnectivityTester(nil))
|
|
if err = usecase.Test(context.Background(), &integrationbiz.IntegrationConfig{
|
|
Kind: integrationbiz.IntegrationKindMQ,
|
|
Provider: provider,
|
|
Values: raw,
|
|
}); err != nil {
|
|
t.Fatalf("test JSON-decoded %s defaults: %v", provider, err)
|
|
}
|
|
}
|