From 89ad7e1ba210a2db3b093ac8a50acd830d4360af Mon Sep 17 00:00:00 2001 From: Tom Wilkie Date: Wed, 8 Jul 2015 14:56:58 +0000 Subject: [PATCH] Move test gossip router out of ipam code for use in gossip dns code. --- ipam/allocator_test.go | 21 +++---- ipam/testutils_test.go | 121 ++------------------------------------ testing/gossip/mocks.go | 125 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 141 insertions(+), 126 deletions(-) create mode 100644 testing/gossip/mocks.go diff --git a/ipam/allocator_test.go b/ipam/allocator_test.go index 3c9075e1..cb5c4b41 100644 --- a/ipam/allocator_test.go +++ b/ipam/allocator_test.go @@ -8,9 +8,10 @@ import ( "time" "github.com/stretchr/testify/require" + "github.com/weaveworks/weave/common" "github.com/weaveworks/weave/net/address" - "github.com/weaveworks/weave/router" + "github.com/weaveworks/weave/testing/gossip" ) const ( @@ -166,13 +167,13 @@ func TestCancel(t *testing.T) { CIDR = "10.0.1.7/26" ) - router := TestGossipRouter{make(map[router.PeerName]chan interface{}), 0.0} + router := gossip.NewTestRouter(0.0) alloc1, subnet := makeAllocator("01:00:00:02:00:00", CIDR, 2) - alloc1.SetInterfaces(router.connect(alloc1.ourName, alloc1)) + alloc1.SetInterfaces(router.Connect(alloc1.ourName, alloc1)) alloc2, _ := makeAllocator("02:00:00:02:00:00", CIDR, 2) - alloc2.SetInterfaces(router.connect(alloc2.ourName, alloc2)) + alloc2.SetInterfaces(router.Connect(alloc2.ourName, alloc2)) alloc1.claimRingForTesting(alloc1, alloc2) alloc2.claimRingForTesting(alloc1, alloc2) @@ -254,17 +255,17 @@ func TestTransfer(t *testing.T) { require.True(t, err == nil, "Failed to get address") router.GossipBroadcast(alloc2.Gossip()) - router.flush() + router.Flush() router.GossipBroadcast(alloc3.Gossip()) - router.flush() - router.removePeer(alloc2.ourName) - router.removePeer(alloc3.ourName) + router.Flush() + router.RemovePeer(alloc2.ourName) + router.RemovePeer(alloc3.ourName) alloc2.Stop() alloc3.Stop() - router.flush() + router.Flush() require.NoError(t, alloc1.AdminTakeoverRanges(alloc2.ourName.String())) require.NoError(t, alloc1.AdminTakeoverRanges(alloc3.ourName.String())) - router.flush() + router.Flush() require.Equal(t, address.Offset(1022), alloc1.NumFreeAddresses(subnet)) diff --git a/ipam/testutils_test.go b/ipam/testutils_test.go index 37855666..294444d6 100644 --- a/ipam/testutils_test.go +++ b/ipam/testutils_test.go @@ -8,9 +8,9 @@ import ( "time" "github.com/stretchr/testify/require" - "github.com/weaveworks/weave/common" "github.com/weaveworks/weave/net/address" "github.com/weaveworks/weave/router" + "github.com/weaveworks/weave/testing/gossip" ) type mockMessage struct { @@ -194,119 +194,8 @@ func AssertNothingSentErr(t *testing.T, ch <-chan error) { } } -// Router to convey gossip from one gossiper to another, for testing -type unicastMessage struct { - sender router.PeerName - buf []byte -} -type broadcastMessage struct { - data router.GossipData -} -type exitMessage struct { - exitChan chan struct{} -} -type flushMessage struct { - flushChan chan struct{} -} - -type TestGossipRouter struct { - gossipChans map[router.PeerName]chan interface{} - loss float32 // 0.0 means no loss -} - -func (grouter *TestGossipRouter) GossipBroadcast(update router.GossipData) error { - for _, gossipChan := range grouter.gossipChans { - select { - case gossipChan <- broadcastMessage{data: update}: - default: // drop the message if we cannot send it - } - } - return nil -} - -func (grouter *TestGossipRouter) flush() { - for _, gossipChan := range grouter.gossipChans { - flushChan := make(chan struct{}) - gossipChan <- flushMessage{flushChan: flushChan} - <-flushChan - } -} - -func (grouter *TestGossipRouter) removePeer(peer router.PeerName) { - gossipChan := grouter.gossipChans[peer] - resultChan := make(chan struct{}) - gossipChan <- exitMessage{exitChan: resultChan} - <-resultChan - delete(grouter.gossipChans, peer) -} - -type TestGossipRouterClient struct { - router *TestGossipRouter - sender router.PeerName -} - -func (grouter *TestGossipRouter) run(gossiper router.Gossiper, gossipChan chan interface{}) { - gossipTimer := time.Tick(10 * time.Second) - for { - select { - case gossip := <-gossipChan: - switch message := gossip.(type) { - case exitMessage: - close(message.exitChan) - return - - case flushMessage: - close(message.flushChan) - - case unicastMessage: - if rand.Float32() > (1.0 - grouter.loss) { - continue - } - if err := gossiper.OnGossipUnicast(message.sender, message.buf); err != nil { - panic(fmt.Sprintf("Error doing gossip unicast to %s: %s", message.sender, err)) - } - - case broadcastMessage: - if rand.Float32() > (1.0 - grouter.loss) { - continue - } - for _, msg := range message.data.Encode() { - if _, err := gossiper.OnGossipBroadcast(msg); err != nil { - panic(fmt.Sprintf("Error doing gossip broadcast: %s", err)) - } - } - } - case <-gossipTimer: - grouter.GossipBroadcast(gossiper.Gossip()) - } - } -} - -func (grouter *TestGossipRouter) connect(sender router.PeerName, gossiper router.Gossiper) router.Gossip { - gossipChan := make(chan interface{}, 100) - - go grouter.run(gossiper, gossipChan) - - grouter.gossipChans[sender] = gossipChan - return TestGossipRouterClient{grouter, sender} -} - -func (client TestGossipRouterClient) GossipUnicast(dstPeerName router.PeerName, buf []byte) error { - select { - case client.router.gossipChans[dstPeerName] <- unicastMessage{sender: client.sender, buf: buf}: - default: // drop the message if we cannot send it - common.Log.Errorf("Dropping message") - } - return nil -} - -func (client TestGossipRouterClient) GossipBroadcast(update router.GossipData) error { - return client.router.GossipBroadcast(update) -} - -func makeNetworkOfAllocators(size int, cidr string) ([]*Allocator, TestGossipRouter, address.Range) { - - gossipRouter := TestGossipRouter{make(map[router.PeerName]chan interface{}), 0.0} +func makeNetworkOfAllocators(size int, cidr string) ([]*Allocator, *gossip.TestRouter, address.Range) { + gossipRouter := gossip.NewTestRouter(0.0) allocs := make([]*Allocator, size) var subnet address.Range @@ -314,13 +203,13 @@ func makeNetworkOfAllocators(size int, cidr string) ([]*Allocator, TestGossipRou var alloc *Allocator alloc, subnet = makeAllocator(fmt.Sprintf("%02d:00:00:02:00:00", i), cidr, uint(size/2+1)) - alloc.SetInterfaces(gossipRouter.connect(alloc.ourName, alloc)) + alloc.SetInterfaces(gossipRouter.Connect(alloc.ourName, alloc)) alloc.Start() allocs[i] = alloc } gossipRouter.GossipBroadcast(allocs[size-1].Gossip()) - gossipRouter.flush() + gossipRouter.Flush() return allocs, gossipRouter, subnet } diff --git a/testing/gossip/mocks.go b/testing/gossip/mocks.go new file mode 100644 index 00000000..d3aad9f3 --- /dev/null +++ b/testing/gossip/mocks.go @@ -0,0 +1,125 @@ +package gossip + +import ( + "fmt" + "math/rand" + "time" + + "github.com/weaveworks/weave/common" + "github.com/weaveworks/weave/router" +) + +// Router to convey gossip from one gossiper to another, for testing +type unicastMessage struct { + sender router.PeerName + buf []byte +} +type broadcastMessage struct { + data router.GossipData +} +type exitMessage struct { + exitChan chan struct{} +} +type flushMessage struct { + flushChan chan struct{} +} + +type TestRouter struct { + gossipChans map[router.PeerName]chan interface{} + loss float32 // 0.0 means no loss +} + +func NewTestRouter(loss float32) *TestRouter { + return &TestRouter{make(map[router.PeerName]chan interface{}, 100), loss} +} + +func (grouter *TestRouter) GossipBroadcast(update router.GossipData) error { + for _, gossipChan := range grouter.gossipChans { + select { + case gossipChan <- broadcastMessage{data: update}: + default: // drop the message if we cannot send it + common.Log.Errorf("Dropping message") + } + } + return nil +} + +func (grouter *TestRouter) Flush() { + for _, gossipChan := range grouter.gossipChans { + flushChan := make(chan struct{}) + gossipChan <- flushMessage{flushChan: flushChan} + <-flushChan + } +} + +func (grouter *TestRouter) RemovePeer(peer router.PeerName) { + gossipChan := grouter.gossipChans[peer] + resultChan := make(chan struct{}) + gossipChan <- exitMessage{exitChan: resultChan} + <-resultChan + delete(grouter.gossipChans, peer) +} + +type TestRouterClient struct { + router *TestRouter + sender router.PeerName +} + +func (grouter *TestRouter) run(gossiper router.Gossiper, gossipChan chan interface{}) { + gossipTimer := time.Tick(10 * time.Second) + for { + select { + case gossip := <-gossipChan: + switch message := gossip.(type) { + case exitMessage: + close(message.exitChan) + return + + case flushMessage: + close(message.flushChan) + + case unicastMessage: + if rand.Float32() > (1.0 - grouter.loss) { + continue + } + if err := gossiper.OnGossipUnicast(message.sender, message.buf); err != nil { + panic(fmt.Sprintf("Error doing gossip unicast to %s: %s", message.sender, err)) + } + + case broadcastMessage: + if rand.Float32() > (1.0 - grouter.loss) { + continue + } + for _, msg := range message.data.Encode() { + if _, err := gossiper.OnGossipBroadcast(msg); err != nil { + panic(fmt.Sprintf("Error doing gossip broadcast: %s", err)) + } + } + } + case <-gossipTimer: + grouter.GossipBroadcast(gossiper.Gossip()) + } + } +} + +func (grouter *TestRouter) Connect(sender router.PeerName, gossiper router.Gossiper) router.Gossip { + gossipChan := make(chan interface{}, 100) + + go grouter.run(gossiper, gossipChan) + + grouter.gossipChans[sender] = gossipChan + return TestRouterClient{grouter, sender} +} + +func (client TestRouterClient) GossipUnicast(dstPeerName router.PeerName, buf []byte) error { + select { + case client.router.gossipChans[dstPeerName] <- unicastMessage{sender: client.sender, buf: buf}: + default: // drop the message if we cannot send it + common.Log.Errorf("Dropping message") + } + return nil +} + +func (client TestRouterClient) GossipBroadcast(update router.GossipData) error { + return client.router.GossipBroadcast(update) +}