// logCommand logs a command and its execution status.
func logCommand(cmd string, args []interface{}, err error, log bool) {
	// Format the command for the log entry.
	formatArgs := func() string {
		if args == nil || len(args) == 0 {
			return "(none)"
		}
		output := make([]string, len(args))
		for i, arg := range args {
			output[i] = string(valueToBytes(arg))
		}
		return strings.Join(output, " / ")
	}
	logOutput := func() string {
		format := "CMD %s ARGS %s %s"
		if err == nil {
			return fmt.Sprintf(format, cmd, formatArgs(), "OK")
		}
		return fmt.Sprintf(format, cmd, formatArgs(), "ERROR "+err.Error())
	}
	// Log positive commands only if wanted, errors always.
	if err != nil {
		if errors.IsError(err, ErrServerResponse) || errors.IsError(err, ErrTimeout) {
			return
		}
		logger.Errorf(logOutput())
	} else if log {
		logger.Infof(logOutput())
	}
}

// Test creation and checking.
func TestIsError(t *testing.T) {
	assert := audit.NewTestingAssertion(t, true)

	ec := 42
	messages := errors.Messages{ec: "test error %d"}
	err := errors.New(ec, messages, 1)

	assert.ErrorMatch(err, `\[ERRORS_TEST:042\] test error 1`)
	assert.True(errors.IsError(err, ec))
	assert.False(errors.IsError(err, 0))

	err = testError("test error 2")

	assert.ErrorMatch(err, "test error 2")
	assert.False(errors.IsError(err, ec))
	assert.False(errors.IsError(err, 0))
}

func TestTransactionPipeline(t *testing.T) {
	assert := audit.NewTestingAssertion(t, true)
	conn, connRestore := connectDatabase(assert)
	defer connRestore()
	ppl, pplRestore := pipelineDatabase(assert)
	defer pplRestore()

	err := ppl.Do("multi")
	assert.Nil(err)
	ppl.Do("set", "tx:a", 1)
	ppl.Do("set", "tx:b", 2)
	ppl.Do("set", "tx:c", 3)
	ppl.Do("exec")
	results, err := ppl.Collect()
	assert.Nil(err)
	assert.Length(results, 5)
	valueB, err := conn.DoInt("get", "tx:b")
	assert.Nil(err)
	assert.Equal(valueB, 2)

	err = ppl.Do("multi")
	assert.Nil(err)
	ppl.Do("set", "tx:d", 4)
	ppl.Do("set", "tx:e", 5)
	ppl.Do("set", "tx:f", 6)
	ppl.Do("discard")
	results, err = ppl.Collect()
	assert.Nil(err)
	assert.Length(results, 5)
	valueE, err := conn.DoValue("get", "tx:e")
	assert.Nil(err)
	assert.True(valueE.IsNil())

	sig := make(chan struct{})
	go func() {
		<-sig
		conn.Do("set", "tx:h", 99)
		sig <- struct{}{}
	}()
	ppl.Do("watch", "tx:h")
	err = ppl.Do("multi")
	assert.Nil(err)
	ppl.Do("set", "tx:g", 4)
	ppl.Do("set", "tx:h", 5)
	sig <- struct{}{}
	ppl.Do("set", "tx:i", 6)
	<-sig
	ppl.Do("exec")
	results, err = ppl.Collect()
	assert.True(errors.IsError(err, redis.ErrTimeout))
	valueH, err := conn.DoInt("get", "tx:h")
	assert.Nil(err)
	assert.Equal(valueH, 99)
}

func TestTransactionConnection(t *testing.T) {
	assert := audit.NewTestingAssertion(t, true)
	conn, restore := connectDatabase(assert)
	defer restore()

	ok, err := conn.DoOK("multi")
	assert.Nil(err)
	assert.True(ok)
	conn.Do("set", "tx:a", 1)
	conn.Do("set", "tx:b", 2)
	conn.Do("set", "tx:c", 3)
	result, err := conn.Do("exec")
	assert.Nil(err)
	assert.Length(result, 3)
	valueB, err := conn.DoInt("get", "tx:b")
	assert.Nil(err)
	assert.Equal(valueB, 2)

	ok, err = conn.DoOK("multi")
	assert.Nil(err)
	assert.True(ok)
	conn.Do("set", "tx:d", 4)
	conn.Do("set", "tx:e", 5)
	conn.Do("set", "tx:f", 6)
	ok, err = conn.DoOK("discard")
	assert.Nil(err)
	assert.True(ok)
	valueE, err := conn.DoValue("get", "tx:e")
	assert.Nil(err)
	assert.True(valueE.IsNil())

	sig := make(chan struct{})
	go func() {
		asyncConn, restore := connectDatabase(assert)
		defer restore()
		<-sig
		asyncConn.Do("set", "tx:h", 99)
		sig <- struct{}{}
	}()
	conn.Do("watch", "tx:h")
	ok, err = conn.DoOK("multi")
	assert.Nil(err)
	assert.True(ok)
	conn.Do("set", "tx:g", 4)
	conn.Do("set", "tx:h", 5)
	sig <- struct{}{}
	conn.Do("set", "tx:i", 6)
	<-sig
	_, err = conn.Do("exec")
	assert.True(errors.IsError(err, redis.ErrTimeout))
	valueH, err := conn.DoInt("get", "tx:h")
	assert.Nil(err)
	assert.Equal(valueH, 99)
}

// TestEnvironmentSubscribeUnsubscribe tests subscribing,
// checking and unsubscribing of cells.
func TestEnvironmentSubscribeUnsubscribe(t *testing.T) {
	assert := audit.NewTestingAssertion(t, true)

	env := cells.NewEnvironment("subscribe-unsubscribe")
	defer env.Stop()

	err := env.StartCell("foo", newCollectBehavior())
	assert.Nil(err)
	err = env.StartCell("bar", newCollectBehavior())
	assert.Nil(err)
	err = env.StartCell("baz", newCollectBehavior())
	assert.Nil(err)
	err = env.StartCell("yadda", newCollectBehavior())
	assert.Nil(err)

	err = env.Subscribe("humpf", "foo")
	assert.True(errors.IsError(err, cells.ErrInvalidID))
	err = env.Subscribe("foo", "humpf")
	assert.True(errors.IsError(err, cells.ErrInvalidID))

	err = env.Subscribe("foo", "bar", "baz")
	assert.Nil(err)
	subs, err := env.Subscribers("foo")
	assert.Nil(err)
	assert.Contents("bar", subs)
	assert.Contents("baz", subs)

	err = env.Unsubscribe("foo", "bar")
	assert.Nil(err)
	subs, err = env.Subscribers("foo")
	assert.Nil(err)
	assert.Contents("baz", subs)

	err = env.Unsubscribe("foo", "baz")
	assert.Nil(err)
	subs, err = env.Subscribers("foo")
	assert.Nil(err)
	assert.Empty(subs)
}

func TestPubSub(t *testing.T) {
	assert := audit.NewTestingAssertion(t, true)
	conn, connRestore := connectDatabase(assert)
	defer connRestore()
	sub, subRestore := subscribeDatabase(assert)
	defer subRestore()

	_, err := conn.Do("subscribe", "pubsub")
	assert.True(errors.IsError(err, redis.ErrUseSubscription))

	err = sub.Subscribe("pubsub")
	assert.Nil(err)
	pv, err := sub.Pop()
	assert.Nil(err)
	assert.Equal(pv.Kind, "subscribe")
	assert.Equal(pv.Channel, "pubsub")
	assert.Equal(pv.Count, 1)

	go func() {
		for i := 0; i < 10; i++ {
			time.Sleep(50 * time.Millisecond)
			receivers, err := conn.DoInt("publish", "pubsub", i)
			assert.Nil(err)
			assert.Equal(receivers, 1)
		}
	}()

	sleep := 1 * time.Millisecond
	for i := 0; i < 10; i++ {
		time.Sleep(sleep)
		pv, err := sub.Pop()
		assert.Nil(err)
		assert.Equal(pv.Kind, "message")
		assert.Equal(pv.Channel, "pubsub")
		value, err := pv.Value.Int()
		assert.Nil(err)
		assert.Equal(value, i)
		sleep *= 2
	}
}

// TestEnvironmentStartStopCell tests starting, checking and
// stopping of cells.
func TestEnvironmentStartStopCell(t *testing.T) {
	assert := audit.NewTestingAssertion(t, true)

	env := cells.NewEnvironment("start-stop")
	defer env.Stop()

	err := env.StartCell("foo", newCollectBehavior())
	assert.Nil(err)

	hasFoo := env.HasCell("foo")
	assert.True(hasFoo)

	err = env.StopCell("foo")
	assert.Nil(err)
	hasFoo = env.HasCell("foo")
	assert.False(hasFoo)

	hasBar := env.HasCell("bar")
	assert.False(hasBar)
	err = env.StopCell("bar")
	assert.True(errors.IsError(err, cells.ErrInvalidID))
	hasBar = env.HasCell("bar")
	assert.False(hasBar)
}

// IsCannotRecoverError checks if an error shows a cell that cannot
// recover.
func IsCannotRecoverError(err error) bool {
	return errors.IsError(err, ErrCannotRecover)
}

// IsCellInitError checks if an error is a cell init error.
func IsCellInitError(err error) bool {
	return errors.IsError(err, ErrCellInit)
}

// IsValidationError checks if the error signals an invalid feed.
func IsValidationError(err error) bool {
	return errors.IsError(err, ErrValidation)
}

// IsInvalidPathError checks if a path cannot be found.
func IsInvalidPathError(err error) bool {
	return errors.IsError(err, ErrInvalidPath)
}

// IsNoTopicError checks if an error shows that an event has no topic..
func IsNoTopicError(err error) bool {
	return errors.IsError(err, ErrNoTopic)
}

// IsEventRecoveringError checks if an error is an error recovering error.
func IsEventRecoveringError(err error) bool {
	return errors.IsError(err, ErrEventRecovering)
}

// IsNoRequestError checks if an error signals that an event is no request.
func IsNoRequestError(err error) bool {
	return errors.IsError(err, ErrNoRequest)
}

// IsNegativeLinesError returns true, if the error shows the
// setting of a negative number of lines to start with.
func IsNegativeLinesError(err error) bool {
	return errors.IsError(err, ErrNegativeLines)
}

// IsNoTargetError returns true, if the error signals that
// no target has been passed.
func IsNoTargetError(err error) bool {
	return errors.IsError(err, ErrNoTarget)
}

// IsNoSourceError returns true, if the error signals that
// no source has been passed.
func IsNoSourceError(err error) bool {
	return errors.IsError(err, ErrNoSource)
}

// IsNoPlainTextError checks if the error signals no plain content
// inside a text element.
func IsNoPlainTextError(err error) bool {
	return errors.IsError(err, ErrNoPlainText)
}

// IsParsingError checks if the error signals a bad formatted value.
func IsParsingError(err error) bool {
	return errors.IsError(err, ErrParsing)
}

// IsDuplicateIDError checks if an error is a cell already exists error.
func IsDuplicateIDError(err error) bool {
	return errors.IsError(err, ErrDuplicateID)
}