// CreateOrUpdate attempts to update the current etcd state with the provided
// allocation.
func (e *Etcd) CreateOrUpdate(snapshot *api.RangeAllocation) error {
	e.lock.Lock()
	defer e.lock.Unlock()

	last := ""
	err := e.helper.GuaranteedUpdate(e.baseKey, &api.RangeAllocation{}, true,
		tools.SimpleUpdate(func(input runtime.Object) (output runtime.Object, err error) {
			existing := input.(*api.RangeAllocation)
			switch {
			case len(snapshot.ResourceVersion) != 0 && len(existing.ResourceVersion) != 0:
				if snapshot.ResourceVersion != existing.ResourceVersion {
					return nil, k8serr.NewConflict(e.kind, "", fmt.Errorf("the provided resource version does not match"))
				}
			case len(existing.ResourceVersion) != 0:
				return nil, k8serr.NewConflict(e.kind, "", fmt.Errorf("another caller has already initialized the resource"))
			}
			last = snapshot.ResourceVersion
			return snapshot, nil
		}),
	)
	if err != nil {
		return etcderr.InterpretUpdateError(err, e.kind, "")
	}
	err = e.alloc.Restore(snapshot.Range, snapshot.Data)
	if err == nil {
		e.last = last
	}
	return err
}

func TestErrorsToAPIStatus(t *testing.T) {
	cases := map[error]api.Status{
		errors.NewAlreadyExists("foo", "bar"): {
			Status:  api.StatusFailure,
			Code:    http.StatusConflict,
			Reason:  "AlreadyExists",
			Message: "foo \"bar\" already exists",
			Details: &api.StatusDetails{
				Kind: "foo",
				ID:   "bar",
			},
		},
		errors.NewConflict("foo", "bar", stderrs.New("failure")): {
			Status:  api.StatusFailure,
			Code:    http.StatusConflict,
			Reason:  "Conflict",
			Message: "foo \"bar\" cannot be updated: failure",
			Details: &api.StatusDetails{
				Kind: "foo",
				ID:   "bar",
			},
		},
	}
	for k, v := range cases {
		actual := errToAPIStatus(k)
		if !reflect.DeepEqual(actual, &v) {
			t.Errorf("%s: Expected %#v, Got %#v", k, v, actual)
		}
	}
}

func TestControllerError(t *testing.T) {
	testCases := map[string]struct {
		err     func() error
		errFn   func(err error) bool
		reactFn testclient.ReactionFunc
		actions int
	}{
		"not found": {
			err:     func() error { return errors.NewNotFound("namespace", "test") },
			errFn:   func(err error) bool { return err == nil },
			actions: 1,
		},
		"unknown": {
			err:     func() error { return fmt.Errorf("unknown") },
			errFn:   func(err error) bool { return err.Error() == "unknown" },
			actions: 1,
		},
		"conflict": {
			actions: 4,
			reactFn: func(a testclient.FakeAction) (runtime.Object, error) {
				if a.Action == "get-namespace" {
					return &kapi.Namespace{ObjectMeta: kapi.ObjectMeta{Name: "test"}}, nil
				}
				return (*kapi.Namespace)(nil), errors.NewConflict("namespace", "test", fmt.Errorf("test conflict"))
			},
			errFn: func(err error) bool {
				return err != nil && strings.Contains(err.Error(), "unable to allocate security info")
			},
		},
	}

	for s, testCase := range testCases {
		client := &testclient.Fake{ReactFn: testCase.reactFn}
		if client.ReactFn == nil {
			client.ReactFn = func(a testclient.FakeAction) (runtime.Object, error) {
				return (*kapi.Namespace)(nil), testCase.err()
			}
		}
		uidr, _ := uid.NewRange(10, 19, 2)
		mcsr, _ := mcs.NewRange("s0:", 10, 2)
		uida := uidallocator.NewInMemory(uidr)
		c := Allocation{
			uid:    uida,
			mcs:    DefaultMCSAllocation(uidr, mcsr, 5),
			client: client.Namespaces(),
		}

		err := c.Next(&kapi.Namespace{ObjectMeta: kapi.ObjectMeta{Name: "test"}})
		if !testCase.errFn(err) {
			t.Errorf("%s: unexpected error: %v", s, err)
		}

		if len(client.Actions) != testCase.actions {
			t.Errorf("%s: expected %d actions: %v", s, testCase.actions, client.Actions)
		}
		if uida.Free() != 5 {
			t.Errorf("%s: should not have allocated uid: %d/%d", s, uida.Free(), uidr.Size())
		}
	}
}

// Update replaces a given Route instance with an existing instance in rs.registry.
func (rs *REST) Update(ctx kapi.Context, obj runtime.Object) (runtime.Object, bool, error) {
	route, ok := obj.(*api.Route)
	if !ok {
		return nil, false, errors.NewBadRequest(fmt.Sprintf("not a route: %#v", obj))
	}
	if !kapi.ValidNamespace(ctx, &route.ObjectMeta) {
		return nil, false, errors.NewConflict("route", route.Namespace, fmt.Errorf("Route.Namespace does not match the provided context"))
	}

	old, err := rs.Get(ctx, route.Name)
	if err != nil {
		return nil, false, err
	}
	if errs := validation.ValidateRouteUpdate(route, old.(*api.Route)); len(errs) > 0 {
		return nil, false, errors.NewInvalid("route", route.Name, errs)
	}

	// TODO: Convert to generic etcd
	// TODO: Call ValidateRouteUpdate->ValidateObjectMetaUpdate
	// TODO: In the UpdateStrategy.PrepareForUpdate, set the HostGeneratedAnnotationKey annotation to "false" if the updated route object modifies the host

	err = rs.registry.UpdateRoute(ctx, route)
	if err != nil {
		return nil, false, err
	}
	out, err := rs.registry.GetRoute(ctx, route.Name)
	return out, false, err
}

// createDockerPullSecretReference updates a service account to reference the dockercfgSecret as a Secret and an ImagePullSecret
func (e *DockercfgController) createDockerPullSecretReference(staleServiceAccount *api.ServiceAccount, dockercfgSecretName string) error {
	liveServiceAccount, err := e.client.ServiceAccounts(staleServiceAccount.Namespace).Get(staleServiceAccount.Name)
	if err != nil {
		return err
	}

	mountableDockercfgSecrets, imageDockercfgPullSecrets := getGeneratedDockercfgSecretNames(liveServiceAccount)
	staleDockercfgMountableSecrets, staleImageDockercfgPullSecrets := getGeneratedDockercfgSecretNames(staleServiceAccount)

	// if we're trying to create a reference based on stale lists of dockercfg secrets, let the caller know
	if !reflect.DeepEqual(staleDockercfgMountableSecrets.List(), mountableDockercfgSecrets.List()) || !reflect.DeepEqual(staleImageDockercfgPullSecrets.List(), imageDockercfgPullSecrets.List()) {
		return kapierrors.NewConflict("ServiceAccount", staleServiceAccount.Name, fmt.Errorf("cannot add reference to %s based on stale data.  decision made for %v,%v, but live version is %v,%v", dockercfgSecretName, staleDockercfgMountableSecrets.List(), staleImageDockercfgPullSecrets.List(), mountableDockercfgSecrets.List(), imageDockercfgPullSecrets.List()))
	}

	changed := false
	if !mountableDockercfgSecrets.Has(dockercfgSecretName) {
		liveServiceAccount.Secrets = append(liveServiceAccount.Secrets, api.ObjectReference{Name: dockercfgSecretName})
		changed = true
	}

	if !imageDockercfgPullSecrets.Has(dockercfgSecretName) {
		liveServiceAccount.ImagePullSecrets = append(liveServiceAccount.ImagePullSecrets, api.LocalObjectReference{Name: dockercfgSecretName})
		changed = true
	}

	if changed {
		if _, err = e.client.ServiceAccounts(liveServiceAccount.Namespace).Update(liveServiceAccount); err != nil {
			// TODO: retry on API conflicts in case the conflict was unrelated to our generated dockercfg secrets?
			return err
		}
	}
	return nil
}

func (rs *REST) Update(ctx api.Context, obj runtime.Object) (<-chan apiserver.RESTResult, error) {
	service := obj.(*api.Service)
	if !api.ValidNamespace(ctx, &service.ObjectMeta) {
		return nil, errors.NewConflict("service", service.Namespace, fmt.Errorf("Service.Namespace does not match the provided context"))
	}
	if errs := validation.ValidateService(service, rs.registry, ctx); len(errs) > 0 {
		return nil, errors.NewInvalid("service", service.Name, errs)
	}
	return apiserver.MakeAsync(func() (runtime.Object, error) {
		cur, err := rs.registry.GetService(ctx, service.Name)
		if err != nil {
			return nil, err
		}
		if service.Spec.PortalIP != cur.Spec.PortalIP {
			// TODO: Would be nice to pass "field is immutable" to users.
			el := errors.ValidationErrorList{errors.NewFieldInvalid("spec.portalIP", service.Spec.PortalIP)}
			return nil, errors.NewInvalid("service", service.Name, el)
		}
		// Copy over non-user fields.
		service.Spec.ProxyPort = cur.Spec.ProxyPort
		// TODO: check to see if external load balancer status changed
		err = rs.registry.UpdateService(ctx, service)
		if err != nil {
			return nil, err
		}
		return rs.registry.GetService(ctx, service.Name)
	}), nil
}

// Update updates a Secret object.
func (rs *REST) Update(ctx api.Context, obj runtime.Object) (runtime.Object, bool, error) {
	secret, ok := obj.(*api.Secret)
	if !ok {
		return nil, false, fmt.Errorf("not a secret: %#v", obj)
	}

	if !api.ValidNamespace(ctx, &secret.ObjectMeta) {
		return nil, false, errors.NewConflict("secret", secret.Namespace, fmt.Errorf("Secret.Namespace does not match the provided context"))
	}

	oldObj, err := rs.registry.Get(ctx, secret.Name)
	if err != nil {
		return nil, false, err
	}

	editSecret := oldObj.(*api.Secret)

	// set the editable fields on the existing object
	editSecret.Labels = secret.Labels
	editSecret.ResourceVersion = secret.ResourceVersion
	editSecret.Annotations = secret.Annotations
	editSecret.Data = secret.Data
	editSecret.Type = secret.Type

	if errs := validation.ValidateSecret(editSecret); len(errs) > 0 {
		return nil, false, errors.NewInvalid("secret", editSecret.Name, errs)
	}

	err = rs.registry.UpdateWithName(ctx, editSecret.Name, editSecret)
	if err != nil {
		return nil, false, err
	}
	out, err := rs.registry.Get(ctx, editSecret.Name)
	return out, false, err
}

// Create a Secret object
func (rs *REST) Create(ctx api.Context, obj runtime.Object) (runtime.Object, error) {
	secret, ok := obj.(*api.Secret)
	if !ok {
		return nil, fmt.Errorf("invalid object type")
	}

	if !api.ValidNamespace(ctx, &secret.ObjectMeta) {
		return nil, errors.NewConflict("secret", secret.Namespace, fmt.Errorf("Secret.Namespace does not match the provided context"))
	}

	if len(secret.Name) == 0 {
		secret.Name = string(util.NewUUID())
	}

	if errs := validation.ValidateSecret(secret); len(errs) > 0 {
		return nil, errors.NewInvalid("secret", secret.Name, errs)
	}
	api.FillObjectMetaSystemFields(ctx, &secret.ObjectMeta)

	err := rs.registry.CreateWithName(ctx, secret.Name, secret)
	if err != nil {
		return nil, err
	}
	return rs.registry.Get(ctx, secret.Name)
}

// Create registers the given ReplicationController.
func (rs *REST) Create(ctx api.Context, obj runtime.Object) (<-chan apiserver.RESTResult, error) {
	controller, ok := obj.(*api.ReplicationController)
	if !ok {
		return nil, fmt.Errorf("not a replication controller: %#v", obj)
	}
	if !api.ValidNamespace(ctx, &controller.ObjectMeta) {
		return nil, errors.NewConflict("controller", controller.Namespace, fmt.Errorf("Controller.Namespace does not match the provided context"))
	}

	if len(controller.Name) == 0 {
		controller.Name = util.NewUUID().String()
	}
	// Pod Manifest ID should be assigned by the pod API
	controller.DesiredState.PodTemplate.DesiredState.Manifest.ID = ""
	if errs := validation.ValidateReplicationController(controller); len(errs) > 0 {
		return nil, errors.NewInvalid("replicationController", controller.Name, errs)
	}

	controller.CreationTimestamp = util.Now()

	return apiserver.MakeAsync(func() (runtime.Object, error) {
		err := rs.registry.CreateController(ctx, controller)
		if err != nil {
			return nil, err
		}
		return rs.registry.GetController(ctx, controller.Name)
	}), nil
}

// Update updates a LimitRange object.
func (rs *REST) Update(ctx api.Context, obj runtime.Object) (runtime.Object, bool, error) {
	limitRange, ok := obj.(*api.LimitRange)
	if !ok {
		return nil, false, fmt.Errorf("invalid object type")
	}

	if !api.ValidNamespace(ctx, &limitRange.ObjectMeta) {
		return nil, false, errors.NewConflict("limitRange", limitRange.Namespace, fmt.Errorf("LimitRange.Namespace does not match the provided context"))
	}

	oldObj, err := rs.registry.Get(ctx, limitRange.Name)
	if err != nil {
		return nil, false, err
	}

	editLimitRange := oldObj.(*api.LimitRange)

	// set the editable fields on the existing object
	editLimitRange.Labels = limitRange.Labels
	editLimitRange.ResourceVersion = limitRange.ResourceVersion
	editLimitRange.Annotations = limitRange.Annotations
	editLimitRange.Spec = limitRange.Spec

	if errs := validation.ValidateLimitRange(editLimitRange); len(errs) > 0 {
		return nil, false, errors.NewInvalid("limitRange", editLimitRange.Name, errs)
	}

	if err := rs.registry.UpdateWithName(ctx, editLimitRange.Name, editLimitRange); err != nil {
		return nil, false, err
	}
	out, err := rs.registry.Get(ctx, editLimitRange.Name)
	return out, false, err
}

// Create registers the given ReplicationController.
func (rs *REST) Create(ctx api.Context, obj runtime.Object) (<-chan apiserver.RESTResult, error) {
	controller, ok := obj.(*api.ReplicationController)
	if !ok {
		return nil, fmt.Errorf("not a replication controller: %#v", obj)
	}
	if !api.ValidNamespace(ctx, &controller.ObjectMeta) {
		return nil, errors.NewConflict("controller", controller.Namespace, fmt.Errorf("Controller.Namespace does not match the provided context"))
	}

	if len(controller.Name) == 0 {
		controller.Name = util.NewUUID().String()
	}
	if errs := validation.ValidateReplicationController(controller); len(errs) > 0 {
		return nil, errors.NewInvalid("replicationController", controller.Name, errs)
	}

	api.FillObjectMetaSystemFields(ctx, &controller.ObjectMeta)

	return apiserver.MakeAsync(func() (runtime.Object, error) {
		err := rs.registry.CreateController(ctx, controller)
		if err != nil {
			return nil, err
		}
		return rs.registry.GetController(ctx, controller.Name)
	}), nil
}

// InterpretUpdateError converts a generic etcd error on a update
// operation into the appropriate API error.
func InterpretUpdateError(err error, kind, name string) error {
	switch {
	case tools.IsEtcdTestFailed(err), tools.IsEtcdNodeExist(err):
		return errors.NewConflict(kind, name, err)
	default:
		return err
	}
}

// assignPod assigns the given pod to the given machine.
func (r *BindingREST) assignPod(ctx api.Context, podID string, machine string, annotations map[string]string) (err error) {
	if _, err = r.setPodHostAndAnnotations(ctx, podID, "", machine, annotations); err != nil {
		err = etcderr.InterpretGetError(err, "pod", podID)
		err = etcderr.InterpretUpdateError(err, "pod", podID)
		if _, ok := err.(*errors.StatusError); !ok {
			err = errors.NewConflict("binding", podID, err)
		}
	}
	return
}

// createBuild is responsible for validating build object and saving it and returning newly created object
func (g *BuildGenerator) createBuild(ctx kapi.Context, build *buildapi.Build) (*buildapi.Build, error) {
	if !kapi.ValidNamespace(ctx, &build.ObjectMeta) {
		return nil, errors.NewConflict("build", build.Namespace, fmt.Errorf("Build.Namespace does not match the provided context"))
	}
	kapi.FillObjectMetaSystemFields(ctx, &build.ObjectMeta)

	err := g.Client.CreateBuild(ctx, build)
	if err != nil {
		return nil, err
	}
	return g.Client.GetBuild(ctx, build.Name)
}

// transformResponse converts an API response into a structured API object.
func (r *Request) transformResponse(resp *http.Response, req *http.Request) ([]byte, bool, error) {
	defer resp.Body.Close()

	body, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return nil, false, err
	}

	// Did the server give us a status response?
	isStatusResponse := false
	var status api.Status
	if err := r.codec.DecodeInto(body, &status); err == nil && status.Status != "" {
		isStatusResponse = true
	}

	switch {
	case resp.StatusCode == http.StatusSwitchingProtocols:
		// no-op, we've been upgraded
	case resp.StatusCode < http.StatusOK || resp.StatusCode > http.StatusPartialContent:
		if !isStatusResponse {
			var err error = &UnexpectedStatusError{
				Request:  req,
				Response: resp,
				Body:     string(body),
			}
			// TODO: handle other error classes we know about
			switch resp.StatusCode {
			case http.StatusConflict:
				if req.Method == "POST" {
					err = errors.NewAlreadyExists(r.resource, r.resourceName)
				} else {
					err = errors.NewConflict(r.resource, r.resourceName, err)
				}
			case http.StatusNotFound:
				err = errors.NewNotFound(r.resource, r.resourceName)
			case http.StatusBadRequest:
				err = errors.NewBadRequest(err.Error())
			}
			return nil, false, err
		}
		return nil, false, errors.FromObject(&status)
	}

	// If the server gave us a status back, look at what it was.
	if isStatusResponse && status.Status != api.StatusSuccess {
		// "Working" requests need to be handled specially.
		// "Failed" requests are clearly just an error and it makes sense to return them as such.
		return nil, false, errors.FromObject(&status)
	}

	created := resp.StatusCode == http.StatusCreated
	return body, created, err
}

// Get retrieves an image by ID that has previously been tagged into an image stream.
// `id` is of the form <repo name>@<image id>.
func (r *REST) Get(ctx kapi.Context, id string) (runtime.Object, error) {
	name, imageID, err := ParseNameAndID(id)
	if err != nil {
		return nil, err
	}

	repo, err := r.imageStreamRegistry.GetImageStream(ctx, name)
	if err != nil {
		return nil, err
	}

	if repo.Status.Tags == nil {
		return nil, errors.NewNotFound("imageStreamImage", imageID)
	}

	set := api.ResolveImageID(repo, imageID)
	switch len(set) {
	case 1:
		imageName := set.List()[0]
		image, err := r.imageRegistry.GetImage(ctx, imageName)
		if err != nil {
			return nil, err
		}
		imageWithMetadata, err := api.ImageWithMetadata(*image)
		if err != nil {
			return nil, err
		}

		if d, err := digest.ParseDigest(imageName); err == nil {
			imageName = d.Hex()
		}
		if len(imageName) > 7 {
			imageName = imageName[:7]
		}

		isi := api.ImageStreamImage{
			ObjectMeta: kapi.ObjectMeta{
				Namespace: kapi.NamespaceValue(ctx),
				Name:      fmt.Sprintf("%[email protected]%s", name, imageName),
			},
			Image: *imageWithMetadata,
		}

		return &isi, nil
	case 0:
		return nil, errors.NewNotFound("imageStreamImage", imageID)
	default:
		return nil, errors.NewConflict("imageStreamImage", imageID, fmt.Errorf("multiple images match the prefix %q: %s", imageID, strings.Join(set.List(), ", ")))
	}
}

func (rs *REST) Update(ctx api.Context, obj runtime.Object) (<-chan apiserver.RESTResult, error) {
	pod := obj.(*api.Pod)
	if !api.ValidNamespace(ctx, &pod.ObjectMeta) {
		return nil, errors.NewConflict("pod", pod.Namespace, fmt.Errorf("Pod.Namespace does not match the provided context"))
	}
	if errs := validation.ValidatePod(pod); len(errs) > 0 {
		return nil, errors.NewInvalid("pod", pod.Name, errs)
	}
	return apiserver.MakeAsync(func() (runtime.Object, error) {
		if err := rs.registry.UpdatePod(ctx, pod); err != nil {
			return nil, err
		}
		return rs.registry.GetPod(ctx, pod.Name)
	}), nil
}

// ValidateService tests if required fields in the service are set.
func ValidateService(service *api.Service, lister ServiceLister, ctx api.Context) errs.ValidationErrorList {
	allErrs := errs.ValidationErrorList{}
	if len(service.Name) == 0 {
		allErrs = append(allErrs, errs.NewFieldRequired("name", service.Name))
	} else if !util.IsDNS952Label(service.Name) {
		allErrs = append(allErrs, errs.NewFieldInvalid("name", service.Name, ""))
	}
	if !util.IsDNSSubdomain(service.Namespace) {
		allErrs = append(allErrs, errs.NewFieldInvalid("namespace", service.Namespace, ""))
	}
	if !util.IsValidPortNum(service.Spec.Port) {
		allErrs = append(allErrs, errs.NewFieldInvalid("spec.port", service.Spec.Port, ""))
	}
	if len(service.Spec.Protocol) == 0 {
		service.Spec.Protocol = "TCP"
	} else if !supportedPortProtocols.Has(strings.ToUpper(string(service.Spec.Protocol))) {
		allErrs = append(allErrs, errs.NewFieldNotSupported("spec.protocol", service.Spec.Protocol))
	}

	if service.Spec.Selector != nil {
		allErrs = append(allErrs, validateLabels(service.Spec.Selector, "spec.selector")...)
	}
	allErrs = append(allErrs, validateLabels(service.Labels, "labels")...)

	if service.Spec.CreateExternalLoadBalancer {
		services, err := lister.ListServices(ctx)
		if err != nil {
			allErrs = append(allErrs, errs.NewInternalError(err))
		} else {
			for i := range services.Items {
				if services.Items[i].Name != service.Name &&
					services.Items[i].Spec.CreateExternalLoadBalancer &&
					services.Items[i].Spec.Port == service.Spec.Port {
					allErrs = append(allErrs, errs.NewConflict("service", service.Name, fmt.Errorf("port: %d is already in use", service.Spec.Port)))
					break
				}
			}
		}
	}
	if service.Spec.SessionAffinity == "" {
		service.Spec.SessionAffinity = api.AffinityTypeNone
	} else if !supportedSessionAffinityType.Has(string(service.Spec.SessionAffinity)) {
		allErrs = append(allErrs, errs.NewFieldNotSupported("spec.sessionAffinity", service.Spec.SessionAffinity))
	}

	return allErrs
}

func TestBuildConfigUpdateError(t *testing.T) {
	// valid configuration, but build creation fails, in that situation the buildconfig should not be updated
	buildcfg := mockBuildConfig("registry.com/namespace/imagename", "registry.com/namespace/imagename", "testImageStream", "testTag")
	imageStream := mockImageStream("testImageStream", "registry.com/namespace/imagename", map[string]string{"testTag": "newImageID123"})
	image := mockImage("[email protected]", "registry.com/namespace/[email protected]")
	controller := mockImageChangeController(buildcfg, imageStream, image)
	bcInstantiator := controller.BuildConfigInstantiator.(*buildConfigInstantiator)
	bcUpdater := bcInstantiator.buildConfigUpdater
	bcUpdater.err = kerrors.NewConflict("BuildConfig", buildcfg.Name, errors.New("foo"))

	err := controller.HandleImageRepo(imageStream)
	if len(bcInstantiator.name) == 0 {
		t.Error("Expected build generation when new image was created!")
	}
	if err == nil || !strings.Contains(err.Error(), "will be retried") {
		t.Fatalf("Expected 'will be retried' from HandleImageRepo, got %s", err.Error())
	}
}

func (rs *REST) Update(ctx api.Context, obj runtime.Object) (runtime.Object, bool, error) {
	service := obj.(*api.Service)
	if !api.ValidNamespace(ctx, &service.ObjectMeta) {
		return nil, false, errors.NewConflict("service", service.Namespace, fmt.Errorf("Service.Namespace does not match the provided context"))
	}

	oldService, err := rs.registry.GetService(ctx, service.Name)
	if err != nil {
		return nil, false, err
	}

	// Copy over non-user fields
	// TODO: make this a merge function
	if errs := validation.ValidateServiceUpdate(oldService, service); len(errs) > 0 {
		return nil, false, errors.NewInvalid("service", service.Name, errs)
	}
	out, err := rs.registry.UpdateService(ctx, service)
	return out, false, err
}