more tests for node.go

node/node.go

@@ -216,11 +216,10 @@ func (w *configWatcher) Start() {
 
 // An active streaming node, managing the local icecast server.
 type RadioNode struct {
-	config *clusterConfig
-
+	client    autoradio.EtcdClient
+	config    *clusterConfig
 	name      string
 	ips       []net.IP
-	client    autoradio.EtcdClient
 	me        *masterelection.MasterElection
 	watcher   *configWatcher
 	icecast   Controller
@@ -234,8 +233,7 @@ func NewRadioNode(name string, ips []net.IP, netDev string, bwLimit float64, cli
 	config := newClusterConfig()
 
 	// Network updates trigger icecast reconfiguration. This
-	// channel is used as an 'event', no more than one entry will
-	// be queued.
+	// channel is used as an event signal.
 	upch := make(chan bool, 1)
 
 	// MasterElection changes trigger an update.

node/node_test.go

@@ -49,26 +49,26 @@ func startTestNodes(n int, etcd autoradio.EtcdClient) []*RadioNode {
 }
 
 func loadTestData(etcd autoradio.EtcdClient) {
-	etcd.Set(autoradio.MountPrefix+"/test.ogg",
+	etcd.Set(autoradio.MountPrefix+"test.ogg",
 		`{"Name": "/test.ogg", "Username": "source1", "Password": "foo"}`,
 		86400)
 }
 
-func countMasters(nodes []*RadioNode) int {
-	var masters int
-	for _, n := range nodes {
-		if n.me.IsMaster() {
-			masters++
-		}
-	}
-	return masters
-}
-
 func TestRadioNode_MasterElection(t *testing.T) {
 	etcd := util.NewTestEtcdClient()
 	loadTestData(etcd)
 	nodes := startTestNodes(3, etcd)
 
+	countMasters := func(nodes []*RadioNode) int {
+		var masters int
+		for _, n := range nodes {
+			if n.me.IsMaster() {
+				masters++
+			}
+		}
+		return masters
+	}
+
 	// Shut down the nodes one by one, and verify that there is a
 	// single master among the remaining ones.
 	for i := 0; i < 3; i++ {
@@ -79,3 +79,25 @@ func TestRadioNode_MasterElection(t *testing.T) {
 		time.Sleep(10 * time.Millisecond)
 	}
 }
+
+func TestRadioNode_ConfigChangePropagation(t *testing.T) {
+	etcd := util.NewTestEtcdClient()
+	loadTestData(etcd)
+	nodes := startTestNodes(3, etcd)
+
+	// Wait a bit and modify the stream. Check that the change
+	// propagates correctly.
+	time.Sleep(100 * time.Millisecond)
+	etcd.Set(autoradio.MountPrefix+"test.ogg",
+		`{"Name": "/test.ogg", "Username": "source2", "Password": "bar"}`,
+		86400)
+	time.Sleep(100 * time.Millisecond)
+
+	for i := 0; i < 3; i++ {
+		username := nodes[i].config.GetMount("/test.ogg").Username
+		if username != "source2" {
+			t.Errorf("change did not propagate to node %d", i+1)
+		}
+		nodes[i].Stop()
+	}
+}

util/mock_etcd.go

@@ -120,6 +120,7 @@ func (s *testEtcdServer) Get(key string, recursive, boh bool) (*etcd.Response, e
 		EtcdIndex: s.index,
 	}
 	var nodes []*etcd.Node
+	key = strings.TrimSuffix(key, "/")
 	keyDirPfx := key + "/"
 	for path, datum := range s.data {
 		if path == key || strings.HasPrefix(path, keyDirPfx) {