chore: add more tests

services/dedup/badger/badger_test.go

@@ -0,0 +1,51 @@
+package badger
+
+import (
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/rudderlabs/rudder-go-kit/config"
+	"github.com/rudderlabs/rudder-go-kit/logger"
+	"github.com/rudderlabs/rudder-go-kit/stats"
+	"github.com/rudderlabs/rudder-server/services/dedup/types"
+	"github.com/rudderlabs/rudder-server/utils/misc"
+)
+
+func Test_Badger(t *testing.T) {
+	config.Reset()
+	logger.Reset()
+	misc.Init()
+
+	dbPath := os.TempDir() + "/dedup_test"
+	defer func() { _ = os.RemoveAll(dbPath) }()
+	_ = os.RemoveAll(dbPath)
+	conf := config.New()
+	t.Setenv("RUDDER_TMPDIR", dbPath)
+	badger := NewBadgerDB(conf, stats.NOP, DefaultPath())
+	require.NotNil(t, badger)
+	defer badger.Close()
+	t.Run("Same messageID should be deduped from badger", func(t *testing.T) {
+		key1 := types.KeyValue{Key: "a", Value: 1, WorkspaceId: "test"}
+		key2 := types.KeyValue{Key: "a", Value: 1, WorkspaceId: "test"}
+		found, _, err := badger.Get(key1)
+		require.Nil(t, err)
+		require.True(t, found)
+		err = badger.Commit([]string{key1.Key})
+		require.NoError(t, err)
+		found, _, err = badger.Get(key2)
+		require.Nil(t, err)
+		require.False(t, found)
+	})
+	t.Run("Same messageID should be deduped from cache", func(t *testing.T) {
+		key1 := types.KeyValue{Key: "b", Value: 1, WorkspaceId: "test"}
+		key2 := types.KeyValue{Key: "b", Value: 1, WorkspaceId: "test"}
+		found, _, err := badger.Get(key1)
+		require.Nil(t, err)
+		require.True(t, found)
+		found, _, err = badger.Get(key2)
+		require.Nil(t, err)
+		require.False(t, found)
+	})
+}

services/dedup/mirrorBadger/mirrorBadger_test.go

@@ -0,0 +1,63 @@
+package mirrorBadger
+
+import (
+	"os"
+	"strings"
+	"testing"
+
+	"github.com/ory/dockertest/v3"
+	"github.com/stretchr/testify/require"
+
+	"github.com/rudderlabs/rudder-go-kit/config"
+	"github.com/rudderlabs/rudder-go-kit/logger"
+	"github.com/rudderlabs/rudder-go-kit/stats"
+	"github.com/rudderlabs/rudder-go-kit/testhelper/docker/resource/scylla"
+	"github.com/rudderlabs/rudder-go-kit/testhelper/rand"
+	"github.com/rudderlabs/rudder-server/services/dedup/types"
+	"github.com/rudderlabs/rudder-server/utils/misc"
+)
+
+func Test_MirrorBadger(t *testing.T) {
+	config.Reset()
+	logger.Reset()
+	misc.Init()
+
+	dbPath := os.TempDir() + "/dedup_test"
+	defer func() { _ = os.RemoveAll(dbPath) }()
+	_ = os.RemoveAll(dbPath)
+	conf := config.New()
+	t.Setenv("RUDDER_TMPDIR", dbPath)
+	pool, err := dockertest.NewPool("")
+	require.NoError(t, err)
+	keySpace := strings.ToUpper(rand.String(5))
+	resource, err := scylla.Setup(pool, t, scylla.WithKeyspace(keySpace))
+	require.NoError(t, err)
+	conf.Set("Scylla.Hosts", resource.URL)
+	conf.Set("Scylla.Keyspace", keySpace)
+	mirrorBadger, err := NewMirrorBadger(conf, stats.NOP)
+	require.Nil(t, err)
+	require.NotNil(t, mirrorBadger)
+	defer mirrorBadger.Close()
+	t.Run("Same messageID should be deduped from badger", func(t *testing.T) {
+		key1 := types.KeyValue{Key: "a", Value: 1, WorkspaceId: "test"}
+		key2 := types.KeyValue{Key: "a", Value: 1, WorkspaceId: "test"}
+		found, _, err := mirrorBadger.Get(key1)
+		require.Nil(t, err)
+		require.True(t, found)
+		err = mirrorBadger.Commit([]string{key1.Key})
+		require.NoError(t, err)
+		found, _, err = mirrorBadger.Get(key2)
+		require.Nil(t, err)
+		require.False(t, found)
+	})
+	t.Run("Same messageID should be deduped from cache", func(t *testing.T) {
+		key1 := types.KeyValue{Key: "b", Value: 1, WorkspaceId: "test"}
+		key2 := types.KeyValue{Key: "b", Value: 1, WorkspaceId: "test"}
+		found, _, err := mirrorBadger.Get(key1)
+		require.Nil(t, err)
+		require.True(t, found)
+		found, _, err = mirrorBadger.Get(key2)
+		require.Nil(t, err)
+		require.False(t, found)
+	})
+}

services/dedup/mirrorScylla/mirrorScylla_test.go

@@ -0,0 +1,77 @@
+package mirrorScylla
+
+import (
+	"os"
+	"strings"
+	"testing"
+
+	"github.com/ory/dockertest/v3"
+	"github.com/stretchr/testify/require"
+
+	"github.com/rudderlabs/rudder-go-kit/config"
+	"github.com/rudderlabs/rudder-go-kit/logger"
+	"github.com/rudderlabs/rudder-go-kit/stats"
+	"github.com/rudderlabs/rudder-go-kit/testhelper/docker/resource/scylla"
+	"github.com/rudderlabs/rudder-go-kit/testhelper/rand"
+	"github.com/rudderlabs/rudder-server/services/dedup/types"
+	"github.com/rudderlabs/rudder-server/utils/misc"
+)
+
+func Test_MirrorBadger(t *testing.T) {
+	config.Reset()
+	logger.Reset()
+	misc.Init()
+
+	dbPath := os.TempDir() + "/dedup_test"
+	defer func() { _ = os.RemoveAll(dbPath) }()
+	_ = os.RemoveAll(dbPath)
+	conf := config.New()
+	t.Setenv("RUDDER_TMPDIR", dbPath)
+	pool, err := dockertest.NewPool("")
+	require.NoError(t, err)
+	keySpace := strings.ToUpper(rand.String(5))
+	resource, err := scylla.Setup(pool, t, scylla.WithKeyspace(keySpace))
+	require.NoError(t, err)
+	conf.Set("Scylla.Hosts", resource.URL)
+	conf.Set("Scylla.Keyspace", keySpace)
+	mirrorScylla, err := NewMirrorScylla(conf, stats.NOP)
+	require.Nil(t, err)
+	require.NotNil(t, mirrorScylla)
+	defer mirrorScylla.Close()
+	t.Run("Same messageID should not be deduped for different workspace", func(t *testing.T) {
+		key1 := types.KeyValue{Key: "a", Value: 1, WorkspaceId: "test1"}
+		key2 := types.KeyValue{Key: "a", Value: 1, WorkspaceId: "test2"}
+		found, _, err := mirrorScylla.Get(key1)
+		require.Nil(t, err)
+		require.True(t, found)
+		err = mirrorScylla.Commit([]string{key1.Key})
+		require.NoError(t, err)
+		found, _, err = mirrorScylla.Get(key2)
+		require.Nil(t, err)
+		require.True(t, found)
+		err = mirrorScylla.Commit([]string{key2.Key})
+		require.NoError(t, err)
+	})
+	t.Run("Same messageID should be deduped for same workspace", func(t *testing.T) {
+		key1 := types.KeyValue{Key: "a", Value: 1, WorkspaceId: "test"}
+		key2 := types.KeyValue{Key: "a", Value: 1, WorkspaceId: "test"}
+		found, _, err := mirrorScylla.Get(key1)
+		require.Nil(t, err)
+		require.True(t, found)
+		err = mirrorScylla.Commit([]string{key1.Key})
+		require.NoError(t, err)
+		found, _, err = mirrorScylla.Get(key2)
+		require.Nil(t, err)
+		require.False(t, found)
+	})
+	t.Run("Same messageID should be deduped for same workspace from cache", func(t *testing.T) {
+		key1 := types.KeyValue{Key: "b", Value: 1, WorkspaceId: "test"}
+		key2 := types.KeyValue{Key: "b", Value: 1, WorkspaceId: "test"}
+		found, _, err := mirrorScylla.Get(key1)
+		require.Nil(t, err)
+		require.True(t, found)
+		found, _, err = mirrorScylla.Get(key2)
+		require.Nil(t, err)
+		require.False(t, found)
+	})
+}

services/dedup/scylla/scylla_test.go

@@ -0,0 +1,66 @@
+package scylla
+
+import (
+	"testing"
+
+	"github.com/ory/dockertest/v3"
+	"github.com/stretchr/testify/require"
+
+	"github.com/rudderlabs/rudder-go-kit/config"
+	"github.com/rudderlabs/rudder-go-kit/stats"
+	"github.com/rudderlabs/rudder-go-kit/testhelper/docker/resource/scylla"
+	"github.com/rudderlabs/rudder-go-kit/testhelper/rand"
+	"github.com/rudderlabs/rudder-server/services/dedup/types"
+)
+
+func Test_Scylla(t *testing.T) {
+	pool, err := dockertest.NewPool("")
+	require.NoError(t, err)
+	keySpace := rand.String(5)
+	scyllaContainer, err := scylla.Setup(pool, t, scylla.WithKeyspace(keySpace))
+	require.NoError(t, err)
+	require.NotNil(t, scyllaContainer)
+	conf := config.New()
+	conf.Set("Scylla.Hosts", scyllaContainer.URL)
+	conf.Set("Scylla.Keyspace", keySpace)
+	scylla, err := New(conf, stats.NOP)
+	require.NoError(t, err)
+	require.NotNil(t, scylla)
+	defer scylla.Close()
+	t.Run("Same messageID should not be deduped for different workspace", func(t *testing.T) {
+		key1 := types.KeyValue{Key: "a", Value: 1, WorkspaceId: "test1"}
+		key2 := types.KeyValue{Key: "a", Value: 1, WorkspaceId: "test2"}
+		found, _, err := scylla.Get(key1)
+		require.Nil(t, err)
+		require.True(t, found)
+		err = scylla.Commit([]string{key1.Key})
+		require.NoError(t, err)
+		found, _, err = scylla.Get(key2)
+		require.Nil(t, err)
+		require.True(t, found)
+		err = scylla.Commit([]string{key2.Key})
+		require.NoError(t, err)
+	})
+	t.Run("Same messageID should be deduped for same workspace", func(t *testing.T) {
+		key1 := types.KeyValue{Key: "a", Value: 1, WorkspaceId: "test"}
+		key2 := types.KeyValue{Key: "a", Value: 1, WorkspaceId: "test"}
+		found, _, err := scylla.Get(key1)
+		require.Nil(t, err)
+		require.True(t, found)
+		err = scylla.Commit([]string{key1.Key})
+		require.NoError(t, err)
+		found, _, err = scylla.Get(key2)
+		require.Nil(t, err)
+		require.False(t, found)
+	})
+	t.Run("Same messageID should be deduped for same workspace from cache", func(t *testing.T) {
+		key1 := types.KeyValue{Key: "b", Value: 1, WorkspaceId: "test"}
+		key2 := types.KeyValue{Key: "b", Value: 1, WorkspaceId: "test"}
+		found, _, err := scylla.Get(key1)
+		require.Nil(t, err)
+		require.True(t, found)
+		found, _, err = scylla.Get(key2)
+		require.Nil(t, err)
+		require.False(t, found)
+	})
+}