func TestControllerNoReplicaUpdate(t *testing.T) {
	// Setup a fake server to listen for requests, and run the rc manager in steady state
	fakeHandler := util.FakeHandler{
		StatusCode:   200,
		ResponseBody: "",
	}
	testServer := httptest.NewServer(&fakeHandler)
	defer testServer.Close()
	client := client.NewOrDie(&client.Config{Host: testServer.URL, Version: testapi.Version()})
	manager := NewReplicationManager(client, BurstReplicas)

	// Steady state for the replication controller, no Status.Replicas updates expected
	activePods := 5
	rc := newReplicationController(activePods)
	manager.controllerStore.Store.Add(rc)
	rc.Status = api.ReplicationControllerStatus{Replicas: activePods}
	newPodList(manager.podStore.Store, activePods, api.PodRunning, rc)

	fakePodControl := FakePodControl{}
	manager.podControl = &fakePodControl
	manager.syncReplicationController(getKey(rc, t))

	validateSyncReplication(t, &fakePodControl, 0, 0)
	if fakeHandler.RequestReceived != nil {
		t.Errorf("Unexpected update when pods and rcs are in a steady state")
	}
}

func TestSyncEndpointsProtocolUDP(t *testing.T) {
	ns := "other"
	testServer, endpointsHandler := makeTestServer(t, ns,
		serverResponse{http.StatusOK, &api.Endpoints{
			ObjectMeta: api.ObjectMeta{
				Name:            "foo",
				Namespace:       ns,
				ResourceVersion: "1",
			},
			Subsets: []api.EndpointSubset{{
				Addresses: []api.EndpointAddress{{IP: "6.7.8.9"}},
				Ports:     []api.EndpointPort{{Port: 1000, Protocol: "UDP"}},
			}},
		}})
	defer testServer.Close()
	client := client.NewOrDie(&client.Config{Host: testServer.URL, Version: testapi.Version()})
	endpoints := NewEndpointController(client)
	endpoints.serviceStore.Store.Add(&api.Service{
		ObjectMeta: api.ObjectMeta{Name: "foo", Namespace: ns},
		Spec: api.ServiceSpec{
			Selector: map[string]string{},
			Ports:    []api.ServicePort{{Port: 80}},
		},
	})
	endpoints.syncService(ns + "/foo")
	endpointsHandler.ValidateRequestCount(t, 0)
}

func TestSyncEndpointsItemsPreexistingIdentical(t *testing.T) {
	ns := api.NamespaceDefault
	testServer, endpointsHandler := makeTestServer(t, api.NamespaceDefault,
		serverResponse{http.StatusOK, &api.Endpoints{
			ObjectMeta: api.ObjectMeta{
				ResourceVersion: "1",
				Name:            "foo",
				Namespace:       ns,
			},
			Subsets: []api.EndpointSubset{{
				Addresses: []api.EndpointAddress{{IP: "1.2.3.4", TargetRef: &api.ObjectReference{Kind: "Pod", Name: "pod0", Namespace: ns}}},
				Ports:     []api.EndpointPort{{Port: 8080, Protocol: "TCP"}},
			}},
		}})
	defer testServer.Close()
	client := client.NewOrDie(&client.Config{Host: testServer.URL, Version: testapi.Version()})
	endpoints := NewEndpointController(client)
	addPods(endpoints.podStore.Store, api.NamespaceDefault, 1, 1)
	endpoints.serviceStore.Store.Add(&api.Service{
		ObjectMeta: api.ObjectMeta{Name: "foo", Namespace: api.NamespaceDefault},
		Spec: api.ServiceSpec{
			Selector: map[string]string{"foo": "bar"},
			Ports:    []api.ServicePort{{Port: 80, Protocol: "TCP", TargetPort: util.NewIntOrStringFromInt(8080)}},
		},
	})
	endpoints.syncService(ns + "/foo")
	endpointsHandler.ValidateRequest(t, testapi.ResourcePathWithNamespaceQuery("endpoints", api.NamespaceDefault, "foo"), "GET", nil)
}

func TestSyncEndpointsPodError(t *testing.T) {
	serviceList := api.ServiceList{
		Items: []api.Service{
			{
				Selector: map[string]string{
					"foo": "bar",
				},
			},
		},
	}
	testServer := makeTestServer(t,
		serverResponse{http.StatusInternalServerError, api.PodList{}},
		serverResponse{http.StatusOK, serviceList})
	client := client.NewOrDie(testServer.URL, nil)
	serviceRegistry := registrytest.ServiceRegistry{
		List: api.ServiceList{
			Items: []api.Service{
				{
					Selector: map[string]string{
						"foo": "bar",
					},
				},
			},
		},
	}
	endpoints := NewEndpointController(&serviceRegistry, client)
	if err := endpoints.SyncServiceEndpoints(); err == nil {
		t.Error("Unexpected non-error")
	}
}

func TestCheckLeftoverEndpoints(t *testing.T) {
	ns := api.NamespaceDefault
	// Note that this requests *all* endpoints, therefore the NamespaceAll
	// below.
	testServer, _ := makeTestServer(t, api.NamespaceAll,
		serverResponse{http.StatusOK, &api.EndpointsList{
			ListMeta: api.ListMeta{
				ResourceVersion: "1",
			},
			Items: []api.Endpoints{{
				ObjectMeta: api.ObjectMeta{
					Name:            "foo",
					Namespace:       ns,
					ResourceVersion: "1",
				},
				Subsets: []api.EndpointSubset{{
					Addresses: []api.EndpointAddress{{IP: "6.7.8.9"}},
					Ports:     []api.EndpointPort{{Port: 1000}},
				}},
			}},
		}})
	defer testServer.Close()
	client := client.NewOrDie(&client.Config{Host: testServer.URL, Version: testapi.Version()})
	endpoints := NewEndpointController(client)
	endpoints.checkLeftoverEndpoints()

	if e, a := 1, endpoints.queue.Len(); e != a {
		t.Fatalf("Expected %v, got %v", e, a)
	}
	got, _ := endpoints.queue.Get()
	if e, a := ns+"/foo", got; e != a {
		t.Errorf("Expected %v, got %v", e, a)
	}
}

func NewAPIFactory() (*cmdutil.Factory, *testFactory, runtime.Codec) {
	t := &testFactory{
		Validator: validation.NullSchema{},
	}
	return &cmdutil.Factory{
		Object: func() (meta.RESTMapper, runtime.ObjectTyper) {
			return latest.RESTMapper, api.Scheme
		},
		Client: func() (*client.Client, error) {
			// Swap out the HTTP client out of the client with the fake's version.
			fakeClient := t.Client.(*client.FakeRESTClient)
			c := client.NewOrDie(t.ClientConfig)
			c.Client = fakeClient.Client
			return c, t.Err
		},
		RESTClient: func(*meta.RESTMapping) (resource.RESTClient, error) {
			return t.Client, t.Err
		},
		Describer: func(*meta.RESTMapping) (kubectl.Describer, error) {
			return t.Describer, t.Err
		},
		Printer: func(mapping *meta.RESTMapping, noHeaders bool) (kubectl.ResourcePrinter, error) {
			return t.Printer, t.Err
		},
		Validator: func() (validation.Schema, error) {
			return t.Validator, t.Err
		},
		DefaultNamespace: func() (string, error) {
			return t.Namespace, t.Err
		},
		ClientConfig: func() (*client.Config, error) {
			return t.ClientConfig, t.Err
		},
	}, t, testapi.Codec()
}

func k8sClientFactory() *k8sClient {
	if len(*addr) > 0 && len(*user) > 0 && len(*pword) > 0 {
		config := client.Config{
			Host:     *addr,
			Username: *user,
			Password: *pword,
			Insecure: true,
		}
		return &k8sClient{client.NewOrDie(&config)}
	} else {
		kubernetesService := os.Getenv("KUBERNETES_SERVICE_HOST")
		if kubernetesService == "" {
			glog.Fatalf("Please specify the Kubernetes server with --server")
		}
		apiServer := fmt.Sprintf("https://%s:%s", kubernetesService, os.Getenv("KUBERNETES_SERVICE_PORT"))

		token, err := ioutil.ReadFile("/var/run/secrets/kubernetes.io/serviceaccount/token")
		if err != nil {
			glog.Fatalf("No service account token found")
		}

		config := client.Config{
			Host:        apiServer,
			BearerToken: string(token),
			Insecure:    true,
		}

		c, err := client.New(&config)
		if err != nil {
			glog.Fatalf("Failed to make client: %v", err)
		}
		return &k8sClient{c}
	}
}

func TestSyncReplicationControllerCreates(t *testing.T) {
	body := runtime.EncodeOrDie(testapi.Codec(), newPodList(0))
	fakePodHandler := util.FakeHandler{
		StatusCode:   200,
		ResponseBody: string(body),
	}
	fakePodControl := FakePodControl{}

	controller := newReplicationController(2)
	fakeUpdateHandler := util.FakeHandler{
		StatusCode:   200,
		ResponseBody: runtime.EncodeOrDie(testapi.Codec(), &controller),
		T:            t,
	}

	testServerMux := http.NewServeMux()
	testServerMux.Handle("/api/"+testapi.Version()+"/pods/", &fakePodHandler)
	testServerMux.Handle(fmt.Sprintf("/api/"+testapi.Version()+"/replicationControllers/%s", controller.Name), &fakeUpdateHandler)
	testServer := httptest.NewServer(testServerMux)
	defer testServer.Close()
	client := client.NewOrDie(&client.Config{Host: testServer.URL, Version: testapi.Version()})

	manager := NewReplicationManager(client)
	manager.podControl = &fakePodControl
	manager.syncReplicationController(controller)
	validateSyncReplication(t, &fakePodControl, 2, 0)

	// No Status.Replicas update expected even though 2 pods were just created,
	// because the controller manager can't observe the pods till the next sync cycle.
	if fakeUpdateHandler.RequestReceived != nil {
		t.Errorf("Unexpected updates for controller via %v",
			fakeUpdateHandler.RequestReceived.URL)
	}
}

func main() {
	flag.Parse()

	bearerToken, err := ioutil.ReadFile(*argBearerTokenFile)
	if err != nil {
		log.Fatal(err)
	}

	config := client.Config{
		Host:        *argMaster,
		Insecure:    *argInsecure,
		BearerToken: string(bearerToken),
		Version:     *argApiVersion,
	}

	if _, err := os.Stat(*argCaCertFile); err == nil {
		config.Insecure = false
		config.TLSClientConfig = client.TLSClientConfig{
			CAFile: *argCaCertFile,
		}
	}

	client := client.NewOrDie(&config)
	done := make(chan bool)
	go watchNodes(client)
	<-done
}

func TestControllerUpdateReplicas(t *testing.T) {
	// Insufficient number of pods in the system, and Status.Replicas is wrong;
	// Status.Replica should update to match number of pods in system, 1 new pod should be created.
	rc := newReplicationController(5)
	rc.Status = api.ReplicationControllerStatus{Replicas: 2}
	activePods := 4

	testServer, fakeUpdateHandler := makeTestServer(t, api.NamespaceDefault, rc.Name,
		serverResponse{http.StatusOK, newPodList(activePods)},
		serverResponse{http.StatusOK, &api.ReplicationControllerList{
			Items: []api.ReplicationController{rc},
		}},
		serverResponse{http.StatusOK, &rc})
	defer testServer.Close()
	client := client.NewOrDie(&client.Config{Host: testServer.URL, Version: testapi.Version()})
	manager := NewReplicationManager(client)
	fakePodControl := FakePodControl{}
	manager.podControl = &fakePodControl

	manager.synchronize()

	// Status.Replicas should go up from 2->4 even though we created 5-4=1 pod
	rc.Status = api.ReplicationControllerStatus{Replicas: 4}
	// These are set by default.
	rc.Spec.Selector = rc.Spec.Template.Labels
	rc.Labels = rc.Spec.Template.Labels

	decRc := runtime.EncodeOrDie(testapi.Codec(), &rc)
	fakeUpdateHandler.ValidateRequest(t, testapi.ResourcePathWithNamespaceQuery(replicationControllerResourceName(), rc.Namespace, rc.Name), "PUT", &decRc)
	validateSyncReplication(t, &fakePodControl, 1, 0)
}

func TestControllerNoReplicaUpdate(t *testing.T) {
	// Steady state for the replication controller, no Status.Replicas updates expected
	rc := newReplicationController(5)
	rc.Status = api.ReplicationControllerStatus{Replicas: 5}
	activePods := 5

	testServer, fakeUpdateHandler := makeTestServer(t, api.NamespaceDefault, rc.Name,
		serverResponse{http.StatusOK, newPodList(activePods)},
		serverResponse{http.StatusOK, &api.ReplicationControllerList{
			Items: []api.ReplicationController{rc},
		}},
		serverResponse{http.StatusOK, &rc})
	defer testServer.Close()
	client := client.NewOrDie(&client.Config{Host: testServer.URL, Version: testapi.Version()})
	manager := NewReplicationManager(client)
	fakePodControl := FakePodControl{}
	manager.podControl = &fakePodControl

	manager.synchronize()

	validateSyncReplication(t, &fakePodControl, 0, 0)
	if fakeUpdateHandler.RequestReceived != nil {
		t.Errorf("Unexpected updates for controller via %v",
			fakeUpdateHandler.RequestReceived.URL)
	}
}

func TestSynchronize(t *testing.T) {
	controllerSpec1 := newReplicationController(4)
	controllerSpec2 := newReplicationController(3)
	controllerSpec2.Name = "bar"
	controllerSpec2.Spec.Template.ObjectMeta.Labels = map[string]string{
		"name": "bar",
		"type": "production",
	}

	testServer, _ := makeTestServer(t, api.NamespaceDefault, "",
		serverResponse{http.StatusOK, newPodList(0)},
		serverResponse{http.StatusOK, &api.ReplicationControllerList{
			Items: []api.ReplicationController{
				controllerSpec1,
				controllerSpec2,
			}}},
		serverResponse{http.StatusInternalServerError, &api.ReplicationController{}})
	defer testServer.Close()
	client := client.NewOrDie(&client.Config{Host: testServer.URL, Version: testapi.Version()})
	manager := NewReplicationManager(client)
	fakePodControl := FakePodControl{}
	manager.podControl = &fakePodControl

	manager.synchronize()

	validateSyncReplication(t, &fakePodControl, 7, 0)
}

func TestControllerUpdateReplicas(t *testing.T) {
	// This is a happy server just to record the PUT request we expect for status.Replicas
	fakeHandler := util.FakeHandler{
		StatusCode:   200,
		ResponseBody: "",
	}
	testServer := httptest.NewServer(&fakeHandler)
	defer testServer.Close()

	client := client.NewOrDie(&client.Config{Host: testServer.URL, Version: testapi.Version()})
	manager := NewReplicationManager(client, BurstReplicas)

	// Insufficient number of pods in the system, and Status.Replicas is wrong;
	// Status.Replica should update to match number of pods in system, 1 new pod should be created.
	rc := newReplicationController(5)
	manager.controllerStore.Store.Add(rc)
	rc.Status = api.ReplicationControllerStatus{Replicas: 2}
	newPodList(manager.podStore.Store, 4, api.PodRunning, rc)
	response := runtime.EncodeOrDie(testapi.Codec(), rc)
	fakeHandler.ResponseBody = response

	fakePodControl := FakePodControl{}
	manager.podControl = &fakePodControl

	manager.syncReplicationController(getKey(rc, t))

	// Status.Replicas should go up from 2->4 even though we created 5-4=1 pod
	rc.Status = api.ReplicationControllerStatus{Replicas: 4}

	decRc := runtime.EncodeOrDie(testapi.Codec(), rc)
	fakeHandler.ValidateRequest(t, testapi.ResourcePath(replicationControllerResourceName(), rc.Namespace, rc.Name), "PUT", &decRc)
	validateSyncReplication(t, &fakePodControl, 1, 0)
}

func TestDeleteControllerAndExpectations(t *testing.T) {
	client := client.NewOrDie(&client.Config{Host: "", Version: testapi.Version()})
	manager := NewReplicationManager(client, 10)
	manager.podStoreSynced = alwaysReady

	rc := newReplicationController(1)
	manager.controllerStore.Store.Add(rc)

	fakePodControl := FakePodControl{}
	manager.podControl = &fakePodControl

	// This should set expectations for the rc
	manager.syncReplicationController(getKey(rc, t))
	validateSyncReplication(t, &fakePodControl, 1, 0)
	fakePodControl.clear()

	// This is to simulate a concurrent addPod, that has a handle on the expectations
	// as the controller deletes it.
	podExp, exists, err := manager.expectations.GetExpectations(rc)
	if !exists || err != nil {
		t.Errorf("No expectations found for rc")
	}
	manager.controllerStore.Delete(rc)
	manager.syncReplicationController(getKey(rc, t))

	if _, exists, err = manager.expectations.GetExpectations(rc); exists {
		t.Errorf("Found expectaions, expected none since the rc has been deleted.")
	}

	// This should have no effect, since we've deleted the rc.
	podExp.Seen(1, 0)
	manager.podStore.Store.Replace(make([]interface{}, 0))
	manager.syncReplicationController(getKey(rc, t))
	validateSyncReplication(t, &fakePodControl, 0, 0)
}

func TestSyncEndpointsItems(t *testing.T) {
	body, _ := json.Marshal(newPodList(1))
	fakeHandler := util.FakeHandler{
		StatusCode:   200,
		ResponseBody: string(body),
	}
	testServer := httptest.NewTLSServer(&fakeHandler)
	client := client.NewOrDie(testServer.URL, nil)
	serviceRegistry := registrytest.ServiceRegistry{
		List: api.ServiceList{
			Items: []api.Service{
				{
					Selector: map[string]string{
						"foo": "bar",
					},
				},
			},
		},
	}
	endpoints := NewEndpointController(&serviceRegistry, client)
	if err := endpoints.SyncServiceEndpoints(); err != nil {
		t.Errorf("unexpected error: %v", err)
	}
	if len(serviceRegistry.Endpoints.Endpoints) != 1 {
		t.Errorf("Unexpected endpoints update: %#v", serviceRegistry.Endpoints)
	}
}

func TestSyncEndpointsProtocolUDP(t *testing.T) {
	serviceList := api.ServiceList{
		Items: []api.Service{
			{
				ObjectMeta: api.ObjectMeta{Name: "foo", Namespace: "other"},
				Spec: api.ServiceSpec{
					Selector: map[string]string{},
					Protocol: api.ProtocolUDP,
				},
			},
		},
	}
	testServer, endpointsHandler := makeTestServer(t,
		serverResponse{http.StatusOK, newPodList(0)},
		serverResponse{http.StatusOK, &serviceList},
		serverResponse{http.StatusOK, &api.Endpoints{
			ObjectMeta: api.ObjectMeta{
				Name:            "foo",
				ResourceVersion: "1",
			},
			Protocol:  api.ProtocolUDP,
			Endpoints: []api.Endpoint{{IP: "6.7.8.9", Port: 1000}},
		}})
	defer testServer.Close()
	client := client.NewOrDie(&client.Config{Host: testServer.URL, Version: testapi.Version()})
	endpoints := NewEndpointController(client)
	if err := endpoints.SyncServiceEndpoints(); err != nil {
		t.Errorf("unexpected error: %v", err)
	}
	endpointsHandler.ValidateRequestCount(t, 0)
}

func newKubeSource(pollDuration time.Duration) (*kubeSource, error) {
	if len(*argMaster) == 0 {
		return nil, fmt.Errorf("kubernetes_master flag not specified")
	}

	if !(strings.HasPrefix(*argMaster, "http://") || strings.HasPrefix(*argMaster, "https://")) {
		*argMaster = "http://" + *argMaster
	}

	kubeClient := kube_client.NewOrDie(&kube_client.Config{
		Host:     *argMaster,
		Version:  kubeClientVersion,
		Insecure: *argMasterInsecure,
	})

	nodesApi, err := nodes.NewKubeNodes(kubeClient)
	if err != nil {
		return nil, err
	}
	glog.Infof("Using Kubernetes client with master %q and version %s\n", *argMaster, kubeClientVersion)
	glog.Infof("Using kubelet port %q", *argKubeletPort)

	return &kubeSource{
		lastQuery:    time.Now(),
		pollDuration: pollDuration,
		kubeletPort:  *argKubeletPort,
		kubeletApi:   datasource.NewKubelet(),
		nodesApi:     nodesApi,
		podsApi:      newPodsApi(kubeClient),
		podErrors:    make(map[podInstance]int),
	}, nil
}

func newApiClient(addr string, port int) *client.Client {
	apiServerURL := fmt.Sprintf("http://%s:%d", addr, port)
	cl := client.NewOrDie(&client.Config{Host: apiServerURL, Version: testapi.Version()})
	cl.PollPeriod = time.Second * 1
	cl.Sync = true
	return cl
}

func TestBind(t *testing.T) {
	table := []struct {
		binding *api.Binding
	}{
		{binding: &api.Binding{
			ObjectMeta: api.ObjectMeta{
				Namespace: api.NamespaceDefault,
				Name:      "foo",
			},
			Target: api.ObjectReference{
				Name: "foohost.kubernetes.mydomain.com",
			},
		}},
	}

	for _, item := range table {
		handler := util.FakeHandler{
			StatusCode:   200,
			ResponseBody: "",
			T:            t,
		}
		server := httptest.NewServer(&handler)
		defer server.Close()
		client := client.NewOrDie(&client.Config{Host: server.URL, Version: testapi.Version()})
		b := binder{client}

		if err := b.Bind(item.binding); err != nil {
			t.Errorf("Unexpected error: %v", err)
			continue
		}
		expectedBody := runtime.EncodeOrDie(testapi.Codec(), item.binding)
		handler.ValidateRequest(t, testapi.ResourcePath("bindings", api.NamespaceDefault, ""), "POST", &expectedBody)
	}
}

func TestSyncEndpointsItems(t *testing.T) {
	serviceList := api.ServiceList{
		Items: []api.Service{
			{
				ObjectMeta: api.ObjectMeta{Name: "foo"},
				Spec: api.ServiceSpec{
					Selector: map[string]string{
						"foo": "bar",
					},
				},
			},
		},
	}
	testServer, endpointsHandler := makeTestServer(t,
		serverResponse{http.StatusOK, newPodList(1)},
		serverResponse{http.StatusOK, serviceList},
		serverResponse{http.StatusOK, api.Endpoints{}})
	defer testServer.Close()
	client := client.NewOrDie(&client.Config{Host: testServer.URL, Version: testapi.Version()})
	endpoints := NewEndpointController(client)
	if err := endpoints.SyncServiceEndpoints(); err != nil {
		t.Errorf("unexpected error: %v", err)
	}
	data := runtime.EncodeOrDie(testapi.Codec(), &api.Endpoints{
		ObjectMeta: api.ObjectMeta{
			ResourceVersion: "",
		},
		Endpoints: []string{"1.2.3.4:8080"},
	})
	endpointsHandler.ValidateRequest(t, "/api/"+testapi.Version()+"/endpoints", "POST", &data)
}