p2p/testing, p2p/protocols: refactored test session setup and driver

This commit is contained in:
nolash 2017-03-17 11:40:32 +01:00 committed by Lewis Marshall
parent 11aeaec420
commit d4a08fdf7e
4 changed files with 125 additions and 174 deletions

View file

@ -126,8 +126,8 @@ func newProtocol(pp *p2ptest.TestPeerPool, wg *sync.WaitGroup) func(adapters.Nod
} }
} }
func protocolTester(t *testing.T, pp *p2ptest.TestPeerPool, wg *sync.WaitGroup) *p2ptest.ExchangeSession { func protocolTester(t *testing.T, pp *p2ptest.TestPeerPool, wg *sync.WaitGroup) *p2ptest.ProtocolTester {
id := p2ptest.RandomNodeId() id := adapters.RandomNodeId()
return p2ptest.NewProtocolTester(t, id, 2, newProtocol(pp, wg)) return p2ptest.NewProtocolTester(t, id, 2, newProtocol(pp, wg))
} }

View file

@ -2,31 +2,17 @@ package testing
import ( import (
"fmt" "fmt"
"sync"
"testing"
"time" "time"
"sync"
"github.com/ethereum/go-ethereum/logger/glog"
"github.com/ethereum/go-ethereum/p2p/adapters" "github.com/ethereum/go-ethereum/p2p/adapters"
"github.com/ethereum/go-ethereum/logger"
"github.com/ethereum/go-ethereum/logger/glog"
) )
// ExchangeTestSession assumes a network with a protocol running on multiple peer connection type ProtocolSession struct {
// and is used to test scanarios of message exchange among a select array of nodes TestNodeAdapter
// the scenarios are sets of exchanges, each with a trigger and an expectation
// This rigid regime is suitable for
// * unit testing protocol message exchanges (nodes are peers of a local node)
// * testing routed messaging between remote non-connected nodes within a group
type ExchangeTestSession struct {
lock sync.Mutex
Ids []*adapters.NodeId Ids []*adapters.NodeId
TestNetAdapter
TestMessenger
t *testing.T
}
// implemented by simulations/
type TestNetAdapter interface {
GetPeer(id *adapters.NodeId) *adapters.Peer
} }
type TestMessenger interface { type TestMessenger interface {
@ -34,7 +20,16 @@ type TestMessenger interface {
TriggerMsg(uint64, interface{}) error TriggerMsg(uint64, interface{}) error
} }
type TestNodeAdapter interface {
GetPeer(id *adapters.NodeId) *adapters.Peer
Connect([]byte) error
}
// exchanges are the basic units of protocol tests // exchanges are the basic units of protocol tests
// the triggers and expects in the arrays are run immediately and asynchronously
// thus one cannot have multiple expects for the SAME peer with the DIFFERENT messagetypes
// because it's unpredictable which expect will receive which message
// (with expect #1 and #2, messages might be sent #2 and #1, and both expects will complain about wrong message code)
// an exchange is defined on a session // an exchange is defined on a session
type Exchange struct { type Exchange struct {
Label string Label string
@ -62,23 +57,17 @@ type Disconnect struct {
Error error // disconnect reason Error error // disconnect reason
} }
// NewExchangeTestSession takes a network session and Messenger
// and returns an exchange session test driver that can func NewProtocolSession(na adapters.NodeAdapter, ids []*adapters.NodeId) *ProtocolSession {
// be used to unit test protocol communications ps := &ProtocolSession {
// it allows for resource-driven scenario testing TestNodeAdapter: na.(TestNodeAdapter),
// disconnect reason errors are written in session.Errs Ids: ids,
// (correcponding to session.Peers)
//func NewExchangeTestSession(t *testing.T, n TestNetAdapter, m TestMessenger, ids []*adapters.NodeId) *ExchangeTestSession {
func NewExchangeTestSession(t *testing.T, n TestNetAdapter, ids []*adapters.NodeId) *ExchangeTestSession {
return &ExchangeTestSession{
Ids: ids,
TestNetAdapter: n,
t: t,
} }
return ps
} }
// trigger sends messages from peers // trigger sends messages from peers
func (self *ExchangeTestSession) trigger(trig Trigger) error { func (self *ProtocolSession) trigger(trig Trigger) error {
peer := self.GetPeer(trig.Peer) peer := self.GetPeer(trig.Peer)
if peer == nil { if peer == nil {
panic(fmt.Sprintf("trigger: peer %v does not exist (1- %v)", trig.Peer, len(self.Ids))) panic(fmt.Sprintf("trigger: peer %v does not exist (1- %v)", trig.Peer, len(self.Ids)))
@ -90,9 +79,9 @@ func (self *ExchangeTestSession) trigger(trig Trigger) error {
errc := make(chan error) errc := make(chan error)
go func() { go func() {
glog.V(6).Infof("trigger %v (%v)....", trig.Msg, trig.Code) glog.V(logger.Detail).Infof("trigger %v (%v)....", trig.Msg, trig.Code)
errc <- m.(TestMessenger).TriggerMsg(trig.Code, trig.Msg) errc <- m.(TestMessenger).TriggerMsg(trig.Code, trig.Msg)
glog.V(6).Infof("triggered %v (%v)", trig.Msg, trig.Code) glog.V(logger.Detail).Infof("triggered %v (%v)", trig.Msg, trig.Code)
}() }()
t := trig.Timeout t := trig.Timeout
@ -108,12 +97,8 @@ func (self *ExchangeTestSession) trigger(trig Trigger) error {
} }
} }
func Key(id []byte) string {
return string(id)
}
// expect checks an expectation // expect checks an expectation
func (self *ExchangeTestSession) expect(exp Expect) error { func (self *ProtocolSession) expect(exp Expect) error {
if exp.Msg == nil { if exp.Msg == nil {
panic("no message to expect") panic("no message to expect")
} }
@ -128,7 +113,7 @@ func (self *ExchangeTestSession) expect(exp Expect) error {
errc := make(chan error) errc := make(chan error)
go func() { go func() {
glog.V(6).Infof("waiting for msg, %v", exp.Msg) glog.V(logger.Detail).Infof("waiting for msg, %v", exp.Msg)
errc <- m.(TestMessenger).ExpectMsg(exp.Code, exp.Msg) errc <- m.(TestMessenger).ExpectMsg(exp.Code, exp.Msg)
}() }()
@ -139,17 +124,18 @@ func (self *ExchangeTestSession) expect(exp Expect) error {
alarm := time.NewTimer(t) alarm := time.NewTimer(t)
select { select {
case err := <-errc: case err := <-errc:
glog.V(6).Infof("expected msg arrives with error %v", err) glog.V(logger.Detail).Infof("expected msg arrives with error %v", err)
return err return err
case <-alarm.C: case <-alarm.C:
glog.V(6).Infof("caught timeout") glog.V(logger.Detail).Infof("caught timeout")
return fmt.Errorf("timout expecting %v sent to peer %v", exp.Msg, exp.Peer) return fmt.Errorf("timout expecting %v sent to peer %v", exp.Msg, exp.Peer)
} }
// fatal upon encountering first exchange error // fatal upon encountering first exchange error
} }
// TestExchange tests a series of exchanges againsts the session // TestExchange tests a series of exchanges againsts the session
func (self *ExchangeTestSession) TestExchanges(exchanges ...Exchange) { func (self *ProtocolSession) TestExchanges(exchanges ...Exchange) error {
// launch all triggers of this exchanges // launch all triggers of this exchanges
for i, e := range exchanges { for i, e := range exchanges {
@ -177,7 +163,7 @@ func (self *ExchangeTestSession) TestExchanges(exchanges ...Exchange) {
defer wg.Done() defer wg.Done()
err := self.expect(exp) err := self.expect(exp)
if err != nil { if err != nil {
glog.V(6).Infof("expect msg fails %v", err) glog.V(logger.Detail).Infof("expect msg fails %v", err)
errc <- err errc <- err
} }
}(ex) }(ex)
@ -195,17 +181,18 @@ func (self *ExchangeTestSession) TestExchanges(exchanges ...Exchange) {
case err := <-errc: case err := <-errc:
if err != nil { if err != nil {
self.t.Fatalf("exchange failed with: %v", err) return fmt.Errorf("exchange failed with: %v", err)
} else { } else {
glog.V(6).Infof("exchange %v: '%v' run successfully", i, e.Label) glog.V(logger.Detail).Infof("exchange %v: '%v' run successfully", i, e.Label)
} }
case <-alarm.C: case <-alarm.C:
self.t.Fatalf("exchange %v: '%v' timed out", i, e.Label) return fmt.Errorf("exchange %v: '%v' timed out", i, e.Label)
} }
} }
return nil
} }
func (self *ExchangeTestSession) TestDisconnected(disconnects ...*Disconnect) { func (self *ProtocolSession) TestDisconnected(disconnects ...*Disconnect) error {
for _, disconnect := range disconnects { for _, disconnect := range disconnects {
id := disconnect.Peer id := disconnect.Peer
err := disconnect.Error err := disconnect.Error
@ -215,10 +202,11 @@ func (self *ExchangeTestSession) TestDisconnected(disconnects ...*Disconnect) {
select { select {
case derr := <-peer.Errc: case derr := <-peer.Errc:
if !((err == nil && derr == nil) || err != nil && derr != nil && err.Error() == derr.Error()) { if !((err == nil && derr == nil) || err != nil && derr != nil && err.Error() == derr.Error()) {
self.t.Fatalf("unexpected error on peer %v. expected '%v', got '%v'", id, err, derr) return fmt.Errorf("unexpected error on peer %v. expected '%v', got '%v'", id, err, derr)
} }
case <-alarm.C: case <-alarm.C:
self.t.Fatalf("timed out waiting for peer %v to disconnect", id) return fmt.Errorf("timed out waiting for peer %v to disconnect", id)
} }
} }
return nil
} }

View file

@ -0,0 +1,84 @@
package testing
import (
"testing"
"github.com/ethereum/go-ethereum/p2p/adapters"
"github.com/ethereum/go-ethereum/p2p/simulations"
"github.com/ethereum/go-ethereum/logger"
"github.com/ethereum/go-ethereum/logger/glog"
)
type ProtocolTester struct {
*ProtocolSession
network *simulations.Network
na adapters.NodeAdapter
}
func NewProtocolTester(t *testing.T, id *adapters.NodeId, n int, run func(id adapters.NodeAdapter) adapters.ProtoCall) *ProtocolTester {
simPipe := adapters.NewSimPipe
net := simulations.NewNetwork(&simulations.NetworkConfig{})
naf := func(conf *simulations.NodeConfig) adapters.NodeAdapter {
na := adapters.NewSimNode(conf.Id, net, simPipe)
if conf.Id.NodeID == id.NodeID {
glog.V(logger.Detail).Infof("adapter run function set to protocol for node %v (=%v)", conf.Id, id)
na.Run = run(na)
}
return na
}
net.SetNaf(naf)
err := net.NewNode(&simulations.NodeConfig{Id: id})
if err != nil {
panic(err.Error())
}
//na := net.GetNode(id).Adapter()
na := net.GetNodeAdapter(id)
ids := adapters.RandomNodeIds(n)
ps := NewProtocolSession(na, ids)
self := &ProtocolTester{
ProtocolSession: ps,
network: net,
na: na,
}
self.Connect(ids...)
return self
}
func (self *ProtocolTester) Start(id *adapters.NodeId) error {
err := self.network.NewNode(&simulations.NodeConfig{Id: id})
if err != nil {
return err
}
node := self.network.GetNode(id)
if node == nil {
glog.V(logger.Detail).Infof("node for peer %v not found", id)
return nil
}
if node.Adapter() == nil {
glog.V(logger.Detail).Infof("node adapter for peer %v not found", id)
return nil
}
return nil
}
func (self *ProtocolTester) Connect(ids ...*adapters.NodeId) {
for _, id := range ids {
glog.V(logger.Detail).Infof("start node %v", id)
err := self.Start(id)
if err != nil {
glog.V(logger.Detail).Infof("error starting peer %v: %v", id, err)
}
glog.V(logger.Detail).Infof("connect to %v", id)
err = self.na.Connect(id.Bytes())
if err != nil {
glog.V(logger.Detail).Infof("error connecting to peer %v: %v", id, err)
}
}
}

View file

@ -1,121 +0,0 @@
package testing
import (
"testing"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/logger/glog"
"github.com/ethereum/go-ethereum/p2p/adapters"
"github.com/ethereum/go-ethereum/p2p/discover"
"github.com/ethereum/go-ethereum/p2p/simulations"
)
type PeerAdapter interface {
adapters.NodeAdapter
TestMessenger
TestNetAdapter
}
type ExchangeSession struct {
network *simulations.Network
na adapters.NodeAdapter
*ExchangeTestSession
}
// NewProtocolTester returns an exchange test session
// this is a resource driver for protocol message exchange
// scenarios expressed as expects and triggers
// see p2p/protocols/exhange_test.go for an example
// this is used primarily to unit test protocols or protocol modules
// correct message exchange, forwarding, and broadcast
// higher level or network behaviour should be tested with network simulators
func NewProtocolTester(t *testing.T, id *adapters.NodeId, n int, run func(id adapters.NodeAdapter) adapters.ProtoCall) *ExchangeSession {
simPipe := adapters.NewSimPipe
network := simulations.NewNetwork(&simulations.NetworkConfig{})
naf := func(conf *simulations.NodeConfig) adapters.NodeAdapter {
na := adapters.NewSimNode(conf.Id, network, simPipe)
if conf.Id.NodeID == id.NodeID {
glog.V(6).Infof("adapter run function set to protocol for node %v (=%v)", conf.Id, id)
na.Run = run(na)
}
return na
}
network.SetNaf(naf)
// setup a simulated network of n nodes
// Startup pivot node
err := network.NewNode(&simulations.NodeConfig{Id: id})
if err != nil {
panic(err.Error())
}
glog.V(6).Infof("network created")
na := network.GetNode(id).Adapter()
//s := NewExchangeTestSession(t, na.(TestNetAdapter), na.Messenger().(TestMessenger), nil)
s := NewExchangeTestSession(t, na.(TestNetAdapter), nil)
self := &ExchangeSession{
network: network,
na: na,
ExchangeTestSession: s,
}
ids := RandomNodeIds(n)
self.Connect(ids...)
// Start up connections to virual nodes serving as endpoints for sending/receiving messages for peers
return self
}
//
func (self *ExchangeSession) Start(id *adapters.NodeId) error {
err := self.network.NewNode(&simulations.NodeConfig{Id: id})
if err != nil {
return err
}
node := self.network.GetNode(id)
if node == nil {
glog.V(6).Infof("node for peer %v not found", id)
return nil
}
if node.Adapter() == nil {
glog.V(6).Infof("node adapter for peer %v not found", id)
return nil
}
self.Ids = append(self.Ids, id)
return nil
}
func (self *ExchangeSession) Connect(ids ...*adapters.NodeId) {
for _, id := range ids {
glog.V(6).Infof("start node %v", id)
err := self.Start(id)
if err != nil {
glog.V(6).Infof("error starting peer %v: %v", id, err)
}
glog.V(6).Infof("connect to %v", id)
err = self.na.Connect(id.Bytes())
if err != nil {
glog.V(6).Infof("error connecting to peer %v: %v", id, err)
}
}
}
// func (self *ExchangeSession) Id(i int) *adapters.NodeId {
// return self.network.Nodes[i].Id
// }
func RandomNodeId() *adapters.NodeId {
key, err := crypto.GenerateKey()
if err != nil {
panic("unable to generate key")
}
var id discover.NodeID
pubkey := crypto.FromECDSAPub(&key.PublicKey)
copy(id[:], pubkey[1:])
return &adapters.NodeId{id}
}
func RandomNodeIds(n int) []*adapters.NodeId {
var ids []*adapters.NodeId
for i := 0; i < n; i++ {
ids = append(ids, RandomNodeId())
}
return ids
}