frostfs-node/pkg/local_object_storage/pilorama/split_test.go

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) {
	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"))

	testSortedByFilename := func(t *testing.T, root MultiNode, last *string, batchSize int) ([]MultiNodeInfo, *string) {
		res, last, err := f.TreeSortedByFilename(context.Background(), d.CID, treeID, root, last, batchSize)
		require.NoError(t, err)
		return res, last
	}

	t.Run("test sorted listing, full children branch", func(t *testing.T) {
		t.Run("big batch size", func(t *testing.T) {
			res, _ := testSortedByFilename(t, MultiNode{RootID}, nil, 10)
			require.Equal(t, 3, len(res))
			require.Equal(t, MultiNode{1, 5}, res[0].Children)
			require.Equal(t, MultiNode{9}, res[1].Children)
			require.Equal(t, MultiNode{10}, res[2].Children)

			t.Run("multi-root", func(t *testing.T) {
				res, _ := testSortedByFilename(t, MultiNode{1, 5}, nil, 10)
				require.Equal(t, 3, len(res))
				require.Equal(t, MultiNode{3, 7}, res[0].Children)
				require.Equal(t, MultiNode{2}, res[1].Children)
				require.Equal(t, MultiNode{6}, res[2].Children)
			})
		})
		t.Run("small batch size", func(t *testing.T) {
			res, last := testSortedByFilename(t, MultiNode{RootID}, nil, 1)
			require.Equal(t, 1, len(res))
			require.Equal(t, MultiNode{1, 5}, res[0].Children)

			res, last = testSortedByFilename(t, MultiNode{RootID}, last, 1)
			require.Equal(t, 1, len(res))
			require.Equal(t, MultiNode{9}, res[0].Children)

			res, last = testSortedByFilename(t, MultiNode{RootID}, last, 1)
			require.Equal(t, 1, len(res))
			require.Equal(t, MultiNode{10}, res[0].Children)

			res, _ = testSortedByFilename(t, MultiNode{RootID}, last, 1)
			require.Equal(t, 0, len(res))

			t.Run("multi-root", func(t *testing.T) {
				res, last := testSortedByFilename(t, MultiNode{1, 5}, nil, 1)
				require.Equal(t, 1, len(res))
				require.Equal(t, MultiNode{3, 7}, res[0].Children)

				res, last = testSortedByFilename(t, MultiNode{1, 5}, last, 1)
				require.Equal(t, 1, len(res))
				require.Equal(t, MultiNode{2}, res[0].Children)

				res, last = testSortedByFilename(t, MultiNode{1, 5}, last, 1)
				require.Equal(t, 1, len(res))
				require.Equal(t, MultiNode{6}, res[0].Children)

				res, _ = testSortedByFilename(t, MultiNode{RootID}, last, 1)
				require.Equal(t, 0, len(res))
			})
		})
	})
}