From 034061dbfc8159682dff4bcb3171bbd3d196fa5a Mon Sep 17 00:00:00 2001
From: Evgenii Stratonikov <e.stratonikov@yadro.com>
Date: Mon, 8 Jul 2024 13:48:39 +0300
Subject: [PATCH] [#1234] pilorama: Add test for duplicate directory behaviour

When AddByPath() is called concurrently on 2 different nodes,
internal path components may be created twice. This violates some
of our assumptions in GetByPath() and, indirectly, in S3 handling of
GetSubTree() results.

Add a test for the correct behaviour, fixes will follow.

Signed-off-by: Evgenii Stratonikov <e.stratonikov@yadro.com>
---
 .../pilorama/split_test.go                    | 99 +++++++++++++++++++
 1 file changed, 99 insertions(+)
 create mode 100644 pkg/local_object_storage/pilorama/split_test.go

diff --git a/pkg/local_object_storage/pilorama/split_test.go b/pkg/local_object_storage/pilorama/split_test.go
new file mode 100644
index 000000000..85c6c3097
--- /dev/null
+++ b/pkg/local_object_storage/pilorama/split_test.go
@@ -0,0 +1,99 @@
+package pilorama
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	cidtest "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id/test"
+	"github.com/stretchr/testify/require"
+)
+
+func TestDuplicateDirectory(t *testing.T) {
+	t.Skip()
+	for i := range providers {
+		if providers[i].name == "inmemory" {
+			continue
+		}
+		t.Run(providers[i].name, func(t *testing.T) {
+			testDuplicateDirectory(t, providers[i].construct(t))
+		})
+	}
+}
+
+func testDuplicateDirectory(t *testing.T, f Forest) {
+	ctx := context.Background()
+	d := CIDDescriptor{CID: cidtest.ID(), Size: 1}
+	treeID := "sometree"
+
+	treeApply := func(t *testing.T, parent, child uint64, filename string, internal bool) {
+		// Nothing magic here, we add items in order and children are unique.
+		// This simplifies function interface a bit.
+		ts := child
+
+		kv := []KeyValue{{Key: AttributeFilename, Value: []byte(filename)}}
+		if !internal {
+			kv = append(kv, KeyValue{Key: "uniqueAttr", Value: []byte{byte(child)}})
+		}
+
+		err := f.TreeApply(ctx, d.CID, treeID, &Move{
+			Parent: parent,
+			Child:  child,
+			Meta: Meta{
+				Time:  ts,
+				Items: kv,
+			},
+		}, true)
+		require.NoError(t, err)
+	}
+
+	// The following tree is constructed:
+	//      0
+	// [1]  |-- dir1 (internal)
+	// [2]     |-- value1
+	// [3]     |-- dir3 (internal)
+	// [4]        |-- value3
+	// [5]  |-- dir1 (internal)
+	// [6]     |-- value2
+	// [7]     |-- dir3 (internal)
+	// [8]        |-- value4
+	// [9]  |-- dir2 (internal)
+	// [10] |-- value0
+	treeApply(t, RootID, 1, "dir1", true)
+	treeApply(t, 1, 2, "value1", false)
+	treeApply(t, 1, 3, "dir3", true)
+	treeApply(t, 3, 4, "value3", false)
+	treeApply(t, RootID, 5, "dir1", true)
+	treeApply(t, 5, 6, "value2", false)
+	treeApply(t, 5, 7, "dir3", true)
+	treeApply(t, 7, 8, "value4", false)
+	treeApply(t, RootID, 9, "dir2", true)
+	treeApply(t, RootID, 10, "value0", false)
+
+	// The compacted view:
+	// 0
+	// [1,5] |-- dir1 (internal)
+	// [2]      |-- value1
+	// [3,7]    |-- dir3 (internal)
+	// [4]          |-- value3
+	// [8]          |-- value4
+	// [6]      |-- value2
+	// [9]   |-- dir2 (internal)
+	// [10]  |-- value0
+	testGetByPath := func(t *testing.T, p string) []byte {
+		pp := strings.Split(p, "/")
+		nodes, err := f.TreeGetByPath(context.Background(), d.CID, treeID, AttributeFilename, pp, false)
+		require.NoError(t, err)
+		require.Equal(t, 1, len(nodes))
+
+		meta, _, err := f.TreeGetMeta(ctx, d.CID, treeID, nodes[0])
+		require.NoError(t, err)
+		require.Equal(t, []byte(pp[len(pp)-1]), meta.GetAttr(AttributeFilename))
+		return meta.GetAttr("uniqueAttr")
+	}
+
+	require.Equal(t, []byte{2}, testGetByPath(t, "dir1/value1"))
+	require.Equal(t, []byte{4}, testGetByPath(t, "dir1/dir3/value3"))
+	require.Equal(t, []byte{8}, testGetByPath(t, "dir1/dir3/value4"))
+	require.Equal(t, []byte{10}, testGetByPath(t, "value0"))
+}