func TestEndpoints(t *testing.T) {
	endpoint := api.Endpoints{
		ObjectMeta: api.ObjectMeta{Name: "bar", ResourceVersion: "2"},
		Subsets: []api.EndpointSubset{{
			Addresses: []api.EndpointAddress{{IP: "127.0.0.1"}},
			Ports:     []api.EndpointPort{{Port: 9000}},
		}},
	}

	fakeWatch := watch.NewFake()
	fakeClient := &testclient.Fake{Watch: fakeWatch}
	endpoints := make(chan EndpointsUpdate)
	source := SourceAPI{
		s: servicesReflector{watcher: fakeClient.Services(api.NamespaceAll)},
		e: endpointsReflector{watcher: fakeClient.Endpoints(api.NamespaceAll), endpoints: endpoints}}
	resourceVersion := "1"
	go func() {
		// called twice
		source.e.run(&resourceVersion)
		source.e.run(&resourceVersion)
	}()

	// test adding an endpoint to the watch
	fakeWatch.Add(&endpoint)
	if !reflect.DeepEqual(fakeClient.Actions, []testclient.FakeAction{{"watch-endpoints", "1"}}) {
		t.Errorf("expected call to watch-endpoints, got %#v", fakeClient)
	}

	actual := <-endpoints
	expected := EndpointsUpdate{Op: ADD, Endpoints: []api.Endpoints{endpoint}}
	if !reflect.DeepEqual(expected, actual) {
		t.Errorf("expected %#v, got %#v", expected, actual)
	}

	// verify that a delete results in a config change
	fakeWatch.Delete(&endpoint)
	actual = <-endpoints
	expected = EndpointsUpdate{Op: REMOVE, Endpoints: []api.Endpoints{endpoint}}
	if !reflect.DeepEqual(expected, actual) {
		t.Errorf("expected %#v, got %#v", expected, actual)
	}

	// verify that closing the channel results in a new call to WatchEndpoints with a higher resource version
	newFakeWatch := watch.NewFake()
	fakeClient.Watch = newFakeWatch
	fakeWatch.Stop()

	newFakeWatch.Add(&endpoint)
	if !reflect.DeepEqual(fakeClient.Actions, []testclient.FakeAction{{"watch-endpoints", "1"}, {"watch-endpoints", "2"}}) {
		t.Errorf("expected call to watch-endpoints, got %#v", fakeClient)
	}
}

func TestRunUntil(t *testing.T) {
	stopCh := make(chan struct{})
	store := NewStore(MetaNamespaceKeyFunc)
	r := NewReflector(&testLW{}, &api.Pod{}, store, 0)
	fw := watch.NewFake()
	r.listerWatcher = &testLW{
		WatchFunc: func(rv string) (watch.Interface, error) {
			return fw, nil
		},
		ListFunc: func() (runtime.Object, error) {
			return &api.PodList{ListMeta: api.ListMeta{ResourceVersion: "1"}}, nil
		},
	}
	r.RunUntil(stopCh)
	// Synchronously add a dummy pod into the watch channel so we
	// know the RunUntil go routine is in the watch handler.
	fw.Add(&api.Pod{ObjectMeta: api.ObjectMeta{Name: "bar"}})
	stopCh <- struct{}{}
	select {
	case _, ok := <-fw.ResultChan():
		if ok {
			t.Errorf("Watch channel left open after stopping the watch")
		}
	case <-time.After(100 * time.Millisecond):
		t.Errorf("the cancellation is at least 99 milliseconds late")
		break
	}
}

func TestWatchControllers(t *testing.T) {
	client := FakeWatcher{watch.NewFake(), &client.Fake{}}
	manager := NewReplicationManager(client)
	var testControllerSpec api.ReplicationController
	received := make(chan struct{})
	manager.syncHandler = func(controllerSpec api.ReplicationController) error {
		if !reflect.DeepEqual(controllerSpec, testControllerSpec) {
			t.Errorf("Expected %#v, but got %#v", testControllerSpec, controllerSpec)
		}
		close(received)
		return nil
	}

	resourceVersion := uint64(0)
	go manager.watchControllers(&resourceVersion)

	// Test normal case
	testControllerSpec.ID = "foo"
	client.w.Add(&testControllerSpec)

	select {
	case <-received:
	case <-time.After(10 * time.Millisecond):
		t.Errorf("Expected 1 call but got 0")
	}
}

func TestWatchControllers(t *testing.T) {
	fakeWatch := watch.NewFake()
	client := &client.Fake{Watch: fakeWatch}
	manager := NewReplicationManager(client)
	var testControllerSpec api.ReplicationController
	received := make(chan struct{})
	manager.syncHandler = func(controllerSpec api.ReplicationController) error {
		if !api.Semantic.DeepDerivative(controllerSpec, testControllerSpec) {
			t.Errorf("Expected %#v, but got %#v", testControllerSpec, controllerSpec)
		}
		close(received)
		return nil
	}

	resourceVersion := ""
	go manager.watchControllers(&resourceVersion)

	// Test normal case
	testControllerSpec.Name = "foo"

	fakeWatch.Add(&testControllerSpec)

	select {
	case <-received:
	case <-time.After(10 * time.Millisecond):
		t.Errorf("Expected 1 call but got 0")
	}
}

// Implement ResourceWatcher.
func (storage *SimpleRESTStorage) WatchAll() (watch.Interface, error) {
	if err := storage.errors["watchAll"]; err != nil {
		return nil, err
	}
	storage.fakeWatch = watch.NewFake()
	return storage.fakeWatch, nil
}

// Implement ResourceWatcher.
func (storage *SimpleRESTStorage) WatchSingle(id string) (watch.Interface, error) {
	storage.requestedID = id
	if err := storage.errors["watchSingle"]; err != nil {
		return nil, err
	}
	storage.fakeWatch = watch.NewFake()
	return storage.fakeWatch, nil
}

func TestServices(t *testing.T) {
	service := api.Service{ObjectMeta: api.ObjectMeta{Name: "bar", ResourceVersion: "2"}}

	fakeWatch := watch.NewFake()
	fakeClient := &testclient.Fake{Watch: fakeWatch}
	services := make(chan ServiceUpdate)
	source := SourceAPI{
		s: servicesReflector{watcher: fakeClient.Services(api.NamespaceAll), services: services},
		e: endpointsReflector{watcher: fakeClient.Endpoints(api.NamespaceAll)}}
	resourceVersion := "1"
	go func() {
		// called twice
		source.s.run(&resourceVersion)
		source.s.run(&resourceVersion)
	}()

	// test adding a service to the watch
	fakeWatch.Add(&service)
	if !reflect.DeepEqual(fakeClient.Actions, []testclient.FakeAction{{"watch-services", "1"}}) {
		t.Errorf("expected call to watch-services, got %#v", fakeClient)
	}

	actual := <-services
	expected := ServiceUpdate{Op: ADD, Services: []api.Service{service}}
	if !reflect.DeepEqual(expected, actual) {
		t.Errorf("expected %#v, got %#v", expected, actual)
	}

	// verify that a delete results in a config change
	fakeWatch.Delete(&service)
	actual = <-services
	expected = ServiceUpdate{Op: REMOVE, Services: []api.Service{service}}
	if !reflect.DeepEqual(expected, actual) {
		t.Errorf("expected %#v, got %#v", expected, actual)
	}

	// verify that closing the channel results in a new call to WatchServices with a higher resource version
	newFakeWatch := watch.NewFake()
	fakeClient.Watch = newFakeWatch
	fakeWatch.Stop()

	newFakeWatch.Add(&service)
	if !reflect.DeepEqual(fakeClient.Actions, []testclient.FakeAction{{"watch-services", "1"}, {"watch-services", "2"}}) {
		t.Errorf("expected call to watch-endpoints, got %#v", fakeClient)
	}
}

func TestNewSourceApiserver(t *testing.T) {
	podv1 := api.Pod{
		ObjectMeta: api.ObjectMeta{Name: "p"},
		Spec:       api.PodSpec{Containers: []api.Container{{Image: "image/one"}}}}
	podv2 := api.Pod{
		ObjectMeta: api.ObjectMeta{Name: "p"},
		Spec:       api.PodSpec{Containers: []api.Container{{Image: "image/two"}}}}

	expectedBoundPodv1 := api.BoundPod{
		ObjectMeta: api.ObjectMeta{Name: "p", SelfLink: "/api/v1beta1/boundPods/p"},
		Spec:       api.PodSpec{Containers: []api.Container{{Image: "image/one"}}}}
	expectedBoundPodv2 := api.BoundPod{
		ObjectMeta: api.ObjectMeta{Name: "p", SelfLink: "/api/v1beta1/boundPods/p"},
		Spec:       api.PodSpec{Containers: []api.Container{{Image: "image/two"}}}}

	// Setup fake api client.
	fakeWatch := watch.NewFake()
	lw := fakePodLW{
		listResp:  &api.PodList{Items: []api.Pod{podv1}},
		watchResp: fakeWatch,
	}

	ch := make(chan interface{})

	newSourceApiserverFromLW(lw, ch)

	got, ok := <-ch
	if !ok {
		t.Errorf("Unable to read from channel when expected")
	}
	update := got.(kubelet.PodUpdate)
	expected := CreatePodUpdate(kubelet.SET, kubelet.ApiserverSource, expectedBoundPodv1)
	if !api.Semantic.DeepEqual(expected, update) {
		t.Errorf("Expected %#v; Got %#v", expected, update)
	}

	fakeWatch.Modify(&podv2)
	got, ok = <-ch
	if !ok {
		t.Errorf("Unable to read from channel when expected")
	}
	update = got.(kubelet.PodUpdate)
	expected = CreatePodUpdate(kubelet.SET, kubelet.ApiserverSource, expectedBoundPodv2)
	if !api.Semantic.DeepEqual(expected, update) {
		t.Fatalf("Expected %#v, Got %#v", expected, update)
	}

	fakeWatch.Delete(&podv2)
	got, ok = <-ch
	if !ok {
		t.Errorf("Unable to read from channel when expected")
	}
	update = got.(kubelet.PodUpdate)
	expected = CreatePodUpdate(kubelet.SET, kubelet.ApiserverSource)
	if !api.Semantic.DeepEqual(expected, update) {
		t.Errorf("Expected %#v, Got %#v", expected, update)
	}
}

func TestReflector_listAndWatch(t *testing.T) {
	createdFakes := make(chan *watch.FakeWatcher)

	// The ListFunc says that it's at revision 1. Therefore, we expect our WatchFunc
	// to get called at the beginning of the watch with 1, and again with 3 when we
	// inject an error.
	expectedRVs := []string{"1", "3"}
	lw := &testLW{
		WatchFunc: func(rv string) (watch.Interface, error) {
			fw := watch.NewFake()
			if e, a := expectedRVs[0], rv; e != a {
				t.Errorf("Expected rv %v, but got %v", e, a)
			}
			expectedRVs = expectedRVs[1:]
			// channel is not buffered because the for loop below needs to block. But
			// we don't want to block here, so report the new fake via a go routine.
			go func() { createdFakes <- fw }()
			return fw, nil
		},
		ListFunc: func() (runtime.Object, error) {
			return &api.PodList{ListMeta: api.ListMeta{ResourceVersion: "1"}}, nil
		},
	}
	s := NewFIFO(MetaNamespaceKeyFunc)
	r := NewReflector(lw, &api.Pod{}, s)
	go r.listAndWatch()

	ids := []string{"foo", "bar", "baz", "qux", "zoo"}
	var fw *watch.FakeWatcher
	for i, id := range ids {
		if fw == nil {
			fw = <-createdFakes
		}
		sendingRV := strconv.FormatUint(uint64(i+2), 10)
		fw.Add(&api.Pod{ObjectMeta: api.ObjectMeta{Name: id, ResourceVersion: sendingRV}})
		if sendingRV == "3" {
			// Inject a failure.
			fw.Stop()
			fw = nil
		}
	}

	// Verify we received the right ids with the right resource versions.
	for i, id := range ids {
		pod := s.Pop().(*api.Pod)
		if e, a := id, pod.Name; e != a {
			t.Errorf("%v: Expected %v, got %v", i, e, a)
		}
		if e, a := strconv.FormatUint(uint64(i+2), 10), pod.ResourceVersion; e != a {
			t.Errorf("%v: Expected %v, got %v", i, e, a)
		}
	}

	if len(expectedRVs) != 0 {
		t.Error("called watchStarter an unexpected number of times")
	}
}

func TestServices(t *testing.T) {
	service := api.Service{JSONBase: api.JSONBase{ID: "bar", ResourceVersion: uint64(2)}}

	fakeWatch := watch.NewFake()
	fakeClient := &client.Fake{Watch: fakeWatch}
	services := make(chan ServiceUpdate)
	source := SourceAPI{client: fakeClient, services: services}
	resourceVersion := uint64(1)
	go func() {
		// called twice
		source.runServices(&resourceVersion)
		source.runServices(&resourceVersion)
	}()

	// test adding a service to the watch
	fakeWatch.Add(&service)
	if !reflect.DeepEqual(fakeClient.Actions, []client.FakeAction{{"watch-services", uint64(1)}}) {
		t.Errorf("expected call to watch-services, got %#v", fakeClient)
	}

	actual := <-services
	expected := ServiceUpdate{Op: ADD, Services: []api.Service{service}}
	if !reflect.DeepEqual(expected, actual) {
		t.Errorf("expected %#v, got %#v", expected, actual)
	}

	// verify that a delete results in a config change
	fakeWatch.Delete(&service)
	actual = <-services
	expected = ServiceUpdate{Op: REMOVE, Services: []api.Service{service}}
	if !reflect.DeepEqual(expected, actual) {
		t.Errorf("expected %#v, got %#v", expected, actual)
	}

	// verify that closing the channel results in a new call to WatchServices with a higher resource version
	newFakeWatch := watch.NewFake()
	fakeClient.Watch = newFakeWatch
	fakeWatch.Stop()

	newFakeWatch.Add(&service)
	if !reflect.DeepEqual(fakeClient.Actions, []client.FakeAction{{"watch-services", uint64(1)}, {"watch-services", uint64(3)}}) {
		t.Errorf("expected call to watch-endpoints, got %#v", fakeClient)
	}
}

func TestEndpoints(t *testing.T) {
	endpoint := api.Endpoints{JSONBase: api.JSONBase{ID: "bar", ResourceVersion: uint64(2)}, Endpoints: []string{"127.0.0.1:9000"}}

	fakeWatch := watch.NewFake()
	fakeClient := &client.Fake{Watch: fakeWatch}
	endpoints := make(chan EndpointsUpdate)
	source := SourceAPI{client: fakeClient, endpoints: endpoints}
	resourceVersion := uint64(1)
	go func() {
		// called twice
		source.runEndpoints(&resourceVersion)
		source.runEndpoints(&resourceVersion)
	}()

	// test adding an endpoint to the watch
	fakeWatch.Add(&endpoint)
	if !reflect.DeepEqual(fakeClient.Actions, []client.FakeAction{{"watch-endpoints", uint64(1)}}) {
		t.Errorf("expected call to watch-endpoints, got %#v", fakeClient)
	}

	actual := <-endpoints
	expected := EndpointsUpdate{Op: ADD, Endpoints: []api.Endpoints{endpoint}}
	if !reflect.DeepEqual(expected, actual) {
		t.Errorf("expected %#v, got %#v", expected, actual)
	}

	// verify that a delete results in a config change
	fakeWatch.Delete(&endpoint)
	actual = <-endpoints
	expected = EndpointsUpdate{Op: REMOVE, Endpoints: []api.Endpoints{endpoint}}
	if !reflect.DeepEqual(expected, actual) {
		t.Errorf("expected %#v, got %#v", expected, actual)
	}

	// verify that closing the channel results in a new call to WatchEndpoints with a higher resource version
	newFakeWatch := watch.NewFake()
	fakeClient.Watch = newFakeWatch
	fakeWatch.Stop()

	newFakeWatch.Add(&endpoint)
	if !reflect.DeepEqual(fakeClient.Actions, []client.FakeAction{{"watch-endpoints", uint64(1)}, {"watch-endpoints", uint64(3)}}) {
		t.Errorf("expected call to watch-endpoints, got %#v", fakeClient)
	}
}

// Implement ResourceWatcher.
func (storage *SimpleRESTStorage) Watch(label, field labels.Selector, resourceVersion uint64) (watch.Interface, error) {
	storage.requestedLabelSelector = label
	storage.requestedFieldSelector = field
	storage.requestedResourceVersion = resourceVersion
	if err := storage.errors["watch"]; err != nil {
		return nil, err
	}
	storage.fakeWatch = watch.NewFake()
	return storage.fakeWatch, nil
}

func TestReflector_watchHandler(t *testing.T) {
	s := NewStore(MetaNamespaceKeyFunc)
	g := NewReflector(&testLW{}, &api.Pod{}, s, 0)
	fw := watch.NewFake()
	s.Add(&api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}})
	s.Add(&api.Pod{ObjectMeta: api.ObjectMeta{Name: "bar"}})
	go func() {
		fw.Add(&api.Service{ObjectMeta: api.ObjectMeta{Name: "rejected"}})
		fw.Delete(&api.Pod{ObjectMeta: api.ObjectMeta{Name: "foo"}})
		fw.Modify(&api.Pod{ObjectMeta: api.ObjectMeta{Name: "bar", ResourceVersion: "55"}})
		fw.Add(&api.Pod{ObjectMeta: api.ObjectMeta{Name: "baz", ResourceVersion: "32"}})
		fw.Stop()
	}()
	var resumeRV string
	err := g.watchHandler(fw, &resumeRV, neverExitWatch, util.NeverStop)
	if err != nil {
		t.Errorf("unexpected error %v", err)
	}

	mkPod := func(id string, rv string) *api.Pod {
		return &api.Pod{ObjectMeta: api.ObjectMeta{Name: id, ResourceVersion: rv}}
	}

	table := []struct {
		Pod    *api.Pod
		exists bool
	}{
		{mkPod("foo", ""), false},
		{mkPod("rejected", ""), false},
		{mkPod("bar", "55"), true},
		{mkPod("baz", "32"), true},
	}
	for _, item := range table {
		obj, exists, _ := s.Get(item.Pod)
		if e, a := item.exists, exists; e != a {
			t.Errorf("%v: expected %v, got %v", item.Pod, e, a)
		}
		if !exists {
			continue
		}
		if e, a := item.Pod.ResourceVersion, obj.(*api.Pod).ResourceVersion; e != a {
			t.Errorf("%v: expected %v, got %v", item.Pod, e, a)
		}
	}

	// RV should send the last version we see.
	if e, a := "32", resumeRV; e != a {
		t.Errorf("expected %v, got %v", e, a)
	}

	// last sync resource version should be the last version synced with store
	if e, a := "32", g.LastSyncResourceVersion(); e != a {
		t.Errorf("expected %v, got %v", e, a)
	}
}

// Implement ResourceWatcher.
func (storage *SimpleRESTStorage) Watch(ctx api.Context, label, field labels.Selector, resourceVersion string) (watch.Interface, error) {
	storage.requestedLabelSelector = label
	storage.requestedFieldSelector = field
	storage.requestedResourceVersion = resourceVersion
	storage.requestedResourceNamespace = api.Namespace(ctx)
	if err := storage.errors["watch"]; err != nil {
		return nil, err
	}
	storage.fakeWatch = watch.NewFake()
	return storage.fakeWatch, nil
}

func TestReflectorStopWatch(t *testing.T) {
	s := NewStore(MetaNamespaceKeyFunc)
	g := NewReflector(&testLW{}, &api.Pod{}, s, 0)
	fw := watch.NewFake()
	var resumeRV string
	stopWatch := make(chan struct{}, 1)
	stopWatch <- struct{}{}
	err := g.watchHandler(fw, &resumeRV, neverExitWatch, stopWatch)
	if err != errorStopRequested {
		t.Errorf("expected stop error, got %q", err)
	}
}

func TestWatchHTTPTimeout(t *testing.T) {
	watcher := watch.NewFake()
	timeoutCh := make(chan time.Time)
	done := make(chan struct{})

	// Setup a new watchserver
	watchServer := &WatchServer{
		watcher,
		newCodec,
		func(obj runtime.Object) {},
		&fakeTimeoutFactory{timeoutCh, done},
	}

	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
		watchServer.ServeHTTP(w, req)
	}))
	defer s.Close()

	// Setup a client
	dest, _ := url.Parse(s.URL)
	dest.Path = "/api/" + newVersion + "/simple"
	dest.RawQuery = "watch=true"

	req, _ := http.NewRequest("GET", dest.String(), nil)
	client := http.Client{}
	resp, err := client.Do(req)
	watcher.Add(&Simple{TypeMeta: api.TypeMeta{APIVersion: newVersion}})

	// Make sure we can actually watch an endpoint
	decoder := json.NewDecoder(resp.Body)
	var got watchJSON
	err = decoder.Decode(&got)
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}

	// Timeout and check for leaks
	close(timeoutCh)
	select {
	case <-done:
		if !watcher.Stopped {
			t.Errorf("Leaked watch on timeout")
		}
	case <-time.After(100 * time.Millisecond):
		t.Errorf("Failed to stop watcher after 100ms of timeout signal")
	}

	// Make sure we can't receive any more events through the timeout watch
	err = decoder.Decode(&got)
	if err != io.EOF {
		t.Errorf("Unexpected non-error")
	}
}

func TestReflector_watchHandlerTimeout(t *testing.T) {
	s := NewStore(MetaNamespaceKeyFunc)
	g := NewReflector(&testLW{}, &api.Pod{}, s, 0)
	fw := watch.NewFake()
	var resumeRV string
	exit := make(chan time.Time, 1)
	exit <- time.Now()
	err := g.watchHandler(fw, &resumeRV, exit, util.NeverStop)
	if err != errorResyncRequested {
		t.Errorf("expected timeout error, but got %q", err)
	}
}

func TestReflector_Run(t *testing.T) {
	createdFakes := make(chan *watch.FakeWatcher)

	// Expect our starter to get called at the beginning of the watch with 0, and again with 3 when we
	// inject an error at 2.
	expectedRVs := []uint64{0, 3}
	watchStarter := func(rv uint64) (watch.Interface, error) {
		fw := watch.NewFake()
		if e, a := expectedRVs[0], rv; e != a {
			t.Errorf("Expected rv %v, but got %v", e, a)
		}
		expectedRVs = expectedRVs[1:]
		// channel is not buffered because the for loop below needs to block. But
		// we don't want to block here, so report the new fake via a go routine.
		go func() { createdFakes <- fw }()
		return fw, nil
	}
	s := NewFIFO()
	r := NewReflector(watchStarter, &api.Pod{}, s)
	r.period = 0
	r.Run()

	ids := []string{"foo", "bar", "baz", "qux", "zoo"}
	var fw *watch.FakeWatcher
	for i, id := range ids {
		if fw == nil {
			fw = <-createdFakes
		}
		sendingRV := uint64(i + 1)
		fw.Add(&api.Pod{JSONBase: api.JSONBase{ID: id, ResourceVersion: sendingRV}})
		if sendingRV == 2 {
			// Inject a failure.
			fw.Stop()
			fw = nil
		}
	}

	// Verify we received the right ids with the right resource versions.
	for i, id := range ids {
		pod := s.Pop().(*api.Pod)
		if e, a := id, pod.ID; e != a {
			t.Errorf("%v: Expected %v, got %v", i, e, a)
		}
		if e, a := uint64(i+1), pod.ResourceVersion; e != a {
			t.Errorf("%v: Expected %v, got %v", i, e, a)
		}
	}

	if len(expectedRVs) != 0 {
		t.Error("called watchStarter an unexpected number of times")
	}
}

func TestReflector_watchHandlerError(t *testing.T) {
	s := NewStore(MetaNamespaceKeyFunc)
	g := NewReflector(&testLW{}, &api.Pod{}, s)
	fw := watch.NewFake()
	go func() {
		fw.Stop()
	}()
	var resumeRV string
	err := g.watchHandler(fw, &resumeRV)
	if err == nil {
		t.Errorf("unexpected non-error")
	}
}

func TestUpdatePods(t *testing.T) {
	fakeWatch := watch.NewFake()
	client := &testclient.Fake{Watch: fakeWatch}
	manager := NewReplicationManager(client, BurstReplicas)
	manager.podStoreSynced = alwaysReady

	received := make(chan string)

	manager.syncHandler = func(key string) error {
		obj, exists, err := manager.rcStore.Store.GetByKey(key)
		if !exists || err != nil {
			t.Errorf("Expected to find controller under key %v", key)
		}
		received <- obj.(*api.ReplicationController).Name
		return nil
	}

	stopCh := make(chan struct{})
	defer close(stopCh)
	go util.Until(manager.worker, 10*time.Millisecond, stopCh)

	// Put 2 rcs and one pod into the controller's stores
	testControllerSpec1 := newReplicationController(1)
	manager.rcStore.Store.Add(testControllerSpec1)
	testControllerSpec2 := *testControllerSpec1
	testControllerSpec2.Spec.Selector = map[string]string{"bar": "foo"}
	testControllerSpec2.Name = "barfoo"
	manager.rcStore.Store.Add(&testControllerSpec2)

	// Put one pod in the podStore
	pod1 := newPodList(manager.podStore.Store, 1, api.PodRunning, testControllerSpec1).Items[0]
	pod2 := pod1
	pod2.Labels = testControllerSpec2.Spec.Selector

	// Send an update of the same pod with modified labels, and confirm we get a sync request for
	// both controllers
	manager.updatePod(&pod1, &pod2)

	expected := util.NewStringSet(testControllerSpec1.Name, testControllerSpec2.Name)
	for _, name := range expected.List() {
		t.Logf("Expecting update for %+v", name)
		select {
		case got := <-received:
			if !expected.Has(got) {
				t.Errorf("Expected keys %#v got %v", expected, got)
			}
		case <-time.After(controllerTimeout):
			t.Errorf("Expected update notifications for controllers within 100ms each")
		}
	}
}