neofs: Add test for consistency

The test requires real neofs cluster to execute. We are checking the object after uploading.

Signed-off-by: Evgenii Baidakov <[email protected]>

internal/neofs/neofs_test.go

@@ -5,7 +5,10 @@ import (
 	"context"
 	"crypto/rand"
 	"fmt"
+	"io"
+	"runtime"
 	"strconv"
+	"sync"
 	"testing"
 	"time"
 
@@ -133,3 +136,101 @@ func deleteContainer(ctx context.Context, id cid.ID, signer user.Signer, p *pool
 	var prm client.PrmContainerDelete
 	return p.ContainerDelete(ctx, id, signer, prm)
 }
+
+func TestConcurrencyAndConsistency(t *testing.T) {
+	t.Skip("Required connection to NeoFS cluster")
+
+	ctx, cancel := context.WithCancel(context.Background())
+
+	pk, err := keys.NewPrivateKey()
+	require.NoError(t, err)
+	signer := user.NewAutoIDSignerRFC6979(pk.PrivateKey)
+
+	anonPk, err := keys.NewPrivateKey()
+	require.NoError(t, err)
+	anonSigner := user.NewAutoIDSignerRFC6979(anonPk.PrivateKey)
+
+	var prm pool.InitParameters
+	prm.SetSigner(signer)
+	prm.AddNode(pool.NewNodeParam(1, "localhost:8080", 1))
+
+	p, err := pool.NewPool(prm)
+	require.NoError(t, err)
+
+	require.NoError(t, p.Dial(ctx))
+
+	ni, err := p.NetworkInfo(ctx, client.PrmNetworkInfo{})
+	require.NoError(t, err)
+
+	gorutines := runtime.GOMAXPROCS(0)
+
+	neofsCfg := Config{
+		MaxObjectSize:        int64(ni.MaxObjectSize()),
+		IsSlicerEnabled:      false,
+		IsHomomorphicEnabled: !ni.HomomorphicHashingDisabled(),
+	}
+
+	neo := NewNeoFS(p, signer, anonSigner, neofsCfg, ni)
+
+	var createParams layer.PrmObjectCreate
+	createParams.Creator = signer.UserID()
+
+	wg := sync.WaitGroup{}
+	wg.Add(gorutines)
+
+	for i := 0; i < gorutines; i++ {
+		go func() {
+			uploadDownload(ctx, t, neo, p, signer, createParams, &wg)
+		}()
+	}
+
+	<-time.After(30 * time.Second)
+	cancel()
+	wg.Wait()
+}
+
+func uploadDownload(ctx context.Context, t *testing.T, neo *NeoFS, p *pool.Pool, signer user.Signer, createParams layer.PrmObjectCreate, wg *sync.WaitGroup) {
+	defer wg.Done()
+
+	payload := make([]byte, 32*1024)
+
+	id, err := createContainer(ctx, signer, p)
+	require.NoError(t, err)
+	createParams.Container = id
+
+	defer func() {
+		_ = deleteContainer(ctx, id, signer, p)
+	}()
+
+	// separate context to operations to catch ContextCanceled only in Select.
+	opContext := context.Background()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+		}
+
+		_, err = rand.Read(payload)
+		require.NoError(t, err)
+
+		createParams.Payload = bytes.NewReader(payload)
+		createParams.CreationTime = time.Now()
+
+		objID, err := neo.CreateObject(opContext, createParams)
+		require.NoError(t, err)
+
+		var objReadPrm layer.PrmObjectRead
+		objReadPrm.Object = objID
+		objReadPrm.Container = id
+
+		op, err := neo.ReadObject(opContext, objReadPrm)
+		require.NoError(t, err)
+
+		pl, err := io.ReadAll(op.Payload)
+		require.NoError(t, err)
+
+		require.True(t, bytes.Equal(payload, pl))
+	}
+}