package kafka

import (
	"fmt"

	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"

	"github.com/pingcap/test-infra/sdk/e2etest/framework"
	sdkresource "github.com/pingcap/test-infra/sdk/resource"
)

var _ = Describe("KafkaCluster Test", func() {
	Context("KafkaCluster.Create", func() {
		f := framework.NewDefaultFramework("testbed-kafkacluster-create")
		It("KafkaCluster's testbed should equal to testbed's name", func() {
			kafka := f.CreateKafkaCluster(f.CreateTestbed())
			Expect(kafka.Testbed).Should(Equal(f.Testbed()))
		})

		It("KafkaCluster should be ready after a while", func() {
			kafka := f.CreateKafkaCluster(f.CreateTestbed())
			err := f.WaitForKafkaClusterReady(kafka, framework.TestContext.ResourceWaitReadyTimeout)
			framework.ExpectNoError(err)
		})

		It("KafkaCluster replicas num should be right", func() {
			kafka := f.CreateKafkaClusterWithMultiBrokers(f.CreateTestbed(), 3, 3)
			err := f.WaitForKafkaClusterReady(kafka, framework.TestContext.ResourceWaitReadyTimeout)
			framework.ExpectNoError(err)
			Expect(kafka.Ready()).Should(Equal(true))
			Expect(kafka.Spec.ZkSpec.Replicas).Should(Equal(int32(3)))
			Expect(kafka.Spec.KafkaSpec.Replicas).Should(Equal(int32(3)))
		})

		It("KafkaCluster should work well", func() {
			testbed := f.CreateTestbed()
			kafka := f.CreateKafkaCluster(testbed)
			err := f.WaitForKafkaClusterReady(kafka, framework.TestContext.ResourceWaitReadyTimeout)
			framework.ExpectNoError(err)

			kafkacat := f.CreateKafkacatWorkloadNode(testbed)
			err = f.WaitForWorkloadNodeReady(kafkacat, framework.TestContext.ResourceWaitReadyTimeout)
			framework.ExpectNoError(err)
			u, _ := kafka.ServiceURL()
			sout, serr, exitcode, err := kafkacat.Exec(sdkresource.WorkloadNodeExecOptions{Command: fmt.Sprintf("kcat -b %s -L", u.Host)})
			framework.ExpectNoError(err)
			framework.Logf(sout)
			Expect(serr).Should(Equal(""))
			Expect(exitcode).Should(Equal(0))
		})
	})

	Context("KafkaCluster.Get", func() {
		f := framework.NewDefaultFramework("testbed-kafkacluster-get")
		It("KafkaCluster's testbed should equal to testbed's name", func() {
			kafka := f.CreateKafkaCluster(f.CreateTestbed())
			kafka = f.GetKafkaCluster(kafka.Handle())
			Expect(kafka.Testbed).Should(Equal(f.Testbed()))
		})

		It("KafkaCluster with get should be ready after a while", func() {
			kafka := f.CreateKafkaCluster(f.CreateTestbed())
			kafka = f.GetKafkaCluster(kafka.Handle())

			err := f.WaitForKafkaClusterReady(kafka, framework.TestContext.ResourceWaitReadyTimeout)
			framework.ExpectNoError(err)
		})
	})

	Context("KafkaCluster.Delete", func() {
		f := framework.NewDefaultFramework("testbed-kafkacluster-delete")
		It("KafkaCluster's delete should clean all resources", func() {
			kafka := f.CreateKafkaCluster(f.CreateTestbed())
			err := f.WaitForKafkaClusterReady(kafka, framework.TestContext.ResourceWaitReadyTimeout)
			framework.ExpectNoError(err)
			f.DeleteKafkaClusterWaited(kafka)

			pods := f.AllPods()
			Expect(len(pods.Items)).Should(Equal(0))
			Expect(len(f.AllPVCs().Items)).Should(Equal(0))
			Expect(len(f.AllServices().Items)).Should(Equal(0))
		})
	})

	Context("KafkaCluster.Exec", func() {
		f := framework.NewDefaultFramework("testbed-kafkacluster-exec")
		It("KafkaCluster's exec an normal command should work well", func() {
			kafka := f.CreateKafkaCluster(f.CreateTestbed())
			err := f.WaitForKafkaClusterReady(kafka, framework.TestContext.ResourceWaitReadyTimeout)
			framework.ExpectNoError(err)

			_, _, exitCode, _ := kafka.Exec(sdkresource.KafkaClusterExecOptions{
				Command: "echo hello > tmpfile",
				Target:  sdkresource.Kafka,
				Peer:    0,
			})
			Expect(exitCode).Should(Equal(0))

			stdout, _, exitCode, _ := kafka.Exec(sdkresource.KafkaClusterExecOptions{
				Command: "cat tmpfile",
				Target:  sdkresource.Kafka,
				Peer:    0,
			})
			Expect(exitCode).Should(Equal(0))
			Expect(stdout).Should(Equal("hello"))
		})
	})
})