frostfs-node/pkg/morph/event/listener_test.go

package event

import (
	"context"
	"fmt"
	"testing"

	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/morph/subscriber"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util/logger/test"
	"github.com/nspcc-dev/neo-go/pkg/core/block"
	"github.com/nspcc-dev/neo-go/pkg/core/state"
	"github.com/nspcc-dev/neo-go/pkg/neorpc/result"
	"github.com/nspcc-dev/neo-go/pkg/util"
	"github.com/stretchr/testify/require"
)

func TestEventHandling(t *testing.T) {
	blockCh := make(chan *block.Block)
	notificationCh := make(chan *state.ContainedNotificationEvent)
	notaryRequestsCh := make(chan *result.NotaryRequestEvent)

	l, err := NewListener(ListenerParams{
		Logger: test.NewLogger(t),
		Subscriber: &testSubscriber{
			blockCh:          blockCh,
			notificationCh:   notificationCh,
			notaryRequestsCh: notaryRequestsCh,
		},
		WorkerPoolCapacity: 10,
	})
	require.NoError(t, err, "failed to create listener")

	list := l.(*listener)

	blockHandled := make(chan bool)
	handledBlocks := make([]*block.Block, 0)
	l.RegisterBlockHandler(func(b *block.Block) {
		handledBlocks = append(handledBlocks, b)
		blockHandled <- true
	})

	key := scriptHashWithType{
		scriptHashValue: scriptHashValue{
			hash: util.Uint160{100},
		},
		typeValue: typeValue{
			typ: TypeFromString("notification type"),
		},
	}

	l.SetNotificationParser(NotificationParserInfo{
		scriptHashWithType: key,
		p: func(cne *state.ContainedNotificationEvent) (Event, error) {
			return testNotificationEvent{source: cne}, nil
		},
	})

	notificationHandled := make(chan bool)
	handledNotifications := make([]Event, 0)
	l.RegisterNotificationHandler(NotificationHandlerInfo{
		scriptHashWithType: key,
		h: func(e Event) {
			handledNotifications = append(handledNotifications, e)
			notificationHandled <- true
		},
	})

	go list.Listen(context.Background())

	t.Run("handles block events", func(t *testing.T) {
		block := &block.Block{}

		blockCh <- block

		<-blockHandled

		require.Equal(t, 1, len(handledBlocks), "invalid handled blocks length")
		require.Equal(t, block, handledBlocks[0], "invalid handled block")
	})

	t.Run("handles notifications", func(t *testing.T) {
		notification := &state.ContainedNotificationEvent{
			Container: util.Uint256{49},
			NotificationEvent: state.NotificationEvent{
				ScriptHash: util.Uint160{100},
				Name:       "notification type",
			},
		}

		notificationCh <- notification

		<-notificationHandled
		require.EqualValues(t, []Event{testNotificationEvent{source: notification}}, handledNotifications, "invalid handled notifications")
	})
}

func TestErrorPassing(t *testing.T) {
	blockCh := make(chan *block.Block)
	notificationCh := make(chan *state.ContainedNotificationEvent)
	notaryRequestsCh := make(chan *result.NotaryRequestEvent)

	t.Run("notification error", func(t *testing.T) {
		nErr := fmt.Errorf("notification error")
		l, err := NewListener(ListenerParams{
			Logger: test.NewLogger(t),
			Subscriber: &testSubscriber{
				blockCh:          blockCh,
				notificationCh:   notificationCh,
				notaryRequestsCh: notaryRequestsCh,

				notificationErr: nErr,
			},
			WorkerPoolCapacity: 10,
		})
		require.NoError(t, err, "failed to create listener")

		errCh := make(chan error)

		go l.ListenWithError(context.Background(), errCh)

		err = <-errCh

		require.ErrorIs(t, err, nErr, "invalid notification error")
	})

	t.Run("block error", func(t *testing.T) {
		bErr := fmt.Errorf("notification error")
		l, err := NewListener(ListenerParams{
			Logger: test.NewLogger(t),
			Subscriber: &testSubscriber{
				blockCh:          blockCh,
				notificationCh:   notificationCh,
				notaryRequestsCh: notaryRequestsCh,

				blockErr: bErr,
			},
			WorkerPoolCapacity: 10,
		})
		require.NoError(t, err, "failed to create listener")
		l.RegisterBlockHandler(func(b *block.Block) {})

		errCh := make(chan error)

		go l.ListenWithError(context.Background(), errCh)

		err = <-errCh

		require.ErrorIs(t, err, bErr, "invalid block error")
	})
}

type testSubscriber struct {
	blockCh          chan *block.Block
	notificationCh   chan *state.ContainedNotificationEvent
	notaryRequestsCh chan *result.NotaryRequestEvent

	blockErr        error
	notificationErr error
}

func (s *testSubscriber) SubscribeForNotification(...util.Uint160) error {
	return s.notificationErr
}
func (s *testSubscriber) UnsubscribeForNotification() {}
func (s *testSubscriber) BlockNotifications() error {
	return s.blockErr
}

func (s *testSubscriber) SubscribeForNotaryRequests(mainTXSigner util.Uint160) error {
	return nil
}

func (s *testSubscriber) NotificationChannels() subscriber.NotificationChannels {
	return subscriber.NotificationChannels{
		BlockCh:          s.blockCh,
		NotificationsCh:  s.notificationCh,
		NotaryRequestsCh: s.notaryRequestsCh,
	}
}

func (s *testSubscriber) Close() {}

type testNotificationEvent struct {
	source *state.ContainedNotificationEvent
}

func (e testNotificationEvent) MorphEvent() {}