From 20d4a867fd81c9d0ea8cb1a2800ff77511cb85d8 Mon Sep 17 00:00:00 2001 From: zelig Date: Sat, 13 Jan 2018 10:51:46 +0100 Subject: [PATCH] swarm/network: all unit tests pass for streamer * separate retrieve request tests * do not subscribe to requests automatically * simplify request test * add Stream to Subcribe --- swarm/network/request_test.go | 146 +++++++++++++++++++++++++++ swarm/network/streamer.go | 8 -- swarm/network/streamer_test.go | 179 +-------------------------------- 3 files changed, 148 insertions(+), 185 deletions(-) create mode 100644 swarm/network/request_test.go diff --git a/swarm/network/request_test.go b/swarm/network/request_test.go new file mode 100644 index 0000000000..8b848611e3 --- /dev/null +++ b/swarm/network/request_test.go @@ -0,0 +1,146 @@ +// Copyright 2016 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 . + +package network + +import ( + "testing" + "time" + + p2ptest "github.com/ethereum/go-ethereum/p2p/testing" + "github.com/ethereum/go-ethereum/swarm/storage" +) + +func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) { + // TODO: we only need streamer + tester, streamer, _, teardown, err := newStreamerTester(t) + defer teardown() + if err != nil { + t.Fatal(err) + } + + err = waitForPeers(streamer, 1*time.Second) + if err != nil { + t.Fatal("timeout: peer is not created") + } + + peerId := tester.IDs[0] + + chunk := storage.NewChunk(storage.Key(hash0[:]), nil) + + peer := streamer.getPeer(peerId) + + peer.handleSubscribeMsg(&SubscribeMsg{ + Stream: retrieveRequestStream, + Key: nil, + From: 0, + To: 0, + Priority: Top, + }) + + err = tester.TestExchanges(p2ptest.Exchange{ + Label: "RetrieveRequestMsg", + Triggers: []p2ptest.Trigger{ + p2ptest.Trigger{ + Code: 5, + Msg: &RetrieveRequestMsg{ + Key: chunk.Key[:], + }, + Peer: peerId, + }, + }, + Expects: []p2ptest.Expect{ + p2ptest.Expect{ + Code: 1, + Msg: &OfferedHashesMsg{ + HandoverProof: nil, + Hashes: nil, + From: 0, + To: 0, + }, + Peer: peerId, + }, + }, + }) + + expectedError := "exchange 0: 'RetrieveRequestMsg' timed out" + if err == nil || err.Error() != expectedError { + t.Fatalf("Expected error %v, got %v", expectedError, err) + } +} + +func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) { + // TODO: we only need streamer + tester, streamer, localStore, teardown, err := newStreamerTester(t) + defer teardown() + if err != nil { + t.Fatal(err) + } + + err = waitForPeers(streamer, 1*time.Second) + if err != nil { + t.Fatal("timeout: peer is not created") + } + + peerId := tester.IDs[0] + + chunk := storage.NewChunk(storage.Key(hash0[:]), nil) + + peer := streamer.getPeer(peerId) + + peer.handleSubscribeMsg(&SubscribeMsg{ + Stream: retrieveRequestStream, + Key: nil, + From: 0, + To: 0, + Priority: Top, + }) + + chunk.SData = hash0[:] + localStore.Put(chunk) + + err = tester.TestExchanges(p2ptest.Exchange{ + Label: "RetrieveRequestMsg", + Triggers: []p2ptest.Trigger{ + p2ptest.Trigger{ + Code: 5, + Msg: &RetrieveRequestMsg{ + Key: chunk.Key[:], + }, + Peer: peerId, + }, + }, + Expects: []p2ptest.Expect{ + p2ptest.Expect{ + Code: 1, + Msg: &OfferedHashesMsg{ + HandoverProof: nil, + Hashes: chunk.Key[:], + From: 0, + // TODO: why is this 32??? + To: 32, + Key: []byte{}, + Stream: retrieveRequestStream, + }, + Peer: peerId, + }, + }, + }) + + if err != nil { + t.Fatal(err) + } +} diff --git a/swarm/network/streamer.go b/swarm/network/streamer.go index 365222ab17..24dbc3b379 100644 --- a/swarm/network/streamer.go +++ b/swarm/network/streamer.go @@ -667,18 +667,10 @@ var StreamerSpec = &protocols.Spec{ func (s *Streamer) Run(p *bzzPeer) error { sp := NewStreamerPeer(p, s) // load saved intervals - // autosubscribe to request handler to serve request only for non-light nodes - // sp.handleSubscribeMsg(&SubscribeMsg{ - // Stream: retrieveRequeststring, - // Priority: uint8(Top), - // }) - // subscribe to request handling ; only with non-light nodes s.setPeer(sp) defer s.deletePeer(sp) - - s.Subscribe(sp.ID(), retrieveRequestStream, nil, 0, 0, Top, true) defer close(sp.quit) return sp.Run(sp.HandleMsg) } diff --git a/swarm/network/streamer_test.go b/swarm/network/streamer_test.go index 6a75a163fc..a28a0bf260 100644 --- a/swarm/network/streamer_test.go +++ b/swarm/network/streamer_test.go @@ -24,7 +24,7 @@ import ( "testing" "time" - sha3 "github.com/ethereum/go-ethereum/crypto/sha3" + "github.com/ethereum/go-ethereum/crypto/sha3" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/protocols" @@ -68,7 +68,6 @@ func newStreamerTester(t *testing.T) (*p2ptest.ProtocolTester, *Streamer, *stora to.On(bzzPeer) return streamer.Run(bzzPeer) } - protocolTester := p2ptest.NewProtocolTester(t, NewNodeIDFromAddr(addr), 1, run) return protocolTester, streamer, localStore, teardown, nil } @@ -226,6 +225,7 @@ func TestStreamerUpstreamSubscribeMsgExchange(t *testing.T) { p2ptest.Expect{ Code: 1, Msg: &OfferedHashesMsg{ + Stream: "foo", HandoverProof: nil, Hashes: make([]byte, HashSize), From: 6, @@ -383,181 +383,6 @@ func TestRetrieveRequest(t *testing.T) { } } -func TestUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) { - // TODO: we only need streamer - tester, streamer, _, teardown, err := newStreamerTester(t) - defer teardown() - if err != nil { - t.Fatal(err) - } - - // streamer.RegisterOutgoingStreamer("foo", func(p *StreamerPeer, t []byte) (OutgoingStreamer, error) { - // return &testOutgoingStreamer{ - // t: t, - // }, nil - // }) - - err = waitForPeers(streamer, 1*time.Second) - if err != nil { - t.Fatal("timeout: peer is not created") - } - - peerId := tester.IDs[0] - - err = tester.TestExchanges(p2ptest.Exchange{ - Label: "SubscribeMsg", - Expects: []p2ptest.Expect{ - p2ptest.Expect{ - Code: 4, - Msg: &SubscribeMsg{ - Stream: retrieveRequestStream, - Key: nil, - From: 0, - To: 0, - Priority: Top, - }, - Peer: peerId, - }, - }, - }) - - if err != nil { - t.Fatalf("Expected no error, got %v", err) - } - - chunk := storage.NewChunk(storage.Key(hash0[:]), nil) - - peer := streamer.getPeer(peerId) - - peer.handleSubscribeMsg(&SubscribeMsg{ - Stream: retrieveRequestStream, - Key: nil, - From: 0, - To: 0, - Priority: Top, - }) - - err = tester.TestExchanges(p2ptest.Exchange{ - Label: "RetrieveRequestMsg", - Triggers: []p2ptest.Trigger{ - p2ptest.Trigger{ - Code: 5, - Msg: &RetrieveRequestMsg{ - Key: chunk.Key[:], - }, - Peer: peerId, - }, - }, - Expects: []p2ptest.Expect{ - p2ptest.Expect{ - Code: 1, - Msg: &OfferedHashesMsg{ - HandoverProof: nil, - Hashes: nil, - From: 0, - To: 0, - }, - Peer: peerId, - }, - }, - }) - - expectedError := "exchange 0: 'RetrieveRequestMsg' timed out" - if err == nil || err.Error() != expectedError { - t.Fatalf("Expected error %v, got %v", expectedError, err) - } -} - -func TestUpstreamRetrieveRequestMsgExchange(t *testing.T) { - // TODO: we only need streamer - tester, streamer, localStore, teardown, err := newStreamerTester(t) - defer teardown() - if err != nil { - t.Fatal(err) - } - - // streamer.RegisterOutgoingStreamer("foo", func(p *StreamerPeer, t []byte) (OutgoingStreamer, error) { - // return &testOutgoingStreamer{ - // t: t, - // }, nil - // }) - - err = waitForPeers(streamer, 1*time.Second) - if err != nil { - t.Fatal("timeout: peer is not created") - } - - peerId := tester.IDs[0] - - err = tester.TestExchanges(p2ptest.Exchange{ - Label: "SubscribeMsg", - Expects: []p2ptest.Expect{ - p2ptest.Expect{ - Code: 4, - Msg: &SubscribeMsg{ - Stream: retrieveRequestStream, - Key: nil, - From: 0, - To: 0, - Priority: Top, - }, - Peer: peerId, - }, - }, - }) - - if err != nil { - t.Fatalf("Expected no error, got %v", err) - } - - chunk := storage.NewChunk(storage.Key(hash0[:]), nil) - - peer := streamer.getPeer(peerId) - - peer.handleSubscribeMsg(&SubscribeMsg{ - Stream: retrieveRequestStream, - Key: nil, - From: 0, - To: 0, - Priority: Top, - }) - - chunk.SData = hash0[:] - localStore.Put(chunk) - - err = tester.TestExchanges(p2ptest.Exchange{ - Label: "RetrieveRequestMsg", - Triggers: []p2ptest.Trigger{ - p2ptest.Trigger{ - Code: 5, - Msg: &RetrieveRequestMsg{ - Key: chunk.Key[:], - }, - Peer: peerId, - }, - }, - Expects: []p2ptest.Expect{ - p2ptest.Expect{ - Code: 1, - Msg: &OfferedHashesMsg{ - HandoverProof: nil, - Hashes: chunk.Key[:], - From: 0, - // TODO: why is this 32??? - To: 32, - Key: []byte{}, - Stream: retrieveRequestStream, - }, - Peer: peerId, - }, - }, - }) - - if err != nil { - t.Fatal(err) - } -} - func waitForPeers(streamer *Streamer, timeout time.Duration) error { ticker := time.NewTicker(10 * time.Millisecond) timeoutTimer := time.NewTimer(timeout)