Merge lp:~pedronis/ubuntu-push/takethebus-and-fixes into lp:ubuntu-push/automatic

[Samuele Pedroni]

TestTakeTheBusWorks doesn't block anymore, fixed leaking of webcheckers

TestTakeTheBusWorks doesn't block anymore, fixed leaking of webcheckers,

to get there support using channels to control WatchSignal in tests,

don't reuse testing Endpoint retvals,

sadly bit of whack a mole:

144 + // XXX this may be stolen by an old watch => dead lock

[John Lenton (chipaca)]

review: Approve

[John Lenton (chipaca)]

review: Approve

=== modified file 'bus/connectivity/connectivity.go'
--- bus/connectivity/connectivity.go	2015-01-22 11:05:37 +0000
+++ bus/connectivity/connectivity.go	2015-02-25 11:05:41 +0000
@@ -87,15 +87,15 @@
 		}
 		cs.log.Debugf("got initial state of %s", initial)
 
+		conCh, err = nm.WatchPrimaryConnection()
+		if err != nil {
+			cs.log.Debugf("failed to set up the connection watch: %s", err)
+			goto Continue
+		}
+
 		primary = nm.GetPrimaryConnection()
 		cs.log.Debugf("primary connection starts as %#v", primary)
 
-		conCh, err = nm.WatchPrimaryConnection()
-		if err != nil {
-			cs.log.Debugf("failed to set up the connection watch: %s", err)
-			goto Continue
-		}
-
 		cs.networkStateCh = stateCh
 		cs.networkConCh = conCh
 
@@ -155,7 +155,12 @@
 		case <-cs.timer.C:
 			if cs.currentState == networkmanager.ConnectedGlobal {
 				log.Debugf("connectivity: timer signal, state: ConnectedGlobal, checking...")
-				cs.webgetCh = make(chan bool)
+				// use a buffered channel, otherwise
+				// we may leak webcheckers that cannot
+				// send their result because we have
+				// cleared webgetCh and wont receive
+				// on it
+				cs.webgetCh = make(chan bool, 1)
 				go cs.webget(cs.webgetCh)
 			}
 

=== modified file 'bus/connectivity/connectivity_test.go'
--- bus/connectivity/connectivity_test.go	2015-01-21 17:55:35 +0000
+++ bus/connectivity/connectivity_test.go	2015-02-25 11:05:41 +0000
@@ -58,32 +58,52 @@
 	s.log = helpers.NewTestLogger(c, "debug")
 }
 
+var (
+	helloCon      = dbus.ObjectPath("hello")
+	helloConProps = map[string]dbus.Variant{"PrimaryConnection": dbus.Variant{helloCon}}
+)
+
 /*
    tests for connectedState's Start() method
 */
 
 // when given a working config and bus, Start() will work
 func (s *ConnSuite) TestStartWorks(c *C) {
-	endp := testingbus.NewTestingEndpoint(condition.Work(true), condition.Work(true), uint32(networkmanager.Connecting))
+	endp := testingbus.NewTestingEndpoint(condition.Work(true), condition.Work(true), uint32(networkmanager.Connecting), helloCon)
 	cs := connectedState{config: ConnectivityConfig{}, log: s.log, endp: endp}
 
+	nopTicker := make(chan []interface{})
+	testingbus.SetWatchSource(endp, "StateChanged", nopTicker)
+	testingbus.SetWatchSource(endp, "PropertiesChanged", nopTicker)
+	defer close(nopTicker)
+
 	c.Check(cs.start(), Equals, networkmanager.Connecting)
 }
 
 // if the bus fails a couple of times, we're still OK
 func (s *ConnSuite) TestStartRetriesConnect(c *C) {
-	endp := testingbus.NewTestingEndpoint(condition.Fail2Work(2), condition.Work(true), uint32(networkmanager.Connecting))
+	endp := testingbus.NewTestingEndpoint(condition.Fail2Work(2), condition.Work(true), uint32(networkmanager.Connecting), helloCon)
 	cs := connectedState{config: ConnectivityConfig{}, log: s.log, endp: endp}
 
+	nopTicker := make(chan []interface{})
+	testingbus.SetWatchSource(endp, "StateChanged", nopTicker)
+	testingbus.SetWatchSource(endp, "PropertiesChanged", nopTicker)
+	defer close(nopTicker)
+
 	c.Check(cs.start(), Equals, networkmanager.Connecting)
 	c.Check(cs.connAttempts, Equals, uint32(3)) // 1 more than the Fail2Work
 }
 
 // when the calls to NetworkManager fails for a bit, we're still OK
 func (s *ConnSuite) TestStartRetriesCall(c *C) {
-	endp := testingbus.NewTestingEndpoint(condition.Work(true), condition.Fail2Work(5), uint32(networkmanager.Connecting))
+	endp := testingbus.NewTestingEndpoint(condition.Work(true), condition.Fail2Work(5), uint32(networkmanager.Connecting), helloCon)
 	cs := connectedState{config: ConnectivityConfig{}, log: s.log, endp: endp}
 
+	nopTicker := make(chan []interface{})
+	testingbus.SetWatchSource(endp, "StateChanged", nopTicker)
+	testingbus.SetWatchSource(endp, "PropertiesChanged", nopTicker)
+	defer close(nopTicker)
+
 	c.Check(cs.start(), Equals, networkmanager.Connecting)
 
 	c.Check(cs.connAttempts, Equals, uint32(6))
@@ -94,9 +114,17 @@
 	cond := condition.Chain(3, condition.Work(true), 1, condition.Work(false),
 		1, condition.Work(true))
 
-	endp := testingbus.NewTestingEndpoint(condition.Work(true), cond, uint32(networkmanager.Connecting))
+	endp := testingbus.NewTestingEndpoint(condition.Work(true), cond,
+		uint32(networkmanager.Connecting), helloCon,
+		uint32(networkmanager.Connecting), helloCon,
+	)
 	cs := connectedState{config: ConnectivityConfig{}, log: s.log, endp: endp}
 
+	nopTicker := make(chan []interface{})
+	testingbus.SetWatchSource(endp, "StateChanged", nopTicker)
+	testingbus.SetWatchSource(endp, "PropertiesChanged", nopTicker)
+	defer close(nopTicker)
+
 	c.Check(cs.start(), Equals, networkmanager.Connecting)
 }
 
@@ -105,17 +133,26 @@
 // watch, we recover and try again.
 func (s *ConnSuite) TestStartRetriesWatch(c *C) {
 	nmcond := condition.Chain(
-		1, condition.Work(true), // 1 call to nm works
+		2, condition.Work(true), // 2 call to nm works
 		1, condition.Work(false), // 1 call to nm fails
 		0, condition.Work(true)) // and everything works from there on
 	endp := testingbus.NewTestingEndpoint(condition.Work(true), nmcond,
 		uint32(networkmanager.Connecting),
-		uint32(networkmanager.ConnectedGlobal))
+		uint32(networkmanager.Connecting),
+		helloCon,
+	)
 	cs := connectedState{config: ConnectivityConfig{}, log: s.log, endp: endp}
+	watchTicker := make(chan []interface{}, 1)
+	nopTicker := make(chan []interface{})
+	testingbus.SetWatchSource(endp, "StateChanged", watchTicker)
+	testingbus.SetWatchSource(endp, "PropertiesChanged", nopTicker)
+	defer close(nopTicker)
+	defer close(watchTicker)
 
 	c.Check(cs.start(), Equals, networkmanager.Connecting)
 	c.Check(cs.connAttempts, Equals, uint32(2))
-	c.Check(<-cs.networkStateCh, Equals, networkmanager.Connecting)
+	// XXX this may be stolen by an old watch => dead lock
+	watchTicker <- []interface{}{uint32(networkmanager.ConnectedGlobal)}
 	c.Check(<-cs.networkStateCh, Equals, networkmanager.ConnectedGlobal)
 }
 
@@ -285,12 +322,16 @@
 	}
 
 	endp := testingbus.NewTestingEndpoint(condition.Work(true), condition.Work(true),
-		uint32(networkmanager.ConnectedGlobal),
-		uint32(networkmanager.Disconnected),
+		uint32(networkmanager.Disconnected),
+		helloCon,
+		uint32(networkmanager.Disconnected),
+		helloCon,
 	)
 
-	watchTicker := make(chan bool)
-	testingbus.SetWatchTicker(endp, watchTicker)
+	watchTicker := make(chan []interface{})
+	testingbus.SetWatchSource(endp, "StateChanged", watchTicker)
+	nopTicker := make(chan []interface{})
+	testingbus.SetWatchSource(endp, "PropertiesChanged", nopTicker)
 
 	out := make(chan bool)
 	dt := time.Second / 10
@@ -298,19 +339,31 @@
 	go ConnectedState(endp, cfg, s.log, out)
 	var v bool
 	expecteds := []struct {
-		p bool
-		s string
-		n int
+		p    bool
+		s    string
+		todo string
 	}{
-		{false, "first state is always false", 0},
-		{true, "then it should be true as per ConnectedGlobal above", 0},
-		{false, "then it should be false (Disconnected)", 2},
-		{false, "then it should be false again because it's restarted", 2},
+		{false, "first state is always false", ""},
+		{true, "then it should be true as per ConnectedGlobal above", "ConnectedGlobal"},
+		{false, "then it should be false (Disconnected)", "Disconnected"},
+		{false, "then it should be false again because it's restarted", "close"},
 	}
 
+	defer func() {
+		if watchTicker != nil {
+			close(watchTicker)
+		}
+	}()
+	defer close(nopTicker)
 	for i, expected := range expecteds {
-		for j := 0; j < expected.n; j++ {
-			watchTicker <- true
+		switch expected.todo {
+		case "ConnectedGlobal":
+			watchTicker <- []interface{}{uint32(networkmanager.ConnectedGlobal)}
+		case "Disconnected":
+			watchTicker <- []interface{}{uint32(networkmanager.Disconnected)}
+		case "close":
+			close(watchTicker)
+			watchTicker = nil
 		}
 		timer.Reset(dt)
 		select {
@@ -335,11 +388,13 @@
 
 	endp := testingbus.NewTestingEndpoint(condition.Work(true), condition.Work(true),
 		uint32(networkmanager.ConnectedGlobal),
-		map[string]dbus.Variant{"PrimaryConnection": dbus.Variant{dbus.ObjectPath("hello")}},
+		helloCon,
 	)
 
-	watchTicker := make(chan bool)
-	testingbus.SetWatchTicker(endp, watchTicker)
+	watchTicker := make(chan []interface{})
+	testingbus.SetWatchSource(endp, "PropertiesChanged", watchTicker)
+	nopTicker := make(chan []interface{})
+	testingbus.SetWatchSource(endp, "StateChanged", nopTicker)
 
 	out := make(chan bool)
 	dt := time.Second / 10
@@ -347,19 +402,25 @@
 	go ConnectedState(endp, cfg, s.log, out)
 	var v bool
 	expecteds := []struct {
-		p bool
-		s string
-		n int
+		p           bool
+		s           string
+		changedConn bool
 	}{
-		{false, "first state is always false", 0},
-		{true, "then it should be true as per ConnectedGlobal above", 0},
-		{false, "then, false (PrimaryConnection changed)", 2},
-		{true, "then it should be true (webcheck passed)", 0},
+		{false, "first state is always false", false},
+		{true, "then it should be true as per ConnectedGlobal above", false},
+		{false, "then, false (PrimaryConnection changed)", true},
+		{true, "then it should be true (webcheck passed)", false},
 	}
 
+	defer func() {
+		if watchTicker != nil {
+			close(watchTicker)
+		}
+	}()
+	defer close(nopTicker)
 	for i, expected := range expecteds {
-		for j := 0; j < expected.n; j++ {
-			watchTicker <- true
+		if expected.changedConn {
+			watchTicker <- []interface{}{helloConProps}
 		}
 		timer.Reset(dt)
 		select {

=== modified file 'bus/notifications/raw_test.go'
--- bus/notifications/raw_test.go	2014-08-15 10:33:04 +0000
+++ bus/notifications/raw_test.go	2015-02-25 11:05:41 +0000
@@ -111,14 +111,16 @@
 	errstr string
 	endp   bus.Endpoint
 	works  bool
+	src    chan []interface{}
 }
 
 func (s *RawSuite) TestWatchActionsToleratesDBusWeirdness(c *C) {
 	X := func(errstr string, args ...interface{}) tst {
-		endp := testibus.NewMultiValuedTestingEndpoint(nil, condition.Work(true), args)
-		// stop the endpoint from closing the channel:
-		testibus.SetWatchTicker(endp, make(chan bool))
-		return tst{errstr, endp, errstr == ""}
+		endp := testibus.NewMultiValuedTestingEndpoint(nil, condition.Work(true))
+		src := make(chan []interface{}, 1)
+		testibus.SetWatchSource(endp, "ActionInvoked", src)
+		src <- args
+		return tst{errstr, endp, errstr == "", src}
 	}
 
 	ts := []tst{
@@ -146,6 +148,7 @@
 		}
 		c.Check(s.log.Captured(), Matches, `(?ms).*`+t.errstr+`.*`)
 		s.log.ResetCapture()
+		close(t.src)
 	}
 
 }

=== modified file 'bus/testing/testing_endpoint.go'
--- bus/testing/testing_endpoint.go	2015-01-29 09:48:40 +0000
+++ bus/testing/testing_endpoint.go	2015-02-25 11:05:41 +0000
@@ -36,13 +36,15 @@
 }
 
 type testingEndpoint struct {
-	dialCond    condition.Interface
-	callCond    condition.Interface
-	retvals     [][]interface{}
-	watchTicker chan bool
-	watchLck    sync.RWMutex
-	callArgs    []callArgs
-	callArgsLck sync.RWMutex
+	dialCond     condition.Interface
+	callCond     condition.Interface
+	usedLck      sync.Mutex
+	used         int
+	retvals      [][]interface{}
+	watchSources map[string]chan []interface{}
+	watchLck     sync.RWMutex
+	callArgs     []callArgs
+	callArgsLck  sync.RWMutex
 }
 
 // Build a bus.Endpoint that calls OK() on its condition before returning
@@ -51,7 +53,7 @@
 // NOTE: Call() always returns the first return value; Watch() will provide
 // each of them in turn, irrespective of whether Call has been called.
 func NewMultiValuedTestingEndpoint(dialCond condition.Interface, callCond condition.Interface, retvalses ...[]interface{}) bus.Endpoint {
-	return &testingEndpoint{dialCond: dialCond, callCond: callCond, retvals: retvalses}
+	return &testingEndpoint{dialCond: dialCond, callCond: callCond, retvals: retvalses, watchSources: make(map[string]chan []interface{})}
 }
 
 func NewTestingEndpoint(dialCond condition.Interface, callCond condition.Interface, retvals ...interface{}) bus.Endpoint {
@@ -59,15 +61,15 @@
 	for i, x := range retvals {
 		retvalses[i] = []interface{}{x}
 	}
-	return &testingEndpoint{dialCond: dialCond, callCond: callCond, retvals: retvalses}
+	return &testingEndpoint{dialCond: dialCond, callCond: callCond, retvals: retvalses, watchSources: make(map[string]chan []interface{})}
 }
 
-// If SetWatchTicker is called with a non-nil watchTicker, it is used
-// instead of the default timeout to wait while sending values over
-// WatchSignal. Set it to nil again to restore default behaviour.
-func SetWatchTicker(tc bus.Endpoint, watchTicker chan bool) {
+// If SetWatchSource is called with a non-nil watchSource, it is used
+// instead of the default timeout and retvals to get values to send
+// over WatchSignal. Set it to nil again to restore default behaviour.
+func SetWatchSource(tc bus.Endpoint, member string, watchSource chan []interface{}) {
 	tc.(*testingEndpoint).watchLck.Lock()
-	tc.(*testingEndpoint).watchTicker = watchTicker
+	tc.(*testingEndpoint).watchSources[member] = watchSource
 	tc.(*testingEndpoint).watchLck.Unlock()
 }
 
@@ -80,23 +82,29 @@
 
 // See Endpoint's WatchSignal. This WatchSignal will check its condition to
 // decide whether to return an error, or provide each of its return values
+// or values from the previously set watchSource for member.
 func (tc *testingEndpoint) WatchSignal(member string, f func(...interface{}), d func()) error {
 	if tc.callCond.OK() {
 		go func() {
-			for _, v := range tc.retvals {
+			tc.watchLck.RLock()
+			source := tc.watchSources[member]
+			tc.watchLck.RUnlock()
+			if source == nil {
+				tc.usedLck.Lock()
+				idx := tc.used
+				tc.used++
+				tc.usedLck.Unlock()
+				source = make(chan []interface{})
+				go func() {
+					for _, v := range tc.retvals[idx:] {
+						source <- v
+						time.Sleep(10 * time.Millisecond)
+					}
+					close(source)
+				}()
+			}
+			for v := range source {
 				f(v...)
-				tc.watchLck.RLock()
-				ticker := tc.watchTicker
-				tc.watchLck.RUnlock()
-				if ticker != nil {
-					_, ok := <-ticker
-					if !ok {
-						// bail out
-						return
-					}
-				} else {
-					time.Sleep(10 * time.Millisecond)
-				}
 			}
 			d()
 		}()
@@ -116,20 +124,24 @@
 	if tc.callCond.OK() {
 		expected := len(rvs)
 		var provided int
-		if len(tc.retvals) == 0 {
+		tc.usedLck.Lock()
+		idx := tc.used
+		tc.used++
+		tc.usedLck.Unlock()
+		if len(tc.retvals) <= idx {
 			if expected != 0 {
 				panic("No return values provided!")
 			}
 			provided = 0
 		} else {
-			provided = len(tc.retvals[0])
+			provided = len(tc.retvals[idx])
 		}
 		if provided != expected {
 			return errors.New("provided/expected return vals mismatch")
 		}
 		if provided != 0 {
 			x := dbus.NewMethodCallMessage("", "", "", "")
-			err := x.AppendArgs(tc.retvals[0]...)
+			err := x.AppendArgs(tc.retvals[idx]...)
 			if err != nil {
 				return err
 			}

=== modified file 'bus/testing/testing_endpoint_test.go'
--- bus/testing/testing_endpoint_test.go	2014-07-04 23:00:42 +0000
+++ bus/testing/testing_endpoint_test.go	2015-02-25 11:05:41 +0000
@@ -134,21 +134,22 @@
 	c.Check(e, NotNil)
 }
 
-// Test WatchSignal can use the WatchTicker instead of a timeout (if
+// Test WatchSignal can use a watchSource instead of a timeout and retvals (if
 // the former is not nil)
-func (s *TestingEndpointSuite) TestWatchTicker(c *C) {
-	watchTicker := make(chan bool, 3)
-	watchTicker <- true
-	watchTicker <- true
-	watchTicker <- true
+func (s *TestingEndpointSuite) TestWatchSources(c *C) {
+	watchTicker := make(chan []interface{}, 3)
+	watchTicker <- []interface{}{true}
+	watchTicker <- []interface{}{true}
+	watchTicker <- []interface{}{true}
 	c.Assert(len(watchTicker), Equals, 3)
 
 	endp := NewTestingEndpoint(nil, condition.Work(true), 0, 0)
-	SetWatchTicker(endp, watchTicker)
+	SetWatchSource(endp, "what", watchTicker)
 	ch := make(chan int)
 	e := endp.WatchSignal("what", func(us ...interface{}) {}, func() { close(ch) })
 	c.Check(e, IsNil)
 
+	close(watchTicker)
 	// wait for the destructor to be called
 	select {
 	case <-time.Tick(10 * time.Millisecond):
@@ -156,8 +157,8 @@
 	case <-ch:
 	}
 
-	// now if all went well, the ticker will have been tuck twice.
-	c.Assert(len(watchTicker), Equals, 1)
+	// now if all went well, the ticker will have been exhausted.
+	c.Assert(len(watchTicker), Equals, 0)
 }
 
 // Tests that GetProperty() works

=== modified file 'client/client_test.go'
--- client/client_test.go	2015-02-06 13:09:16 +0000
+++ client/client_test.go	2015-02-25 11:05:41 +0000
@@ -31,6 +31,7 @@
 	"testing"
 	"time"
 
+	"launchpad.net/go-dbus/v1"
 	. "launchpad.net/gocheck"
 
 	"launchpad.net/ubuntu-push/accounts"
@@ -657,13 +658,19 @@
 	// testing endpoints
 	cCond := condition.Fail2Work(7)
 	cEndp := testibus.NewTestingEndpoint(cCond, condition.Work(true),
-		uint32(networkmanager.ConnectedGlobal),
+		uint32(networkmanager.Connecting),
+		dbus.ObjectPath("hello"),
+		uint32(networkmanager.Connecting),
+		dbus.ObjectPath("hello"),
 	)
 	siCond := condition.Fail2Work(2)
 	siEndp := testibus.NewMultiValuedTestingEndpoint(siCond, condition.Work(true), []interface{}{int32(101), "mako", "daily", "Unknown", map[string]string{}})
-	tickerCh := make(chan bool)
-	testibus.SetWatchTicker(cEndp, tickerCh)
+	tickerCh := make(chan []interface{})
+	nopTickerCh := make(chan []interface{})
+	testibus.SetWatchSource(cEndp, "StateChanged", tickerCh)
+	testibus.SetWatchSource(cEndp, "PropertiesChanged", nopTickerCh)
 	defer close(tickerCh)
+	defer close(nopTickerCh)
 	// ok, create the thing
 	cli := NewPushClient(cs.configPath, cs.leveldbPath)
 	cli.log = cs.log
@@ -679,6 +686,7 @@
 	c.Assert(cli.takeTheBus(), IsNil)
 
 	c.Check(takeNextBool(cli.connCh), Equals, false)
+	tickerCh <- []interface{}{uint32(networkmanager.ConnectedGlobal)}
 	c.Check(takeNextBool(cli.connCh), Equals, true)
 	// the connectivity endpoint retried until connected
 	c.Check(cCond.OK(), Equals, true)

=== modified file 'client/service/postal_test.go'
--- client/service/postal_test.go	2014-09-09 22:54:04 +0000
+++ client/service/postal_test.go	2015-02-25 11:05:41 +0000
@@ -552,6 +552,10 @@
 	svc.fallbackVibration = &launch_helper.Vibration{Pattern: []uint32{1}}
 	c.Assert(svc.Start(), IsNil)
 
+	nopTicker := make(chan []interface{})
+	testibus.SetWatchSource(endp, "ActionInvoked", nopTicker)
+	defer close(nopTicker)
+
 	// Persist is false so we just check the log
 	card := &launch_helper.Card{Icon: "icon-value", Summary: "summary-value", Body: "body-value", Popup: true, Persist: false}
 	vib := json.RawMessage(`true`)
@@ -837,6 +841,10 @@
 }
 
 func (ps *postalSuite) TestBlacklisted(c *C) {
+	ps.winStackBus = testibus.NewTestingEndpoint(condition.Work(true), condition.Work(true), []windowstack.WindowsInfo{},
+		[]windowstack.WindowsInfo{},
+		[]windowstack.WindowsInfo{},
+		[]windowstack.WindowsInfo{})
 	svc := ps.replaceBuses(NewPostalService(ps.cfg, ps.log))
 	svc.Start()
 	ps.blacklisted = false