Fix package validation on Windows

.ci/Jenkinsfile

@@ -50,11 +50,28 @@ pipeline {
      Test the source code.
      */
     stage('Test') {
-      steps {
-        cleanup()
-        withMageEnv(){
-          dir("${BASE_DIR}"){
-            sh(label: 'Test', script: 'make test')
+      parallel {
+        stage('Test on Linux') {
+          options { skipDefaultCheckout() }
+          steps {
+            cleanup()
+            withMageEnv(){
+              dir("${BASE_DIR}"){
+                sh(label: 'Test', script: 'make test')
+              }
+            }
+          }
+        }
+        stage('Test on Windows') {
+          agent { label 'windows-10' }
+          options { skipDefaultCheckout() }
+          steps {
+            cleanup()
+            withMageEnv(){
+              dir("${BASE_DIR}"){
+                bat(label: 'Test', script: 'go test -v ./code/go/...')
+              }
+            }
           }
         }
       }

code/go/internal/fspath/fspath.go

@@ -7,7 +7,7 @@ package fspath
 import (
 	"io/fs"
 	"os"
-	"path/filepath"
+	"path"
 )
 
 // FS implements the fs interface and can also show a path where the fs is located.
@@ -26,7 +26,7 @@ type fsDir struct {
 
 // Path returns a path for the given names, based on the location of the file system.
 func (fs *fsDir) Path(names ...string) string {
-	return filepath.Join(append([]string{fs.path}, names...)...)
+	return path.Join(append([]string{fs.path}, names...)...)
 }
 
 // DirFS returns a file system for a directory, it keeps the path to implement the FS interface.

code/go/internal/validator/folder_item_content.go

@@ -5,7 +5,7 @@
 package validator
 
 import (
-	"bytes"
+	"bufio"
 	"encoding/json"
 	"fmt"
 	"io/fs"
@@ -55,16 +55,18 @@ func validateYAMLDashes(fsys fs.FS, path string) error {
 	}
 	defer f.Close()
 
-	dashes := []byte("---\n")
-	b := make([]byte, len(dashes))
-	_, err = f.Read(b)
-	if err != nil {
+	scanner := bufio.NewScanner(f)
+	// A small buffer should be enough to check if the document starts with three dashes.
+	buf := make([]byte, 8)
+	scanner.Buffer(buf, len(buf))
+	scanner.Scan()
+	if err := scanner.Err(); err != bufio.ErrTooLong && err != nil {
 		return err
 	}
-
-	if !bytes.Equal(dashes, b) {
+	if scanner.Text() != "---" {
 		return errors.New("document dashes are required (start the document with '---')")
 	}
+
 	return nil
 }
 

code/go/internal/validator/folder_item_spec.go

@@ -7,7 +7,7 @@ package validator
 import (
 	"fmt"
 	"io/fs"
-	"path/filepath"
+	"path"
 	"regexp"
 	"sync"
 
@@ -103,7 +103,7 @@ func (s *folderItemSpec) validateSchema(schemaFsys fs.FS, fsys fs.FS, folderSpec
 		return nil // item's schema is not defined
 	}
 
-	schemaPath := filepath.Join(filepath.Dir(folderSpecPath), s.Ref)
+	schemaPath := path.Join(path.Dir(folderSpecPath), s.Ref)
 	schemaLoader := yamlschema.NewReferenceLoaderFileSystem("file:///"+schemaPath, schemaFsys)
 
 	// validation with schema
@@ -120,8 +120,8 @@ func (s *folderItemSpec) validateSchema(schemaFsys fs.FS, fsys fs.FS, folderSpec
 		formatCheckersMutex.Unlock()
 	}()
 
-	semantic.LoadRelativePathFormatChecker(fsys, filepath.Dir(itemPath), s.Limits.RelativePathSizeLimit)
-	semantic.LoadDataStreamNameFormatChecker(fsys, filepath.Dir(itemPath))
+	semantic.LoadRelativePathFormatChecker(fsys, path.Dir(itemPath), s.Limits.RelativePathSizeLimit)
+	semantic.LoadDataStreamNameFormatChecker(fsys, path.Dir(itemPath))
 	result, err := gojsonschema.Validate(schemaLoader, documentLoader)
 	if err != nil {
 		return ve.ValidationErrors{err}

code/go/internal/validator/folder_spec.go

@@ -9,7 +9,7 @@ import (
 	"io/fs"
 	"io/ioutil"
 	"log"
-	"path/filepath"
+	"path"
 	"regexp"
 	"strings"
 
@@ -71,32 +71,32 @@ func (s *folderSpec) load(fs fs.FS, specPath string) error {
 	return nil
 }
 
-func (s *folderSpec) validate(pkg *Package, path string) ve.ValidationErrors {
+func (s *folderSpec) validate(pkg *Package, folderPath string) ve.ValidationErrors {
 	var errs ve.ValidationErrors
-	files, err := fs.ReadDir(pkg, path)
+	files, err := fs.ReadDir(pkg, folderPath)
 	if err != nil {
-		errs = append(errs, errors.Wrapf(err, "could not read folder [%s]", pkg.Path(path)))
+		errs = append(errs, errors.Wrapf(err, "could not read folder [%s]", pkg.Path(folderPath)))
 		return errs
 	}
 
 	// This is not taking into account if the folder is for development. Enforce
 	// this limit in all cases to avoid having to read too many files.
 	if contentsLimit := s.Limits.TotalContentsLimit; contentsLimit > 0 && len(files) > contentsLimit {
-		errs = append(errs, errors.Errorf("folder [%s] exceeds the limit of %d files", pkg.Path(path), contentsLimit))
+		errs = append(errs, errors.Errorf("folder [%s] exceeds the limit of %d files", pkg.Path(folderPath), contentsLimit))
 		return errs
 	}
 
 	// Don't enable beta features for packages marked as GA.
 	switch s.Release {
-		case "", "ga": // do nothing
-		case "beta":
-			if pkg.Version.Major() > 0 && pkg.Version.Prerelease() == "" {
-				errs = append(errs, errors.Errorf("spec for [%s] defines beta features which can't be enabled for packages with a stable semantic version", pkg.Path(path)))
-			} else {
-				log.Printf("Warning: package with non-stable semantic version and active beta features (enabled in [%s]) can't be released as stable version.", pkg.Path(path))
-			}
-		default:
-			errs = append(errs, errors.Errorf("unsupport release level, supported values: beta, ga"))
+	case "", "ga": // do nothing
+	case "beta":
+		if pkg.Version.Major() > 0 && pkg.Version.Prerelease() == "" {
+			errs = append(errs, errors.Errorf("spec for [%s] defines beta features which can't be enabled for packages with a stable semantic version", pkg.Path(folderPath)))
+		} else {
+			log.Printf("Warning: package with non-stable semantic version and active beta features (enabled in [%s]) can't be released as stable version.", pkg.Path(folderPath))
+		}
+	default:
+		errs = append(errs, errors.Errorf("unsupport release level, supported values: beta, ga"))
 	}
 
 	for _, file := range files {
@@ -113,15 +113,15 @@ func (s *folderSpec) validate(pkg *Package, path string) ve.ValidationErrors {
 				if !s.DevelopmentFolder && strings.Contains(fileName, "-") {
 					errs = append(errs,
 						fmt.Errorf(`file "%s" is invalid: directory name inside package %s contains -: %s`,
-							pkg.Path(path, fileName), pkg.Name, fileName))
+							pkg.Path(folderPath, fileName), pkg.Name, fileName))
 				}
 			}
 			continue
 		}
 
 		if itemSpec == nil && !s.AdditionalContents {
 			// No spec found for current folder item and we do not allow additional contents in folder.
-			errs = append(errs, fmt.Errorf("item [%s] is not allowed in folder [%s]", fileName, pkg.Path(path)))
+			errs = append(errs, fmt.Errorf("item [%s] is not allowed in folder [%s]", fileName, pkg.Path(folderPath)))
 			continue
 		}
 
@@ -145,7 +145,7 @@ func (s *folderSpec) validate(pkg *Package, path string) ve.ValidationErrors {
 			// Inherit limits from parent directory.
 			subFolderSpec.Limits = s.Limits
 			if itemSpec.Ref != "" {
-				subFolderSpecPath := filepath.Join(filepath.Dir(s.specPath), itemSpec.Ref)
+				subFolderSpecPath := path.Join(path.Dir(s.specPath), itemSpec.Ref)
 				err := subFolderSpec.load(s.fs, subFolderSpecPath)
 				if err != nil {
 					errs = append(errs, err)
@@ -163,7 +163,7 @@ func (s *folderSpec) validate(pkg *Package, path string) ve.ValidationErrors {
 				subFolderSpec.DevelopmentFolder = true
 			}
 
-			subFolderPath := filepath.Join(path, fileName)
+			subFolderPath := path.Join(folderPath, fileName)
 			subErrs := subFolderSpec.validate(pkg, subFolderPath)
 			if len(subErrs) > 0 {
 				errs = append(errs, subErrs...)
@@ -180,7 +180,7 @@ func (s *folderSpec) validate(pkg *Package, path string) ve.ValidationErrors {
 				continue
 			}
 
-			itemPath := filepath.Join(path, file.Name())
+			itemPath := path.Join(folderPath, file.Name())
 			itemValidationErrs := itemSpec.validate(s.fs, pkg, s.specPath, itemPath)
 			if itemValidationErrs != nil {
 				for _, ive := range itemValidationErrs {
@@ -199,7 +199,7 @@ func (s *folderSpec) validate(pkg *Package, path string) ve.ValidationErrors {
 	}
 
 	if sizeLimit := s.Limits.TotalSizeLimit; sizeLimit > 0 && s.totalSize > sizeLimit {
-		errs = append(errs, errors.Errorf("folder [%s] exceeds the total size limit of %s", pkg.Path(path), sizeLimit))
+		errs = append(errs, errors.Errorf("folder [%s] exceeds the total size limit of %s", pkg.Path(folderPath), sizeLimit))
 	}
 
 	// validate that required items in spec are all accounted for
@@ -217,9 +217,9 @@ func (s *folderSpec) validate(pkg *Package, path string) ve.ValidationErrors {
 		if !fileFound {
 			var err error
 			if itemSpec.Name != "" {
-				err = fmt.Errorf("expecting to find [%s] %s in folder [%s]", itemSpec.Name, itemSpec.ItemType, pkg.Path(path))
+				err = fmt.Errorf("expecting to find [%s] %s in folder [%s]", itemSpec.Name, itemSpec.ItemType, pkg.Path(folderPath))
 			} else if itemSpec.Pattern != "" {
-				err = fmt.Errorf("expecting to find %s matching pattern [%s] in folder [%s]", itemSpec.ItemType, itemSpec.Pattern, pkg.Path(path))
+				err = fmt.Errorf("expecting to find %s matching pattern [%s] in folder [%s]", itemSpec.ItemType, itemSpec.Pattern, pkg.Path(folderPath))
 			}
 			errs = append(errs, err)
 		}

code/go/internal/validator/package.go

@@ -8,7 +8,7 @@ import (
 	"fmt"
 	"io/fs"
 	"os"
-	"path/filepath"
+	"path"
 
 	"github.com/Masterminds/semver/v3"
 	"github.com/pkg/errors"
@@ -33,7 +33,7 @@ func (p *Package) Open(name string) (fs.File, error) {
 
 // Path returns a path meaningful for the user.
 func (p *Package) Path(names ...string) string {
-	return filepath.Join(append([]string{p.location}, names...)...)
+	return path.Join(append([]string{p.location}, names...)...)
 }
 
 // NewPackage creates a new Package from a path to the package's root folder

code/go/internal/validator/package_test.go

@@ -47,4 +47,3 @@ func TestNewPackage(t *testing.T) {
 		})
 	}
 }
-

code/go/internal/validator/semantic/format_checkers.go

@@ -6,7 +6,7 @@ package semantic
 
 import (
 	"io/fs"
-	"path/filepath"
+	"path"
 
 	"github.com/xeipuuv/gojsonschema"
 
@@ -40,7 +40,7 @@ func (r relativePathChecker) IsFormat(input interface{}) bool {
 		return false
 	}
 
-	path := filepath.Join(r.currentPath, asString)
+	path := path.Join(r.currentPath, asString)
 	info, err := fs.Stat(r.fsys, path)
 	if err != nil {
 		return false
@@ -77,7 +77,7 @@ func UnloadRelativePathFormatChecker() {
 func LoadDataStreamNameFormatChecker(fsys fs.FS, currentPath string) {
 	gojsonschema.FormatCheckers.Add(DataStreamNameFormat, relativePathChecker{
 		fsys:        fsys,
-		currentPath: filepath.Join(currentPath, "data_stream"),
+		currentPath: path.Join(currentPath, "data_stream"),
 	})
 }
 

code/go/internal/validator/semantic/types.go

@@ -5,10 +5,9 @@
 package semantic
 
 import (
-	"fmt"
 	"io/fs"
 	"os"
-	"path/filepath"
+	"path"
 	"strings"
 
 	"github.com/pkg/errors"
@@ -84,7 +83,7 @@ func listFieldsFiles(fsys fspath.FS) ([]string, error) {
 	}
 
 	for _, dataStream := range dataStreams {
-		fieldsDir := filepath.Join(dataStreamDir, dataStream, "fields")
+		fieldsDir := path.Join(dataStreamDir, dataStream, "fields")
 		fs, err := fs.ReadDir(fsys, fieldsDir)
 		if errors.Is(err, os.ErrNotExist) {
 			continue
@@ -94,7 +93,7 @@ func listFieldsFiles(fsys fspath.FS) ([]string, error) {
 		}
 
 		for _, f := range fs {
-			fieldsFiles = append(fieldsFiles, filepath.Join(fieldsDir, f.Name()))
+			fieldsFiles = append(fieldsFiles, path.Join(fieldsDir, f.Name()))
 		}
 	}
 
@@ -116,13 +115,13 @@ func unmarshalFields(fsys fspath.FS, fieldsPath string) (fields, error) {
 }
 
 func dataStreamFromFieldsPath(pkgRoot, fieldsFile string) (string, error) {
-	dataStreamPath := filepath.Clean(filepath.Join(pkgRoot, "data_stream"))
-	relPath, err := filepath.Rel(dataStreamPath, filepath.Clean(fieldsFile))
-	if err != nil {
-		return "", fmt.Errorf("looking for fields file (%s) in data streams path (%s): %w", fieldsFile, dataStreamPath, err)
+	dataStreamPath := path.Clean(path.Join(pkgRoot, "data_stream")) + "/"
+	if !strings.HasPrefix(path.Clean(fieldsFile), dataStreamPath) {
+		return "", errors.Errorf("%q is not a fields file in a data stream of %q", fieldsFile, dataStreamPath)
 	}
+	relPath := strings.TrimPrefix(path.Clean(fieldsFile), dataStreamPath)
 
-	parts := strings.SplitN(relPath, string(filepath.Separator), 2)
+	parts := strings.SplitN(relPath, "/", 2)
 	if len(parts) != 2 {
 		return "", errors.Errorf("could not find data stream for fields file %s", fieldsFile)
 	}

code/go/internal/validator/semantic/types_test.go

@@ -0,0 +1,54 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package semantic
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestDataStreamFromFieldsPath(t *testing.T) {
+	cases := []struct {
+		pkgRoot    string
+		fieldsFile string
+		expected   string
+		fail       bool
+	}{
+		{
+			pkgRoot:    "package",
+			fieldsFile: "package/data_stream/foo/fields/some-fields.yml",
+			expected:   "foo",
+		},
+		{
+			pkgRoot:    "package/",
+			fieldsFile: "package/data_stream/foo/fields/some-fields.yml",
+			expected:   "foo",
+		},
+		{
+			pkgRoot:    "/package/",
+			fieldsFile: "/package/data_stream/foo/fields/some-fields.yml",
+			expected:   "foo",
+		},
+		{
+			pkgRoot:    "/package/",
+			fieldsFile: "/package/fields/some-fields.yml",
+			fail:       true,
+		},
+	}
+
+	for _, c := range cases {
+		t.Run(c.pkgRoot+"_"+c.fieldsFile, func(t *testing.T) {
+			dataStream, err := dataStreamFromFieldsPath(c.pkgRoot, c.fieldsFile)
+			if c.fail {
+				assert.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				assert.Equal(t, c.expected, dataStream)
+			}
+		})
+	}
+}