// Copyright 2018 The go-ethereum Authors // This file is part of the go-ethereum library. // // The go-ethereum library is free software: you can redistribute it and/or modify // it under the terms of the GNU Lesser General Public License as published by // the Free Software Foundation, either version 3 of the License, or // (at your option) any later version. // // The go-ethereum library is distributed in the hope that it will be useful, // but WITHOUT ANY WARRANTY; without even the implied warranty of // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the // GNU Lesser General Public License for more details. // // You should have received a copy of the GNU Lesser General Public License // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>. package stream import ( "bytes" "context" "errors" "fmt" "strconv" "strings" "sync" "testing" "time" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/node" "github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/p2p/simulations/adapters" p2ptest "github.com/ethereum/go-ethereum/p2p/testing" "github.com/ethereum/go-ethereum/swarm/network" "github.com/ethereum/go-ethereum/swarm/network/simulation" "github.com/ethereum/go-ethereum/swarm/state" "golang.org/x/crypto/sha3" ) func TestStreamerSubscribe(t *testing.T) { tester, streamer, _, teardown, err := newStreamerTester(nil) defer teardown() if err != nil { t.Fatal(err) } stream := NewStream("foo", "", true) err = streamer.Subscribe(tester.Nodes[0].ID(), stream, NewRange(0, 0), Top) if err == nil || err.Error() != "stream foo not registered" { t.Fatalf("Expected error %v, got %v", "stream foo not registered", err) } } func TestStreamerRequestSubscription(t *testing.T) { tester, streamer, _, teardown, err := newStreamerTester(nil) defer teardown() if err != nil { t.Fatal(err) } stream := NewStream("foo", "", false) err = streamer.RequestSubscription(tester.Nodes[0].ID(), stream, &Range{}, Top) if err == nil || err.Error() != "stream foo not registered" { t.Fatalf("Expected error %v, got %v", "stream foo not registered", err) } } var ( hash0 = sha3.Sum256([]byte{0}) hash1 = sha3.Sum256([]byte{1}) hash2 = sha3.Sum256([]byte{2}) hashesTmp = append(hash0[:], hash1[:]...) hashes = append(hashesTmp, hash2[:]...) corruptHashes = append(hashes[:40]) ) type testClient struct { t string wait0 chan bool wait2 chan bool batchDone chan bool receivedHashes map[string][]byte } func newTestClient(t string) *testClient { return &testClient{ t: t, wait0: make(chan bool), wait2: make(chan bool), batchDone: make(chan bool), receivedHashes: make(map[string][]byte), } } func (self *testClient) NeedData(ctx context.Context, hash []byte) func(context.Context) error { self.receivedHashes[string(hash)] = hash if bytes.Equal(hash, hash0[:]) { return func(context.Context) error { <-self.wait0 return nil } } else if bytes.Equal(hash, hash2[:]) { return func(context.Context) error { <-self.wait2 return nil } } return nil } func (self *testClient) BatchDone(Stream, uint64, []byte, []byte) func() (*TakeoverProof, error) { close(self.batchDone) return nil } func (self *testClient) Close() {} type testServer struct { t string sessionIndex uint64 } func newTestServer(t string, sessionIndex uint64) *testServer { return &testServer{ t: t, sessionIndex: sessionIndex, } } func (s *testServer) SessionIndex() (uint64, error) { return s.sessionIndex, nil } func (self *testServer) SetNextBatch(from uint64, to uint64) ([]byte, uint64, uint64, *HandoverProof, error) { return make([]byte, HashSize), from + 1, to + 1, nil, nil } func (self *testServer) GetData(context.Context, []byte) ([]byte, error) { return nil, nil } func (self *testServer) Close() { } func TestStreamerDownstreamSubscribeUnsubscribeMsgExchange(t *testing.T) { tester, streamer, _, teardown, err := newStreamerTester(nil) defer teardown() if err != nil { t.Fatal(err) } streamer.RegisterClientFunc("foo", func(p *Peer, t string, live bool) (Client, error) { return newTestClient(t), nil }) node := tester.Nodes[0] stream := NewStream("foo", "", true) err = streamer.Subscribe(node.ID(), stream, NewRange(5, 8), Top) if err != nil { t.Fatalf("Expected no error, got %v", err) } err = tester.TestExchanges( p2ptest.Exchange{ Label: "Subscribe message", Expects: []p2ptest.Expect{ { Code: 4, Msg: &SubscribeMsg{ Stream: stream, History: NewRange(5, 8), Priority: Top, }, Peer: node.ID(), }, }, }, // trigger OfferedHashesMsg to actually create the client p2ptest.Exchange{ Label: "OfferedHashes message", Triggers: []p2ptest.Trigger{ { Code: 1, Msg: &OfferedHashesMsg{ HandoverProof: &HandoverProof{ Handover: &Handover{}, }, Hashes: hashes, From: 5, To: 8, Stream: stream, }, Peer: node.ID(), }, }, Expects: []p2ptest.Expect{ { Code: 2, Msg: &WantedHashesMsg{ Stream: stream, Want: []byte{5}, From: 9, To: 0, }, Peer: node.ID(), }, }, }, ) if err != nil { t.Fatal(err) } err = streamer.Unsubscribe(node.ID(), stream) if err != nil { t.Fatalf("Expected no error, got %v", err) } err = tester.TestExchanges(p2ptest.Exchange{ Label: "Unsubscribe message", Expects: []p2ptest.Expect{ { Code: 0, Msg: &UnsubscribeMsg{ Stream: stream, }, Peer: node.ID(), }, }, }) if err != nil { t.Fatal(err) } } func TestStreamerUpstreamSubscribeUnsubscribeMsgExchange(t *testing.T) { tester, streamer, _, teardown, err := newStreamerTester(nil) defer teardown() if err != nil { t.Fatal(err) } stream := NewStream("foo", "", false) streamer.RegisterServerFunc("foo", func(p *Peer, t string, live bool) (Server, error) { return newTestServer(t, 10), nil }) node := tester.Nodes[0] err = tester.TestExchanges(p2ptest.Exchange{ Label: "Subscribe message", Triggers: []p2ptest.Trigger{ { Code: 4, Msg: &SubscribeMsg{ Stream: stream, History: NewRange(5, 8), Priority: Top, }, Peer: node.ID(), }, }, Expects: []p2ptest.Expect{ { Code: 1, Msg: &OfferedHashesMsg{ Stream: stream, HandoverProof: &HandoverProof{ Handover: &Handover{}, }, Hashes: make([]byte, HashSize), From: 6, To: 9, }, Peer: node.ID(), }, }, }) if err != nil { t.Fatal(err) } err = tester.TestExchanges(p2ptest.Exchange{ Label: "unsubscribe message", Triggers: []p2ptest.Trigger{ { Code: 0, Msg: &UnsubscribeMsg{ Stream: stream, }, Peer: node.ID(), }, }, }) if err != nil { t.Fatal(err) } } func TestStreamerUpstreamSubscribeUnsubscribeMsgExchangeLive(t *testing.T) { tester, streamer, _, teardown, err := newStreamerTester(nil) defer teardown() if err != nil { t.Fatal(err) } stream := NewStream("foo", "", true) streamer.RegisterServerFunc("foo", func(p *Peer, t string, live bool) (Server, error) { return newTestServer(t, 0), nil }) node := tester.Nodes[0] err = tester.TestExchanges(p2ptest.Exchange{ Label: "Subscribe message", Triggers: []p2ptest.Trigger{ { Code: 4, Msg: &SubscribeMsg{ Stream: stream, Priority: Top, }, Peer: node.ID(), }, }, Expects: []p2ptest.Expect{ { Code: 1, Msg: &OfferedHashesMsg{ Stream: stream, HandoverProof: &HandoverProof{ Handover: &Handover{}, }, Hashes: make([]byte, HashSize), From: 1, To: 0, }, Peer: node.ID(), }, }, }) if err != nil { t.Fatal(err) } err = tester.TestExchanges(p2ptest.Exchange{ Label: "unsubscribe message", Triggers: []p2ptest.Trigger{ { Code: 0, Msg: &UnsubscribeMsg{ Stream: stream, }, Peer: node.ID(), }, }, }) if err != nil { t.Fatal(err) } } func TestStreamerUpstreamSubscribeErrorMsgExchange(t *testing.T) { tester, streamer, _, teardown, err := newStreamerTester(nil) defer teardown() if err != nil { t.Fatal(err) } streamer.RegisterServerFunc("foo", func(p *Peer, t string, live bool) (Server, error) { return newTestServer(t, 0), nil }) stream := NewStream("bar", "", true) node := tester.Nodes[0] err = tester.TestExchanges(p2ptest.Exchange{ Label: "Subscribe message", Triggers: []p2ptest.Trigger{ { Code: 4, Msg: &SubscribeMsg{ Stream: stream, History: NewRange(5, 8), Priority: Top, }, Peer: node.ID(), }, }, Expects: []p2ptest.Expect{ { Code: 7, Msg: &SubscribeErrorMsg{ Error: "stream bar not registered", }, Peer: node.ID(), }, }, }) if err != nil { t.Fatal(err) } } func TestStreamerUpstreamSubscribeLiveAndHistory(t *testing.T) { tester, streamer, _, teardown, err := newStreamerTester(nil) defer teardown() if err != nil { t.Fatal(err) } stream := NewStream("foo", "", true) streamer.RegisterServerFunc("foo", func(p *Peer, t string, live bool) (Server, error) { return newTestServer(t, 10), nil }) node := tester.Nodes[0] err = tester.TestExchanges(p2ptest.Exchange{ Label: "Subscribe message", Triggers: []p2ptest.Trigger{ { Code: 4, Msg: &SubscribeMsg{ Stream: stream, History: NewRange(5, 8), Priority: Top, }, Peer: node.ID(), }, }, Expects: []p2ptest.Expect{ { Code: 1, Msg: &OfferedHashesMsg{ Stream: NewStream("foo", "", false), HandoverProof: &HandoverProof{ Handover: &Handover{}, }, Hashes: make([]byte, HashSize), From: 6, To: 9, }, Peer: node.ID(), }, { Code: 1, Msg: &OfferedHashesMsg{ Stream: stream, HandoverProof: &HandoverProof{ Handover: &Handover{}, }, From: 11, To: 0, Hashes: make([]byte, HashSize), }, Peer: node.ID(), }, }, }) if err != nil { t.Fatal(err) } } func TestStreamerDownstreamCorruptHashesMsgExchange(t *testing.T) { tester, streamer, _, teardown, err := newStreamerTester(nil) defer teardown() if err != nil { t.Fatal(err) } stream := NewStream("foo", "", true) var tc *testClient streamer.RegisterClientFunc("foo", func(p *Peer, t string, live bool) (Client, error) { tc = newTestClient(t) return tc, nil }) node := tester.Nodes[0] err = streamer.Subscribe(node.ID(), stream, NewRange(5, 8), Top) if err != nil { t.Fatalf("Expected no error, got %v", err) } err = tester.TestExchanges(p2ptest.Exchange{ Label: "Subscribe message", Expects: []p2ptest.Expect{ { Code: 4, Msg: &SubscribeMsg{ Stream: stream, History: NewRange(5, 8), Priority: Top, }, Peer: node.ID(), }, }, }, p2ptest.Exchange{ Label: "Corrupt offered hash message", Triggers: []p2ptest.Trigger{ { Code: 1, Msg: &OfferedHashesMsg{ HandoverProof: &HandoverProof{ Handover: &Handover{}, }, Hashes: corruptHashes, From: 5, To: 8, Stream: stream, }, Peer: node.ID(), }, }, }) if err != nil { t.Fatal(err) } expectedError := errors.New("Message handler error: (msg code 1): error invalid hashes length (len: 40)") if err := tester.TestDisconnected(&p2ptest.Disconnect{Peer: node.ID(), Error: expectedError}); err != nil { t.Fatal(err) } } func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) { tester, streamer, _, teardown, err := newStreamerTester(nil) defer teardown() if err != nil { t.Fatal(err) } stream := NewStream("foo", "", true) var tc *testClient streamer.RegisterClientFunc("foo", func(p *Peer, t string, live bool) (Client, error) { tc = newTestClient(t) return tc, nil }) node := tester.Nodes[0] err = streamer.Subscribe(node.ID(), stream, NewRange(5, 8), Top) if err != nil { t.Fatalf("Expected no error, got %v", err) } err = tester.TestExchanges(p2ptest.Exchange{ Label: "Subscribe message", Expects: []p2ptest.Expect{ { Code: 4, Msg: &SubscribeMsg{ Stream: stream, History: NewRange(5, 8), Priority: Top, }, Peer: node.ID(), }, }, }, p2ptest.Exchange{ Label: "WantedHashes message", Triggers: []p2ptest.Trigger{ { Code: 1, Msg: &OfferedHashesMsg{ HandoverProof: &HandoverProof{ Handover: &Handover{}, }, Hashes: hashes, From: 5, To: 8, Stream: stream, }, Peer: node.ID(), }, }, Expects: []p2ptest.Expect{ { Code: 2, Msg: &WantedHashesMsg{ Stream: stream, Want: []byte{5}, From: 9, To: 0, }, Peer: node.ID(), }, }, }) if err != nil { t.Fatal(err) } if len(tc.receivedHashes) != 3 { t.Fatalf("Expected number of received hashes %v, got %v", 3, len(tc.receivedHashes)) } close(tc.wait0) timeout := time.NewTimer(100 * time.Millisecond) defer timeout.Stop() select { case <-tc.batchDone: t.Fatal("batch done early") case <-timeout.C: } close(tc.wait2) timeout2 := time.NewTimer(10000 * time.Millisecond) defer timeout2.Stop() select { case <-tc.batchDone: case <-timeout2.C: t.Fatal("timeout waiting batchdone call") } } func TestStreamerRequestSubscriptionQuitMsgExchange(t *testing.T) { tester, streamer, _, teardown, err := newStreamerTester(nil) defer teardown() if err != nil { t.Fatal(err) } streamer.RegisterServerFunc("foo", func(p *Peer, t string, live bool) (Server, error) { return newTestServer(t, 10), nil }) node := tester.Nodes[0] stream := NewStream("foo", "", true) err = streamer.RequestSubscription(node.ID(), stream, NewRange(5, 8), Top) if err != nil { t.Fatalf("Expected no error, got %v", err) } err = tester.TestExchanges( p2ptest.Exchange{ Label: "RequestSubscription message", Expects: []p2ptest.Expect{ { Code: 8, Msg: &RequestSubscriptionMsg{ Stream: stream, History: NewRange(5, 8), Priority: Top, }, Peer: node.ID(), }, }, }, p2ptest.Exchange{ Label: "Subscribe message", Triggers: []p2ptest.Trigger{ { Code: 4, Msg: &SubscribeMsg{ Stream: stream, History: NewRange(5, 8), Priority: Top, }, Peer: node.ID(), }, }, Expects: []p2ptest.Expect{ { Code: 1, Msg: &OfferedHashesMsg{ Stream: NewStream("foo", "", false), HandoverProof: &HandoverProof{ Handover: &Handover{}, }, Hashes: make([]byte, HashSize), From: 6, To: 9, }, Peer: node.ID(), }, { Code: 1, Msg: &OfferedHashesMsg{ Stream: stream, HandoverProof: &HandoverProof{ Handover: &Handover{}, }, From: 11, To: 0, Hashes: make([]byte, HashSize), }, Peer: node.ID(), }, }, }, ) if err != nil { t.Fatal(err) } err = streamer.Quit(node.ID(), stream) if err != nil { t.Fatalf("Expected no error, got %v", err) } err = tester.TestExchanges(p2ptest.Exchange{ Label: "Quit message", Expects: []p2ptest.Expect{ { Code: 9, Msg: &QuitMsg{ Stream: stream, }, Peer: node.ID(), }, }, }) if err != nil { t.Fatal(err) } historyStream := getHistoryStream(stream) err = streamer.Quit(node.ID(), historyStream) if err != nil { t.Fatalf("Expected no error, got %v", err) } err = tester.TestExchanges(p2ptest.Exchange{ Label: "Quit message", Expects: []p2ptest.Expect{ { Code: 9, Msg: &QuitMsg{ Stream: historyStream, }, Peer: node.ID(), }, }, }) if err != nil { t.Fatal(err) } } // TestMaxPeerServersWithUnsubscribe creates a registry with a limited // number of stream servers, and performs a test with subscriptions and // unsubscriptions, checking if unsubscriptions will remove streams, // leaving place for new streams. func TestMaxPeerServersWithUnsubscribe(t *testing.T) { var maxPeerServers = 6 tester, streamer, _, teardown, err := newStreamerTester(&RegistryOptions{ Retrieval: RetrievalDisabled, Syncing: SyncingDisabled, MaxPeerServers: maxPeerServers, }) defer teardown() if err != nil { t.Fatal(err) } streamer.RegisterServerFunc("foo", func(p *Peer, t string, live bool) (Server, error) { return newTestServer(t, 0), nil }) node := tester.Nodes[0] for i := 0; i < maxPeerServers+10; i++ { stream := NewStream("foo", strconv.Itoa(i), true) err = tester.TestExchanges(p2ptest.Exchange{ Label: "Subscribe message", Triggers: []p2ptest.Trigger{ { Code: 4, Msg: &SubscribeMsg{ Stream: stream, Priority: Top, }, Peer: node.ID(), }, }, Expects: []p2ptest.Expect{ { Code: 1, Msg: &OfferedHashesMsg{ Stream: stream, HandoverProof: &HandoverProof{ Handover: &Handover{}, }, Hashes: make([]byte, HashSize), From: 1, To: 0, }, Peer: node.ID(), }, }, }) if err != nil { t.Fatal(err) } err = tester.TestExchanges(p2ptest.Exchange{ Label: "unsubscribe message", Triggers: []p2ptest.Trigger{ { Code: 0, Msg: &UnsubscribeMsg{ Stream: stream, }, Peer: node.ID(), }, }, }) if err != nil { t.Fatal(err) } } } // TestMaxPeerServersWithoutUnsubscribe creates a registry with a limited // number of stream servers, and performs subscriptions to detect subscriptions // error message exchange. func TestMaxPeerServersWithoutUnsubscribe(t *testing.T) { var maxPeerServers = 6 tester, streamer, _, teardown, err := newStreamerTester(&RegistryOptions{ MaxPeerServers: maxPeerServers, }) defer teardown() if err != nil { t.Fatal(err) } streamer.RegisterServerFunc("foo", func(p *Peer, t string, live bool) (Server, error) { return newTestServer(t, 0), nil }) node := tester.Nodes[0] for i := 0; i < maxPeerServers+10; i++ { stream := NewStream("foo", strconv.Itoa(i), true) if i >= maxPeerServers { err = tester.TestExchanges(p2ptest.Exchange{ Label: "Subscribe message", Triggers: []p2ptest.Trigger{ { Code: 4, Msg: &SubscribeMsg{ Stream: stream, Priority: Top, }, Peer: node.ID(), }, }, Expects: []p2ptest.Expect{ { Code: 7, Msg: &SubscribeErrorMsg{ Error: ErrMaxPeerServers.Error(), }, Peer: node.ID(), }, }, }) if err != nil { t.Fatal(err) } continue } err = tester.TestExchanges(p2ptest.Exchange{ Label: "Subscribe message", Triggers: []p2ptest.Trigger{ { Code: 4, Msg: &SubscribeMsg{ Stream: stream, Priority: Top, }, Peer: node.ID(), }, }, Expects: []p2ptest.Expect{ { Code: 1, Msg: &OfferedHashesMsg{ Stream: stream, HandoverProof: &HandoverProof{ Handover: &Handover{}, }, Hashes: make([]byte, HashSize), From: 1, To: 0, }, Peer: node.ID(), }, }, }) if err != nil { t.Fatal(err) } } } //TestHasPriceImplementation is to check that the Registry has a //`Price` interface implementation func TestHasPriceImplementation(t *testing.T) { _, r, _, teardown, err := newStreamerTester(&RegistryOptions{ Retrieval: RetrievalDisabled, Syncing: SyncingDisabled, }) defer teardown() if err != nil { t.Fatal(err) } if r.prices == nil { t.Fatal("No prices implementation available for the stream protocol") } pricesInstance, ok := r.prices.(*StreamerPrices) if !ok { t.Fatal("`Registry` does not have the expected Prices instance") } price := pricesInstance.Price(&ChunkDeliveryMsgRetrieval{}) if price == nil || price.Value == 0 || price.Value != pricesInstance.getChunkDeliveryMsgRetrievalPrice() { t.Fatal("No prices set for chunk delivery msg") } price = pricesInstance.Price(&RetrieveRequestMsg{}) if price == nil || price.Value == 0 || price.Value != pricesInstance.getRetrieveRequestMsgPrice() { t.Fatal("No prices set for chunk delivery msg") } } /* TestRequestPeerSubscriptions is a unit test for stream's pull sync subscriptions. The test does: * assign each connected peer to a bin map * build up a known kademlia in advance * run the EachConn function, which returns supposed subscription bins * store all supposed bins per peer in a map * check that all peers have the expected subscriptions This kad table and its peers are copied from network.TestKademliaCase1, it represents an edge case but for the purpose of testing the syncing subscriptions it is just fine. Addresses used in this test are discovered as part of the simulation network in higher level tests for streaming. They were generated randomly. The resulting kademlia looks like this: ========================================================================= Fri Dec 21 20:02:39 UTC 2018 KΛÐΞMLIΛ hive: queen's address: 7efef1 population: 12 (12), MinProxBinSize: 2, MinBinSize: 2, MaxBinSize: 4 000 2 8196 835f | 2 8196 (0) 835f (0) 001 2 2690 28f0 | 2 2690 (0) 28f0 (0) 002 2 4d72 4a45 | 2 4d72 (0) 4a45 (0) 003 1 646e | 1 646e (0) 004 3 769c 76d1 7656 | 3 769c (0) 76d1 (0) 7656 (0) ============ DEPTH: 5 ========================================== 005 1 7a48 | 1 7a48 (0) 006 1 7cbd | 1 7cbd (0) 007 0 | 0 008 0 | 0 009 0 | 0 010 0 | 0 011 0 | 0 012 0 | 0 013 0 | 0 014 0 | 0 015 0 | 0 ========================================================================= */ func TestRequestPeerSubscriptions(t *testing.T) { // the pivot address; this is the actual kademlia node pivotAddr := "7efef1c41d77f843ad167be95f6660567eb8a4a59f39240000cce2e0d65baf8e" // a map of bin number to addresses from the given kademlia binMap := make(map[int][]string) binMap[0] = []string{ "835fbbf1d16ba7347b6e2fc552d6e982148d29c624ea20383850df3c810fa8fc", "81968a2d8fb39114342ee1da85254ec51e0608d7f0f6997c2a8354c260a71009", } binMap[1] = []string{ "28f0bc1b44658548d6e05dd16d4c2fe77f1da5d48b6774bc4263b045725d0c19", "2690a910c33ee37b91eb6c4e0731d1d345e2dc3b46d308503a6e85bbc242c69e", } binMap[2] = []string{ "4a45f1fc63e1a9cb9dfa44c98da2f3d20c2923e5d75ff60b2db9d1bdb0c54d51", "4d72a04ddeb851a68cd197ef9a92a3e2ff01fbbff638e64929dd1a9c2e150112", } binMap[3] = []string{ "646e9540c84f6a2f9cf6585d45a4c219573b4fd1b64a3c9a1386fc5cf98c0d4d", } binMap[4] = []string{ "7656caccdc79cd8d7ce66d415cc96a718e8271c62fb35746bfc2b49faf3eebf3", "76d1e83c71ca246d042e37ff1db181f2776265fbcfdc890ce230bfa617c9c2f0", "769ce86aa90b518b7ed382f9fdacfbed93574e18dc98fe6c342e4f9f409c2d5a", } binMap[5] = []string{ "7a48f75f8ca60487ae42d6f92b785581b40b91f2da551ae73d5eae46640e02e8", } binMap[6] = []string{ "7cbd42350bde8e18ae5b955b5450f8e2cef3419f92fbf5598160c60fd78619f0", } // create the pivot's kademlia addr := common.FromHex(pivotAddr) k := network.NewKademlia(addr, network.NewKadParams()) // construct the peers and the kademlia for _, binaddrs := range binMap { for _, a := range binaddrs { addr := common.FromHex(a) k.On(network.NewPeer(&network.BzzPeer{BzzAddr: &network.BzzAddr{OAddr: addr}}, k)) } } // TODO: check kad table is same // currently k.String() prints date so it will never be the same :) // --> implement JSON representation of kad table log.Debug(k.String()) // simulate that we would do subscriptions: just store the bin numbers fakeSubscriptions := make(map[string][]int) //after the test, we need to reset the subscriptionFunc to the default defer func() { subscriptionFunc = doRequestSubscription }() // define the function which should run for each connection // instead of doing real subscriptions, we just store the bin numbers subscriptionFunc = func(r *Registry, p *network.Peer, bin uint8, subs map[enode.ID]map[Stream]struct{}) bool { // get the peer ID peerstr := fmt.Sprintf("%x", p.Over()) // create the array of bins per peer if _, ok := fakeSubscriptions[peerstr]; !ok { fakeSubscriptions[peerstr] = make([]int, 0) } // store the (fake) bin subscription log.Debug(fmt.Sprintf("Adding fake subscription for peer %s with bin %d", peerstr, bin)) fakeSubscriptions[peerstr] = append(fakeSubscriptions[peerstr], int(bin)) return true } // create just a simple Registry object in order to be able to call... r := &Registry{} r.requestPeerSubscriptions(k, nil) // calculate the kademlia depth kdepth := k.NeighbourhoodDepth() // now, check that all peers have the expected (fake) subscriptions // iterate the bin map for bin, peers := range binMap { // for every peer... for _, peer := range peers { // ...get its (fake) subscriptions fakeSubsForPeer := fakeSubscriptions[peer] // if the peer's bin is shallower than the kademlia depth... if bin < kdepth { // (iterate all (fake) subscriptions) for _, subbin := range fakeSubsForPeer { // ...only the peer's bin should be "subscribed" // (and thus have only one subscription) if subbin != bin || len(fakeSubsForPeer) != 1 { t.Fatalf("Did not get expected subscription for bin < depth; bin of peer %s: %d, subscription: %d", peer, bin, subbin) } } } else { //if the peer's bin is equal or higher than the kademlia depth... // (iterate all (fake) subscriptions) for i, subbin := range fakeSubsForPeer { // ...each bin from the peer's bin number up to k.MaxProxDisplay should be "subscribed" // as we start from depth we can use the iteration index to check if subbin != i+kdepth { t.Fatalf("Did not get expected subscription for bin > depth; bin of peer %s: %d, subscription: %d", peer, bin, subbin) } // the last "subscription" should be k.MaxProxDisplay if i == len(fakeSubsForPeer)-1 && subbin != k.MaxProxDisplay { t.Fatalf("Expected last subscription to be: %d, but is: %d", k.MaxProxDisplay, subbin) } } } } } // print some output for p, subs := range fakeSubscriptions { log.Debug(fmt.Sprintf("Peer %s has the following fake subscriptions: ", p)) for _, bin := range subs { log.Debug(fmt.Sprintf("%d,", bin)) } } } // TestGetSubscriptions is a unit test for the api.GetPeerSubscriptions() function func TestGetSubscriptions(t *testing.T) { // create an amount of dummy peers testPeerCount := 8 // every peer will have this amount of dummy servers testServerCount := 4 // the peerMap which will store this data for the registry peerMap := make(map[enode.ID]*Peer) // create the registry r := &Registry{} api := NewAPI(r) // call once, at this point should be empty regs := api.GetPeerSubscriptions() if len(regs) != 0 { t.Fatal("Expected subscription count to be 0, but it is not") } // now create a number of dummy servers for each node for i := 0; i < testPeerCount; i++ { addr := network.RandomAddr() id := addr.ID() p := &Peer{} p.servers = make(map[Stream]*server) for k := 0; k < testServerCount; k++ { s := Stream{ Name: strconv.Itoa(k), Key: "", Live: false, } p.servers[s] = &server{} } peerMap[id] = p } r.peers = peerMap // call the subscriptions again regs = api.GetPeerSubscriptions() // count how many (fake) subscriptions there are cnt := 0 for _, reg := range regs { for range reg { cnt++ } } // check expected value expectedCount := testPeerCount * testServerCount if cnt != expectedCount { t.Fatalf("Expected %d subscriptions, but got %d", expectedCount, cnt) } } /* TestGetSubscriptionsRPC sets up a simulation network of `nodeCount` nodes, starts the simulation, waits for SyncUpdateDelay in order to kick off stream registration, then tests that there are subscriptions. */ func TestGetSubscriptionsRPC(t *testing.T) { // arbitrarily set to 4 nodeCount := 4 // run with more nodes if `longrunning` flag is set if *longrunning { nodeCount = 64 } // set the syncUpdateDelay for sync registrations to start syncUpdateDelay := 200 * time.Millisecond // holds the msg code for SubscribeMsg var subscribeMsgCode uint64 var ok bool var expectedMsgCount = 0 // this channel signalizes that the expected amount of subscriptiosn is done allSubscriptionsDone := make(chan struct{}) lock := sync.RWMutex{} // after the test, we need to reset the subscriptionFunc to the default defer func() { subscriptionFunc = doRequestSubscription }() // we use this subscriptionFunc for this test: just increases count and calls the actual subscription subscriptionFunc = func(r *Registry, p *network.Peer, bin uint8, subs map[enode.ID]map[Stream]struct{}) bool { lock.Lock() expectedMsgCount++ lock.Unlock() doRequestSubscription(r, p, bin, subs) return true } // create a standard sim sim := simulation.New(map[string]simulation.ServiceFunc{ "streamer": func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) { addr, netStore, delivery, clean, err := newNetStoreAndDeliveryWithRequestFunc(ctx, bucket, dummyRequestFromPeers) if err != nil { return nil, nil, err } // configure so that sync registrations actually happen r := NewRegistry(addr.ID(), delivery, netStore, state.NewInmemoryStore(), &RegistryOptions{ Retrieval: RetrievalEnabled, Syncing: SyncingAutoSubscribe, //enable sync registrations SyncUpdateDelay: syncUpdateDelay, }, nil) // get the SubscribeMsg code subscribeMsgCode, ok = r.GetSpec().GetCode(SubscribeMsg{}) if !ok { t.Fatal("Message code for SubscribeMsg not found") } cleanup = func() { r.Close() clean() } return r, cleanup, nil }, }) defer sim.Close() ctx, cancelSimRun := context.WithTimeout(context.Background(), 1*time.Minute) defer cancelSimRun() // upload a snapshot err := sim.UploadSnapshot(fmt.Sprintf("testing/snapshot_%d.json", nodeCount)) if err != nil { t.Fatal(err) } // setup the filter for SubscribeMsg msgs := sim.PeerEvents( context.Background(), sim.NodeIDs(), simulation.NewPeerEventsFilter().ReceivedMessages().Protocol("stream").MsgCode(subscribeMsgCode), ) // strategy: listen to all SubscribeMsg events; after every event we wait // if after `waitDuration` no more messages are being received, we assume the // subscription phase has terminated! // the loop in this go routine will either wait for new message events // or times out after 1 second, which signals that we are not receiving // any new subscriptions any more go func() { //for long running sims, waiting 1 sec will not be enough waitDuration := time.Duration(nodeCount/16) * time.Second for { select { case <-ctx.Done(): return case m := <-msgs: // just reset the loop if m.Error != nil { log.Error("stream message", "err", m.Error) continue } log.Trace("stream message", "node", m.NodeID, "peer", m.PeerID) case <-time.After(waitDuration): // one second passed, don't assume more subscriptions allSubscriptionsDone <- struct{}{} log.Info("All subscriptions received") return } } }() //run the simulation result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) error { log.Info("Simulation running") nodes := sim.Net.Nodes //wait until all subscriptions are done select { case <-allSubscriptionsDone: case <-ctx.Done(): t.Fatal("Context timed out") } log.Debug("Expected message count: ", "expectedMsgCount", expectedMsgCount) //now iterate again, this time we call each node via RPC to get its subscriptions realCount := 0 for _, node := range nodes { //create rpc client client, err := node.Client() if err != nil { t.Fatalf("create node 1 rpc client fail: %v", err) } //ask it for subscriptions pstreams := make(map[string][]string) err = client.Call(&pstreams, "stream_getPeerSubscriptions") if err != nil { t.Fatal(err) } //length of the subscriptions can not be smaller than number of peers log.Debug("node subscriptions", "node", node.String()) for p, ps := range pstreams { log.Debug("... with", "peer", p) for _, s := range ps { log.Debug(".......", "stream", s) // each node also has subscriptions to RETRIEVE_REQUEST streams, // we need to ignore those, we are only counting SYNC streams if !strings.HasPrefix(s, "RETRIEVE_REQUEST") { realCount++ } } } } // every node is mutually subscribed to each other, so the actual count is half of it if realCount/2 != expectedMsgCount { return fmt.Errorf("Real subscriptions and expected amount don't match; real: %d, expected: %d", realCount/2, expectedMsgCount) } return nil }) if result.Error != nil { t.Fatal(result.Error) } }