achuprov/frostfs-node
1
0
Fork 0
forked from TrueCloudLab/frostfs-node
Code Pull requests Activity

[#280] ir: Add container processor unit tests

Browse source 
Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>
This commit is contained in:
Dmitrii Stepanov 2023-04-26 12:05:33 +03:00 committed by Evgenii Stratonikov
parent 686f01bce5
commit 5010b35466
11 changed files with 409 additions and 51 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
pkg/innerring/processors/container/handlers.go
View file

@ -36,7 +36,7 @@ func (cp *Processor) handleDelete(ev event.Event) {
// send an event to the worker pool
err := cp.pool.Submit(func() { cp.processContainerDelete(&del) })
err := cp.pool.Submit(func() { cp.processContainerDelete(del) })
if err != nil {
// there system can be moved into controlled degradation stage
cp.log.Warn(logs.ContainerContainerProcessorWorkerPoolDrained,

337
pkg/innerring/processors/container/handlers_test.go Normal file
View file

@ -0,0 +1,337 @@
package container
import (
"crypto/ecdsa"
"encoding/hex"
"testing"
"time"
containercore "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/container"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/morph/client"
cntClient "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/morph/client/container"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/morph/client/frostfsid"
containerEvent "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/morph/event/container"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util/logger/test"
apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
containerSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/acl"
cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
frostfscrypto "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/crypto"
frostfsecdsa "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/crypto/ecdsa"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/eacl"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/netmap"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/user"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/version"
"github.com/nspcc-dev/neo-go/pkg/crypto/keys"
"github.com/nspcc-dev/neo-go/pkg/network/payload"
"github.com/nspcc-dev/neo-go/pkg/util"
"github.com/stretchr/testify/require"
)
func TestPutEvent(t *testing.T) {
t.Parallel()
nst := &testNetworkState{
homHashDisabled: true,
epoch: 100,
}
cc := &testContainerClient{
get: make(map[string]*containercore.Container),
}
proc, err := New(&Params{
Log: test.NewLogger(t, true),
PoolSize: 2,
AlphabetState: &testAlphabetState{isAlphabet: true},
FrostFSIDClient: &testIDClient{},
NotaryDisabled: true,
NetworkState: nst,
ContainerClient: cc,
})
require.NoError(t, err, "failed to create processor")
p, err := keys.NewPrivateKey()
require.NoError(t, err)
var usr user.ID
user.IDFromKey(&usr, (ecdsa.PublicKey)(*p.PublicKey()))
var pp netmap.PlacementPolicy
pp.AddReplicas(netmap.ReplicaDescriptor{})
var cnr containerSDK.Container
cnr.Init()
cnr.SetOwner(usr)
cnr.SetPlacementPolicy(pp)
cnr.SetBasicACL(acl.Private)
containerSDK.DisableHomomorphicHashing(&cnr)
event := &testPutEvent{
cnr: &cnr,
pk: p,
st: nil,
}
proc.handlePut(event)
for proc.pool.Running() > 0 {
time.Sleep(10 * time.Millisecond)
}
var expectedPut cntClient.PutPrm
expectedPut.SetContainer(cnr.Marshal())
expectedPut.SetKey(p.PublicKey().Bytes())
expectedPut.SetSignature(p.Sign(cnr.Marshal()))
expectedPut.SetZone("container")
require.EqualValues(t, []cntClient.PutPrm{expectedPut}, cc.put, "invalid put requests")
}
func TestDeleteEvent(t *testing.T) {
t.Parallel()
nst := &testNetworkState{
homHashDisabled: true,
epoch: 100,
}
cc := &testContainerClient{
get: make(map[string]*containercore.Container),
}
p, err := keys.NewPrivateKey()
require.NoError(t, err)
idc := &testIDClient{
publicKeys: []*keys.PublicKey{
p.PublicKey(),
},
}
proc, err := New(&Params{
Log: test.NewLogger(t, true),
PoolSize: 2,
AlphabetState: &testAlphabetState{isAlphabet: true},
FrostFSIDClient: idc,
NotaryDisabled: true,
NetworkState: nst,
ContainerClient: cc,
})
require.NoError(t, err, "failed to create processor")
var usr user.ID
user.IDFromKey(&usr, (ecdsa.PublicKey)(*p.PublicKey()))
var pp netmap.PlacementPolicy
pp.AddReplicas(netmap.ReplicaDescriptor{})
var cnr containerSDK.Container
cnr.Init()
cnr.SetOwner(usr)
cnr.SetPlacementPolicy(pp)
cnr.SetBasicACL(acl.Private)
containerSDK.DisableHomomorphicHashing(&cnr)
var cid cid.ID
containerSDK.CalculateID(&cid, cnr)
cidBin := make([]byte, 32)
cid.Encode(cidBin)
ev := containerEvent.Delete{
ContainerIDValue: cidBin,
SignatureValue: p.Sign(cidBin),
}
var signature frostfscrypto.Signature
signer := frostfsecdsa.Signer(p.PrivateKey)
require.NoError(t, signature.Calculate(signer, ev.ContainerID()), "failed to calculate signature")
cc.get[hex.EncodeToString(ev.ContainerID())] = &containercore.Container{
Value: cnr,
Signature: signature,
}
proc.handleDelete(ev)
for proc.pool.Running() > 0 {
time.Sleep(10 * time.Millisecond)
}
var expectedDelete cntClient.DeletePrm
expectedDelete.SetCID(ev.ContainerID())
expectedDelete.SetSignature(ev.Signature())
require.EqualValues(t, []cntClient.DeletePrm{expectedDelete}, cc.delete, "invalid delete requests")
}
func TestSetEACLEvent(t *testing.T) {
t.Parallel()
nst := &testNetworkState{
homHashDisabled: true,
epoch: 100,
}
cc := &testContainerClient{
get: make(map[string]*containercore.Container),
}
proc, err := New(&Params{
Log: test.NewLogger(t, true),
PoolSize: 2,
AlphabetState: &testAlphabetState{isAlphabet: true},
FrostFSIDClient: &testIDClient{},
NotaryDisabled: true,
NetworkState: nst,
ContainerClient: cc,
})
require.NoError(t, err, "failed to create processor")
p, err := keys.NewPrivateKey()
require.NoError(t, err)
var usr user.ID
user.IDFromKey(&usr, (ecdsa.PublicKey)(*p.PublicKey()))
var pp netmap.PlacementPolicy
pp.AddReplicas(netmap.ReplicaDescriptor{})
var cnr containerSDK.Container
cnr.Init()
cnr.SetOwner(usr)
cnr.SetPlacementPolicy(pp)
cnr.SetBasicACL(acl.PrivateExtended)
containerSDK.DisableHomomorphicHashing(&cnr)
var cid cid.ID
containerSDK.CalculateID(&cid, cnr)
cidBytes := make([]byte, 32)
cid.Encode(cidBytes)
var signature frostfscrypto.Signature
signer := frostfsecdsa.Signer(p.PrivateKey)
require.NoError(t, signature.Calculate(signer, cidBytes), "failed to calculate signature")
cc.get[hex.EncodeToString(cidBytes)] = &containercore.Container{
Value: cnr,
Signature: signature,
}
table := eacl.NewTable()
table.SetCID(cid)
table.SetVersion(version.Current())
r := &eacl.Record{}
r.AddObjectContainerIDFilter(eacl.MatchStringEqual, cid)
table.AddRecord(r)
event := containerEvent.SetEACL{
TableValue: table.ToV2().StableMarshal(nil),
PublicKeyValue: p.PublicKey().Bytes(),
SignatureValue: p.Sign(table.ToV2().StableMarshal(nil)),
}
proc.handleSetEACL(event)
for proc.pool.Running() > 0 {
time.Sleep(10 * time.Millisecond)
}
var expectedPutEACL cntClient.PutEACLPrm
expectedPutEACL.SetTable(table.ToV2().StableMarshal(nil))
expectedPutEACL.SetKey(p.PublicKey().Bytes())
expectedPutEACL.SetSignature(p.Sign(table.ToV2().StableMarshal(nil)))
require.EqualValues(t, []cntClient.PutEACLPrm{expectedPutEACL}, cc.putEACL, "invalid set EACL requests")
}
type testAlphabetState struct {
isAlphabet bool
}
func (s *testAlphabetState) IsAlphabet() bool {
return s.isAlphabet
}
type testNetworkState struct {
homHashDisabled bool
epoch uint64
}
func (s *testNetworkState) HomomorphicHashDisabled() (bool, error) {
return s.homHashDisabled, nil
}
func (s *testNetworkState) Epoch() (uint64, error) {
return s.epoch, nil
}
type testContainerClient struct {
contractAddress util.Uint160
put []cntClient.PutPrm
get map[string]*containercore.Container
delete []cntClient.DeletePrm
putEACL []cntClient.PutEACLPrm
}
func (c *testContainerClient) ContractAddress() util.Uint160 {
return c.contractAddress
}
func (c *testContainerClient) Morph() *client.Client {
return nil
}
func (c *testContainerClient) Put(p cntClient.PutPrm) error {
c.put = append(c.put, p)
return nil
}
func (c *testContainerClient) Get(cid []byte) (*containercore.Container, error) {
key := hex.EncodeToString(cid)
if cont, found := c.get[key]; found {
return cont, nil
}
return nil, apistatus.ContainerNotFound{}
}
func (c *testContainerClient) Delete(p cntClient.DeletePrm) error {
c.delete = append(c.delete, p)
return nil
}
func (c *testContainerClient) PutEACL(p cntClient.PutEACLPrm) error {
c.putEACL = append(c.putEACL, p)
return nil
}
type testIDClient struct {
publicKeys keys.PublicKeys
}
func (c *testIDClient) AccountKeys(p frostfsid.AccountKeysPrm) (keys.PublicKeys, error) {
return c.publicKeys, nil
}
var _ putEvent = &testPutEvent{}
type testPutEvent struct {
cnr *containerSDK.Container
pk *keys.PrivateKey
st []byte
}
func (e *testPutEvent) MorphEvent() {}
func (e *testPutEvent) Container() []byte {
return e.cnr.Marshal()
}
func (e *testPutEvent) PublicKey() []byte {
return e.pk.PublicKey().Bytes()
}
func (e *testPutEvent) Signature() []byte {
return e.pk.Sign(e.cnr.Marshal())
}
func (e *testPutEvent) SessionToken() []byte {
return e.st
}
func (e *testPutEvent) NotaryRequest() *payload.P2PNotaryRequest {
return nil
}

6
pkg/innerring/processors/container/process_container.go
View file

@ -120,7 +120,7 @@ func (cp *Processor) approvePutContainer(ctx *putContainerContext) {
// Process delete container operation from the user by checking container sanity
// and sending approve tx back to morph.
func (cp *Processor) processContainerDelete(e *containerEvent.Delete) {
func (cp *Processor) processContainerDelete(e containerEvent.Delete) {
if !cp.alphabetState.IsAlphabet() {
cp.log.Info(logs.ContainerNonAlphabetModeIgnoreContainerDelete)
return
@ -138,7 +138,7 @@ func (cp *Processor) processContainerDelete(e *containerEvent.Delete) {
cp.approveDeleteContainer(e)
}
func (cp *Processor) checkDeleteContainer(e *containerEvent.Delete) error {
func (cp *Processor) checkDeleteContainer(e containerEvent.Delete) error {
binCnr := e.ContainerID()
var idCnr cid.ID
@ -170,7 +170,7 @@ func (cp *Processor) checkDeleteContainer(e *containerEvent.Delete) error {
return nil
}
func (cp *Processor) approveDeleteContainer(e *containerEvent.Delete) {
func (cp *Processor) approveDeleteContainer(e containerEvent.Delete) {
var err error
prm := cntClient.DeletePrm{}

8
pkg/innerring/processors/container/process_eacl.go
View file

@ -6,13 +6,13 @@ import (
"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
cntClient "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/morph/client/container"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/morph/event/container"
containerEvent "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/morph/event/container"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/eacl"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/session"
"go.uber.org/zap"
)
func (cp *Processor) processSetEACL(e container.SetEACL) {
func (cp *Processor) processSetEACL(e containerEvent.SetEACL) {
if !cp.alphabetState.IsAlphabet() {
cp.log.Info(logs.ContainerNonAlphabetModeIgnoreSetEACL)
return
@ -30,7 +30,7 @@ func (cp *Processor) processSetEACL(e container.SetEACL) {
cp.approveSetEACL(e)
}
func (cp *Processor) checkSetEACL(e container.SetEACL) error {
func (cp *Processor) checkSetEACL(e containerEvent.SetEACL) error {
binTable := e.Table()
// unmarshal table
@ -74,7 +74,7 @@ func (cp *Processor) checkSetEACL(e container.SetEACL) error {
return nil
}
func (cp *Processor) approveSetEACL(e container.SetEACL) {
func (cp *Processor) approveSetEACL(e containerEvent.SetEACL) {
var err error
prm := cntClient.PutEACLPrm{}

27
pkg/innerring/processors/container/processor.go
View file

@ -5,12 +5,16 @@ import (
"fmt"
"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/morph/client/container"
containercore "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/container"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/morph/client"
cntClient "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/morph/client/container"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/morph/client/frostfsid"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/morph/event"
containerEvent "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/morph/event/container"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util/logger"
"github.com/nspcc-dev/neo-go/pkg/core/mempoolevent"
"github.com/nspcc-dev/neo-go/pkg/crypto/keys"
"github.com/nspcc-dev/neo-go/pkg/util"
"github.com/panjf2000/ants/v2"
"go.uber.org/zap"
)
@ -21,13 +25,26 @@ type (
IsAlphabet() bool
}
ContClient interface {
ContractAddress() util.Uint160
Morph() *client.Client
Put(p cntClient.PutPrm) error
Get(cid []byte) (*containercore.Container, error)
Delete(p cntClient.DeletePrm) error
PutEACL(p cntClient.PutEACLPrm) error
}
IDClient interface {
AccountKeys(p frostfsid.AccountKeysPrm) (keys.PublicKeys, error)
}
// Processor of events produced by container contract in the sidechain.
Processor struct {
log *logger.Logger
pool *ants.Pool
alphabetState AlphabetState
cnrClient *container.Client // notary must be enabled
idClient *frostfsid.Client
cnrClient ContClient // notary must be enabled
idClient IDClient
netState NetworkState
notaryDisabled bool
}
@ -37,8 +54,8 @@ type (
Log *logger.Logger
PoolSize int
AlphabetState AlphabetState
ContainerClient *container.Client
FrostFSIDClient *frostfsid.Client
ContainerClient ContClient
FrostFSIDClient IDClient
NetworkState NetworkState
NotaryDisabled bool
}

6
pkg/morph/client/container/get.go
View file

@ -26,8 +26,12 @@ func AsContainerSource(w *Client) containercore.Source {
return (*containerSource)(w)
}
type getContainer interface {
Get(cid []byte) (*containercore.Container, error)
}
// Get marshals container ID, and passes it to Wrapper's Get method.
func Get(c *Client, cnr cid.ID) (*containercore.Container, error) {
func Get(c getContainer, cnr cid.ID) (*containercore.Container, error) {
binCnr := make([]byte, sha256.Size)
cnr.Encode(binCnr)

22
pkg/morph/event/container/delete.go
View file

@ -12,34 +12,34 @@ import (
// Delete structure of container.Delete notification from morph chain.
type Delete struct {
containerID []byte
signature []byte
token []byte
ContainerIDValue []byte
SignatureValue []byte
TokenValue []byte
// For notary notifications only.
// Contains raw transactions of notary request.
notaryRequest *payload.P2PNotaryRequest
NotaryRequestValue *payload.P2PNotaryRequest
}
// MorphEvent implements Neo:Morph Event interface.
func (Delete) MorphEvent() {}
// ContainerID is a marshalled container structure, defined in API.
func (d Delete) ContainerID() []byte { return d.containerID }
func (d Delete) ContainerID() []byte { return d.ContainerIDValue }
// Signature of marshalled container by container owner.
func (d Delete) Signature() []byte { return d.signature }
func (d Delete) Signature() []byte { return d.SignatureValue }
// SessionToken returns binary token of the session
// within which the eACL was set.
func (d Delete) SessionToken() []byte {
return d.token
return d.TokenValue
}
// NotaryRequest returns raw notary request if notification
// was received via notary service. Otherwise, returns nil.
func (d Delete) NotaryRequest() *payload.P2PNotaryRequest {
return d.notaryRequest
return d.NotaryRequestValue
}
const expectedItemNumDelete = 3
@ -63,19 +63,19 @@ func ParseDelete(e *state.ContainedNotificationEvent) (event.Event, error) {
}
// parse container
ev.containerID, err = client.BytesFromStackItem(params[0])
ev.ContainerIDValue, err = client.BytesFromStackItem(params[0])
if err != nil {
return nil, fmt.Errorf("could not get container: %w", err)
}
// parse signature
ev.signature, err = client.BytesFromStackItem(params[1])
ev.SignatureValue, err = client.BytesFromStackItem(params[1])
if err != nil {
return nil, fmt.Errorf("could not get signature: %w", err)
}
// parse session token
ev.token, err = client.BytesFromStackItem(params[2])
ev.TokenValue, err = client.BytesFromStackItem(params[2])
if err != nil {
return nil, fmt.Errorf("could not get session token: %w", err)
}

8
pkg/morph/event/container/delete_notary.go
View file

@ -7,19 +7,19 @@ import (
func (d *Delete) setContainerID(v []byte) {
if v != nil {
d.containerID = v
d.ContainerIDValue = v
}
}
func (d *Delete) setSignature(v []byte) {
if v != nil {
d.signature = v
d.SignatureValue = v
}
}
func (d *Delete) setToken(v []byte) {
if v != nil {
d.token = v
d.TokenValue = v
}
}
@ -62,7 +62,7 @@ func ParseDeleteNotary(ne event.NotaryEvent) (event.Event, error) {
}
}
ev.notaryRequest = ne.Raw()
ev.NotaryRequestValue = ne.Raw()
return ev, nil
}

6
pkg/morph/event/container/delete_test.go
View file

@ -63,9 +63,9 @@ func TestParseDelete(t *testing.T) {
require.NoError(t, err)
require.Equal(t, Delete{
containerID: containerID,
signature: signature,
token: token,
ContainerIDValue: containerID,
SignatureValue: signature,
TokenValue: token,
}, ev)
})
}

28
pkg/morph/event/container/eacl.go
View file

@ -12,14 +12,14 @@ import (
// SetEACL represents structure of notification about
// modified eACL table coming from FrostFS Container contract.
type SetEACL struct {
table []byte
signature []byte
publicKey []byte
token []byte
TableValue []byte
SignatureValue []byte
PublicKeyValue []byte
TokenValue []byte
// For notary notifications only.
// Contains raw transactions of notary request.
notaryRequest *payload.P2PNotaryRequest
NotaryRequestValue *payload.P2PNotaryRequest
}
// MorphEvent implements Neo:Morph Event interface.
@ -27,30 +27,30 @@ func (SetEACL) MorphEvent() {}
// Table returns returns eACL table in a binary FrostFS API format.
func (x SetEACL) Table() []byte {
return x.table
return x.TableValue
}
// Signature returns signature of the binary table.
func (x SetEACL) Signature() []byte {
return x.signature
return x.SignatureValue
}
// PublicKey returns public keys of container
// owner in a binary format.
func (x SetEACL) PublicKey() []byte {
return x.publicKey
return x.PublicKeyValue
}
// SessionToken returns binary token of the session
// within which the eACL was set.
func (x SetEACL) SessionToken() []byte {
return x.token
return x.TokenValue
}
// NotaryRequest returns raw notary request if notification
// was received via notary service. Otherwise, returns nil.
func (x SetEACL) NotaryRequest() *payload.P2PNotaryRequest {
return x.notaryRequest
return x.NotaryRequestValue
}
const expectedItemNumEACL = 4
@ -74,25 +74,25 @@ func ParseSetEACL(e *state.ContainedNotificationEvent) (event.Event, error) {
}
// parse table
ev.table, err = client.BytesFromStackItem(params[0])
ev.TableValue, err = client.BytesFromStackItem(params[0])
if err != nil {
return nil, fmt.Errorf("could not parse binary table: %w", err)
}
// parse signature
ev.signature, err = client.BytesFromStackItem(params[1])
ev.SignatureValue, err = client.BytesFromStackItem(params[1])
if err != nil {
return nil, fmt.Errorf("could not parse table signature: %w", err)
}
// parse public key
ev.publicKey, err = client.BytesFromStackItem(params[2])
ev.PublicKeyValue, err = client.BytesFromStackItem(params[2])
if err != nil {
return nil, fmt.Errorf("could not parse binary public key: %w", err)
}
// parse session token
ev.token, err = client.BytesFromStackItem(params[3])
ev.TokenValue, err = client.BytesFromStackItem(params[3])
if err != nil {
return nil, fmt.Errorf("could not get session token: %w", err)
}

10
pkg/morph/event/container/eacl_notary.go
View file

@ -7,25 +7,25 @@ import (
func (x *SetEACL) setTable(v []byte) {
if v != nil {
x.table = v
x.TableValue = v
}
}
func (x *SetEACL) setSignature(v []byte) {
if v != nil {
x.signature = v
x.SignatureValue = v
}
}
func (x *SetEACL) setPublicKey(v []byte) {
if v != nil {
x.publicKey = v
x.PublicKeyValue = v
}
}
func (x *SetEACL) setToken(v []byte) {
if v != nil {
x.token = v
x.TokenValue = v
}
}
@ -69,7 +69,7 @@ func ParseSetEACLNotary(ne event.NotaryEvent) (event.Event, error) {
}
}
ev.notaryRequest = ne.Raw()
ev.NotaryRequestValue = ne.Raw()
return ev, nil
}