Fix expected dataset check in test runners

internal/packages/packages.go

@@ -254,6 +254,12 @@ func ReadDataStreamManifest(path string) (*DataStreamManifest, error) {
 	return &m, nil
 }
 
+// ReadDataStreamManifestFromPackageRoot reads and parses the manifest of the given
+// data stream from the given package root.
+func ReadDataStreamManifestFromPackageRoot(packageRoot string, name string) (*DataStreamManifest, error) {
+	return ReadDataStreamManifest(filepath.Join(packageRoot, "data_stream", name, DataStreamManifestFile))
+}
+
 // GetPipelineNameOrDefault returns the name of the data stream's pipeline, if one is explicitly defined in the
 // data stream manifest. If not, the default pipeline name is returned.
 func (dsm *DataStreamManifest) GetPipelineNameOrDefault() string {

internal/testrunner/runners/pipeline/runner.go

@@ -99,6 +99,16 @@ func (r *runner) run() ([]testrunner.TestResult, error) {
 		return nil, errors.Wrap(err, "failed to read manifest")
 	}
 
+	dsManifest, err := packages.ReadDataStreamManifestFromPackageRoot(r.options.PackageRootPath, r.options.TestFolder.DataStream)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to read data stream manifest")
+	}
+
+	expectedDataset := dsManifest.Dataset
+	if expectedDataset == "" {
+		expectedDataset = pkgManifest.Name + "." + r.options.TestFolder.DataStream
+	}
+
 	results := make([]testrunner.TestResult, 0)
 	for _, testCaseFile := range testCaseFiles {
 		tr := testrunner.TestResult{
@@ -140,7 +150,6 @@ func (r *runner) run() ([]testrunner.TestResult, error) {
 		result := &testResult{events: processedEvents}
 
 		tr.TimeElapsed = time.Since(startTime)
-		expectedDataset := pkgManifest.Name + "." + r.options.TestFolder.DataStream
 		fieldsValidator, err := fields.CreateValidatorForDirectory(dataStreamPath,
 			fields.WithSpecVersion(pkgManifest.SpecVersion),
 			fields.WithNumericKeywordFields(tc.config.NumericKeywordFields),

internal/testrunner/runners/static/runner.go

@@ -70,33 +70,33 @@ func (r runner) run() ([]testrunner.TestResult, error) {
 		return result.WithError(errors.Wrap(err, "failed to read manifest"))
 	}
 
-	var results []testrunner.TestResult
-	results = append(results, r.verifySampleEvent(pkgManifest)...)
-	return results, nil
+	return r.verifySampleEvent(pkgManifest), nil
 }
 
 func (r runner) verifySampleEvent(pkgManifest *packages.PackageManifest) []testrunner.TestResult {
-	dataStreamPath := filepath.Join(r.options.PackageRootPath, "data_stream", r.options.TestFolder.DataStream)
-	sampleEventPath := filepath.Join(dataStreamPath, sampleEventJSON)
-	_, err := os.Stat(sampleEventPath)
-	if errors.Is(err, os.ErrNotExist) {
-		return []testrunner.TestResult{} // nothing to succeed, nothing to skip
-	}
-
 	resultComposer := testrunner.NewResultComposer(testrunner.TestResult{
 		Name:       "Verify " + sampleEventJSON,
 		TestType:   TestType,
 		Package:    r.options.TestFolder.Package,
 		DataStream: r.options.TestFolder.DataStream,
 	})
 
+	sampleEventPath, found, err := r.getSampleEventPath()
 	if err != nil {
-		results, _ := resultComposer.WithError(errors.Wrap(err, "stat file failed"))
+		results, _ := resultComposer.WithError(err)
 		return results
 	}
+	if !found {
+		// Nothing to do.
+		return []testrunner.TestResult{}
+	}
 
-	expectedDataset := pkgManifest.Name + "." + r.options.TestFolder.DataStream
-	fieldsValidator, err := fields.CreateValidatorForDirectory(dataStreamPath,
+	expectedDataset, err := r.getExpectedDataset(pkgManifest)
+	if err != nil {
+		results, _ := resultComposer.WithError(err)
+		return results
+	}
+	fieldsValidator, err := fields.CreateValidatorForDirectory(filepath.Dir(sampleEventPath),
 		fields.WithSpecVersion(pkgManifest.SpecVersion),
 		fields.WithDefaultNumericConversion(),
 		fields.WithExpectedDataset(expectedDataset),
@@ -125,6 +125,46 @@ func (r runner) verifySampleEvent(pkgManifest *packages.PackageManifest) []testr
 	return results
 }
 
+func (r runner) getSampleEventPath() (string, bool, error) {
+	var sampleEventPath string
+	if r.options.TestFolder.DataStream != "" {
+		sampleEventPath = filepath.Join(
+			r.options.PackageRootPath,
+			"data_stream",
+			r.options.TestFolder.DataStream,
+			sampleEventJSON)
+	} else {
+		sampleEventPath = filepath.Join(r.options.PackageRootPath, sampleEventJSON)
+	}
+	_, err := os.Stat(sampleEventPath)
+	if errors.Is(err, os.ErrNotExist) {
+		return "", false, nil
+	}
+	if err != nil {
+		return "", false, errors.Wrap(err, "stat file failed")
+	}
+	return sampleEventPath, true, nil
+}
+
+func (r runner) getExpectedDataset(pkgManifest *packages.PackageManifest) (string, error) {
+	dsName := r.options.TestFolder.DataStream
+	if dsName == "" {
+		// TODO: This should return the package name plus the policy name, but we don't know
+		// what policy created this event, so we cannot reliably know it here. Skip the check
+		// by now.
+		return "", nil
+	}
+
+	dataStreamManifest, err := packages.ReadDataStreamManifestFromPackageRoot(r.options.PackageRootPath, dsName)
+	if err != nil {
+		return "", errors.Wrap(err, "failed to read data stream manifest")
+	}
+	if ds := dataStreamManifest.Dataset; ds != "" {
+		return ds, nil
+	}
+	return pkgManifest.Name + "." + dsName, nil
+}
+
 func (r runner) TearDown() error {
 	return nil // it's a static test runner, no state is stored
 }

internal/testrunner/runners/system/runner.go

@@ -464,7 +464,7 @@ func (r *runner) runTest(config *testConfig, ctxt servicedeployer.ServiceContext
 	// Validate fields in docs
 	var expectedDataset string
 	if ds := r.options.TestFolder.DataStream; ds != "" {
-		expectedDataset = pkgManifest.Name + "." + ds
+		expectedDataset = getDataStreamDataset(*pkgManifest, *dataStreamManifest)
 	} else {
 		expectedDataset = pkgManifest.Name + "." + policyTemplateName
 	}

test/packages/other/with_dataset/changelog.yml

@@ -1,4 +1,4 @@
-- version: "0.0.1"
+- version: "0.0.2"
   changes:
     - description: initial release
       type: enhancement

test/packages/other/with_dataset/manifest.yml

@@ -1,15 +1,14 @@
-format_version: 1.0.0
+format_version: 2.2.0
 name: with_dataset
 title: With dataset test
-version: 0.0.1
+version: 0.0.2
 description: Package that defines system test for data streams with an overwriting dataset
 categories:
   - custom
-release: experimental
-license: basic
 type: integration
 conditions:
   kibana.version: '^8.0.0'
+  elastic.subscription: basic
 policy_templates:
   - name: sample
     title: Sample logs