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") } }