Cherry-pick elastic#19177 to 7.x: Introduce statestore internal testi…

…ng support (elastic#19690)

libbeat/statestore/internal/storecompliance/reg.go

@@ -0,0 +1,113 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you 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 storecompliance
+
+import (
+	"testing"
+
+	"github.com/elastic/beats/v7/libbeat/statestore/backend"
+)
+
+// Registry helper for writing tests.
+// The registry uses a testing.T and provides some MustX methods that fail if
+// an error occured.
+type Registry struct {
+	T testing.TB
+	backend.Registry
+}
+
+// Store helper for writing tests.
+// The store needs a reference to the Registry with the current test context.
+// The Store provides additional helpers for reopening the store, MustX methods
+// that will fail the test if an error has occured.
+type Store struct {
+	backend.Store
+
+	Registry *Registry
+	name     string
+}
+
+// Access uses the backend Registry to create a new Store.
+func (r *Registry) Access(name string) (*Store, error) {
+	s, err := r.Registry.Access(name)
+	if err != nil {
+		return nil, err
+	}
+	return &Store{Store: s, Registry: r, name: name}, nil
+}
+
+// MustAccess opens a Store. It fails the test if an error has occured.
+func (r *Registry) MustAccess(name string) *Store {
+	store, err := r.Access(name)
+	must(r.T, err, "open store")
+	return store
+}
+
+// Close closes the testing store.
+func (s *Store) Close() {
+	err := s.Store.Close()
+	must(s.Registry.T, err, "closing store %q failed", s.name)
+}
+
+// ReopenIf reopens the store if b is true.
+func (s *Store) ReopenIf(b bool) {
+	if b {
+		s.Reopen()
+	}
+}
+
+// Reopen reopens the store by closing the backend store and using the registry
+// backend to access the same store again.
+func (s *Store) Reopen() {
+	t := s.Registry.T
+
+	s.Close()
+	if t.Failed() {
+		t.Fatal("Test already failed")
+	}
+
+	store, err := s.Registry.Registry.Access(s.name)
+	must(s.Registry.T, err, "reopen failed")
+
+	s.Store = store
+}
+
+// MustHave fails the test if an error occured in a call to Has.
+func (s *Store) MustHave(key string) bool {
+	b, err := s.Has(key)
+	must(s.Registry.T, err, "unexpected error on store/has call")
+	return b
+}
+
+// MustGet fails the test if an error occured in a call to Get.
+func (s *Store) MustGet(key string, into interface{}) {
+	err := s.Get(key, into)
+	must(s.Registry.T, err, "unexpected error on store/get call")
+}
+
+// MustSet fails the test if an error occured in a call to Set.
+func (s *Store) MustSet(key string, from interface{}) {
+	err := s.Set(key, from)
+	must(s.Registry.T, err, "unexpected error on store/set call")
+}
+
+// MustRemove fails the test if an error occured in a call to Remove.
+func (s *Store) MustRemove(key string) {
+	err := s.Store.Remove(key)
+	must(s.Registry.T, err, "unexpected error remove key")
+}

libbeat/statestore/internal/storecompliance/storecompliance.go

@@ -0,0 +1,188 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you 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 storecompliance provides a common test suite that a store
+// implementation must succeed in order to be compliant to the beats
+// statestore. The Internal tests are used by statestore/storetest and
+// statestore/backend/memlog.
+//
+// The package adds the `-keep` and `-dir <path>` CLI flags:
+//   - `-dir <path>`: configure path where to create test folders in (defaults
+//     to OS specific temporary directory)
+//   - `-keep`: The test directories will not be deleted after a test has
+//     finished. The test directory is added to the test logs.
+//
+package storecompliance
+
+import (
+	"errors"
+	"flag"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/elastic/beats/v7/libbeat/statestore/backend"
+)
+
+// BackendFactory is used by TestBackendCompliance to create
+// store instances for testing. Each store will be configured
+// with an unique temporary directory.
+type BackendFactory func(testPath string) (backend.Registry, error)
+
+var defaultTempDir string
+var keepTmpDir bool
+
+func init() {
+	flag.StringVar(&defaultTempDir, "dir", "", "Temporary directory for use by the test")
+	flag.BoolVar(&keepTmpDir, "keep", false, "Keep temporary test directories")
+}
+
+// TestBackendCompliance runs a set of tests the verifies that the store
+// implementation can be used by beats.
+// Most tests are executed twice if they modify data. Once with keeping the
+// store open between operations, and once with reopening the store between
+// updates.
+// For a store backend that supports different 'modes' that can impact storage,
+// the compliance tests should be run with the different modes enabled.
+//
+// Note: The tests only check for interoperability. Implementations should add
+// additional tests as well.
+func TestBackendCompliance(t *testing.T, factory BackendFactory) {
+	t.Run("init and close registry", WithPath(factory, func(t *testing.T, reg *Registry) {
+	}))
+
+	t.Run("open stores", WithPath(factory, func(t *testing.T, reg *Registry) {
+		store := reg.MustAccess("test1")
+		defer store.Close()
+
+		store2 := reg.MustAccess("test2")
+		defer store2.Close()
+	}))
+
+	t.Run("set-get", withBackend(factory, testSetGet))
+	t.Run("remove", withBackend(factory, testRemove))
+	t.Run("iteration", withBackend(factory, testIteration))
+}
+
+func testSetGet(t *testing.T, factory BackendFactory) {
+	t.Run("unknown key", WithStore(factory, func(t *testing.T, store *Store) {
+		has := store.MustHave("key")
+		assert.False(t, has)
+	}))
+
+	runWithBools(t, "reopen", func(t *testing.T, reopen bool) {
+		t.Run("has key after set", WithStore(factory, func(t *testing.T, store *Store) {
+			type entry struct{ A int }
+			store.MustSet("key", entry{A: 1})
+
+			store.ReopenIf(reopen)
+
+			has := store.MustHave("key")
+			assert.True(t, has)
+		}))
+
+		t.Run("set and get one entry only", WithStore(factory, func(t *testing.T, store *Store) {
+			type entry struct{ A int }
+			key := "key"
+			value := entry{A: 1}
+
+			store.MustSet(key, value)
+			store.ReopenIf(reopen)
+
+			var actual entry
+			store.MustGet(key, &actual)
+			assert.Equal(t, value, actual)
+		}))
+	})
+}
+
+func testRemove(t *testing.T, factory BackendFactory) {
+	t.Run("no error when removing unknown key", WithStore(factory, func(t *testing.T, store *Store) {
+		store.MustRemove("key")
+	}))
+
+	runWithBools(t, "reopen", func(t *testing.T, reopen bool) {
+		t.Run("remove key", WithStore(factory, func(t *testing.T, store *Store) {
+			type entry struct{ A int }
+			key := "key"
+			store.MustSet(key, entry{A: 1})
+			store.ReopenIf(reopen)
+			store.MustRemove(key)
+			store.ReopenIf(reopen)
+			has := store.MustHave(key)
+			assert.False(t, has)
+		}))
+	})
+}
+
+func testIteration(t *testing.T, factory BackendFactory) {
+	data := map[string]interface{}{
+		"a": map[string]interface{}{"field": "hello"},
+		"b": map[string]interface{}{"field": "world"},
+	}
+
+	addTestData := func(store *Store, reopen bool, data map[string]interface{}) {
+		for k, v := range data {
+			store.MustSet(k, v)
+		}
+		store.ReopenIf(reopen)
+	}
+
+	runWithBools(t, "reopen", func(t *testing.T, reopen bool) {
+		t.Run("all keys", WithStore(factory, func(t *testing.T, store *Store) {
+			addTestData(store, reopen, data)
+
+			got := map[string]interface{}{}
+			err := store.Each(func(key string, dec backend.ValueDecoder) (bool, error) {
+				var tmp interface{}
+				if err := dec.Decode(&tmp); err != nil {
+					return false, err
+				}
+
+				got[key] = tmp
+				return true, nil
+			})
+
+			assert.NoError(t, err)
+			assert.Equal(t, data, got)
+		}))
+
+		t.Run("stop on error", WithStore(factory, func(t *testing.T, store *Store) {
+			addTestData(store, reopen, data)
+
+			count := 0
+			err := store.Each(func(_ string, _ backend.ValueDecoder) (bool, error) {
+				count++
+				return true, errors.New("oops")
+			})
+			assert.Equal(t, 1, count)
+			assert.Error(t, err)
+		}))
+
+		t.Run("stop on bool", WithStore(factory, func(t *testing.T, store *Store) {
+			addTestData(store, reopen, data)
+
+			count := 0
+			err := store.Each(func(_ string, _ backend.ValueDecoder) (bool, error) {
+				count++
+				return false, nil
+			})
+			assert.Equal(t, 1, count)
+			assert.NoError(t, err)
+		}))
+	})
+}