func TestGetEndpointsMissingService(t *testing.T) {
	registry := &registrytest.ServiceRegistry{
		Err: errors.NewNotFound("service", "foo"),
	}
	storage := NewREST(registry)
	ctx := api.NewContext()
	// returns service not found
	_, err := storage.Get(ctx, "foo")
	if !errors.IsNotFound(err) || !reflect.DeepEqual(err, errors.NewNotFound("service", "foo")) {
		t.Errorf("expected NotFound error, got %#v", err)
	}

	// returns empty endpoints
	registry.Err = nil
	registry.Service = &api.Service{
		ObjectMeta: api.ObjectMeta{Name: "foo"},
	}
	obj, err := storage.Get(ctx, "foo")
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if obj.(*api.Endpoints).Endpoints != nil {
		t.Errorf("unexpected endpoints: %#v", obj)
	}
}

// Delete removes a tag from a stream. `id` is of the format <stream name>:<tag>.
// The associated image that the tag points to is *not* deleted.
// The tag history remains intact and is not deleted.
func (r *REST) Delete(ctx kapi.Context, id string) (runtime.Object, error) {
	name, tag, err := nameAndTag(id)
	if err != nil {
		return nil, err
	}

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

	if stream.Spec.Tags == nil {
		return nil, errors.NewNotFound("imageStreamTag", tag)
	}

	_, ok := stream.Spec.Tags[tag]
	if !ok {
		return nil, errors.NewNotFound("imageStreamTag", tag)
	}

	delete(stream.Spec.Tags, tag)

	_, err = r.imageStreamRegistry.UpdateImageStream(ctx, stream)
	if err != nil {
		return nil, fmt.Errorf("Error removing tag from image stream: %s", err)
	}

	return &kapi.Status{Status: kapi.StatusSuccess}, nil
}

func (o objects) Kind(kind, name string) (runtime.Object, error) {
	empty, _ := o.creater.New("", kind)
	nilValue := reflect.Zero(reflect.TypeOf(empty)).Interface().(runtime.Object)

	arr, ok := o.types[kind]
	if !ok {
		if strings.HasSuffix(kind, "List") {
			itemKind := kind[:len(kind)-4]
			arr, ok := o.types[itemKind]
			if !ok {
				return empty, nil
			}
			out, err := o.creater.New("", kind)
			if err != nil {
				return nilValue, err
			}
			if err := runtime.SetList(out, arr); err != nil {
				return nilValue, err
			}
			if out, err = o.copier.Copy(out); err != nil {
				return nilValue, err
			}
			return out, nil
		}
		return nilValue, errors.NewNotFound(kind, name)
	}

	index := o.last[kind]
	if index >= len(arr) {
		index = len(arr) - 1
	}
	if index < 0 {
		return nilValue, errors.NewNotFound(kind, name)
	}
	out, err := o.copier.Copy(arr[index])
	if err != nil {
		return nilValue, err
	}
	o.last[kind] = index + 1

	if status, ok := out.(*api.Status); ok {
		if status.Details != nil {
			status.Details.Kind = kind
		}
		if status.Status != api.StatusSuccess {
			return nilValue, &errors.StatusError{*status}
		}
	}

	return out, nil
}

// 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 (m *VirtualStorage) Get(ctx kapi.Context, name string) (runtime.Object, error) {
	policyBinding, err := m.getPolicyBindingOwningRoleBinding(ctx, name)
	if err != nil && kapierrors.IsNotFound(err) {
		return nil, kapierrors.NewNotFound("RoleBinding", name)
	}
	if err != nil {
		return nil, err
	}

	binding, exists := policyBinding.RoleBindings[name]
	if !exists {
		return nil, kapierrors.NewNotFound("RoleBinding", name)
	}
	return binding, nil
}

func (m *VirtualStorage) Get(ctx kapi.Context, name string) (runtime.Object, error) {
	policy, err := m.PolicyStorage.GetPolicy(ctx, authorizationapi.PolicyName)
	if err != nil && kapierrors.IsNotFound(err) {
		return nil, kapierrors.NewNotFound("Role", name)
	}
	if err != nil {
		return nil, err
	}

	role, exists := policy.Roles[name]
	if !exists {
		return nil, kapierrors.NewNotFound("Role", name)
	}

	return role, nil
}

func TestUpdateMissing(t *testing.T) {
	storage := map[string]RESTStorage{}
	ID := "id"
	simpleStorage := SimpleRESTStorage{
		errors: map[string]error{"update": apierrs.NewNotFound("simple", ID)},
	}
	storage["simple"] = &simpleStorage
	handler := Handle(storage, codec, "/prefix/version", selfLinker)
	server := httptest.NewServer(handler)
	defer server.Close()

	item := &Simple{
		Other: "bar",
	}
	body, err := codec.Encode(item)
	if err != nil {
		t.Errorf("unexpected error: %v", err)
	}

	client := http.Client{}
	request, err := http.NewRequest("PUT", server.URL+"/prefix/version/simple/"+ID, bytes.NewReader(body))
	response, err := client.Do(request)
	if err != nil {
		t.Errorf("unexpected error: %v", err)
	}

	if response.StatusCode != http.StatusNotFound {
		t.Errorf("Unexpected response %#v", response)
	}
}

func TestCreateMissingDeploymentConfig(t *testing.T) {
	rest := REST{
		generator: Client{
			GRFn: func(from, to *deployapi.DeploymentConfig, spec *deployapi.DeploymentConfigRollbackSpec) (*deployapi.DeploymentConfig, error) {
				t.Fatal("unexpected call to generator")
				return nil, errors.New("something terrible happened")
			},
			RCFn: func(ctx kapi.Context, name string) (*kapi.ReplicationController, error) {
				deployment, _ := deployutil.MakeDeployment(deploytest.OkDeploymentConfig(1), kapi.Codec)
				return deployment, nil
			},
			DCFn: func(ctx kapi.Context, name string) (*deployapi.DeploymentConfig, error) {
				return nil, kerrors.NewNotFound("deploymentConfig", name)
			},
		},
		codec: api.Codec,
	}

	obj, err := rest.Create(kapi.NewDefaultContext(), &deployapi.DeploymentConfigRollback{
		Spec: deployapi.DeploymentConfigRollbackSpec{
			From: kapi.ObjectReference{
				Name:      "deployment",
				Namespace: kapi.NamespaceDefault,
			},
		},
	})

	if err == nil {
		t.Errorf("Expected an error")
	}

	if obj != nil {
		t.Error("Unexpected result obj")
	}
}

func (m *VirtualStorage) getReferencedRole(roleRef kapi.ObjectReference) (*authorizationapi.Role, error) {
	ctx := kapi.WithNamespace(kapi.NewContext(), roleRef.Namespace)

	var policy *authorizationapi.Policy
	var err error
	switch {
	case len(roleRef.Namespace) == 0:
		var clusterPolicy *authorizationapi.ClusterPolicy
		clusterPolicy, err = m.ClusterPolicyRegistry.GetClusterPolicy(ctx, authorizationapi.PolicyName)
		policy = authorizationapi.ToPolicy(clusterPolicy)
	default:
		policy, err = m.PolicyRegistry.GetPolicy(ctx, authorizationapi.PolicyName)
	}

	if err != nil {
		return nil, err
	}

	role, exists := policy.Roles[roleRef.Name]
	if !exists {
		return nil, kapierrors.NewNotFound("Role", roleRef.Name)
	}

	return role, nil
}

func validateHostDir(hostDir *HostDirectory) errs.ErrorList {
	allErrs := errs.ErrorList{}
	if hostDir.Path == "" {
		allErrs = append(allErrs, errs.NewNotFound("path", hostDir.Path))
	}
	return allErrs
}

func validateVolumeMounts(mounts []VolumeMount, volumes util.StringSet) errs.ErrorList {
	allErrs := errs.ErrorList{}

	for i := range mounts {
		mnt := &mounts[i] // so we can set default values
		if len(mnt.Name) == 0 {
			allErrs = append(allErrs, errs.NewInvalid("VolumeMount.Name", mnt.Name))
		} else if !volumes.Has(mnt.Name) {
			allErrs = append(allErrs, errs.NewNotFound("VolumeMount.Name", mnt.Name))
		}
		if len(mnt.MountPath) == 0 {
			// Backwards compat.
			if len(mnt.Path) == 0 {
				allErrs = append(allErrs, errs.NewInvalid("VolumeMount.MountPath", mnt.MountPath))
			} else {
				glog.Warning("DEPRECATED: VolumeMount.Path has been replaced by VolumeMount.MountPath")
				mnt.MountPath = mnt.Path
				mnt.Path = ""
			}
		}
		if len(mnt.MountType) != 0 {
			glog.Warning("DEPRECATED: VolumeMount.MountType will be removed. The Volume struct will handle types")
		}
	}
	return allErrs
}

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())
		}
	}
}

func Test_errToAPIStatus(t *testing.T) {
	err := errors.NewNotFound("foo", "bar")
	status := errToAPIStatus(err)
	if status.Reason != api.StatusReasonNotFound || status.Status != api.StatusFailure {
		t.Errorf("unexpected status object: %#v", status)
	}
}

func validateHostDir(hostDir *api.HostDir) errs.ValidationErrorList {
	allErrs := errs.ValidationErrorList{}
	if hostDir.Path == "" {
		allErrs = append(allErrs, errs.NewNotFound("path", hostDir.Path))
	}
	return allErrs
}

func TestCreateNotFound(t *testing.T) {
	handler := Handle(map[string]RESTStorage{
		"simple": &SimpleRESTStorage{
			// storage.Create can fail with not found error in theory.
			// See https://github.com/GoogleCloudPlatform/kubernetes/pull/486#discussion_r15037092.
			errors: map[string]error{"create": apierrs.NewNotFound("simple", "id")},
		},
	}, codec, "/prefix/version", selfLinker)
	server := httptest.NewServer(handler)
	defer server.Close()
	client := http.Client{}

	simple := &Simple{Other: "foo"}
	data, _ := codec.Encode(simple)
	request, err := http.NewRequest("POST", server.URL+"/prefix/version/simple", bytes.NewBuffer(data))
	if err != nil {
		t.Errorf("unexpected error: %v", err)
	}

	response, err := client.Do(request)
	if err != nil {
		t.Errorf("unexpected error: %v", err)
	}

	if response.StatusCode != http.StatusNotFound {
		t.Errorf("Unexpected response %#v", response)
	}
}

func (c *mockBinderClient) GetPersistentVolumeClaim(namespace, name string) (*api.PersistentVolumeClaim, error) {
	if c.claim != nil {
		return c.claim, nil
	} else {
		return nil, errors.NewNotFound("persistentVolume", name)
	}
}

// DeletePod deletes an existing pod specified by its ID.
func (r *Registry) DeletePod(podID string) error {
	var pod api.Pod
	podKey := makePodKey(podID)
	err := r.ExtractObj(podKey, &pod, false)
	if tools.IsEtcdNotFound(err) {
		return errors.NewNotFound("pod", podID)
	}
	if err != nil {
		return err
	}
	// First delete the pod, so a scheduler doesn't notice it getting removed from the
	// machine and attempt to put it somewhere.
	err = r.Delete(podKey, true)
	if tools.IsEtcdNotFound(err) {
		return errors.NewNotFound("pod", podID)
	}
	if err != nil {
		return err
	}
	machine := pod.DesiredState.Host
	if machine == "" {
		// Pod was never scheduled anywhere, just return.
		return nil
	}
	// Next, remove the pod from the machine atomically.
	contKey := makeContainerKey(machine)
	return r.AtomicUpdate(contKey, &api.ContainerManifestList{}, func(in interface{}) (interface{}, error) {
		manifests := in.(*api.ContainerManifestList)
		newManifests := make([]api.ContainerManifest, 0, len(manifests.Items))
		found := false
		for _, manifest := range manifests.Items {
			if manifest.ID != podID {
				newManifests = append(newManifests, manifest)
			} else {
				found = true
			}
		}
		if !found {
			// This really shouldn't happen, it indicates something is broken, and likely
			// there is a lost pod somewhere.
			// However it is "deleted" so log it and move on
			glog.Infof("Couldn't find: %s in %#v", podID, manifests)
		}
		manifests.Items = newManifests
		return manifests, nil
	})
}

// InterpretGetError converts a generic etcd error on a retrieval
// operation into the appropriate API error.
func InterpretGetError(err error, kind, name string) error {
	switch {
	case etcdstorage.IsEtcdNotFound(err):
		return errors.NewNotFound(kind, name)
	default:
		return err
	}
}

// InterpretDeleteError converts a generic etcd error on a delete
// operation into the appropriate API error.
func InterpretDeleteError(err error, kind, name string) error {
	switch {
	case tools.IsEtcdNotFound(err):
		return errors.NewNotFound(kind, name)
	default:
		return err
	}
}

// DeleteController deletes a ReplicationController specified by its ID.
func (r *Registry) DeleteController(controllerID string) error {
	key := makeControllerKey(controllerID)
	err := r.Delete(key, false)
	if tools.IsEtcdNotFound(err) {
		return errors.NewNotFound("replicationController", controllerID)
	}
	return err
}