pkg/acquisition: refact configuration validation and tests (#4187)

.gitattributes

@@ -0,0 +1,5 @@
+# Under windows, don't mess with line endings at least for yaml test files,
+# otherwise the reported line numbers don't match the expected errors.
+
+**/testdata/**/*.yaml  text eol=lf
+**/testdata/**/*.yml   text eol=lf

cmd/crowdsec-cli/clisetup/setup/detect_test.go

@@ -345,29 +345,6 @@ detect:
       datasource: {}`,
 			want:    nil,
 			wantErr: "invalid acquisition spec for wizard: datasource configuration is empty",
-		}, {
-			name: "missing acquisition file name",
-			config: `
-detect:
-  wizard:
-    acquisition_spec:
-      filename: something.yaml
-      datasource:
-        labels:
-          type: something`,
-			want:    nil,
-			wantErr: "invalid acquisition spec for wizard: source field is required",
-		}, {
-			name: "source is unknown",
-			config: `
-detect:
-  foobar:
-    acquisition_spec:
-      filename: wombat.yaml
-      datasource:
-        source: wombat`,
-			want:    nil,
-			wantErr: "invalid acquisition spec for foobar: unknown data source wombat",
 		}, {
 			name: "source is misplaced",
 			config: `
@@ -379,131 +356,6 @@ detect:
       source: file`,
 			want:    nil,
 			wantErr: "yaml: unmarshal errors:\n  line 7: field source not found in type setup.AcquisitionSpec",
-		}, {
-			name: "source is mismatched",
-			config: `
-detect:
-  foobar:
-    acquisition_spec:
-      filename: journalctl.yaml
-      datasource:
-        source: journalctl
-        filename: /path/to/file.log`,
-			want:    nil,
-			wantErr: `invalid acquisition spec for foobar: cannot parse: [1:1] unknown field "filename"`,
-		}, {
-			name: "source file: required fields",
-			config: `
-detect:
-  foobar:
-    acquisition_spec:
-      filename: file.yaml
-      datasource:
-        source: file`,
-			want:    nil,
-			wantErr: "invalid acquisition spec for foobar: no filename or filenames configuration provided",
-		}, {
-			name: "source journalctl: required fields",
-			config: `
-detect:
-  foobar:
-    acquisition_spec:
-      filename: foobar.yaml
-      datasource:
-        source: journalctl`,
-			want:    nil,
-			wantErr: "invalid acquisition spec for foobar: journalctl_filter is required",
-		}, {
-			name: "source cloudwatch: required fields",
-			config: `
-detect:
-  foobar:
-    acquisition_spec:
-      filename: cloudwatch.yaml
-      datasource:
-        source: cloudwatch`,
-			want:    nil,
-			wantErr: "invalid acquisition spec for foobar: group_name is mandatory for CloudwatchSource",
-		}, {
-			name: "source syslog: all fields are optional",
-			config: `
-detect:
-  foobar:
-    acquisition_spec:
-      filename: syslog.yaml
-      datasource:
-        source: syslog`,
-			want: &Setup{
-				Plans: []ServicePlan{
-					{
-						Name: "foobar",
-						InstallRecommendation: InstallRecommendation{
-							AcquisitionSpec: AcquisitionSpec{
-								Filename: "syslog.yaml",
-								Datasource: DatasourceConfig{
-									"source": "syslog",
-								},
-							},
-						},
-					},
-				},
-			},
-		}, {
-			name: "source docker: required fields",
-			config: `
-detect:
-  foobar:
-    acquisition_spec:
-      filename: docker.yaml
-      datasource:
-        source: docker`,
-			want:    nil,
-			wantErr: "invalid acquisition spec for foobar: no containers or services configuration provided",
-		}, {
-			name: "source kinesis: required fields (enhanced fanout=false)",
-			config: `
-detect:
-  foobar:
-    acquisition_spec:
-      filename: kinesis.yaml
-      datasource:
-        source: kinesis`,
-			want:    nil,
-			wantErr: "invalid acquisition spec for foobar: stream_name is mandatory when use_enhanced_fanout is false",
-		}, {
-			name: "source kinesis: required fields (enhanced fanout=true)",
-			config: `
-detect:
-  foobar:
-    acquisition_spec:
-      filename: kinesis.yaml
-      datasource:
-        source: kinesis
-        use_enhanced_fanout: true`,
-			want:    nil,
-			wantErr: "invalid acquisition spec for foobar: stream_arn is mandatory when use_enhanced_fanout is true",
-		}, {
-			name: "source kafka: required fields",
-			config: `
-detect:
-  foobar:
-    acquisition_spec:
-      filename: kafka.yaml
-      datasource:
-        source: kafka`,
-			want:    nil,
-			wantErr: "invalid acquisition spec for foobar: cannot create a kafka reader with an empty list of broker addresses",
-		}, {
-			name: "source loki: required fields",
-			config: `
-detect:
-  foobar:
-    acquisition_spec:
-      filename: loki.yaml
-      datasource:
-        source: loki`,
-			want:    nil,
-			wantErr: "invalid acquisition spec for foobar: loki query is mandatory",
 		},
 	}
 

go.mod

@@ -71,6 +71,7 @@ require (
 	github.com/prometheus/common v0.66.1
 	github.com/r3labs/diff/v2 v2.15.1
 	github.com/sanity-io/litter v1.5.8
+	github.com/santhosh-tekuri/jsonschema/v6 v6.0.2
 	github.com/segmentio/kafka-go v0.4.48
 	github.com/shirou/gopsutil/v4 v4.25.8
 	github.com/sirupsen/logrus v1.9.3

go.sum

@@ -148,6 +148,8 @@ github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5QvfrDyIgxBk=
 github.com/distribution/reference v0.6.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E=
+github.com/dlclark/regexp2 v1.11.0 h1:G/nrcoOa7ZXlpoa/91N3X7mM3r8eIlMBBJZvsz/mxKI=
+github.com/dlclark/regexp2 v1.11.0/go.mod h1:DHkYz0B9wPfa6wondMfaivmHpzrQ3v9q8cnmRbL6yW8=
 github.com/docker/go-connections v0.6.0 h1:LlMG9azAe1TqfR7sO+NJttz1gy6KO7VJBh+pMmjSD94=
 github.com/docker/go-connections v0.6.0/go.mod h1:AahvXYshr6JgfUJGdDCs2b5EZG/vmaMAntpSFH5BFKE=
 github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4=
@@ -508,6 +510,8 @@ github.com/russross/blackfriday/v2 v2.1.0 h1:JIOH55/0cWyOuilr9/qlrm0BSXldqnqwMsf
 github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
 github.com/sanity-io/litter v1.5.8 h1:uM/2lKrWdGbRXDrIq08Lh9XtVYoeGtcQxk9rtQ7+rYg=
 github.com/sanity-io/litter v1.5.8/go.mod h1:9gzJgR2i4ZpjZHsKvUXIRQVk7P+yM3e+jAF7bU2UI5U=
+github.com/santhosh-tekuri/jsonschema/v6 v6.0.2 h1:KRzFb2m7YtdldCEkzs6KqmJw4nqEVZGK7IN2kJkjTuQ=
+github.com/santhosh-tekuri/jsonschema/v6 v6.0.2/go.mod h1:JXeL+ps8p7/KNMjDQk3TCwPpBy0wYklyWTfbkIzdIFU=
 github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=
 github.com/segmentio/kafka-go v0.4.48 h1:9jyu9CWK4W5W+SroCe8EffbrRZVqAOkuaLd/ApID4Vs=
 github.com/segmentio/kafka-go v0.4.48/go.mod h1:HjF6XbOKh0Pjlkr5GVZxt6CsjjwnmhVOfURM5KMd8qg=

pkg/acquisition/acquisition.go

@@ -10,6 +10,7 @@ import (
 	"os"
 	"time"
 	"slices"
+	"strconv"
 	"strings"
 
 	"github.com/cenkalti/backoff/v5"
@@ -89,6 +90,9 @@ func DataSourceConfigure(
 
 	if lapiClientAware, ok := dataSrc.(types.LAPIClientAware); ok {
 		cConfig := csconfig.GetConfig()
+		if cConfig.API == nil {
+			return nil, errors.New("crowdsec configuration not loaded while initializing appsec - this is a bug, plese report")
+		}
 		lapiClientAware.SetClientConfig(cConfig.API.Client)
 	}
 
@@ -146,7 +150,7 @@ func LoadAcquisitionFromDSN(
 	subLogger := log.StandardLogger().WithField("type", labels["type"])
 
 	if err = dsnConf.ConfigureByDSN(ctx, dsn, labels, subLogger, uniqueID); err != nil {
-		return nil, fmt.Errorf("configuring datasource for %q: %w", dsn, err)
+		return nil, fmt.Errorf("datasource for %q: %w", dsn, err)
 	}
 
 	return dataSrc, nil
@@ -202,6 +206,115 @@ func detectType(r io.Reader) (string, error) {
 	}
 }
 
+type ParsedSourceConfig struct {
+	Common    configuration.DataSourceCommonCfg
+	Source    types.DataSource
+	Transform *vm.Program
+	SourceMissing bool      // the "source" field was missing, and detected
+	SourceOverridden string // the "source" field was not missing, but didn't match the detected one
+}
+
+var ErrEmptyYAMLDocument = errors.New("empty yaml document")
+
+// ParseSourceConfig validates and configures one YAML document.
+//
+// It does not expand env variables, they must already be expanded.
+//
+// - return sentinel error for empty/comment-only documents
+// - backward-compat source auto-detection (filename/filenames/journalctl_filter)
+// - validate common fields
+// - delegate per-source config validation to the appropriate module
+// - compile transform expression
+func ParseSourceConfig(ctx context.Context, yamlDoc []byte, metricsLevel metrics.AcquisitionMetricsLevel, hub *cwhub.Hub) (*ParsedSourceConfig, error) {
+	detectedType, err := detectType(bytes.NewReader(yamlDoc))
+	if err != nil {
+		return nil, err
+	}
+
+	// if there are not keys or only comments, the document will be skipped
+	empty, err := csyaml.IsEmptyYAML(bytes.NewReader(yamlDoc))
+	if err != nil {
+		return nil, err
+	}
+
+	if empty {
+		return nil, ErrEmptyYAMLDocument
+	}
+
+	var sub configuration.DataSourceCommonCfg
+
+	// can't be strict here, the doc contains specific datasource config too but we won't collect them now.
+	if err = yaml.UnmarshalWithOptions(yamlDoc, &sub); err != nil {
+		return nil, fmt.Errorf("failed to parse: %w", errors.New(yaml.FormatError(err, false, false)))
+	}
+
+	parsed := &ParsedSourceConfig{}
+
+	// report that the user did not specify a source
+	if sub.Source == "" {
+		parsed.SourceMissing = true
+	}
+
+	// report that the user specified a source that doesn't match with one detected from the presence of other fields
+	if detectedType != "" {
+		if sub.Source != "" && sub.Source != detectedType {
+			parsed.SourceOverridden = sub.Source
+		}
+
+		sub.Source = detectedType
+	}
+
+	parsed.Common = sub
+
+	// could not detect, alas
+	if sub.Source == "" {
+		return nil, errors.New("missing 'source' field")
+	}
+
+	// pre-check that the source is valid
+	_, err = registry.LookupFactory(sub.Source)
+	if err != nil {
+		return nil, err
+	}
+
+	// check for labels now, an error for missing labels has lower priority
+	// than missing or unknown source type
+	if len(sub.Labels) == 0 && sub.Source != "docker" {
+		// docker is the only source that does not require labels
+		return nil, errors.New("missing labels")
+	}
+
+	uniqueID := uuid.NewString()
+	sub.UniqueId = uniqueID
+
+	src, err := DataSourceConfigure(ctx, sub, yamlDoc, metricsLevel, hub)
+	if err != nil {
+		return nil, fmt.Errorf("datasource of type %s: %w", sub.Source, err)
+	}
+	parsed.Source = src
+
+	if sub.TransformExpr != "" {
+		vm, err := expr.Compile(sub.TransformExpr, exprhelpers.GetExprOptions(map[string]any{"evt": &pipeline.Event{}})...)
+		if err != nil {
+			return nil, fmt.Errorf("while compiling transform expression '%s' for datasource %s: %w", sub.TransformExpr, sub.Source, err)
+		}
+ 
+		parsed.Transform = vm
+	}
+
+	return parsed, nil
+}
+
+func formatConfigLocation(acquisFile string, withPos bool, idx int) string {
+	ret := acquisFile
+
+	if withPos {
+		ret += " (position " + strconv.Itoa(idx) + ")"
+	}
+
+	return ret
+}
+
 // sourcesFromFile reads and parses one acquisition file into DataSources.
 func sourcesFromFile(
 	ctx context.Context,
@@ -235,84 +348,42 @@ func sourcesFromFile(
 	idx := -1
 
 	for _, yamlDoc := range documents {
-		detectedType, err := detectType(bytes.NewReader(yamlDoc))
-		if err != nil {
-			return nil, fmt.Errorf("failed to parse %s: %w", yamlFile.Name(), err)
-		}
-
 		idx += 1
 
-		var sub configuration.DataSourceCommonCfg
+		loc := formatConfigLocation(acquisFile, len(documents) > 1, idx)
 
-		// can't be strict here, the doc contains specific datasource config too but we won't collect them now.
-		if err = yaml.UnmarshalWithOptions(yamlDoc, &sub); err != nil {
-			return nil, fmt.Errorf("failed to parse %s: %w", yamlFile.Name(), errors.New(yaml.FormatError(err, false, false)))
-		}
+		parsed, err := ParseSourceConfig(ctx, yamlDoc, metricsLevel, hub)
 
-		// for backward compat ('type' was not mandatory, detect it)
-		if guessType := detectedType; guessType != "" {
-			log.Debugf("datasource type missing in %s (position %d): detected 'source=%s'", acquisFile, idx, guessType)
-
-			if sub.Source != "" && sub.Source != guessType {
-				log.Warnf("datasource type mismatch in %s (position %d): found '%s' but should probably be '%s'", acquisFile, idx, sub.Source, guessType)
+		// report data source detection, it can be required to understand an error
+		if parsed != nil {
+			if parsed.SourceMissing {
+				log.Debugf("%s: datasource type missing, detected 'source=%s'", loc, parsed.Common.Source)
 			}
 
-			sub.Source = guessType
-		}
-
-		// it's an empty item, skip it
-
-		empty, err := csyaml.IsEmptyYAML(bytes.NewReader(yamlDoc))
-		if err != nil {
-			return nil, fmt.Errorf("failed to parse %s (position %d): %w", acquisFile, idx, err)
-		}
-
-		if empty {
-			// there are no keys or only comments, skip the document
-			continue
-		}
-
-		if len(sub.Labels) == 0 {
-			if sub.Source != "docker" {
-				// docker is the only source that can be empty
-				return nil, fmt.Errorf("missing labels in %s (position %d)", acquisFile, idx)
+			if parsed.SourceOverridden != "" {
+				log.Warnf("%s: datasource type mismatch: found '%s' but should probably be '%s'", loc, parsed.SourceOverridden, parsed.Common.Source)
 			}
 		}
 
-		if sub.Source == "" {
-			return nil, fmt.Errorf("missing 'source' field in %s (position %d)", acquisFile, idx)
-		}
-
-		// pre-check that the source is valid
-		_, err = registry.LookupFactory(sub.Source)
 		if err != nil {
-			return nil, fmt.Errorf("in file %s (position %d) - %w", acquisFile, idx, err)
-		}
-
-		uniqueID := uuid.NewString()
-		sub.UniqueId = uniqueID
-
-		src, err := DataSourceConfigure(ctx, sub, yamlDoc, metricsLevel, hub)
-		if err != nil {
-			var dserr *DataSourceUnavailableError
-			if errors.As(err, &dserr) {
-				log.Error(err)
+			if errors.Is(err, ErrEmptyYAMLDocument) {
 				continue
 			}
 
-			return nil, fmt.Errorf("configuring datasource of type %s from %s (position %d): %w", sub.Source, acquisFile, idx, err)
-		}
-
-		if sub.TransformExpr != "" {
-			vm, err := expr.Compile(sub.TransformExpr, exprhelpers.GetExprOptions(map[string]any{"evt": &pipeline.Event{}})...)
-			if err != nil {
-				return nil, fmt.Errorf("while compiling transform expression '%s' for datasource %s in %s (position %d): %w", sub.TransformExpr, sub.Source, acquisFile, idx, err)
+			var dserr *DataSourceUnavailableError
+			if errors.As(err, &dserr) {
+				log.Error(fmt.Errorf("%s: %w", loc, err))
+				continue
 			}
 
-			transformRuntimes[uniqueID] = vm
+			return nil, fmt.Errorf("%s: %w", loc, err)
 		}
 
-		sources = append(sources, src)
+		if parsed.Transform != nil {
+			transformRuntimes[parsed.Common.UniqueId] = parsed.Transform
+		}
+
+		sources = append(sources, parsed.Source)
 	}
 
 	return sources, nil

pkg/acquisition/acquisition_test.go

@@ -268,7 +268,7 @@ func TestLoadAcquisitionFromFiles(t *testing.T) {
 			Config: csconfig.CrowdsecServiceCfg{
 				AcquisitionFiles: []string{"testdata/missing_labels.yaml"},
 			},
-			ExpectedError: "missing labels in testdata/missing_labels.yaml",
+			ExpectedError: "testdata/missing_labels.yaml: missing labels",
 		},
 		{
 			TestName: "backward_compat",
@@ -282,14 +282,14 @@ func TestLoadAcquisitionFromFiles(t *testing.T) {
 			Config: csconfig.CrowdsecServiceCfg{
 				AcquisitionFiles: []string{"testdata/bad_source.yaml"},
 			},
-			ExpectedError: "in file testdata/bad_source.yaml (position 0) - unknown data source does_not_exist",
+			ExpectedError: "testdata/bad_source.yaml: unknown data source does_not_exist",
 		},
 		{
 			TestName: "invalid_filetype_config",
 			Config: csconfig.CrowdsecServiceCfg{
 				AcquisitionFiles: []string{"testdata/bad_filetype.yaml"},
 			},
-			ExpectedError: "configuring datasource of type file from testdata/bad_filetype.yaml",
+			ExpectedError: "testdata/bad_filetype.yaml: datasource of type file: cannot parse FileAcquisition configuration: [2:12] string was used where sequence is expected",
 		},
 		{
 			TestName: "from_env",

pkg/acquisition/config_test.go

@@ -0,0 +1,185 @@
+package acquisition
+
+import (
+	"errors"
+	"io/fs"
+	"os"
+	"path/filepath"
+	"regexp"
+	"runtime"
+	"slices"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/goccy/go-yaml"
+
+	"github.com/crowdsecurity/crowdsec/pkg/csconfig"
+	"github.com/crowdsecurity/crowdsec/pkg/cwhub"
+	"github.com/crowdsecurity/crowdsec/pkg/metrics"
+)
+
+var (
+	wantErrLineRE = regexp.MustCompile(`(?m)^\s*#\s*wantErr:\s*(.*?)\s*$`)
+	wantSchemaErrLineRE = regexp.MustCompile(`(?m)^[ \t]*#[ \t]*schemaErr:[ \t]*([^\r\n]*)[ \t]*$`)
+)
+
+func findYAMLFiles(t *testing.T, root string) []string {
+	t.Helper()
+
+	var files []string
+	err := filepath.WalkDir(root, func(path string, d os.DirEntry, err error) error {
+		if err != nil {
+			return err
+		}
+		if d.IsDir() {
+			return nil
+		}
+		ext := strings.ToLower(filepath.Ext(d.Name()))
+		if ext != ".yaml" && ext != ".yml" {
+			return nil
+		}
+		files = append(files, path)
+		return nil
+	})
+	require.NoError(t, err, "walking %q", root)
+
+	slices.Sort(files)
+	return files
+}
+
+func wantErrFromYAML(t *testing.T, fileContent []byte) (want string, found bool) {
+	t.Helper()
+
+	m := wantErrLineRE.FindSubmatch(fileContent)
+	if len(m) == 0 {
+		return "", false
+	}
+
+	return strings.TrimSpace(string(m[1])), true
+}
+
+func wantSchemaErrFromYAML(t *testing.T, fileContent []byte) (want string, found bool) {
+	t.Helper()
+
+	m := wantSchemaErrLineRE.FindSubmatch(fileContent)
+	if len(m) == 0 {
+		return "", false
+	}
+
+	return strings.TrimSpace(string(m[1])), true
+}
+
+func TestParseSourceConfig(t *testing.T) {
+	ctx := t.Context()
+
+	type suite struct {
+		name        string
+		root        string
+		expectValid bool
+	}
+
+	type source struct {
+		Source string
+	}
+
+	// load a configuration, appsec needs it
+	_, _, err := csconfig.NewConfig("./testdata/config.yaml", false, false, true)
+	require.NoError(t, err)
+
+	// load a hub, appsec needs it
+	hub := cwhub.Hub{}
+
+	suites := []suite{
+		{name: "valid", root: filepath.Join("testdata", "valid"), expectValid: true},
+		{name: "invalid", root: filepath.Join("testdata", "invalid"), expectValid: false},
+	}
+
+	for _, s := range suites {
+		t.Run(s.name, func(t *testing.T) {
+			files := findYAMLFiles(t, s.root)
+			require.NotEmpty(t, files, "no YAML files found under %q", s.root)
+
+			for _, path := range files {
+				rel, _ := filepath.Rel(s.root, path)
+
+				fileContent, err := os.ReadFile(path)
+				require.NoError(t, err, "read %q", path)
+
+				t.Run(filepath.ToSlash(rel), func(t *testing.T) {
+					var (
+						so source
+						schema string
+					)
+
+					if err = yaml.Unmarshal(fileContent, &so); err == nil {
+						schema = filepath.Join("schemas", so.Source + ".yaml")
+					}
+
+					if runtime.GOOS == "windows" && strings.Contains(path, "journalctl") {
+						return
+					}
+
+					if runtime.GOOS != "windows" && strings.Contains(path, "wineventlog") {
+						return
+					}
+
+					wantErr, hasWant := wantErrFromYAML(t, fileContent)
+
+					wantSchemaErr, hasWantSchemaErr := wantSchemaErrFromYAML(t, fileContent)
+
+					if s.expectValid {
+						require.False(t, hasWant, "valid config must not include # wantErr: directive")
+						parsed, err := ParseSourceConfig(ctx, fileContent, metrics.AcquisitionMetricsLevelNone, &hub)
+						require.NoError(t, err)
+						require.NotNil(t, parsed)
+						if schema != "" {
+							err = ValidateYAML(fileContent, schema)
+							if !errors.Is(err, fs.ErrNotExist) {
+								// XXX: ignore missing schema
+								require.NoError(t, err)
+							}
+						}
+						return
+					}
+
+					// invalid
+
+					require.True(t, hasWant, "invalid config must include '# wantErr: <exact error>'")
+					require.NotEmpty(t, wantErr, "wantErr directive found but empty")
+
+					parsed, err := ParseSourceConfig(ctx, fileContent, metrics.AcquisitionMetricsLevelNone, &hub)
+					require.Error(t, err, "got no error, expected %q", wantErr)
+					require.Nil(t, parsed)
+					assert.Equal(t, wantErr, err.Error())
+					if schema == "" {
+						return
+					}
+
+					// schema validation
+
+					err = ValidateYAML(fileContent, schema)
+					if errors.Is(err, fs.ErrNotExist) {
+						// XXX: ignore missing schema, for now
+						return
+					}
+
+					// a "schemaErr" comment must be present, even if empty
+					require.True(t, hasWantSchemaErr, "invalid configurations require an exlicit schemaErr comment. it can be empty string if the schema cannot detect the issue")
+					switch {
+					case err == nil && wantSchemaErr != "":
+						require.Error(t, err, "got no schema error, expected %q", wantSchemaErr)
+					case err != nil && wantSchemaErr == "":
+						require.Error(t, err, "got schema error %q, expected nil", err)
+					case err != nil:
+						assert.Contains(t, err.Error(), wantSchemaErr)
+					default:
+						require.NoError(t, err)
+						assert.Empty(t, wantSchemaErr)
+					}
+				})
+			}
+		})
+	}
+}

pkg/acquisition/modules/appsec/config.go

@@ -128,6 +128,10 @@ func (w *Source) Configure(_ context.Context, yamlConfig []byte, logger *log.Ent
 		return fmt.Errorf("unable to parse appsec configuration: %w", err)
 	}
 
+	if w.lapiClientConfig.Credentials == nil {
+		return errors.New("missing lapi client credentials")
+	}
+
 	w.lapiURL = fmt.Sprintf("%sv1/decisions/stream", w.lapiClientConfig.Credentials.URL)
 	w.AuthCache = NewAuthCache()
 

pkg/acquisition/modules/cloudwatch/cloudwatch_test.go

@@ -422,25 +422,6 @@ group_name: test_group
 stream_name: test_stream`,
 			expectedStartErr: "The specified log group does not exist",
 		},
-		{
-			config: `
-xxx: cloudwatch
-labels:
-  type: test_source
-group_name: test_group
-stream_name: test_stream`,
-			expectedCfgErr: `[2:1] unknown field "xxx"`,
-		},
-		{
-			name: "missing_group_name",
-			config: `
-source: cloudwatch
-aws_region: us-east-1
-labels:
-  type: test_source
-stream_name: test_stream`,
-			expectedCfgErr: "group_name is mandatory for CloudwatchSource",
-		},
 	}
 
 	for _, tc := range tests {

pkg/acquisition/modules/docker/docker_test.go

@@ -30,113 +30,6 @@ import (
 const testContainerName = "docker_test"
 const testServiceName = "test_service"
 
-func TestConfigure(t *testing.T) {
-	log.Infof("Test 'TestConfigure'")
-
-	ctx := t.Context()
-
-	tests := []struct {
-		config      string
-		expectedErr string
-	}{
-		{
-			config:      `foobar: asd`,
-			expectedErr: `while parsing DockerAcquisition configuration: [1:1] unknown field "foobar"`,
-		},
-		{
-			config: `
-mode: tail
-source: docker`,
-			expectedErr: "no containers or services configuration provided",
-		},
-		{
-			config: `
-mode: cat
-source: docker
-container_name:
- - toto`,
-			expectedErr: "",
-		},
-		{
-			config: `
-mode: cat
-source: docker
-check_interval: 10s
-container_name:
- - toto`,
-			expectedErr: "",
-		},
-		{
-			config: `
-mode: cat
-source: docker
-service_name:
- - web-service`,
-			expectedErr: "",
-		},
-		{
-			config: `
-mode: cat
-source: docker
-use_container_labels: true
-container_name:
- - toto`,
-			expectedErr: "use_container_labels and container_name, container_id, container_id_regexp, container_name_regexp are mutually exclusive",
-		},
-		{
-			config: `
-mode: cat
-source: docker
-use_service_labels: true
-service_name:
- - web-service`,
-			expectedErr: "use_service_labels and service_name, service_id, service_id_regexp, service_name_regexp are mutually exclusive",
-		},
-		{
-			config: `
-mode: cat
-source: docker
-container_name_regexp:
- - "[invalid"`,
-			expectedErr: "container_name_regexp: error parsing regexp: missing closing ]: `[invalid`",
-		},
-		{
-			config: `
-mode: cat
-source: docker
-container_id_regexp:
- - "*invalid"`,
-			expectedErr: "container_id_regexp: error parsing regexp: missing argument to repetition operator: `*`",
-		},
-		{
-			config: `
-mode: cat
-source: docker
-service_name_regexp:
- - "(?P<invalid"`,
-			expectedErr: "service_name_regexp: error parsing regexp: invalid named capture: `(?P<invalid`",
-		},
-		{
-			config: `
-mode: cat
-source: docker
-service_id_regexp:
- - "+invalid"`,
-			expectedErr: "service_id_regexp: error parsing regexp: missing argument to repetition operator: `+`",
-		},
-	}
-
-	subLogger := log.WithField("type", ModuleName)
-
-	for _, tc := range tests {
-		t.Run(tc.config, func(t *testing.T) {
-			f := Source{}
-			err := f.Configure(ctx, []byte(tc.config), subLogger, metrics.AcquisitionMetricsLevelNone)
-			cstest.RequireErrorContains(t, err, tc.expectedErr)
-		})
-	}
-}
-
 func TestConfigureDSN(t *testing.T) {
 	log.Infof("Test 'TestConfigureDSN'")
 

pkg/acquisition/modules/file/file_test.go

@@ -21,54 +21,6 @@ import (
 	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
 )
 
-func TestBadConfiguration(t *testing.T) {
-	ctx := t.Context()
-
-	tests := []struct {
-		name        string
-		config      string
-		expectedErr string
-	}{
-		{
-			name:        "extra configuration key",
-			config:      "foobar: asd.log",
-			expectedErr: `cannot parse FileAcquisition configuration: [1:1] unknown field "foobar"`,
-		},
-		{
-			name:        "missing filenames",
-			config:      "mode: tail",
-			expectedErr: "no filename or filenames configuration provided",
-		},
-		{
-			name:        "glob syntax error",
-			config:      `filename: "[asd-.log"`,
-			expectedErr: "glob failure: syntax error in pattern",
-		},
-		{
-			name: "bad exclude regexp",
-			config: `filenames: ["asd.log"]
-exclude_regexps: ["as[a-$d"]`,
-			expectedErr: "could not compile regexp as",
-		},
-		{
-			name: "duplicate keys",
-			config: `filenames: ["asd.log"]
-filenames: ["ase.log"]`,
-			expectedErr: `cannot parse FileAcquisition configuration: [2:1] mapping key "filenames" already defined at [1:1]`,
-		},
-	}
-
-	subLogger := log.WithField("type", fileacquisition.ModuleName)
-
-	for _, tc := range tests {
-		t.Run(tc.name, func(t *testing.T) {
-			f := fileacquisition.Source{}
-			err := f.Configure(ctx, []byte(tc.config), subLogger, metrics.AcquisitionMetricsLevelNone)
-			cstest.RequireErrorContains(t, err, tc.expectedErr)
-		})
-	}
-}
-
 func TestConfigureDSN(t *testing.T) {
 	ctx := t.Context()
 
@@ -505,57 +457,6 @@ exclude_regexps: ["\\.gz$"]`
 	hook.Reset()
 }
 
-func TestDiscoveryPollConfiguration(t *testing.T) {
-	ctx := t.Context()
-
-	tests := []struct {
-		name    string
-		config  string
-		wantErr string
-	}{
-		{
-			name: "valid discovery poll config",
-			config: `
-filenames:
- - "tests/test.log"
-discovery_poll_enable: true
-discovery_poll_interval: "30s"
-mode: tail
-`,
-			wantErr: "",
-		},
-		{
-			name: "invalid poll interval",
-			config: `
-filenames:
- - "tests/test.log"
-discovery_poll_enable: true
-discovery_poll_interval: "invalid"
-mode: tail
-`,
-			wantErr: `cannot parse FileAcquisition configuration: time: invalid duration "invalid"`,
-		},
-		{
-			name: "polling disabled",
-			config: `
-filenames:
- - "tests/test.log"
-discovery_poll_enable: false
-mode: tail
-`,
-			wantErr: "",
-		},
-	}
-
-	for _, tc := range tests {
-		t.Run(tc.name, func(t *testing.T) {
-			f := &fileacquisition.Source{}
-			err := f.Configure(ctx, []byte(tc.config), log.NewEntry(log.New()), metrics.AcquisitionMetricsLevelNone)
-			cstest.RequireErrorContains(t, err, tc.wantErr)
-		})
-	}
-}
-
 func TestDiscoveryPolling(t *testing.T) {
 	ctx := t.Context()
 	dir := t.TempDir()

pkg/acquisition/modules/http/http_test.go

@@ -22,8 +22,6 @@ import (
 	"github.com/stretchr/testify/require"
 	"gopkg.in/tomb.v2"
 
-	"github.com/crowdsecurity/go-cs-lib/cstest"
-
 	"github.com/crowdsecurity/crowdsec/pkg/metrics"
 	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
 )
@@ -33,173 +31,12 @@ const (
 	testHTTPServerAddrTLS = "https://127.0.0.1:8080"
 )
 
-func TestConfigure(t *testing.T) {
-	ctx := t.Context()
-
-	tests := []struct {
-		config      string
-		expectedErr string
-	}{
-		{
-			config: `
-timeout: 1m`,
-			expectedErr: "listen_addr or listen_socket is required",
-		},
-		{
-			config: `
-source: http
-listen_addr: 127.0.0.1:8080
-path: wrongpath`,
-			expectedErr: "path must start with /",
-		},
-		{
-			config: `
-source: http
-listen_addr: 127.0.0.1:8080
-path: /test
-auth_type: basic_auth`,
-			expectedErr: "basic_auth is selected, but basic_auth is not provided",
-		},
-		{
-			config: `
-source: http
-listen_addr: 127.0.0.1:8080
-path: /test
-auth_type: headers`,
-			expectedErr: "headers is selected, but headers is not provided",
-		},
-		{
-			config: `
-source: http
-listen_addr: 127.0.0.1:8080
-path: /test
-auth_type: basic_auth
-basic_auth:
-  username: 132`,
-			expectedErr: "basic_auth is selected, but password is not provided",
-		},
-		{
-			config: `
-source: http
-listen_addr: 127.0.0.1:8080
-path: /test
-auth_type: basic_auth
-basic_auth:
-  password: 132`,
-			expectedErr: "basic_auth is selected, but username is not provided",
-		},
-		{
-			config: `
-source: http
-listen_addr: 127.0.0.1:8080
-path: /test
-auth_type: headers
-headers:`,
-			expectedErr: "headers is selected, but headers is not provided",
-		},
-		{
-			config: `
-source: http
-listen_addr: 127.0.0.1:8080
-path: /test
-auth_type: toto`,
-			expectedErr: "invalid auth_type: must be one of basic_auth, headers, mtls",
-		},
-		{
-			config: `
-source: http
-listen_addr: 127.0.0.1:8080
-path: /test
-auth_type: headers
-headers:
-  key: value
-tls:
-  server_key: key`,
-			expectedErr: "server_cert is required",
-		},
-		{
-			config: `
-source: http
-listen_addr: 127.0.0.1:8080
-path: /test
-auth_type: headers
-headers:
-  key: value
-tls:
-  server_cert: cert`,
-			expectedErr: "server_key is required",
-		},
-		{
-			config: `
-source: http
-listen_addr: 127.0.0.1:8080
-path: /test
-auth_type: mtls
-tls:
-  server_cert: cert
-  server_key: key`,
-			expectedErr: "mtls is selected, but ca_cert is not provided",
-		},
-		{
-			config: `
-source: http
-listen_addr: 127.0.0.1:8080
-path: /test
-auth_type: headers
-headers:
-  key: value
-max_body_size: 0`,
-			expectedErr: "max_body_size must be positive",
-		},
-		{
-			config: `
-source: http
-listen_addr: 127.0.0.1:8080
-path: /test
-auth_type: headers
-headers:
-  key: value
-timeout: toto`,
-			expectedErr: `cannot parse: time: invalid duration "toto"`,
-		},
-		{
-			config: `
-source: http
-listen_addr: 127.0.0.1:8080
-path: /test
-auth_type: headers
-headers:
-  key: value
-custom_status_code: 999`,
-			expectedErr: "invalid HTTP status code",
-		},
-	}
-
-	subLogger := log.WithField("type", ModuleName)
-
-	for _, test := range tests {
-		h := Source{}
-		err := h.Configure(ctx, []byte(test.config), subLogger, 0)
-		cstest.AssertErrorContains(t, err, test.expectedErr)
-	}
-}
-
 func TestGetUuid(t *testing.T) {
 	h := Source{}
 	h.Config.UniqueId = "test"
 	assert.Equal(t, "test", h.GetUuid())
 }
 
-func TestUnmarshalConfig(t *testing.T) {
-	h := Source{}
-	err := h.UnmarshalConfig([]byte(`
-source: http
-listen_addr: 127.0.0.1:8080
-path: 15
-	auth_type: headers`))
-	cstest.AssertErrorMessage(t, err, "cannot parse: [5:1] found character '\t' that cannot start any token")
-}
-
 func TestGetMode(t *testing.T) {
 	h := Source{}
 	h.Config.Mode = "test"
@@ -213,7 +50,9 @@ func TestGetName(t *testing.T) {
 
 func SetupAndRunHTTPSource(t *testing.T, h *Source, config []byte, metricLevel metrics.AcquisitionMetricsLevel) (chan pipeline.Event, *prometheus.Registry, *tomb.Tomb) {
 	ctx := t.Context()
-	subLogger := log.WithField("type", ModuleName)
+	subLogger := log.WithFields(log.Fields{
+		"type": ModuleName,
+	})
 	err := h.Configure(ctx, config, subLogger, metricLevel)
 	require.NoError(t, err)
 

pkg/acquisition/modules/journalctl/journalctl_test.go

@@ -21,44 +21,6 @@ import (
 	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
 )
 
-func TestBadConfiguration(t *testing.T) {
-	cstest.SkipOnWindows(t)
-
-	ctx := t.Context()
-
-	tests := []struct {
-		config  string
-		wantErr string
-	}{
-		{
-			config:  `foobar: asd.log`,
-			wantErr: `cannot parse: [1:1] unknown field "foobar"`,
-		},
-		{
-			config: `
-mode: tail
-source: journalctl`,
-			wantErr: "journalctl_filter is required",
-		},
-		{
-			config: `
-mode: cat
-source: journalctl
-journalctl_filter:
- - _UID=42`,
-		},
-	}
-
-	for _, tc := range tests {
-		t.Run(tc.config, func(t *testing.T) {
-			f := Source{}
-			logger, _ := logtest.NewNullLogger()
-			err := f.Configure(ctx, []byte(tc.config), logrus.NewEntry(logger), metrics.AcquisitionMetricsLevelNone)
-			cstest.RequireErrorContains(t, err, tc.wantErr)
-		})
-	}
-}
-
 func TestConfigureDSN(t *testing.T) {
 	cstest.SkipOnWindows(t)
 

pkg/acquisition/modules/kafka/config.go

@@ -49,7 +49,7 @@ func (s *Source) UnmarshalConfig(yamlConfig []byte) error {
 
 	err := yaml.UnmarshalWithOptions(yamlConfig, &s.Config, yaml.Strict())
 	if err != nil {
-		return fmt.Errorf("cannot parse %s datasource configuration: %s", s.GetName(), yaml.FormatError(err, false, false))
+		return fmt.Errorf("cannot parse: %s", yaml.FormatError(err, false, false))
 	}
 
 	if len(s.Config.Brokers) == 0 {
@@ -57,7 +57,7 @@ func (s *Source) UnmarshalConfig(yamlConfig []byte) error {
 	}
 
 	if s.Config.Topic == "" {
-		return fmt.Errorf("cannot create a %s reader with am empty topic", s.GetName())
+		return fmt.Errorf("cannot create a %s reader with an empty topic", s.GetName())
 	}
 
 	if s.Config.Mode == "" {

pkg/acquisition/modules/kafka/kafka_test.go

@@ -18,69 +18,6 @@ import (
 	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
 )
 
-func TestConfigure(t *testing.T) {
-	ctx := t.Context()
-
-	tests := []struct {
-		config      string
-		expectedErr string
-	}{
-		{
-			config: `
-foobar: bla
-source: kafka`,
-			expectedErr: `[2:1] unknown field "foobar"`,
-		},
-		{
-			config:      `source: kafka`,
-			expectedErr: "cannot create a kafka reader with an empty list of broker addresses",
-		},
-		{
-			config: `
-source: kafka
-brokers:
-  - bla
-timeout: 5`,
-			expectedErr: "cannot create a kafka reader with am empty topic",
-		},
-		{
-			config: `
-source: kafka
-brokers:
-  - bla
-topic: toto
-timeout: aa`,
-			expectedErr: "cannot create kafka dialer: strconv.Atoi: parsing \"aa\": invalid syntax",
-		},
-		{
-			config: `
-source: kafka
-brokers:
-  - localhost:9092
-topic: crowdsec`,
-			expectedErr: "",
-		},
-		{
-			config: `
-source: kafka
-brokers:
-  - localhost:9092
-topic: crowdsec
-partition: 1
-group_id: crowdsec`,
-			expectedErr: "cannote create kafka reader: cannot specify both group_id and partition",
-		},
-	}
-
-	subLogger := log.WithField("type", ModuleName)
-
-	for _, test := range tests {
-		k := Source{}
-		err := k.Configure(ctx, []byte(test.config), subLogger, metrics.AcquisitionMetricsLevelNone)
-		cstest.AssertErrorContains(t, err, test.expectedErr)
-	}
-}
-
 func writeToKafka(ctx context.Context, w *kafka.Writer, logs []string) {
 	for idx, log := range logs {
 		err := w.WriteMessages(ctx, kafka.Message{

pkg/acquisition/modules/kinesis/kinesis_test.go

@@ -98,58 +98,6 @@ func TestMain(m *testing.M) {
 	os.Exit(code)
 }
 
-func TestBadConfiguration(t *testing.T) {
-	cstest.SkipOnWindows(t)
-	
-	ctx := t.Context()
-
-	tests := []struct {
-		config      string
-		expectedErr string
-	}{
-		{
-			config:      `source: kinesis`,
-			expectedErr: "stream_name is mandatory when use_enhanced_fanout is false",
-		},
-		{
-			config: `
-source: kinesis
-max_retries: whatev`,
-			expectedErr: "[3:14] cannot unmarshal string into Go struct field Configuration.MaxRetries of type int",
-		},
-		{
-			config: `
-source: kinesis
-use_enhanced_fanout: true`,
-			expectedErr: "stream_arn is mandatory when use_enhanced_fanout is true",
-		},
-		{
-			config: `
-source: kinesis
-use_enhanced_fanout: true
-stream_arn: arn:aws:kinesis:eu-west-1:123456789012:stream/my-stream`,
-			expectedErr: "consumer_name is mandatory when use_enhanced_fanout is true",
-		},
-		{
-			config: `
-source: kinesis
-stream_name: foobar
-stream_arn: arn:aws:kinesis:eu-west-1:123456789012:stream/my-stream`,
-			expectedErr: "stream_arn and stream_name are mutually exclusive",
-		},
-	}
-
-	subLogger := log.WithField("type", ModuleName)
-
-	for _, test := range tests {
-		t.Run(test.config, func(t *testing.T) {
-			f := Source{}
-			err := f.Configure(ctx, []byte(test.config), subLogger, metrics.AcquisitionMetricsLevelNone)
-			cstest.AssertErrorContains(t, err, test.expectedErr)
-		})
-	}
-}
-
 func TestReadFromStream(t *testing.T) {
 	endpoint := cstest.SetAWSTestEnv(t)
 

pkg/acquisition/modules/kubernetesaudit/k8s_audit_test.go

@@ -18,50 +18,6 @@ import (
 	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
 )
 
-func TestBadConfiguration(t *testing.T) {
-	tests := []struct {
-		config      string
-		name        string
-		expectedErr string
-	}{
-		{
-			name: "unknown field",
-			config: `source: k8s-audit
-foobar: asd.log`,
-			expectedErr: `[2:1] unknown field "foobar"`,
-		},
-		{
-			name:        "missing listen_addr",
-			config:      `source: k8s-audit`,
-			expectedErr: "listen_addr cannot be empty",
-		},
-		{
-			name: "missing listen_port",
-			config: `source: k8s-audit
-listen_addr: 0.0.0.0`,
-			expectedErr: "listen_port cannot be empty",
-		},
-		{
-			name: "mismatched types",
-			config: `
-source: k8s-audit
-listen_addr: 0.0.0.0
-listen_port: true
-`,
-			expectedErr: `[4:14] cannot unmarshal bool into Go struct field Configuration.ListenPort of type int`,
-		},
-	}
-
-	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			f := Source{}
-
-			err := f.UnmarshalConfig([]byte(test.config))
-			cstest.RequireErrorContains(t, err, test.expectedErr)
-		})
-	}
-}
-
 func TestInvalidConfig(t *testing.T) {
 	ctx := t.Context()
 	tests := []struct {

pkg/acquisition/modules/loki/loki_test.go

@@ -24,159 +24,6 @@ import (
 	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
 )
 
-func TestConfiguration(t *testing.T) {
-	log.Infof("Test 'TestConfigure'")
-
-	ctx := t.Context()
-
-	tests := []struct {
-		config       string
-		expectedErr  string
-		password     string
-		waitForReady time.Duration
-		delayFor     time.Duration
-		noReadyCheck bool
-		testName     string
-	}{
-		{
-			config:      `foobar: asd`,
-			expectedErr: `[1:1] unknown field "foobar"`,
-			testName:    "Unknown field",
-		},
-		{
-			config: `
-mode: tail
-source: loki`,
-			expectedErr: "loki query is mandatory",
-			testName:    "Missing url",
-		},
-		{
-			config: `
-mode: tail
-source: loki
-url: http://localhost:3100/
-`,
-			expectedErr: "loki query is mandatory",
-			testName:    "Missing query",
-		},
-		{
-			config: `
-mode: tail
-source: loki
-url: http://localhost:3100/
-query: >
-        {server="demo"}
-`,
-			expectedErr: "",
-			testName:    "Correct config",
-		},
-		{
-			config: `
-mode: tail
-source: loki
-url: http://localhost:3100/
-wait_for_ready: 5s
-query: >
-        {server="demo"}
-`,
-			expectedErr:  "",
-			testName:     "Correct config with wait_for_ready",
-			waitForReady: 5 * time.Second,
-		},
-		{
-			config: `
-mode: tail
-source: loki
-url: http://localhost:3100/
-delay_for: 1s
-query: >
-        {server="demo"}
-`,
-			expectedErr: "",
-			testName:    "Correct config with delay_for",
-			delayFor:    1 * time.Second,
-		},
-		{
-			config: `
-mode: tail
-source: loki
-url: http://localhost:3100/
-no_ready_check: true
-query: >
-        {server="demo"}
-`,
-			expectedErr:  "",
-			testName:     "Correct config with no_ready_check",
-			noReadyCheck: true,
-		},
-		{
-			config: `
-mode: tail
-source: loki
-url: http://localhost:3100/
-auth:
-  username: foo
-  password: bar
-query: >
-        {server="demo"}
-`,
-			expectedErr: "",
-			password:    "bar",
-			testName:    "Correct config with password",
-		},
-		{
-			config: `
-mode: tail
-source: loki
-url: http://localhost:3100/
-delay_for: 10s
-query: >
-        {server="demo"}
-`,
-			expectedErr: "delay_for should be a value between 1s and 5s",
-			testName:    "Invalid DelayFor",
-		},
-		{
-			config: `
-source: loki
-no_ready_check: 37
-`,
-			expectedErr: "[3:17] cannot unmarshal uint64 into Go struct field Configuration.NoReadyCheck of type bool",
-			testName:    "type mismatch",
-		},
-	}
-	subLogger := log.WithField("type", loki.ModuleName)
-
-	for _, test := range tests {
-		t.Run(test.testName, func(t *testing.T) {
-			lokiSource := loki.Source{}
-			err := lokiSource.Configure(ctx, []byte(test.config), subLogger, metrics.AcquisitionMetricsLevelNone)
-			cstest.AssertErrorContains(t, err, test.expectedErr)
-
-			if test.password != "" {
-				p := lokiSource.Config.Auth.Password
-				if test.password != p {
-					t.Fatalf("Password mismatch : %s != %s", test.password, p)
-				}
-			}
-
-			if test.waitForReady != 0 {
-				if lokiSource.Config.WaitForReady != test.waitForReady {
-					t.Fatalf("Wrong WaitForReady %v != %v", lokiSource.Config.WaitForReady, test.waitForReady)
-				}
-			}
-
-			if test.delayFor != 0 {
-				if lokiSource.Config.DelayFor != test.delayFor {
-					t.Fatalf("Wrong DelayFor %v != %v", lokiSource.Config.DelayFor, test.delayFor)
-				}
-			}
-
-			assert.Equal(t, test.noReadyCheck, lokiSource.Config.NoReadyCheck)
-		})
-	}
-}
-
 func TestConfigureDSN(t *testing.T) {
 	log.Infof("Test 'TestConfigureDSN'")
 

pkg/acquisition/modules/s3/s3_test.go

@@ -19,118 +19,10 @@ import (
 	"github.com/stretchr/testify/require"
 	"gopkg.in/tomb.v2"
 
-	"github.com/crowdsecurity/go-cs-lib/cstest"
-
 	"github.com/crowdsecurity/crowdsec/pkg/metrics"
 	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
 )
 
-func TestBadConfiguration(t *testing.T) {
-	ctx := t.Context()
-
-	tests := []struct {
-		name        string
-		config      string
-		expectedErr string
-	}{
-		{
-			name: "no bucket",
-			config: `
-source: s3
-`,
-			expectedErr: "bucket_name is required",
-		},
-		{
-			name: "type mismatch",
-			config: `
-source: s3
-max_buffer_size: true
-`,
-			expectedErr: "[3:18] cannot unmarshal bool into Go struct field Configuration.MaxBufferSize of type int",
-		},
-		{
-			name: "invalid polling method",
-			config: `
-source: s3
-bucket_name: foobar
-polling_method: foobar
-`,
-			expectedErr: "invalid polling method foobar",
-		},
-		{
-			name: "no sqs name",
-			config: `
-source: s3
-bucket_name: foobar
-polling_method: sqs
-`,
-			expectedErr: "sqs_name is required when using sqs polling method",
-		},
-		{
-			name: "both bucket and sqs",
-			config: `
-source: s3
-bucket_name: foobar
-polling_method: sqs
-sqs_name: foobar
-`,
-			expectedErr: "bucket_name and sqs_name are mutually exclusive",
-		},
-	}
-
-	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			f := Source{}
-
-			err := f.Configure(ctx, []byte(test.config), nil, metrics.AcquisitionMetricsLevelNone)
-			cstest.RequireErrorContains(t, err, test.expectedErr)
-		})
-	}
-}
-
-func TestGoodConfiguration(t *testing.T) {
-	ctx := t.Context()
-
-	tests := []struct {
-		name   string
-		config string
-	}{
-		{
-			name: "basic",
-			config: `
-source: s3
-bucket_name: foobar
-`,
-		},
-		{
-			name: "polling method",
-			config: `
-source: s3
-polling_method: sqs
-sqs_name: foobar
-`,
-		},
-		{
-			name: "list method",
-			config: `
-source: s3
-bucket_name: foobar
-polling_method: list
-`,
-		},
-	}
-
-	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			f := Source{}
-			logger := log.NewEntry(log.New())
-
-			err := f.Configure(ctx, []byte(test.config), logger, metrics.AcquisitionMetricsLevelNone)
-			require.NoError(t, err)
-		})
-	}
-}
-
 type mockS3Client struct{}
 
 // We add one hour to trick the listing goroutine into thinking the files are new

pkg/acquisition/modules/syslog/syslog_test.go

@@ -18,54 +18,6 @@ import (
 	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
 )
 
-func TestConfigure(t *testing.T) {
-	ctx := t.Context()
-
-	tests := []struct {
-		config      string
-		expectedErr string
-	}{
-		{
-			config: `
-foobar: bla
-source: syslog`,
-			expectedErr: `[2:1] unknown field "foobar"`,
-		},
-		{
-			config:      `source: syslog`,
-			expectedErr: "",
-		},
-		{
-			config: `
-source: syslog
-listen_port: asd`,
-			expectedErr: "[3:14] cannot unmarshal string into Go struct field Configuration.Port of type int",
-		},
-		{
-			config: `
-source: syslog
-listen_port: 424242`,
-			expectedErr: "invalid port 424242",
-		},
-		{
-			config: `
-source: syslog
-listen_addr: 10.0.0`,
-			expectedErr: "invalid listen IP 10.0.0",
-		},
-	}
-
-	subLogger := log.WithField("type", ModuleName)
-
-	for _, test := range tests {
-		t.Run(test.config, func(t *testing.T) {
-			s := Source{}
-			err := s.Configure(ctx, []byte(test.config), subLogger, metrics.AcquisitionMetricsLevelNone)
-			cstest.AssertErrorContains(t, err, test.expectedErr)
-		})
-	}
-}
-
 func writeToSyslog(ctx context.Context, logs []string) error {
 	dialer := &net.Dialer{}
 

pkg/acquisition/modules/victorialogs/victorialogs_test.go

@@ -26,115 +26,6 @@ import (
 	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
 )
 
-func TestConfiguration(t *testing.T) {
-	log.Infof("Test 'TestConfigure'")
-
-	ctx := t.Context()
-
-	tests := []struct {
-		config       string
-		expectedErr  string
-		password     string
-		waitForReady time.Duration
-		testName     string
-	}{
-		{
-			config:      `foobar: asd`,
-			expectedErr: `[1:1] unknown field "foobar"`,
-			testName:    "Unknown field",
-		},
-		{
-			config: `
-mode: tail
-source: victorialogs`,
-			expectedErr: "url is mandatory",
-			testName:    "Missing url",
-		},
-		{
-			config: `
-mode: tail
-source: victorialogs
-url: http://localhost:9428/
-`,
-			expectedErr: "query is mandatory",
-			testName:    "Missing query",
-		},
-		{
-			config: `
-mode: tail
-source: victorialogs
-url: http://localhost:9428/
-query: >
-        {server="demo"}
-limit: true
-`,
-			expectedErr: "[7:8] cannot unmarshal bool into Go struct field Configuration.Limit of type int",
-			testName:    "mismatched type",
-		},
-		{
-			config: `
-mode: tail
-source: victorialogs
-url: http://localhost:9428/
-query: >
-        {server="demo"}
-`,
-			expectedErr: "",
-			testName:    "Correct config",
-		},
-		{
-			config: `
-mode: tail
-source: victorialogs
-url: http://localhost:9428/
-wait_for_ready: 5s
-query: >
-        {server="demo"}
-`,
-			expectedErr:  "",
-			testName:     "Correct config with wait_for_ready",
-			waitForReady: 5 * time.Second,
-		},
-		{
-			config: `
-mode: tail
-source: victorialogs
-url: http://localhost:9428/
-auth:
-  username: foo
-  password: bar
-query: >
-        {server="demo"}
-`,
-			expectedErr: "",
-			password:    "bar",
-			testName:    "Correct config with password",
-		},
-	}
-	subLogger := log.WithField("type", victorialogs.ModuleName)
-
-	for _, test := range tests {
-		t.Run(test.testName, func(t *testing.T) {
-			vlSource := victorialogs.Source{}
-			err := vlSource.Configure(ctx, []byte(test.config), subLogger, metrics.AcquisitionMetricsLevelNone)
-			cstest.AssertErrorContains(t, err, test.expectedErr)
-
-			if test.password != "" {
-				p := vlSource.Config.Auth.Password
-				if test.password != p {
-					t.Fatalf("Password mismatch : %s != %s", test.password, p)
-				}
-			}
-
-			if test.waitForReady != 0 {
-				if vlSource.Config.WaitForReady != test.waitForReady {
-					t.Fatalf("Wrong WaitForReady %v != %v", vlSource.Config.WaitForReady, test.waitForReady)
-				}
-			}
-		})
-	}
-}
-
 func TestConfigureDSN(t *testing.T) {
 	log.Infof("Test 'TestConfigureDSN'")
 

pkg/acquisition/modules/wineventlog/wineventlog_windows_test.go

@@ -63,7 +63,7 @@ event_ids: true`,
 		},
 	}
 
-	subLogger := log.WithField("type", "windowseventlog")
+	subLogger := log.WithField("type", ModuleName)
 	for _, test := range tests {
 		f := Source{}
 		err := f.Configure(ctx, []byte(test.config), subLogger, metrics.AcquisitionMetricsLevelNone)
@@ -122,7 +122,7 @@ event_level: bla`,
 			expectedErr:   "invalid log level",
 		},
 	}
-	subLogger := log.WithField("type", "windowseventlog")
+	subLogger := log.WithField("type", ModuleName)
 	for _, test := range tests {
 		t.Run(test.config, func(t *testing.T) {
 			f := Source{}
@@ -190,7 +190,7 @@ event_ids:
 			expectedLines: nil,
 		},
 	}
-	subLogger := log.WithField("type", "windowseventlog")
+	subLogger := log.WithField("type", ModuleName)
 
 	evthandler, err := eventlog.Open("Application")
 	if err != nil {

pkg/acquisition/schemaval_test.go

@@ -0,0 +1,102 @@
+package acquisition
+
+import (
+	"bytes"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sort"
+	"strings"
+
+	"github.com/santhosh-tekuri/jsonschema/v6"
+	"github.com/goccy/go-yaml"
+)
+
+// format a compact error without schema location for testing purposes.
+func compactSchemaErr(err error) error {
+	var ve *jsonschema.ValidationError
+	if !errors.As(err, &ve) {
+		return err
+	}
+
+	out := ve.BasicOutput()
+	if out == nil || len(out.Errors) == 0 {
+		// Fallback; this may include schema URL, but it's better than losing the error.
+		return err
+	}
+
+	msgs := make([]string, 0, len(out.Errors))
+	for _, u := range out.Errors {
+		if u.Error == nil {
+			continue
+		}
+		loc := u.InstanceLocation
+		if loc == "" {
+			loc = "/"
+		}
+		msgs = append(msgs, fmt.Sprintf("%s: %s", loc, u.Error.String()))
+	}
+
+	sort.Strings(msgs)
+	return fmt.Errorf("%s", strings.Join(msgs, "; "))
+}
+
+// ValidateYAML validates configYAML against schemaPath.
+func ValidateYAML(configYAML []byte, schemaPath string) error {
+	if schemaPath == "" {
+		return errors.New("no schema provided")
+	}
+
+	configJSON, err := yaml.YAMLToJSON(configYAML)
+	if err != nil {
+		return fmt.Errorf("config: YAML->JSON: %w", err)
+	}
+
+	configDoc, err := jsonschema.UnmarshalJSON(bytes.NewReader(configJSON))
+	if err != nil {
+		return fmt.Errorf("config: decode JSON: %w", err)
+	}
+
+	c := jsonschema.NewCompiler()
+	c.DefaultDraft(jsonschema.Draft2020)
+
+	schemaYAML, err := os.ReadFile(schemaPath)
+	if err != nil {
+		return fmt.Errorf("read schema %q: %w", schemaPath, err)
+	}
+
+	schemaJSON, err := yaml.YAMLToJSON(schemaYAML)
+	if err != nil {
+		return fmt.Errorf("schema %q: YAML->JSON: %w", schemaPath, err)
+	}
+
+	schemaDoc, err := jsonschema.UnmarshalJSON(bytes.NewReader(schemaJSON))
+	if err != nil {
+		return fmt.Errorf("schema %q: decode JSON: %w", schemaPath, err)
+	}
+
+	abs, err := filepath.Abs(schemaPath)
+	if err != nil {
+		return fmt.Errorf("abs %q: %w", schemaPath, err)
+	}
+
+	if err := c.AddResource(abs, schemaDoc); err != nil {
+		return fmt.Errorf("add schema resource %q: %w", abs, err)
+	}
+
+	sch, err := c.Compile(abs)
+	if err != nil {
+		return fmt.Errorf("compile schema %q: %w", abs, err)
+	}
+
+	if err := sch.Validate(configDoc); err != nil {
+		var ve *jsonschema.ValidationError
+		if errors.As(err, &ve) {
+			return compactSchemaErr(ve)
+		}
+		return err
+	}
+
+	return nil
+}

pkg/acquisition/testdata/config.yaml

@@ -0,0 +1,6 @@
+# minimal configuration to satisfy appsec
+config_paths:
+  data_dir: /path/to/data
+api:
+  client:
+    insecure_skip_verify: false

pkg/acquisition/testdata/invalid/appsec/appsec_config_required.yaml

@@ -0,0 +1,4 @@
+# wantErr: datasource of type appsec: unable to parse appsec configuration: appsec_config or appsec_config_path must be set
+source: appsec
+labels:
+  type: sometype

pkg/acquisition/testdata/invalid/appsec/common_log_level_invalid.yaml

@@ -0,0 +1,4 @@
+# wantErr: failed to parse: not a valid logrus Level: "toto"
+# schemaErr: /log_level: value must be one of 'panic', 'fatal', 'error', 'warn', 'warning', 'info', 'debug', 'trace'
+source: appsec
+log_level: toto

pkg/acquisition/testdata/invalid/appsec/invalid_appsec_config.yaml

@@ -0,0 +1,6 @@
+# wantErr: datasource of type appsec: unable to parse appsec configuration: cannot parse appsec configuration: [6:6] cannot unmarshal map[string]interface {} into Go struct field Configuration.AppsecConfig of type string
+source: appsec
+labels:
+  type: sometype
+appsec_config:
+  foo: bar

pkg/acquisition/testdata/invalid/appsec/labels_required.yaml

@@ -0,0 +1,2 @@
+# wantErr: missing labels
+source: appsec

pkg/acquisition/testdata/invalid/appsec/xxx.yaml

@@ -0,0 +1,5 @@
+# wantErr: datasource of type appsec: missing lapi client credentials
+source: appsec
+labels:
+  type: sometype
+appsec_config: something

pkg/acquisition/testdata/invalid/cloudwatch/aws_region_required.yaml

@@ -0,0 +1,6 @@
+# wantErr: datasource of type cloudwatch: aws_region is not specified, specify it or aws_config_dir
+source: cloudwatch
+labels:
+  type: sometype
+group_name: testgroup
+

pkg/acquisition/testdata/invalid/cloudwatch/common_log_level_invalid.yaml

@@ -0,0 +1,4 @@
+# wantErr: failed to parse: not a valid logrus Level: "toto"
+# schemaErr: /log_level: value must be one of 'panic', 'fatal', 'error', 'warn', 'warning', 'info', 'debug', 'trace'
+source: cloudwatch
+log_level: toto

pkg/acquisition/testdata/invalid/cloudwatch/group_name_required.yaml

@@ -0,0 +1,5 @@
+# wantErr: datasource of type cloudwatch: group_name is mandatory for CloudwatchSource
+source: cloudwatch
+labels:
+  type: sometype
+

pkg/acquisition/testdata/invalid/cloudwatch/labels_required.yaml

@@ -0,0 +1,2 @@
+# wantErr: missing labels
+source: cloudwatch

pkg/acquisition/testdata/invalid/cloudwatch/unknown_field.yaml

@@ -0,0 +1,5 @@
+# wantErr: datasource of type cloudwatch: cannot parse: [5:1] unknown field "xxx"
+source: cloudwatch
+labels:
+  type: sometype
+xxx: dummy

pkg/acquisition/testdata/invalid/docker/common_log_level_invalid.yaml

@@ -0,0 +1,4 @@
+# wantErr: failed to parse: not a valid logrus Level: "toto"
+# schemaErr: /log_level: value must be one of 'panic', 'fatal', 'error', 'warn', 'warning', 'info', 'debug', 'trace'
+source: docker
+log_level: toto

pkg/acquisition/testdata/invalid/docker/common_mode_invalid.yaml

@@ -0,0 +1,6 @@
+# wantErr: datasource of type docker: unsupported mode server for docker datasource
+# schemaErr: /mode: value must be one of 'tail', 'cat'
+mode: server
+source: docker
+container_name:
+ - toto

pkg/acquisition/testdata/invalid/docker/container_id_regexp_parsing_error.yaml

@@ -0,0 +1,5 @@
+# wantErr: datasource of type docker: container_id_regexp: error parsing regexp: missing argument to repetition operator: `*`
+# schemaErr:
+source: docker
+container_id_regexp:
+ - "*invalid"

pkg/acquisition/testdata/invalid/docker/container_labels_mutual_exclusive.yaml

@@ -0,0 +1,6 @@
+# wantErr: datasource of type docker: use_container_labels and container_name, container_id, container_id_regexp, container_name_regexp are mutually exclusive
+# schemaErr: /: 'allOf' failed; /: 'not' failed
+source: docker
+use_container_labels: true
+container_name:
+ - toto

pkg/acquisition/testdata/invalid/docker/container_name_regexp_parsing_error.yaml

@@ -0,0 +1,5 @@
+# wantErr: datasource of type docker: container_name_regexp: error parsing regexp: missing closing ]: `[invalid`
+# schemaErr:
+source: docker
+container_name_regexp:
+ - "[invalid"

pkg/acquisition/testdata/invalid/docker/containers_services_required.yaml

@@ -0,0 +1,3 @@
+# wantErr: datasource of type docker: no containers or services configuration provided
+# schemaErr: /: 'allOf' failed; /: 'anyOf' failed; /: missing property 'container_id'; /: missing property 'container_id_regexp'; /: missing property 'container_name'; /: missing property 'container_name_regexp'; /: missing property 'service_id'; /: missing property 'service_id_regexp'; /: missing property 'service_name'; /: missing property 'service_name_regexp'; /: missing property 'use_container_labels'; /: missing property 'use_service_labels'
+source: docker

pkg/acquisition/testdata/invalid/docker/service_id_regexp_parsing_error.yaml

@@ -0,0 +1,5 @@
+# wantErr: datasource of type docker: service_id_regexp: error parsing regexp: missing argument to repetition operator: `+`
+# schemaErr:
+source: docker
+service_id_regexp:
+ - "+invalid"

pkg/acquisition/testdata/invalid/docker/service_labels_mutual_exclusive.yaml

@@ -0,0 +1,6 @@
+# wantErr: datasource of type docker: use_service_labels and service_name, service_id, service_id_regexp, service_name_regexp are mutually exclusive
+# schemaErr: /: 'allOf' failed; /: 'not' failed
+source: docker
+use_service_labels: true
+service_name:
+ - web-service

pkg/acquisition/testdata/invalid/docker/service_name_regexp_parsing_error.yaml

@@ -0,0 +1,5 @@
+# wantErr: datasource of type docker: service_name_regexp: error parsing regexp: invalid named capture: `(?P<invalid`
+# schemaErr:
+source: docker
+service_name_regexp:
+ - "(?P<invalid"

pkg/acquisition/testdata/invalid/docker/unknown_field.yaml

@@ -0,0 +1,4 @@
+# wantErr: datasource of type docker: while parsing DockerAcquisition configuration: [4:1] unknown field "foobar"
+# schemaErr: /: additional properties 'foobar' not allowed
+source: docker
+foobar: asd.log

pkg/acquisition/testdata/invalid/empty.yaml

@@ -0,0 +1 @@
+# wantErr: empty yaml document

pkg/acquisition/testdata/invalid/file/bad_exclude_regexp.yaml

@@ -0,0 +1,6 @@
+# wantErr: datasource of type file: could not compile regexp as[a-$d: error parsing regexp: invalid character class range: `a-$`
+source: file
+labels:
+  type: sometype
+filename: "[asd.log]"
+exclude_regexps: ["as[a-$d"]

pkg/acquisition/testdata/invalid/file/common_log_level_invalid.yaml

@@ -0,0 +1,4 @@
+# wantErr: failed to parse: not a valid logrus Level: "toto"
+# schemaErr: /log_level: value must be one of 'panic', 'fatal', 'error', 'warn', 'warning', 'info', 'debug', 'trace'
+source: file
+log_level: toto

pkg/acquisition/testdata/invalid/file/duplicate_keys.yaml

@@ -0,0 +1,6 @@
+# wantErr: position 0: [6:1] mapping key "filenames" already defined at [5:1]
+source: file
+labels:
+  type: sometype
+filenames: "[asd.log]"
+filenames: "[ase.log]"

pkg/acquisition/testdata/invalid/file/filename_filenames_required.yaml

@@ -0,0 +1,4 @@
+# wantErr: datasource of type file: no filename or filenames configuration provided
+source: file
+labels:
+  type: sometype

pkg/acquisition/testdata/invalid/file/glob_invalid_field.yaml

@@ -0,0 +1,5 @@
+# wantErr: position 0: [5:11] sequence end token ']' not found
+source: file
+labels:
+  type: sometype
+filename: [asd-.log

pkg/acquisition/testdata/invalid/file/glob_syntax_error.yaml

@@ -0,0 +1,5 @@
+# wantErr: datasource of type file: glob failure: syntax error in pattern
+source: file
+labels:
+  type: sometype
+filename: "[asd-.log"

pkg/acquisition/testdata/invalid/file/invalid_poll_interval.yaml

@@ -0,0 +1,9 @@
+# wantErr: datasource of type file: cannot parse FileAcquisition configuration: time: invalid duration "invalid"
+source: file
+labels:
+  type: sometype
+filenames:
+  - "tests/test.log"
+discovery_poll_enable: true
+discovery_poll_interval: invalid
+

pkg/acquisition/testdata/invalid/file/labels_required.yaml

@@ -0,0 +1,2 @@
+# wantErr: missing labels
+source: file

pkg/acquisition/testdata/invalid/file/unknown_field.yaml

@@ -0,0 +1,5 @@
+# wantErr: datasource of type file: cannot parse FileAcquisition configuration: [5:1] unknown field "foobar"
+source: file
+labels:
+  type: sometype
+foobar: asd.log

pkg/acquisition/testdata/invalid/http/auth_type_unknown.yaml

@@ -0,0 +1,7 @@
+# wantErr: datasource of type http: invalid auth_type: must be one of basic_auth, headers, mtls
+source: http
+labels:
+  type: sometype
+listen_addr: 127.0.0.1:8080
+path: /test
+auth_type: toto

pkg/acquisition/testdata/invalid/http/basic_auth_missing.yaml

@@ -0,0 +1,7 @@
+# wantErr: datasource of type http: basic_auth is selected, but basic_auth is not provided
+source: http
+labels:
+  type: sometype
+listen_addr: 127.0.0.1:8080
+path: /test
+auth_type: basic_auth

pkg/acquisition/testdata/invalid/http/ca_cert_required.yaml

@@ -0,0 +1,10 @@
+# wantErr: datasource of type http: mtls is selected, but ca_cert is not provided
+source: http
+labels:
+  type: sometype
+listen_addr: 127.0.0.1:8080
+path: /test
+auth_type: mtls
+tls:
+  server_cert: cert
+  server_key: key

pkg/acquisition/testdata/invalid/http/common_log_level_invalid.yaml

@@ -0,0 +1,4 @@
+# wantErr: failed to parse: not a valid logrus Level: "toto"
+# schemaErr: /log_level: value must be one of 'panic', 'fatal', 'error', 'warn', 'warning', 'info', 'debug', 'trace'
+source: http
+log_level: toto

pkg/acquisition/testdata/invalid/http/custom_status_code_invalid.yaml

@@ -0,0 +1,10 @@
+# wantErr: datasource of type http: invalid HTTP status code
+source: http
+labels:
+  type: sometype
+listen_addr: 127.0.0.1:8080
+path: /test
+auth_type: headers
+headers:
+  key: value
+custom_status_code: 999

pkg/acquisition/testdata/invalid/http/duration_invalid.yaml

@@ -0,0 +1,10 @@
+# wantErr: datasource of type http: cannot parse: time: invalid duration "toto"
+source: http
+labels:
+  type: sometype
+listen_addr: 127.0.0.1:8080
+path: /test
+auth_type: headers
+headers:
+  key: value
+timeout: toto

pkg/acquisition/testdata/invalid/http/headers_missing.yaml

@@ -0,0 +1,7 @@
+# wantErr: datasource of type http: headers is selected, but headers is not provided
+source: http
+labels:
+  type: sometype
+listen_addr: 127.0.0.1:8080
+path: /test
+auth_type: headers

pkg/acquisition/testdata/invalid/http/labels_required.yaml

@@ -0,0 +1,2 @@
+# wantErr: missing labels
+source: http

pkg/acquisition/testdata/invalid/http/listen_addr_listen_socket_required.yaml

@@ -0,0 +1,4 @@
+# wantErr: datasource of type http: listen_addr or listen_socket is required
+source: http
+labels:
+  type: sometype

pkg/acquisition/testdata/invalid/http/max_body_size_positive.yaml

@@ -0,0 +1,10 @@
+# wantErr: datasource of type http: max_body_size must be positive
+source: http
+labels:
+  type: sometype
+listen_addr: 127.0.0.1:8080
+path: /test
+auth_type: headers
+headers:
+  key: value
+max_body_size: 0

pkg/acquisition/testdata/invalid/http/password_missing.yaml

@@ -0,0 +1,9 @@
+# wantErr: datasource of type http: basic_auth is selected, but password is not provided
+source: http
+labels:
+  type: sometype
+listen_addr: 127.0.0.1:8080
+path: /test
+auth_type: basic_auth
+basic_auth:
+  username: 132

pkg/acquisition/testdata/invalid/http/path_missing_slash.yaml

@@ -0,0 +1,6 @@
+# wantErr: datasource of type http: path must start with /
+source: http
+labels:
+  type: sometype
+listen_addr: 127.0.0.1:8080
+path: wrongpath

pkg/acquisition/testdata/invalid/http/server_cert_required.yaml

@@ -0,0 +1,11 @@
+# wantErr: datasource of type http: server_cert is required
+source: http
+labels:
+  type: sometype
+listen_addr: 127.0.0.1:8080
+path: /test
+auth_type: headers
+headers:
+  key: value
+tls:
+  server_key: key

pkg/acquisition/testdata/invalid/http/server_key_required.yaml

@@ -0,0 +1,11 @@
+# wantErr: datasource of type http: server_key is required
+source: http
+labels:
+  type: sometype
+listen_addr: 127.0.0.1:8080
+path: /test
+auth_type: headers
+headers:
+  key: value
+tls:
+  server_cert: cert

pkg/acquisition/testdata/invalid/http/username_missing.yaml

@@ -0,0 +1,9 @@
+# wantErr: datasource of type http: basic_auth is selected, but username is not provided
+source: http
+labels:
+  type: sometype
+listen_addr: 127.0.0.1:8080
+path: /test
+auth_type: basic_auth
+basic_auth:
+  password: xyz

pkg/acquisition/testdata/invalid/journalctl/common_log_level_invalid.yaml

@@ -0,0 +1,4 @@
+# wantErr: failed to parse: not a valid logrus Level: "toto"
+# schemaErr: /log_level: value must be one of 'panic', 'fatal', 'error', 'warn', 'warning', 'info', 'debug', 'trace'
+source: journalctl
+log_level: toto

pkg/acquisition/testdata/invalid/journalctl/journalctl_filter_required.yaml

@@ -0,0 +1,4 @@
+# wantErr: datasource of type journalctl: journalctl_filter is required
+source: journalctl
+labels:
+  type: sometype

pkg/acquisition/testdata/invalid/journalctl/labels_required.yaml

@@ -0,0 +1,2 @@
+# wantErr: missing labels
+source: journalctl

pkg/acquisition/testdata/invalid/journalctl/unknown_field.yaml

@@ -0,0 +1,5 @@
+# wantErr: datasource of type journalctl: cannot parse: [3:1] unknown field "filename"
+source: journalctl
+filename: /path/to/file.log
+labels:
+  type: sometype

pkg/acquisition/testdata/invalid/kafka/brokers_required.yaml

@@ -0,0 +1,4 @@
+# wantErr: datasource of type kafka: cannot create a kafka reader with an empty list of broker addresses
+source: kafka
+labels:
+  type: sometype

pkg/acquisition/testdata/invalid/kafka/common_log_level_invalid.yaml

@@ -0,0 +1,4 @@
+# wantErr: failed to parse: not a valid logrus Level: "toto"
+# schemaErr: /log_level: value must be one of 'panic', 'fatal', 'error', 'warn', 'warning', 'info', 'debug', 'trace'
+source: kafka
+log_level: toto

pkg/acquisition/testdata/invalid/kafka/group_id_partition_mutual_exclusive.yaml

@@ -0,0 +1,9 @@
+# wantErr: datasource of type kafka: cannote create kafka reader: cannot specify both group_id and partition
+source: kafka
+labels:
+  type: sometype
+brokers:
+  - localhost:9092
+topic: crowdsec
+partition: 1
+group_id: crowdsec

pkg/acquisition/testdata/invalid/kafka/labels_required.yaml

@@ -0,0 +1,2 @@
+# wantErr: missing labels
+source: kafka

pkg/acquisition/testdata/invalid/kafka/timeout_parser_error.yaml

@@ -0,0 +1,8 @@
+# wantErr: datasource of type kafka: cannot create kafka dialer: strconv.Atoi: parsing "aa": invalid syntax
+source: kafka
+labels:
+  type: sometype
+brokers:
+  - bla
+topic: toto
+timeout: aa

pkg/acquisition/testdata/invalid/kafka/topic_required.yaml

@@ -0,0 +1,6 @@
+# wantErr: datasource of type kafka: cannot create a kafka reader with an empty topic
+source: kafka
+labels:
+  type: sometype
+brokers:
+  - bla

pkg/acquisition/testdata/invalid/kafka/unknown_field.yaml

@@ -0,0 +1,5 @@
+# wantErr: datasource of type kafka: cannot parse: [5:1] unknown field "foobar"
+source: kafka
+labels:
+  type: sometype
+foobar: bla

pkg/acquisition/testdata/invalid/kinesis/common_log_level_invalid.yaml

@@ -0,0 +1,4 @@
+# wantErr: failed to parse: not a valid logrus Level: "toto"
+# schemaErr: /log_level: value must be one of 'panic', 'fatal', 'error', 'warn', 'warning', 'info', 'debug', 'trace'
+source: kinesis
+log_level: toto

pkg/acquisition/testdata/invalid/kinesis/consumer_name_required.yaml

@@ -0,0 +1,6 @@
+# wantErr: datasource of type kinesis: consumer_name is mandatory when use_enhanced_fanout is true
+source: kinesis
+labels:
+  type: sometype
+use_enhanced_fanout: true
+stream_arn: arn:aws:kinesis:eu-west-1:123456789012:stream/my-stream

pkg/acquisition/testdata/invalid/kinesis/labels_required.yaml

@@ -0,0 +1,2 @@
+# wantErr: missing labels
+source: kinesis

pkg/acquisition/testdata/invalid/kinesis/maxretries_invalid.yaml

@@ -0,0 +1,5 @@
+# wantErr: datasource of type kinesis: cannot parse: [5:14] cannot unmarshal string into Go struct field Configuration.MaxRetries of type int
+source: kinesis
+labels:
+  type: sometype
+max_retries: whavev

pkg/acquisition/testdata/invalid/kinesis/stream_arn_required.yaml

@@ -0,0 +1,5 @@
+# wantErr: datasource of type kinesis: stream_arn is mandatory when use_enhanced_fanout is true
+source: kinesis
+use_enhanced_fanout: true
+labels:
+  type: sometype

pkg/acquisition/testdata/invalid/kinesis/stream_arn_stream_name_mutual_exclusive.yaml

@@ -0,0 +1,6 @@
+# wantErr: datasource of type kinesis: stream_arn and stream_name are mutually exclusive
+source: kinesis
+labels:
+  type: sometype
+stream_arn: arn:aws:kinesis:eu-west-1:123456789012:stream/my-stream
+stream_name: foobar

pkg/acquisition/testdata/invalid/kinesis/stream_name_required.yaml

@@ -0,0 +1,4 @@
+# wantErr: datasource of type kinesis: stream_name is mandatory when use_enhanced_fanout is false
+source: kinesis
+labels:
+  type: sometype

pkg/acquisition/testdata/invalid/kubernetesaudit/common_log_level_invalid.yaml

@@ -0,0 +1,4 @@
+# wantErr: failed to parse: not a valid logrus Level: "toto"
+# schemaErr: /log_level: value must be one of 'panic', 'fatal', 'error', 'warn', 'warning', 'info', 'debug', 'trace'
+source: k8s-audit
+log_level: toto

pkg/acquisition/testdata/invalid/kubernetesaudit/labels_required.yaml

@@ -0,0 +1,2 @@
+# wantErr: missing labels
+source: k8s-audit

pkg/acquisition/testdata/invalid/kubernetesaudit/listen_addr_required.yaml

@@ -0,0 +1,4 @@
+# wantErr: datasource of type k8s-audit: listen_addr cannot be empty
+source: k8s-audit
+labels:
+  type: sometype

pkg/acquisition/testdata/invalid/kubernetesaudit/listen_port_invalid.yaml

@@ -0,0 +1,6 @@
+# wantErr: datasource of type k8s-audit: cannot parse: [6:14] cannot unmarshal bool into Go struct field Configuration.ListenPort of type int
+source: k8s-audit
+labels:
+  type: sometype
+listen_addr: 0.0.0.0
+listen_port: true

pkg/acquisition/testdata/invalid/kubernetesaudit/listen_port_required.yaml

@@ -0,0 +1,5 @@
+# wantErr: datasource of type k8s-audit: listen_port cannot be empty
+source: k8s-audit
+labels:
+  type: sometype
+listen_addr: 0.0.0.0

pkg/acquisition/testdata/invalid/kubernetesaudit/unknown_field.yaml

@@ -0,0 +1,6 @@
+# wantErr: datasource of type k8s-audit: cannot parse: [6:1] unknown field "foobar"
+# schemaErr: foo
+source: k8s-audit
+labels:
+  type: sometype
+foobar: asd.log

pkg/acquisition/testdata/invalid/loki/common_log_level_invalid.yaml

@@ -0,0 +1,4 @@
+# wantErr: failed to parse: not a valid logrus Level: "toto"
+# schemaErr: /log_level: value must be one of 'panic', 'fatal', 'error', 'warn', 'warning', 'info', 'debug', 'trace'
+source: loki
+log_level: toto

pkg/acquisition/testdata/invalid/loki/delay_for_invalid.yaml

@@ -0,0 +1,7 @@
+# wantErr: datasource of type loki: delay_for should be a value between 1s and 5s
+source: loki
+labels:
+  type: sometype
+delay_for: 10s
+query: >
+        {server="demo"}

pkg/acquisition/testdata/invalid/loki/labels_required.yaml

@@ -0,0 +1,2 @@
+# wantErr: missing labels
+source: loki

pkg/acquisition/testdata/invalid/loki/no_ready_check_invalid.yaml

@@ -0,0 +1,5 @@
+# wantErr: datasource of type loki: cannot parse loki acquisition configuration: [5:17] cannot unmarshal uint64 into Go struct field Configuration.NoReadyCheck of type bool
+source: loki
+labels:
+  type: sometype
+no_ready_check: 37

pkg/acquisition/testdata/invalid/loki/query_required.yaml

@@ -0,0 +1,4 @@
+# wantErr: datasource of type loki: loki query is mandatory
+source: loki
+labels:
+  type: sometype

pkg/acquisition/testdata/invalid/loki/unknown_field.yaml

@@ -0,0 +1,5 @@
+# wantErr: datasource of type loki: cannot parse loki acquisition configuration: [5:1] unknown field "foobar"
+source: loki
+labels:
+  type: sometype
+foobar: asd

pkg/acquisition/testdata/invalid/mapping_expected.yaml

@@ -0,0 +1,2 @@
+# wantErr: failed to parse: [2:1] string was used where mapping is expected
+blahblah