kra-new/internal/biz/integration/integration_config_communication_test.go

package integration

import (
	"context"
	"encoding/json"
	"errors"
	"testing"
)

type integrationConfigRepoTestDouble struct {
	saves int
}

func (*integrationConfigRepoTestDouble) ListIntegrationConfigs(context.Context, string) ([]*IntegrationConfig, error) {
	return nil, nil
}
func (*integrationConfigRepoTestDouble) FindIntegrationConfig(context.Context, string, string) (*IntegrationConfig, error) {
	return nil, nil
}
func (r *integrationConfigRepoTestDouble) SaveIntegrationConfig(context.Context, *IntegrationConfig) error {
	r.saves++
	return nil
}
func (*integrationConfigRepoTestDouble) DeleteIntegrationConfig(context.Context, string, string) error {
	return nil
}

type integrationConnectionTesterDouble struct {
	calls  int
	config *IntegrationConfig
	err    error
}

func (t *integrationConnectionTesterDouble) TestIntegration(_ context.Context, config *IntegrationConfig) error {
	t.calls++
	t.config = config
	return t.err
}

func TestCommunicationIntegrationDefinitionsAndValidation(t *testing.T) {
	for _, target := range []struct{ kind, provider string }{
		{IntegrationKindMQ, "emqx"},
		{IntegrationKindMQ, "kafka"},
		{IntegrationKindMQ, "rabbitmq"},
		{IntegrationKindWebSocket, "melody"},
	} {
		values := DefaultIntegrationConfig(target.kind, target.provider)
		if len(values) == 0 {
			t.Fatalf("default config missing for %s/%s", target.kind, target.provider)
		}
		if err := ValidateIntegrationConfig(target.kind, target.provider, values); err != nil {
			t.Fatalf("default config invalid for %s/%s: %v", target.kind, target.provider, err)
		}
	}
}

func TestKafkaDefaultConfigSurvivesJSONRoundTrip(t *testing.T) {
	raw, err := json.Marshal(DefaultIntegrationConfig(IntegrationKindMQ, "kafka"))
	if err != nil {
		t.Fatal(err)
	}
	values, err := decodeIntegrationObject(raw)
	if err != nil {
		t.Fatal(err)
	}
	if got := integrationInt64(values, "max_bytes", 0); got != 10485760 {
		t.Fatalf("max_bytes = %d, want 10485760; decoded=%#v", got, values["max_bytes"])
	}
	if err = ValidateIntegrationConfig(IntegrationKindMQ, "kafka", values); err != nil {
		t.Fatalf("JSON-decoded kafka defaults are invalid: %v", err)
	}
}

func TestIntegrationInt64RejectsFractionalAndOverflowValues(t *testing.T) {
	values := map[string]any{"fractional": 1.5, "overflow": json.Number("9223372036854775808")}
	if got := integrationInt64(values, "fractional", -1); got != -1 {
		t.Fatalf("fractional value = %d, want fallback", got)
	}
	if got := integrationInt64(values, "overflow", -1); got != -1 {
		t.Fatalf("overflow value = %d, want fallback", got)
	}
}

func TestCommunicationIntegrationValidationRejectsInvalidValues(t *testing.T) {
	rabbit := DefaultIntegrationConfig(IntegrationKindMQ, "rabbitmq")
	rabbit["port"] = 0
	if err := ValidateIntegrationConfig(IntegrationKindMQ, "rabbitmq", rabbit); err == nil {
		t.Fatal("invalid rabbitmq port was accepted")
	}
	kafka := DefaultIntegrationConfig(IntegrationKindMQ, "kafka")
	kafka["brokers"] = []string{"missing-port"}
	if err := ValidateIntegrationConfig(IntegrationKindMQ, "kafka", kafka); err == nil {
		t.Fatal("invalid kafka broker was accepted")
	}
	kafka = DefaultIntegrationConfig(IntegrationKindMQ, "kafka")
	kafka["max_bytes"] = 0
	if err := ValidateIntegrationConfig(IntegrationKindMQ, "kafka", kafka); err == nil {
		t.Fatal("invalid kafka byte limits were accepted")
	}
	websocket := DefaultIntegrationConfig(IntegrationKindWebSocket, "melody")
	websocket["path"] = "ws"
	if err := ValidateIntegrationConfig(IntegrationKindWebSocket, "melody", websocket); err == nil {
		t.Fatal("invalid websocket path was accepted")
	}
}

func TestIntegrationConfigTestDoesNotPersistCandidate(t *testing.T) {
	repo := &integrationConfigRepoTestDouble{}
	tester := &integrationConnectionTesterDouble{}
	usecase := NewIntegrationConfigUsecase(repo, tester)
	raw, _ := json.Marshal(map[string]any{"path": "/candidate"})

	err := usecase.Test(context.Background(), &IntegrationConfig{
		Kind:     " WebSocket ",
		Provider: " Melody ",
		Enabled:  false,
		Values:   raw,
	})
	if err != nil {
		t.Fatal(err)
	}
	if repo.saves != 0 {
		t.Fatalf("candidate config was persisted %d times", repo.saves)
	}
	if tester.calls != 1 || tester.config == nil {
		t.Fatalf("connection tester calls = %d, config = %#v", tester.calls, tester.config)
	}
	if tester.config.Kind != IntegrationKindWebSocket || tester.config.Provider != "melody" || !tester.config.Enabled {
		t.Fatalf("tested config = %#v", tester.config)
	}
	values := map[string]any{}
	if err = json.Unmarshal(tester.config.Values, &values); err != nil {
		t.Fatal(err)
	}
	if values["path"] != "/candidate" || values["write_wait"] != "10s" {
		t.Fatalf("tested values = %#v", values)
	}
}

func TestIntegrationConfigTestReturnsSafeProviderFailure(t *testing.T) {
	underlying := errors.New("dial tcp 127.0.0.1:9092: connection refused")
	for _, test := range []struct {
		provider string
		name     string
	}{
		{provider: "kafka", name: "Kafka"},
		{provider: "emqx", name: "EMQX"},
		{provider: "rabbitmq", name: "RabbitMQ"},
	} {
		t.Run(test.provider, func(t *testing.T) {
			values := DefaultIntegrationConfig(IntegrationKindMQ, test.provider)
			raw, _ := json.Marshal(values)
			usecase := NewIntegrationConfigUsecase(&integrationConfigRepoTestDouble{}, &integrationConnectionTesterDouble{err: underlying})
			err := usecase.Test(context.Background(), &IntegrationConfig{Kind: IntegrationKindMQ, Provider: test.provider, Values: raw})
			if err == nil || err.Error() != test.name+" 连接测试失败" {
				t.Fatalf("Test() error = %v", err)
			}
			if !errors.Is(err, underlying) {
				t.Fatalf("Test() error does not wrap the provider failure: %v", err)
			}
		})
	}
}

func TestIntegrationConfigRejectsJSONNull(t *testing.T) {
	repo := &integrationConfigRepoTestDouble{}
	usecase := NewIntegrationConfigUsecase(repo, &integrationConnectionTesterDouble{})
	if err := usecase.Save(context.Background(), &IntegrationConfig{Kind: IntegrationKindMQ, Provider: "emqx", Values: json.RawMessage("null")}); err == nil {
		t.Fatal("Save() accepted JSON null as an object")
	}
	if err := usecase.Test(context.Background(), &IntegrationConfig{Kind: IntegrationKindMQ, Provider: "emqx", Values: json.RawMessage("null")}); err == nil {
		t.Fatal("Test() accepted JSON null as an object")
	}
}