From 1e06fa7ca3d0b2e5efee300b2b5d139438eb3701 Mon Sep 17 00:00:00 2001 From: Devaris Date: Sun, 5 Jul 2026 14:35:50 -0700 Subject: [PATCH 1/3] feat(api): wire ConduitError into the API error boundary MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Step 2 of the structured-error rollout: the ConduitError foundation now reaches the API. When an error returned to a gRPC/HTTP handler carries a *ConduitError (anywhere in its chain), status.go emits its gRPC category plus an additive google.rpc.ErrorInfo detail carrying the stable reason, configPath, suggestion, and fix — so an API/MCP/UI consumer gets a machine-actionable error. This is a single integration point (conduitErrorStatus, checked at the top of PipelineError/ConnectorError/ProcessorError/PluginError). It is purely additive: any error that is NOT a ConduitError falls through to the existing sentinel->code mapping, byte-for-byte unchanged (existing tests still pass). Boundaries get migrated to emit ConduitError codes incrementally from here. Tests: a wrapped ConduitError yields the right code + ErrorInfo detail (reason, configPath, suggestion); a sentinel error maps exactly as before. Design: docs/design-documents/20260705-conduit-error-and-structured-output.md Co-Authored-By: Claude Opus 4.8 (1M context) --- pkg/http/api/status/status.go | 29 +++++++++++++++++++++ pkg/http/api/status/status_test.go | 41 ++++++++++++++++++++++++++++++ 2 files changed, 70 insertions(+) diff --git a/pkg/http/api/status/status.go b/pkg/http/api/status/status.go index 3cbde0377..354eb1ba4 100644 --- a/pkg/http/api/status/status.go +++ b/pkg/http/api/status/status.go @@ -17,6 +17,7 @@ package status import ( "github.com/conduitio/conduit/pkg/connector" "github.com/conduitio/conduit/pkg/foundation/cerrors" + "github.com/conduitio/conduit/pkg/foundation/cerrors/conduiterr" "github.com/conduitio/conduit/pkg/orchestrator" "github.com/conduitio/conduit/pkg/pipeline" conn_plugin "github.com/conduitio/conduit/pkg/plugin/connector" @@ -25,7 +26,24 @@ import ( grpcstatus "google.golang.org/grpc/status" ) +// conduitErrorStatus returns the structured gRPC status for a ConduitError in err's +// chain — its gRPC category plus a google.rpc.ErrorInfo detail carrying the stable +// reason and any configPath/suggestion/fix — and true if one was found. This is the +// single point where the ConduitError model reaches the API; boundaries that have +// not yet been migrated fall through to the sentinel-based code mapping below, +// unchanged. +func conduitErrorStatus(err error) (error, bool) { + if ce, ok := conduiterr.Get(err); ok { + return conduiterr.ToStatus(ce).Err(), true + } + return nil, false +} + func PipelineError(err error) error { + if s, ok := conduitErrorStatus(err); ok { + return s + } + var code codes.Code switch { @@ -41,6 +59,10 @@ func PipelineError(err error) error { } func ConnectorError(err error) error { + if s, ok := conduitErrorStatus(err); ok { + return s + } + var code codes.Code switch { @@ -56,6 +78,10 @@ func ConnectorError(err error) error { } func ProcessorError(err error) error { + if s, ok := conduitErrorStatus(err); ok { + return s + } + var code codes.Code switch { @@ -71,6 +97,9 @@ func ProcessorError(err error) error { } func PluginError(err error) error { + if s, ok := conduitErrorStatus(err); ok { + return s + } return grpcstatus.Error(codeFromError(err), err.Error()) } diff --git a/pkg/http/api/status/status_test.go b/pkg/http/api/status/status_test.go index 9e1ce0631..f8840cefe 100644 --- a/pkg/http/api/status/status_test.go +++ b/pkg/http/api/status/status_test.go @@ -19,10 +19,12 @@ import ( "github.com/conduitio/conduit/pkg/connector" "github.com/conduitio/conduit/pkg/foundation/cerrors" + "github.com/conduitio/conduit/pkg/foundation/cerrors/conduiterr" "github.com/conduitio/conduit/pkg/orchestrator" "github.com/conduitio/conduit/pkg/pipeline" "github.com/conduitio/conduit/pkg/processor" "github.com/matryer/is" + "google.golang.org/genproto/googleapis/rpc/errdetails" "google.golang.org/grpc/codes" grpcstatus "google.golang.org/grpc/status" ) @@ -126,6 +128,45 @@ func TestProcessorError(t *testing.T) { } } +// TestConduitErrorFlowsThroughBoundary proves the ConduitError foundation reaches +// the API: a ConduitError (even wrapped) yields a gRPC status with the code's +// category and an additive google.rpc.ErrorInfo detail carrying the stable reason, +// configPath, and suggestion. +func TestConduitErrorFlowsThroughBoundary(t *testing.T) { + is := is.New(t) + + ce := conduiterr.New(conduiterr.CodeConnectorPluginNotFound, "connector plugin not found") + ce.ConfigPath = "/connectors/0/plugin" + ce.Suggestion = "run `conduit connectors install `" + + // wrapped, to prove errors.As finds it through the chain + got := PipelineError(cerrors.Errorf("provisioning failed: %w", ce)) + + st, ok := grpcstatus.FromError(got) + is.True(ok) + is.Equal(st.Code(), codes.NotFound) // the code's gRPC category + + var info *errdetails.ErrorInfo + for _, d := range st.Details() { + if ei, ok := d.(*errdetails.ErrorInfo); ok { + info = ei + } + } + is.True(info != nil) // the structured detail is present + is.Equal(info.GetReason(), conduiterr.CodeConnectorPluginNotFound.Reason()) + is.Equal(info.GetDomain(), "conduit") + is.Equal(info.GetMetadata()["configPath"], "/connectors/0/plugin") + is.Equal(info.GetMetadata()["suggestion"], "run `conduit connectors install `") +} + +// TestNonConduitErrorUnchanged confirms the wiring is additive: a plain sentinel +// error still maps exactly as before, with no detail. +func TestNonConduitErrorUnchanged(t *testing.T) { + is := is.New(t) + got := PipelineError(pipeline.ErrInstanceNotFound) + is.Equal(grpcstatus.Error(codes.NotFound, pipeline.ErrInstanceNotFound.Error()), got) +} + func TestCodeFromError(t *testing.T) { type args struct { err error From 38ac8d0484847c0390ac0b2a6c1eb207b40995cd Mon Sep 17 00:00:00 2001 From: Devaris Date: Sun, 5 Jul 2026 14:56:19 -0700 Subject: [PATCH 2/3] feat(plugin): emit connector.plugin_not_found ConduitError code MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Step 3 of the structured-error rollout — the first real error source migrated to a ConduitError code, exercising the API boundary wired in the previous step. When the builtin connector registry can't find a plugin (unknown name, or a known name with no matching version), it now returns a ConduitError with code `connector.plugin_not_found` and a suggested fix, instead of a bare sentinel / generic wrapped error. The `plugin.ErrPluginNotFound` sentinel remains in the chain, so existing `errors.Is` checks are unaffected (invariant noted at the site; existing tests unchanged). End to end: registry -> ConduitError(code, suggestion) -> orchestrator -> API (status.ConnectorError/PluginError) -> ToStatus -> a google.rpc.ErrorInfo detail carrying reason=connector.plugin_not_found + the suggestion, on the wire. Test augments TestRegistry_NewDispenser_PluginNotFound to assert both the sentinel Is-check AND the new code + suggestion. Design: docs/design-documents/20260705-conduit-error-and-structured-output.md Co-Authored-By: Claude Opus 4.8 (1M context) --- pkg/plugin/connector/builtin/registry.go | 20 +++++++++++++++++-- pkg/plugin/connector/builtin/registry_test.go | 7 ++++++- 2 files changed, 24 insertions(+), 3 deletions(-) diff --git a/pkg/plugin/connector/builtin/registry.go b/pkg/plugin/connector/builtin/registry.go index ad75ad8fe..1f4312e81 100644 --- a/pkg/plugin/connector/builtin/registry.go +++ b/pkg/plugin/connector/builtin/registry.go @@ -16,6 +16,7 @@ package builtin import ( "context" + "fmt" "runtime/debug" file "github.com/conduitio/conduit-connector-file" @@ -28,6 +29,7 @@ import ( sdk "github.com/conduitio/conduit-connector-sdk" "github.com/conduitio/conduit-connector-sdk/schema" "github.com/conduitio/conduit/pkg/foundation/cerrors" + "github.com/conduitio/conduit/pkg/foundation/cerrors/conduiterr" "github.com/conduitio/conduit/pkg/foundation/log" "github.com/conduitio/conduit/pkg/plugin" "github.com/conduitio/conduit/pkg/plugin/connector" @@ -190,7 +192,15 @@ func newFullName(pluginName, pluginVersion string) plugin.FullName { func (r *Registry) NewDispenser(logger log.CtxLogger, fullName plugin.FullName, cfg pconnector.PluginConfig) (connector.Dispenser, error) { versionMap, ok := r.plugins[fullName.PluginName()] if !ok { - return nil, plugin.ErrPluginNotFound + // Invariant: errors.Is(err, plugin.ErrPluginNotFound) still holds — the + // sentinel is wrapped, and the ConduitError adds the machine-actionable code. + err := conduiterr.Wrap( + conduiterr.CodeConnectorPluginNotFound, + fmt.Sprintf("builtin connector plugin %q not found", fullName.PluginName()), + plugin.ErrPluginNotFound, + ) + err.Suggestion = "run `conduit connectors list` to see available built-in connectors" + return nil, err } b, ok := versionMap[fullName.PluginVersion()] if !ok { @@ -198,7 +208,13 @@ func (r *Registry) NewDispenser(logger log.CtxLogger, fullName plugin.FullName, for k := range versionMap { availableVersions = append(availableVersions, k) } - return nil, cerrors.Errorf("could not find builtin plugin %q, only found versions %v: %w", fullName, availableVersions, plugin.ErrPluginNotFound) + err := conduiterr.Wrap( + conduiterr.CodeConnectorPluginNotFound, + fmt.Sprintf("builtin connector plugin %q not found; available versions: %v", fullName, availableVersions), + plugin.ErrPluginNotFound, + ) + err.Suggestion = "check the plugin version, or omit it to use the latest" + return nil, err } return b.dispenserFactory(fullName, cfg, logger), nil diff --git a/pkg/plugin/connector/builtin/registry_test.go b/pkg/plugin/connector/builtin/registry_test.go index df6869cb4..43d27da40 100644 --- a/pkg/plugin/connector/builtin/registry_test.go +++ b/pkg/plugin/connector/builtin/registry_test.go @@ -20,6 +20,7 @@ import ( "github.com/conduitio/conduit-connector-protocol/pconnector" "github.com/conduitio/conduit/pkg/foundation/cerrors" + "github.com/conduitio/conduit/pkg/foundation/cerrors/conduiterr" "github.com/conduitio/conduit/pkg/foundation/log" "github.com/conduitio/conduit/pkg/plugin" "github.com/google/go-cmp/cmp" @@ -98,7 +99,11 @@ func TestRegistry_NewDispenser_PluginNotFound(t *testing.T) { dispenser, err := underTest.NewDispenser(log.Nop(), plugin.FullName(tc.pluginName), pconnector.PluginConfig{}) if tc.wantErr { - is.True(cerrors.Is(err, plugin.ErrPluginNotFound)) + is.True(cerrors.Is(err, plugin.ErrPluginNotFound)) // sentinel still in the chain + ce, ok := conduiterr.Get(err) + is.True(ok) // now also carries a machine-actionable ConduitError code + is.Equal(ce.Code.Reason(), conduiterr.CodeConnectorPluginNotFound.Reason()) + is.True(ce.Suggestion != "") // with a suggested fix is.True(dispenser == nil) return } From 7e8e407a17a3255efa47840811684cf44ffe7e32 Mon Sep 17 00:00:00 2001 From: Devaris Date: Sun, 5 Jul 2026 15:38:03 -0700 Subject: [PATCH 3/3] feat(config): emit ConduitError codes + configPath for pipeline validation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Structured-error rollout: pipeline-config validation now attaches a machine-actionable code and a JSON-pointer configPath to the offending field for every validation error — so an API/MCP/UI consumer gets exactly what is wrong and where (e.g. code=config.field_required, configPath=/connectors/0/processors/0/id), not just a prose string. This delivers the "failing config path" half of the machine-actionable-errors promise. - codes.go registers config.field_required/field_invalid/field_too_long/ id_duplicate and a fieldError helper. - Validate/validateConnectors/validateProcessors build the JSON-pointer per field (validateProcessors takes a pathPrefix so nested connector processors get /connectors/i/processors/j/...). - The original sentinels (ErrMandatoryField, ErrInvalidField, ErrDuplicateID, pipeline/connector over-limit errors) stay in the chain — existing errors.Is checks and tests are unaffected. Codes flow through the API boundary wired earlier: ConduitError -> ToStatus -> google.rpc.ErrorInfo with reason + configPath, on the wire. Design: docs/design-documents/20260705-conduit-error-and-structured-output.md Co-Authored-By: Claude Opus 4.8 (1M context) --- pkg/provisioning/config/codes.go | 39 ++++++++++++++++ pkg/provisioning/config/validate.go | 57 +++++++++++++----------- pkg/provisioning/config/validate_test.go | 52 +++++++++++++++++++++ 3 files changed, 123 insertions(+), 25 deletions(-) create mode 100644 pkg/provisioning/config/codes.go diff --git a/pkg/provisioning/config/codes.go b/pkg/provisioning/config/codes.go new file mode 100644 index 000000000..951eff3ba --- /dev/null +++ b/pkg/provisioning/config/codes.go @@ -0,0 +1,39 @@ +// Copyright © 2026 Meroxa, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package config + +import ( + "github.com/conduitio/conduit/pkg/foundation/cerrors/conduiterr" + "google.golang.org/grpc/codes" +) + +// Pipeline-config validation error codes. Every validation error carries one of +// these codes plus a JSON-pointer configPath to the offending field, so an API, +// MCP, or UI consumer knows exactly what to fix and where. +var ( + CodeFieldRequired = conduiterr.Register("config.field_required", codes.InvalidArgument) + CodeFieldInvalid = conduiterr.Register("config.field_invalid", codes.InvalidArgument) + CodeFieldTooLong = conduiterr.Register("config.field_too_long", codes.InvalidArgument) + CodeIDDuplicate = conduiterr.Register("config.id_duplicate", codes.InvalidArgument) +) + +// fieldError builds a ConduitError carrying a machine-actionable code and the +// JSON-pointer path to the offending field. The sentinel is kept in the chain so +// existing errors.Is checks (ErrMandatoryField, ErrInvalidField, …) still hold. +func fieldError(code conduiterr.Code, configPath, msg string, sentinel error) error { + e := conduiterr.Wrap(code, msg, sentinel) + e.ConfigPath = configPath + return e +} diff --git a/pkg/provisioning/config/validate.go b/pkg/provisioning/config/validate.go index 661b22740..e790e9604 100644 --- a/pkg/provisioning/config/validate.go +++ b/pkg/provisioning/config/validate.go @@ -15,6 +15,8 @@ package config import ( + "fmt" + "github.com/conduitio/conduit/pkg/connector" "github.com/conduitio/conduit/pkg/foundation/cerrors" "github.com/conduitio/conduit/pkg/pipeline" @@ -27,84 +29,89 @@ const ( TypeDestination = "destination" ) -// Validate validates config field values for a pipeline +// Validate validates config field values for a pipeline. Each error carries a +// ConduitError code and a JSON-pointer configPath to the offending field; the +// original sentinel stays in the chain for backward-compatible errors.Is checks. func Validate(cfg Pipeline) error { var errs []error if cfg.ID == "" { - errs = append(errs, cerrors.Errorf(`id is mandatory: %w`, ErrMandatoryField)) + errs = append(errs, fieldError(CodeFieldRequired, "/id", "id is mandatory", ErrMandatoryField)) } if len(cfg.ID) > pipeline.IDLengthLimit { - errs = append(errs, pipeline.ErrIDOverLimit) + errs = append(errs, fieldError(CodeFieldTooLong, "/id", pipeline.ErrIDOverLimit.Error(), pipeline.ErrIDOverLimit)) } if len(cfg.Name) > pipeline.NameLengthLimit { - errs = append(errs, pipeline.ErrNameOverLimit) + errs = append(errs, fieldError(CodeFieldTooLong, "/name", pipeline.ErrNameOverLimit.Error(), pipeline.ErrNameOverLimit)) } if len(cfg.Description) > pipeline.DescriptionLengthLimit { - errs = append(errs, pipeline.ErrDescriptionOverLimit) + errs = append(errs, fieldError(CodeFieldTooLong, "/description", pipeline.ErrDescriptionOverLimit.Error(), pipeline.ErrDescriptionOverLimit)) } if cfg.Status != StatusRunning && cfg.Status != StatusStopped { - errs = append(errs, cerrors.Errorf(`"status" is invalid: %w`, ErrInvalidField)) + errs = append(errs, fieldError(CodeFieldInvalid, "/status", `"status" is invalid`, ErrInvalidField)) } errs = append(errs, validateConnectors(cfg.Connectors)...) - errs = append(errs, validateProcessors(cfg.Processors)...) + errs = append(errs, validateProcessors(cfg.Processors, "/processors")...) return cerrors.Join(errs...) } -// validateConnectors validates config field values for connectors +// validateConnectors validates config field values for connectors. pathPrefix is +// the JSON-pointer to the connectors slice ("/connectors"). func validateConnectors(mp []Connector) []error { var errs []error ids := make(map[string]bool) for i, cfg := range mp { + path := fmt.Sprintf("/connectors/%d", i) if cfg.ID == "" { - errs = append(errs, cerrors.Errorf("connector %d: id is mandatory: %w", i+1, ErrMandatoryField)) + errs = append(errs, fieldError(CodeFieldRequired, path+"/id", fmt.Sprintf("connector %d: id is mandatory", i+1), ErrMandatoryField)) } if len(cfg.ID) > connector.IDLengthLimit { - errs = append(errs, cerrors.Errorf("connector %q: %w", cfg.ID, connector.ErrIDOverLimit)) + errs = append(errs, fieldError(CodeFieldTooLong, path+"/id", fmt.Sprintf("connector %q: %s", cfg.ID, connector.ErrIDOverLimit), connector.ErrIDOverLimit)) } if len(cfg.Name) > connector.NameLengthLimit { - errs = append(errs, cerrors.Errorf("connector %q: %w", cfg.ID, connector.ErrNameOverLimit)) + errs = append(errs, fieldError(CodeFieldTooLong, path+"/name", fmt.Sprintf("connector %q: %s", cfg.ID, connector.ErrNameOverLimit), connector.ErrNameOverLimit)) } if cfg.Plugin == "" { - errs = append(errs, cerrors.Errorf(`connector %q: "plugin" is mandatory: %w`, cfg.ID, ErrMandatoryField)) + errs = append(errs, fieldError(CodeFieldRequired, path+"/plugin", fmt.Sprintf(`connector %q: "plugin" is mandatory`, cfg.ID), ErrMandatoryField)) } if cfg.Type == "" { - errs = append(errs, cerrors.Errorf(`connector %q: "type" is mandatory: %w`, cfg.ID, ErrMandatoryField)) + errs = append(errs, fieldError(CodeFieldRequired, path+"/type", fmt.Sprintf(`connector %q: "type" is mandatory`, cfg.ID), ErrMandatoryField)) } if cfg.Type != "" && cfg.Type != TypeSource && cfg.Type != TypeDestination { - errs = append(errs, cerrors.Errorf(`connector %q: "type" is invalid: %w`, cfg.ID, ErrInvalidField)) + errs = append(errs, fieldError(CodeFieldInvalid, path+"/type", fmt.Sprintf(`connector %q: "type" is invalid`, cfg.ID), ErrInvalidField)) } - pErrs := validateProcessors(cfg.Processors) - for _, pErr := range pErrs { - errs = append(errs, cerrors.Errorf("connector %q: %w", cfg.ID, pErr)) - } + // nested processor errors already carry their own /connectors/i/processors/j path + errs = append(errs, validateProcessors(cfg.Processors, path+"/processors")...) if ids[cfg.ID] { - errs = append(errs, cerrors.Errorf(`connector %q: "id" must be unique: %w`, cfg.ID, ErrDuplicateID)) + errs = append(errs, fieldError(CodeIDDuplicate, path+"/id", fmt.Sprintf(`connector %q: "id" must be unique`, cfg.ID), ErrDuplicateID)) } ids[cfg.ID] = true } return errs } -// validateProcessorsConfig validates config field values for processors -func validateProcessors(mp []Processor) []error { +// validateProcessors validates config field values for processors. pathPrefix is +// the JSON-pointer to the processors slice (e.g. "/processors" or +// "/connectors/0/processors"). +func validateProcessors(mp []Processor, pathPrefix string) []error { var errs []error ids := make(map[string]bool) for i, cfg := range mp { + path := fmt.Sprintf("%s/%d", pathPrefix, i) if cfg.ID == "" { - errs = append(errs, cerrors.Errorf("processor %d: id is mandatory: %w", i+1, ErrMandatoryField)) + errs = append(errs, fieldError(CodeFieldRequired, path+"/id", fmt.Sprintf("processor %d: id is mandatory", i+1), ErrMandatoryField)) } if cfg.Plugin == "" { - errs = append(errs, cerrors.Errorf(`processor %q: "plugin" is mandatory: %w`, cfg.ID, ErrMandatoryField)) + errs = append(errs, fieldError(CodeFieldRequired, path+"/plugin", fmt.Sprintf(`processor %q: "plugin" is mandatory`, cfg.ID), ErrMandatoryField)) } if cfg.Workers < 0 { - errs = append(errs, cerrors.Errorf(`processor %q: "workers" can't be negative: %w`, cfg.ID, ErrInvalidField)) + errs = append(errs, fieldError(CodeFieldInvalid, path+"/workers", fmt.Sprintf(`processor %q: "workers" can't be negative`, cfg.ID), ErrInvalidField)) } if ids[cfg.ID] { - errs = append(errs, cerrors.Errorf(`processor %q: "id" must be unique: %w`, cfg.ID, ErrDuplicateID)) + errs = append(errs, fieldError(CodeIDDuplicate, path+"/id", fmt.Sprintf(`processor %q: "id" must be unique`, cfg.ID), ErrDuplicateID)) } ids[cfg.ID] = true } diff --git a/pkg/provisioning/config/validate_test.go b/pkg/provisioning/config/validate_test.go index 40bcfb6e7..bf0cec7fd 100644 --- a/pkg/provisioning/config/validate_test.go +++ b/pkg/provisioning/config/validate_test.go @@ -20,10 +20,62 @@ import ( "github.com/conduitio/conduit/pkg/connector" "github.com/conduitio/conduit/pkg/foundation/cerrors" + "github.com/conduitio/conduit/pkg/foundation/cerrors/conduiterr" "github.com/conduitio/conduit/pkg/pipeline" "github.com/matryer/is" ) +// TestValidate_ConfigPathAndCode proves each validation error carries a +// machine-actionable code and a JSON-pointer to the offending field — the "failing +// config path" an agent or UI needs to point a user at exactly what to fix. +func TestValidate_ConfigPathAndCode(t *testing.T) { + testCases := []struct { + name string + cfg Pipeline + wantPath string + wantCode conduiterr.Code + }{ + { + name: "missing pipeline id", + cfg: Pipeline{Status: StatusRunning}, + wantPath: "/id", + wantCode: CodeFieldRequired, + }, + { + name: "invalid status", + cfg: Pipeline{ID: "pipeline1", Status: "bogus"}, + wantPath: "/status", + wantCode: CodeFieldInvalid, + }, + { + name: "connector missing plugin", + cfg: Pipeline{ID: "pipeline1", Status: StatusRunning, Connectors: []Connector{{ID: "con1", Type: TypeSource}}}, + wantPath: "/connectors/0/plugin", + wantCode: CodeFieldRequired, + }, + { + name: "nested processor missing id", + cfg: Pipeline{ID: "pipeline1", Status: StatusRunning, Connectors: []Connector{ + {ID: "con1", Plugin: "builtin:file", Type: TypeSource, Processors: []Processor{{Plugin: "field.set"}}}, + }}, + wantPath: "/connectors/0/processors/0/id", + wantCode: CodeFieldRequired, + }, + } + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + is := is.New(t) + err := Validate(tc.cfg) + is.True(err != nil) + + ce, ok := conduiterr.Get(err) + is.True(ok) // carries a ConduitError + is.Equal(ce.Code.Reason(), tc.wantCode.Reason()) + is.Equal(ce.ConfigPath, tc.wantPath) + }) + } +} + func TestValidator_MandatoryFields(t *testing.T) { tests := []struct { name string