observer tests

s7techlab · Sep 22, 2023 · a9586a2 · a9586a2
1 parent 4ffcfd5
commit a9586a2
Show file tree

Hide file tree

Showing 8 changed files with 247 additions and 347 deletions.
diff --git a/client/testing/channels_fetcher_mock.go b/client/testing/channels_fetcher_mock.go
diff --git a/observer/block_peer_test.go b/observer/block_peer_test.go
@@ -1,115 +1,128 @@
 package observer_test
 
-//var (
-//	ctx = context.Background()
-//
-//	channelPeerMock *observer.ChannelPeerMock
-//	blockPeer       *observer.BlockPeer
-//
-//	channelPeerMockConcurrently *observer.ChannelPeerMock
-//	blockPeerConcurrently       *observer.BlockPeer
-//	blocksByChannels            *observer.BlocksByChannels
-//)
-
-//var _ = BeforeSuite(func() {
-//	const closeChannelWhenAllRead = true
-//	blockDelivererMock, err := sdkmocks.NewBlocksDelivererMock(fmt.Sprintf("../%s", testdata.Path), closeChannelWhenAllRead)
-//	Expect(err).ShouldNot(HaveOccurred())
-//
-//	channelPeerMock = observer.NewChannelPeerMock()
-//	for channel := range testdata.TestChannels {
-//		channelPeerMock.UpdateChannelInfo(&observer.ChannelInfo{Channel: channel})
-//	}
-//
-//	blockPeer = observer.NewBlockPeer(channelPeerMock, blockDelivererMock,
-//		observer.WithBlockStopRecreateStream(true), observer.WithBlockPeerObservePeriod(time.Nanosecond))
-//
-//	_, err = blockPeer.Observe(ctx)
-//	Expect(err).ShouldNot(HaveOccurred())
-//
-//	channelPeerMockConcurrently = observer.NewChannelPeerMock()
-//	for channel := range testdata.TestChannels {
-//		channelPeerMockConcurrently.UpdateChannelInfo(&observer.ChannelInfo{Channel: channel})
-//	}
-//
-//	blockPeerConcurrently = observer.NewBlockPeer(channelPeerMockConcurrently, blockDelivererMock,
-//		observer.WithBlockStopRecreateStream(true), observer.WithBlockPeerObservePeriod(time.Nanosecond))
-//
-//	blocksByChannels, err = blockPeerConcurrently.ObserveByChannels(ctx)
-//	Expect(err).ShouldNot(HaveOccurred())
-//})
-//
-//var _ = Describe("Block Peer", func() {
-//	Context("Channels number check", func() {
-//		Context("Block peer", func() {
-//			It("should return current number of channels", func() {
-//				channelObservers := blockPeer.ChannelObservers()
-//				Expect(channelObservers).To(HaveLen(len(testdata.TestChannels)))
-//			})
-//
-//			It("should add channels to channelPeerMock", func() {
-//				newChannels := []string{"channel1", "channel2", "channel3"}
-//				for _, channel := range newChannels {
-//					channelPeerMock.UpdateChannelInfo(&observer.ChannelInfo{Channel: channel})
-//				}
-//
-//				// wait to blockPeer observer
-//				time.Sleep(time.Millisecond * 10)
-//
-//				channelObservers := blockPeer.ChannelObservers()
-//				Expect(channelObservers).To(HaveLen(len(testdata.TestChannels) + len(newChannels)))
-//			})
-//		})
-//
-//		Context("Block peer concurrently", func() {
-//			It("should return current number of channels", func() {
-//				channelObservers := blockPeerConcurrently.ChannelObservers()
-//				Expect(channelObservers).To(HaveLen(len(testdata.TestChannels)))
-//
-//				channelsWithBlocks := blocksByChannels.Observe()
-//
-//				for i := 0; i < len(testdata.TestChannels); i++ {
-//					sampleOrFabcarChannelBlocks := <-channelsWithBlocks
-//					if sampleOrFabcarChannelBlocks.Name == testdata.SampleChannel {
-//						Expect(sampleOrFabcarChannelBlocks.Name).To(Equal(testdata.SampleChannel))
-//					} else {
-//						Expect(sampleOrFabcarChannelBlocks.Name).To(Equal(testdata.FabcarChannel))
-//					}
-//
-//					Expect(sampleOrFabcarChannelBlocks.Blocks).NotTo(BeNil())
-//				}
-//			})
-//
-//			It("should add channels to channelPeerMock", func() {
-//				channel4, channel5, channel6 := "channel4", "channel5", "channel6"
-//				newChannels := []string{channel4, channel5, channel6}
-//				for _, channel := range newChannels {
-//					channelPeerMockConcurrently.UpdateChannelInfo(&observer.ChannelInfo{Channel: channel})
-//				}
-//
-//				// wait to blockPeer observer
-//				time.Sleep(time.Millisecond * 200)
-//
-//				channelObservers := blockPeerConcurrently.ChannelObservers()
-//				Expect(channelObservers).To(HaveLen(len(testdata.TestChannels) + len(newChannels)))
-//
-//				channelsWithBlocks := blocksByChannels.Observe()
-//
-//				for i := 0; i < len(newChannels); i++ {
-//					channel4Or5Or6Blocks := <-channelsWithBlocks
-//
-//					if channel4Or5Or6Blocks.Name == channel4 {
-//						Expect(channel4Or5Or6Blocks.Name).To(Equal(channel4))
-//						Expect(channel4Or5Or6Blocks.Blocks).NotTo(BeNil())
-//					} else if channel4Or5Or6Blocks.Name == channel5 {
-//						Expect(channel4Or5Or6Blocks.Name).To(Equal(channel5))
-//						Expect(channel4Or5Or6Blocks.Blocks).NotTo(BeNil())
-//					} else {
-//						Expect(channel4Or5Or6Blocks.Name).To(Equal(channel6))
-//						Expect(channel4Or5Or6Blocks.Blocks).NotTo(BeNil())
-//					}
-//				}
-//			})
-//		})
-//	})
-//})
+import (
+	"context"
+	"fmt"
+	"time"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	sdkmocks "github.com/s7techlab/hlf-sdk-go/client/testing"
+	"github.com/s7techlab/hlf-sdk-go/observer"
+	testdata "github.com/s7techlab/hlf-sdk-go/testdata/blocks"
+)
+
+var (
+	ctx = context.Background()
+
+	channelPeerMock *observer.ChannelPeerMock
+	blockPeer       *observer.BlockPeer
+
+	channelPeerMockConcurrently *observer.ChannelPeerMock
+	blockPeerConcurrently       *observer.BlockPeer
+	blocksByChannels            *observer.BlocksByChannels
+)
+
+var _ = BeforeSuite(func() {
+	const closeChannelWhenAllRead = true
+	blockDelivererMock, err := sdkmocks.NewBlocksDelivererMock(fmt.Sprintf("../%s", testdata.Path), closeChannelWhenAllRead)
+	Expect(err).ShouldNot(HaveOccurred())
+
+	channelPeerMock = observer.NewChannelPeerMock()
+	for _, channel := range testdata.Channels {
+		channelPeerMock.UpdateChannelInfo(&observer.ChannelInfo{Channel: channel})
+	}
+
+	blockPeer = observer.NewBlockPeer(channelPeerMock, blockDelivererMock,
+		observer.WithBlockStopRecreateStream(true), observer.WithBlockPeerObservePeriod(time.Nanosecond))
+
+	_, err = blockPeer.Observe(ctx)
+	Expect(err).ShouldNot(HaveOccurred())
+
+	channelPeerMockConcurrently = observer.NewChannelPeerMock()
+	for _, channel := range testdata.Channels {
+		channelPeerMockConcurrently.UpdateChannelInfo(&observer.ChannelInfo{Channel: channel})
+	}
+
+	blockPeerConcurrently = observer.NewBlockPeer(channelPeerMockConcurrently, blockDelivererMock,
+		observer.WithBlockStopRecreateStream(true), observer.WithBlockPeerObservePeriod(time.Nanosecond))
+
+	blocksByChannels, err = blockPeerConcurrently.ObserveByChannels(ctx)
+	Expect(err).ShouldNot(HaveOccurred())
+})
+
+var _ = Describe("Block Peer", func() {
+	Context("Channels number check", func() {
+		Context("Block peer", func() {
+			It("should return current number of channels", func() {
+				channelObservers := blockPeer.ChannelObservers()
+				Expect(channelObservers).To(HaveLen(len(testdata.Channels)))
+			})
+
+			It("should add channels to channelPeerMock", func() {
+				newChannels := []string{"channel1", "channel2", "channel3"}
+				for _, channel := range newChannels {
+					channelPeerMock.UpdateChannelInfo(&observer.ChannelInfo{Channel: channel})
+				}
+
+				// wait to blockPeer observer
+				time.Sleep(time.Millisecond * 10)
+
+				channelObservers := blockPeer.ChannelObservers()
+				Expect(channelObservers).To(HaveLen(len(testdata.Channels) + len(newChannels)))
+			})
+		})
+
+		Context("Block peer concurrently", func() {
+			It("should return current number of channels", func() {
+				channelObservers := blockPeerConcurrently.ChannelObservers()
+				Expect(channelObservers).To(HaveLen(len(testdata.Channels)))
+
+				channelsWithBlocks := blocksByChannels.Observe()
+
+				for i := 0; i < len(testdata.Channels); i++ {
+					sampleOrFabcarChannelBlocks := <-channelsWithBlocks
+					if sampleOrFabcarChannelBlocks.Name == testdata.SampleChannel {
+						Expect(sampleOrFabcarChannelBlocks.Name).To(Equal(testdata.SampleChannel))
+					} else {
+						Expect(sampleOrFabcarChannelBlocks.Name).To(Equal(testdata.FabcarChannel))
+					}
+
+					Expect(sampleOrFabcarChannelBlocks.Blocks).NotTo(BeNil())
+				}
+			})
+
+			It("should add channels to channelPeerMock", func() {
+				channel4, channel5, channel6 := "channel4", "channel5", "channel6"
+				newChannels := []string{channel4, channel5, channel6}
+				for _, channel := range newChannels {
+					channelPeerMockConcurrently.UpdateChannelInfo(&observer.ChannelInfo{Channel: channel})
+				}
+
+				// wait to blockPeer observer
+				time.Sleep(time.Millisecond * 200)
+
+				channelObservers := blockPeerConcurrently.ChannelObservers()
+				Expect(channelObservers).To(HaveLen(len(testdata.Channels) + len(newChannels)))
+
+				channelsWithBlocks := blocksByChannels.Observe()
+
+				for i := 0; i < len(newChannels); i++ {
+					channel4Or5Or6Blocks := <-channelsWithBlocks
+
+					if channel4Or5Or6Blocks.Name == channel4 {
+						Expect(channel4Or5Or6Blocks.Name).To(Equal(channel4))
+						Expect(channel4Or5Or6Blocks.Blocks).NotTo(BeNil())
+					} else if channel4Or5Or6Blocks.Name == channel5 {
+						Expect(channel4Or5Or6Blocks.Name).To(Equal(channel5))
+						Expect(channel4Or5Or6Blocks.Blocks).NotTo(BeNil())
+					} else {
+						Expect(channel4Or5Or6Blocks.Name).To(Equal(channel6))
+						Expect(channel4Or5Or6Blocks.Blocks).NotTo(BeNil())
+					}
+				}
+			})
+		})
+	})
+})
diff --git a/observer/channel_peer.go b/observer/channel_peer.go
@@ -108,7 +108,7 @@ func (cp *ChannelPeer) Observe(ctx context.Context) {
 		return
 	}
 
-	// ctxObserve using for nested controll process without stopped primary context
+	// ctxObserve using for nested control process without stopped primary context
 	ctxObserve, cancel := context.WithCancel(context.Background())
 	cp.cancelObserve = cancel
 

diff --git a/observer/channel_peer_fetcher_mock.go b/observer/channel_peer_fetcher_mock.go
@@ -0,0 +1,41 @@
+package observer
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/hyperledger/fabric-protos-go/common"
+	"github.com/hyperledger/fabric-protos-go/peer"
+)
+
+type ChannelPeerFetcherMock struct {
+	channels map[string]uint64
+}
+
+func NewChannelPeerFetcherMock(channels map[string]uint64) *ChannelPeerFetcherMock {
+	return &ChannelPeerFetcherMock{
+		channels: channels,
+	}
+}
+
+func (c *ChannelPeerFetcherMock) GetChannels(ctx context.Context) (*peer.ChannelQueryResponse, error) {
+	var channels []*peer.ChannelInfo
+	for channelName := range c.channels {
+		channels = append(channels, &peer.ChannelInfo{ChannelId: channelName})
+	}
+
+	return &peer.ChannelQueryResponse{
+		Channels: channels,
+	}, nil
+}
+
+func (c *ChannelPeerFetcherMock) GetChainInfo(ctx context.Context, channel string) (*common.BlockchainInfo, error) {
+	chHeight, exists := c.channels[channel]
+	if !exists {
+		return nil, fmt.Errorf("channel '%s' does not exist", channel)
+	}
+
+	return &common.BlockchainInfo{
+		Height: chHeight,
+	}, nil
+}
diff --git a/observer/channel_peer_test.go b/observer/channel_peer_test.go
@@ -0,0 +1,60 @@
+package observer_test
+
+import (
+	"time"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	"github.com/s7techlab/hlf-sdk-go/observer"
+	testdata "github.com/s7techlab/hlf-sdk-go/testdata/blocks"
+)
+
+var _ = Describe("Channel peer", func() {
+	var (
+		channelPeerFetcherMock observer.PeerChannelsFetcher
+	)
+	BeforeEach(func() {
+		channelPeerFetcherMock = observer.NewChannelPeerFetcherMock(testdata.ChannelsHeights)
+	})
+
+	It("default channel peer, no channel matcher", func() {
+		channelPeer, err := observer.NewChannelPeer(channelPeerFetcherMock)
+		Expect(err).To(BeNil())
+
+		channelPeer.Observe(ctx)
+		time.Sleep(time.Millisecond * 100)
+
+		channelsMap := channelPeer.Channels()
+
+		sampleChannelInfo, exist := channelsMap[testdata.SampleChannel]
+		Expect(exist).To(BeTrue())
+		Expect(sampleChannelInfo.Channel).To(Equal(testdata.SampleChannel))
+		Expect(sampleChannelInfo.Height).To(Equal(testdata.SampleChannelHeight))
+
+		fabcarChannelInfo, exist := channelsMap[testdata.FabcarChannel]
+		Expect(exist).To(BeTrue())
+		Expect(fabcarChannelInfo.Channel).To(Equal(testdata.FabcarChannel))
+		Expect(fabcarChannelInfo.Height).To(Equal(testdata.FabcarChannelHeight))
+	})
+
+	It("default channel peer, with channel matcher, exclude Fabcar", func() {
+		channelPeer, err := observer.NewChannelPeer(channelPeerFetcherMock,
+			observer.WithChannels([]observer.ChannelToMatch{{MatchPattern: testdata.SampleChannel}}))
+		Expect(err).To(BeNil())
+
+		channelPeer.Observe(ctx)
+		time.Sleep(time.Millisecond * 100)
+
+		channelsMap := channelPeer.Channels()
+
+		sampleChannelInfo, exist := channelsMap[testdata.SampleChannel]
+		Expect(exist).To(BeTrue())
+		Expect(sampleChannelInfo.Channel).To(Equal(testdata.SampleChannel))
+		Expect(sampleChannelInfo.Height).To(Equal(testdata.SampleChannelHeight))
+
+		fabcarChannelInfo, exist := channelsMap[testdata.FabcarChannel]
+		Expect(exist).To(BeFalse())
+		Expect(fabcarChannelInfo).To(BeNil())
+	})
+})