func TestCreateFailsIfCantAcquireLock(t *testing.T) {
	newRCID := rc_fields.ID("new_rc")
	oldRCID := rc_fields.ID("old_rc")

	rollstore := newRollStore(t, nil)

	update := fields.Update{
		NewRC: newRCID,
		OldRC: oldRCID,
	}

	// Grab an update creation lock on one of the RCs and make sure the
	// creation fails
	session, _, err := rollstore.store.NewSession("conflicting session", nil)
	if err != nil {
		t.Fatalf("Unable to create session for conflicting lock: %s", err)
	}
	defer session.Destroy()

	_, err = rollstore.rcstore.LockForUpdateCreation(update.OldRC, session)
	if err != nil {
		t.Fatalf("Unable to acquire conflicting lock on old rc: %s", err)
	}

	_, err = rollstore.CreateRollingUpdateFromExistingRCs(update, nil, nil)
	if err == nil {
		t.Fatal("Expected update creation to fail due to lock conflict")
	}

	ru, _ := rollstore.Get(fields.ID(newRCID))
	if ru.NewRC != "" || ru.OldRC != "" {
		t.Fatal("New ru shouldn't have been created but it was")
	}
}

func newRollStore(t *testing.T, entries []fields.Update) consulStore {
	storeFields := make(map[string]*api.KVPair)
	for _, u := range entries {
		path, err := RollPath(fields.ID(u.NewRC))
		if err != nil {
			t.Fatalf("Unable to create roll store for test: %s", err)
		}
		json, err := json.Marshal(u)
		if err != nil {
			t.Fatalf("Unable to marshal test field as JSON: %s", err)
		}
		storeFields[path] = &api.KVPair{
			Key:   path,
			Value: json,
		}
	}
	return consulStore{
		kv: &consulutil.FakeKV{
			Entries: storeFields,
		},
		store:   kptest.NewFakePodStore(nil, nil),
		rcstore: rcstore.NewFake(),
		labeler: labels.NewFakeApplicator(),
	}
}

// Attempts to create a rolling update. Checks sessionErrCh for session renewal
// errors just before actually doing the creation to minimize the likelihood of
// race conditions resulting in conflicting RUs
func (s consulStore) attemptRUCreation(u roll_fields.Update, rollLabels klabels.Set, sessionErrCh chan error) (createdRU roll_fields.Update, err error) {
	// If we create an RU, we also want to create its labels. If the second step
	// fails, we want to best-effort remove the RU
	var ruCleanup func()
	defer func() {
		if err != nil && ruCleanup != nil {
			ruCleanup()
		}
	}()

	b, err := json.Marshal(u)
	if err != nil {
		return u, err
	}

	key, err := RollPath(roll_fields.ID(u.NewRC))
	if err != nil {
		return u, err
	}

	// Confirm that our lock session is still valid, and then create the
	// rolling update. If session isn't valid, delete the newRC we just
	// created
	select {
	case err := <-sessionErrCh:
		if err == nil {
			err = util.Errorf("Cannot create ru because session was destroyed")
		}
		return u, err
	default:
		success, _, err := s.kv.CAS(&api.KVPair{
			Key:   key,
			Value: b,
		}, nil)
		if err != nil {
			return u, consulutil.NewKVError("cas", key, err)
		}

		// Shouldn't be possible if our session is still valid, preventing other insertions
		if !success {
			return u, util.Errorf("update with new RC ID %s already exists", u.NewRC)
		}

		ruCleanup = func() {
			err := s.Delete(u.ID())
			if err != nil {
				s.logger.WithError(err).Errorln("Unable to cleanup RU %s after failed labeling attempt", u.ID())
			}
		}
	}

	err = s.labeler.SetLabels(labels.RU, u.ID().String(), rollLabels)
	if err != nil {
		return roll_fields.Update{}, err
	}

	return u, nil
}

func TestCreateRollingUpdateFromOneExistingRCWithIDFailsIfCantAcquireLock(t *testing.T) {
	oldRCID := rc_fields.ID("old_rc")

	rollstore := newRollStore(t, nil)

	// Grab an update creation lock on the old RC and make sure the
	// creation fails
	session, _, err := rollstore.store.NewSession("conflicting session", nil)
	if err != nil {
		t.Fatalf("Unable to create session for conflicting lock: %s", err)
	}
	defer session.Destroy()

	_, err = rollstore.rcstore.LockForUpdateCreation(oldRCID, session)
	if err != nil {
		t.Fatalf("Unable to acquire conflicting lock on old rc: %s", err)
	}

	newUpdate, err := rollstore.CreateRollingUpdateFromOneExistingRCWithID(
		oldRCID,
		1,
		0,
		false,
		0,
		testManifest(),
		testNodeSelector(),
		nil,
		nil,
		nil,
	)

	if err == nil {
		t.Fatalf("Should have erred creating conflicting update")
	}

	update, err := rollstore.Get(fields.ID(newUpdate.NewRC))
	if err != nil {
		t.Fatalf("Should nothave erred checking for update creation: %s", err)
	}

	if update.NewRC != "" {
		t.Fatalf("Update was created but shouldn't have been: %s", err)
	}

	rcs, err := rollstore.rcstore.List()
	if err != nil {
		t.Fatalf("Shouldn't have failed to list RCs: %s", err)
	}

	if len(rcs) != 0 {
		t.Fatalf("There shouldn't be any new RCs after a failed update: expect 0 but were %d", len(rcs))
	}
}

func TestGet(t *testing.T) {
	rollstore := newRollStore(t, []fields.Update{testRollValue(testRCId)})

	entry, err := rollstore.Get(fields.ID(testRCId))
	if err != nil {
		t.Fatalf("Unexpected error retrieving roll from roll store: %s", err)
	}

	if entry.NewRC != testRCId {
		t.Errorf("Expected roll to have NewRC of %s, was %s", testRCId, entry.NewRC)
	}
}

func TestRollLockPath(t *testing.T) {
	rollLockPath, err := RollLockPath(fields.ID(testRCId))
	if err != nil {
		t.Fatalf("Unable to compute roll lock path: %s", err)
	}

	expected := fmt.Sprintf("%s/%s/%s", consulutil.LOCK_TREE, rollTree, testRCId)
	if rollLockPath != expected {
		t.Errorf("Unexpected value for rollLockPath, wanted '%s' got '%s'",
			expected,
			rollLockPath,
		)
	}
}

// Test that if a conflicting update exists, a new one will not be admitted
func TestCreateExistingRCsMutualExclusion(t *testing.T) {
	newRCID := rc_fields.ID("new_rc")
	oldRCID := rc_fields.ID("old_rc")

	conflictingEntry := fields.Update{
		OldRC: newRCID,
		NewRC: rc_fields.ID("some_other_rc"),
	}

	rollstore := newRollStore(t, []fields.Update{conflictingEntry})

	update := fields.Update{
		NewRC: newRCID,
		OldRC: oldRCID,
	}

	_, err := rollstore.CreateRollingUpdateFromExistingRCs(update, nil, nil)
	if err == nil {
		t.Fatal("Expected update creation to fail due to conflict")
	}

	if conflictingErr, ok := err.(*ConflictingRUError); !ok {
		t.Error("Returned error didn't have ConflictingRUError type")
	} else {
		if conflictingErr.ConflictingID != conflictingEntry.ID() {
			t.Errorf("Expected error to have conflicting ID of '%s', was '%s'", conflictingEntry.ID(), conflictingErr.ConflictingID)
		}

		if conflictingErr.ConflictingRCID != conflictingEntry.OldRC {
			t.Errorf("Expected error to have conflicting rc ID of '%s', was '%s'", conflictingEntry.OldRC, conflictingErr.ConflictingRCID)
		}
	}

	ru, _ := rollstore.Get(fields.ID(update.NewRC))
	if ru.NewRC != "" || ru.OldRC != "" {
		t.Fatal("New ru shouldn't have been created but it was")
	}
}

func TestCreateRollingUpdateFromOneExistingRCWithIDMutualExclusion(t *testing.T) {
	rollstore := newRollStore(t, nil)

	// create the old RC
	oldRC, err := rollstore.rcstore.Create(testManifest(), nil, nil)
	if err != nil {
		t.Fatalf("Failed to create old rc: %s", err)
	}

	conflictingEntry, err := rollstore.CreateRollingUpdateFromOneExistingRCWithID(
		oldRC.ID,
		1,
		0,
		false,
		0,
		testManifest(),
		testNodeSelector(),
		nil,
		nil,
		nil,
	)
	if err != nil {
		t.Fatalf("Unable to create conflicting update: %s", err)
	}

	newUpdate, err := rollstore.CreateRollingUpdateFromOneExistingRCWithID(
		oldRC.ID,
		1,
		0,
		false,
		0,
		testManifest(),
		testNodeSelector(),
		nil,
		nil,
		nil,
	)
	if err == nil {
		t.Fatalf("Should have erred creating conflicting update")
	}

	if conflictingErr, ok := err.(*ConflictingRUError); !ok {
		t.Error("Returned error didn't have ConflictingRUError type")
	} else {
		if conflictingErr.ConflictingID != conflictingEntry.ID() {
			t.Errorf("Expected error to have conflicting ID of '%s', was '%s'", conflictingEntry.ID(), conflictingErr.ConflictingID)
		}

		if conflictingErr.ConflictingRCID != conflictingEntry.OldRC {
			t.Errorf("Expected error to have conflicting rc ID of '%s', was '%s'", conflictingEntry.OldRC, conflictingErr.ConflictingRCID)
		}
	}

	update, err := rollstore.Get(fields.ID(newUpdate.NewRC))
	if err != nil {
		t.Fatalf("Should not have erred checking for update creation: %s", err)
	}

	if update.NewRC != "" {
		t.Fatalf("Update was created but shouldn't have been: %s", err)
	}

	rcs, err := rollstore.rcstore.List()
	if err != nil {
		t.Fatalf("Shouldn't have failed to list RCs: %s", err)
	}

	if len(rcs) != 2 {
		t.Fatalf("There shouldn't be any new RCs after a failed update: expect 2 but were %d", len(rcs))
	}
}

func TestCreateRollingUpdateFromOneExistingRCWithID(t *testing.T) {
	oldRCID := rc_fields.ID("old_rc")

	rollstore := newRollStore(t, nil)

	newRCLabels := klabels.Set(map[string]string{
		"some_key": "some_val",
	})

	newUpdate, err := rollstore.CreateRollingUpdateFromOneExistingRCWithID(
		oldRCID,
		1,
		0,
		false,
		0,
		testManifest(),
		testNodeSelector(),
		nil,
		newRCLabels,
		newRCLabels,
	)

	if err != nil {
		t.Fatalf("Unable to create rolling update: %s", err)
	}

	storedUpdate, err := rollstore.Get(fields.ID(newUpdate.NewRC))
	if err != nil {
		t.Fatalf("Unable to retrieve value put in roll store: %s", err)
	}

	if storedUpdate.NewRC != newUpdate.NewRC {
		t.Errorf("Stored update didn't have expected new rc value: wanted '%s' but got '%s'", newUpdate.NewRC, storedUpdate.NewRC)
	}

	if storedUpdate.OldRC != oldRCID {
		t.Errorf("Stored update didn't have expected old rc value: wanted '%s' but got '%s'", oldRCID, storedUpdate.OldRC)
	}

	_, err = rollstore.rcstore.Get(newUpdate.NewRC)
	if err != nil {
		t.Fatalf("Shouldn't have failed to fetch new RC: %s", err)
	}

	rcLabels, err := rollstore.labeler.GetLabels(labels.RC, storedUpdate.NewRC.String())
	if err != nil {
		t.Fatalf("Unable to fetch labels for newly created new RC: %s", err)
	}

	if rcLabels.Labels["some_key"] != "some_val" {
		t.Errorf("Expected labels to be set on new RC")
	}

	ruLabels, err := rollstore.labeler.GetLabels(labels.RU, newUpdate.ID().String())
	if err != nil {
		t.Fatalf("Unable to fetch labels for newly created new RU: %s", err)
	}

	if ruLabels.Labels["some_key"] != "some_val" {
		t.Errorf("Expected labels to be set on new RU")
	}
}