package api_v2

import (
	"context"
	"encoding/json"
	"flag"
	"math/rand"
	"net"
	"net/http"
	"net/url"
	"os"
	"strings"
	"sync/atomic"
	"time"

	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
	. "github.com/pingcap/endless/pkg/matcher"
	. "github.com/pingcap/endless/pkg/util"
	"github.com/pingcap/endless/testcase/rawkv/pkg/chaos"
	"github.com/pingcap/endless/testcase/rawkv/pkg/client"
	"github.com/pingcap/endless/testcase/rawkv/pkg/scheduler"
	"github.com/pingcap/kvproto/pkg/kvrpcpb"
	netproxy "golang.org/x/net/proxy"
	"golang.org/x/sync/errgroup"
	"google.golang.org/grpc"

	pdclient "github.com/pingcap/endless/pkg/client/pd"
	_ "github.com/pingcap/endless/pkg/client/tidb"
	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/pkg/proxy"
	"github.com/pingcap/test-infra/sdk/pkg/util/pdutil"
	"github.com/pingcap/test-infra/sdk/resource"
	"github.com/tikv/client-go/v2/rawkv"
	pd "github.com/tikv/pd/client"
	"go.uber.org/zap"
)

const (
	DEFAULT_PADDING_LEN = 64
	KEY_RANGE           = 1e12 // deprecated
)

// RAW_PREFIX is only used for debug client-go without encoding raw prefix.
var RAW_PREFIX = [...]byte{byte('r'), 0, 0, 0}

// Example of running with parameters:
// `ginkgo -focus correctness_01 -- RawKV -threads=50 -keys=10000 -duration=1m`
var _ = ParameterizedDescribe("RawKV", func(flagSet *flag.FlagSet) {
	ctx := suiteTestCtx
	var (
		flagApiVersionStr string
		flagApiVersion    kvrpcpb.APIVersion
		flagThreads       int64
		flagKeysPerThread int64
		flagBatchSize     int64
		flagPaddingLen    int
		flagPadding       []byte
		flagDuration      time.Duration
		flagLogLevel      string
		flagLogFilename   string
		// Clear storage on startup
		// As the step of generated test data is fixed, it is unnecessary to clear data for correctness
		flagStartupClear bool

		flagShuffleLeader int
		flagShuffleRegion int
		flagRegionChange  int

		flagChaosInterval time.Duration
		flagAnsibleHosts  string
		flagAnsibleKey    string
		flagAnsibleUser   string

		// `ClientWithRetry` must be TRUE for chaos testcase
		// See https://github.com/tikv/client-go/issues/522
		flagClientWithRetry     bool
		flagClientRetryCnt      int           = 10
		flagClientRetryInterval time.Duration = 100 * time.Millisecond
		flagEncodeRawPrefix     bool
		flagRawPrefix           []byte

		tc      resource.TiDBCluster
		pdAddrs []string

		tikvCli       client.Client
		tikvCliAtomic client.Client
		pdCli         *pdutil.Client
	)

	flagSet.StringVar(&flagApiVersionStr, "api_version", "V2", "TiKV API Version (V1, V2)")
	flagSet.Int64Var(&flagThreads, "threads", 20, "Number of threads")
	flagSet.Int64Var(&flagKeysPerThread, "keys", 1e4, "Number of keys per thread")
	flagSet.Int64Var(&flagBatchSize, "batch_size", 10000, "Batch size of batch operations")
	flagSet.DurationVar(&flagDuration, "duration", 10*time.Second, "Duration (e.g. 1m30s) to run")
	flagSet.StringVar(&flagLogLevel, "log_level", "warn", "Log level [debug/info/error]")
	flagSet.StringVar(&flagLogFilename, "log_file", "rawkv_test.log", "Log file path")
	// The generated test data has fix interval between keys. So it is unnecessary to clear data by default.
	flagSet.BoolVar(&flagStartupClear, "clear", false, "Clear storage on startup")

	flagSet.IntVar(&flagShuffleLeader, "shuffle_leader", 1, "Shuffle leader, 1:enable, 0:disable")
	// `shuffle_region` defaults to "disable", as it make regions difficult to split/merge.
	flagSet.IntVar(&flagShuffleRegion, "shuffle_region", 0, "Shuffle region, 1:enable, 0:disable")
	flagSet.IntVar(&flagRegionChange, "region_change", 1, "Region change (split/merge), 1:enable, 0:disable")

	flagSet.DurationVar(&flagChaosInterval, "chaos_interval", 5*time.Minute, "Interval to inject chaos")
	flagSet.StringVar(&flagAnsibleHosts, "ansible_hosts", "hosts", "Hosts file of ansible")
	flagSet.StringVar(&flagAnsibleKey, "ansible_key", "", "Key file of ansible")
	flagSet.StringVar(&flagAnsibleUser, "ansible_user", "root", "Remote user of ansible")

	flagSet.BoolVar(&flagClientWithRetry, "client_with_retry", false, "Use client with internal retry")
	flagSet.BoolVar(&flagEncodeRawPrefix, "encode_raw_prefix", false, "Insert raw prefix. Should be 1 for debug client-go without encoding raw prefix")

	flagSet.IntVar(&flagPaddingLen, "pad", DEFAULT_PADDING_LEN, "Value padding length")

	initFlags := func() {
		if flagEncodeRawPrefix {
			flagRawPrefix = RAW_PREFIX[:]
		}

		flagPadding = []byte(strings.Repeat("0", flagPaddingLen))

		apiVersion, ok := kvrpcpb.APIVersion_value[flagApiVersionStr]
		Expect(ok).Should(BeTrue(), "unexpected api version: %v", flagApiVersionStr)
		flagApiVersion = kvrpcpb.APIVersion(apiVersion)
	}

	var clientCounter uint64
	mustNewClient := func(ctx context.Context, pdAddrs []string, isAtomic bool, rebuildOnError bool) client.Client {
		grpcDialer := grpc.WithContextDialer(func(ctx context.Context, addr string) (net.Conn, error) {
			dialer := proxy.GrpcProxyDialer()
			switch dialer := dialer.(type) {
			case netproxy.ContextDialer:
				return dialer.DialContext(ctx, "tcp", addr)
			default:
				return dialer.Dial("tcp", addr)
			}
		})

		buildClient := func() (cli *rawkv.Client, err error) {
			cli, err = rawkv.NewClientWithOpts(
				ctx, pdAddrs,
				rawkv.WithAPIVersion(flagApiVersion),
				rawkv.WithGRPCDialOptions(grpcDialer),
				rawkv.WithPDOptions(pd.WithGRPCDialOptions(grpcDialer)),
			)
			if err == nil {
				cli.SetAtomicForCAS(isAtomic)
				numClients := atomic.AddUint64(&clientCounter, 1)
				log.Info("Client connected", zap.Uint64("ClusterID", cli.ClusterID()), zap.Uint64("numClients", numClients))
			} else {
				log.Warn("Client connect fail", zap.Error(err), zap.Uint64("numClients", atomic.LoadUint64(&clientCounter)))
			}
			return
		}

		if !flagClientWithRetry {
			return Try(buildClient()).(*rawkv.Client)
		}
		return Try(client.NewClientWithRetry(buildClient, flagClientRetryCnt, flagClientRetryInterval, rebuildOnError)).(client.Client)
	}

	generateTestData := func(keys, values0, values1 [][]byte, keyPrefix string, idx, offset int64) ([][]byte, [][]byte, [][]byte) {
		return GenerateTestData(
			keys, values0, values1,
			flagRawPrefix, keyPrefix,
			flagKeysPerThread, flagThreads,
			flagPadding,
			idx, flagBatchSize, offset,
		)
	}

	schedulerRoutine := func(eg *errgroup.Group, closeCh <-chan struct{}) {
		pdClient := Try(pdclient.NewPDClient(tc)).(*pdclient.Client)

		if flagShuffleLeader == 1 {
			scheduler.ShuffleLeaderRoutine(pdClient, eg, closeCh)
		}

		if flagShuffleRegion == 1 {
			scheduler.ShuffleRegionRoutine(pdClient, eg, closeCh)
		}

		if flagRegionChange == 1 {
			// `pdclient.Client` lacks of `Operators` method. So `pdutil.Client` is used here.
			// TODO: unify usage of PD client.
			pdURL, err := tc.ServiceURL(resource.PDAddr)
			Expect(err).Should(Succeed())
			scheduler.RegionChangeRoutine(pdURL.String(), eg, closeCh)
		}
	}

	clearSchedulers := func() {
		pdClient := Try(pdclient.NewPDClient(tc)).(*pdclient.Client)
		scheduler.ClearTestingSchedulers(pdClient)
	}

	chaosRoutine := func(eg *errgroup.Group, closeCh <-chan struct{}) {
		chaos := Try(chaos.NewAnsibleChaos(flagAnsibleHosts, flagAnsibleKey, flagAnsibleUser, false)).(chaos.Chaos)
		log.Warn("INFO: NewAnsibleChaos", zap.Any("chaos", chaos), zap.Duration("interval", flagChaosInterval))

		workloads := NewChaosWorkloads([]ChaosWorkload{
			{chaos.KillOneTiKV, 10},
			{chaos.KillOnePD, 1},
		})

		ChaosRoutine(workloads, flagChaosInterval, eg, closeCh)
	}

	durationRoutine := func() {
		time.Sleep(flagDuration)
		log.Warn("INFO: duration routine finished")
	}

	// Substitution of `local.TiDBCluster.PeerURLs``. As `local.TiDBCluster`` do not implement this interface.
	// TODO: implement PeerURLs for local.TiDBCluster.
	getPDPeerURLs := func(tc resource.TiDBCluster) (pdAddrs []string) {
		// TODO: return Try(tc.PeerURLs(resource.PDAddr)).([]*url.URL)
		pdurls, ok := os.LookupEnv("TI_PARAM_RES_" + tc.Handle().Name + "_PD_URL")
		if ok { // local environment
			var urls []string
			Try(json.Unmarshal([]byte(pdurls), &urls))

			// verify
			for _, u := range urls {
				Try(url.Parse(u))
			}
			return urls
		}

		urls := Try(tc.PeerURLs(resource.PDAddr)).([]*url.URL)
		pdAddrs = make([]string, 0, len(urls))
		for _, u := range urls {
			pdAddrs = append(pdAddrs, u.String())
		}
		return pdAddrs
	}

	BeforeSuite(func() {
		initFlags()
	})

	BeforeEach(func() {
		rand.Seed(time.Now().UnixNano())

		tc = ctx.Resource("tc").(resource.TiDBCluster)

		fileConf := log.FileLogConfig{Filename: flagLogFilename, MaxSize: 300}
		conf := &log.Config{Level: flagLogLevel, File: fileConf}
		logger, props, _ := log.InitLogger(conf)
		log.ReplaceGlobals(logger, props)

		pdAddrs = getPDPeerURLs(tc)

		tikvCli = mustNewClient(ctx, pdAddrs, false, true)
		tikvCliAtomic = mustNewClient(ctx, pdAddrs, true, true)

		pdCli = pdutil.NewPDClient(http.DefaultClient, Try(tc.ServiceURL(resource.PDAddr)).(*url.URL).String())

		log.Warn("INFO: BeforeEach", zap.Int64("threads", flagThreads), zap.Int64("keysPerThread", flagKeysPerThread),
			zap.Duration("duration", flagDuration), zap.Any("pdAddrs", pdAddrs), zap.Bool("startupClear", flagStartupClear),
			zap.Int("clientRetryCnt", flagClientRetryCnt), zap.Duration("clientRetryInterval", flagClientRetryInterval),
			zap.Any("apiVersion", flagApiVersion),
		)
	})

	AfterEach(func() {
		tikvCli.Close()
		tikvCliAtomic.Close()
	})

	Context("ApiV2", func() {
		TestCase("Self test GenerateCausalityTestData", "self_1", func() {
			keys, values0, values1 := GenerateTestData(nil, nil, nil, RAW_PREFIX[:], "k", 5, 2, []byte("0"), 1, 20, 0) // generated: [0, 50), [50, 100), step 10
			Expect(keys).To(Equal([][]byte{
				[]byte("r\x00\x00\x00k0000000000000050"),
				[]byte("r\x00\x00\x00k0000000000000060"),
				[]byte("r\x00\x00\x00k0000000000000070"),
				[]byte("r\x00\x00\x00k0000000000000080"),
				[]byte("r\x00\x00\x00k0000000000000090"),
			}))
			Expect(values0).To(Equal([][]byte{
				[]byte("50_0"),
				[]byte("60_0"),
				[]byte("70_0"),
				[]byte("80_0"),
				[]byte("90_0"),
			}))
			Expect(values1).To(Equal([][]byte{
				[]byte("51_0"),
				[]byte("61_0"),
				[]byte("71_0"),
				[]byte("81_0"),
				[]byte("91_0"),
			}))

			keys, _, _ = GenerateTestData(nil, nil, nil, RAW_PREFIX[:], "k", 100, 2, []byte("0"), 1, 5, 0) // generated: [0, 1000), [1000, 2000), step 10
			Expect(keys).To(Equal([][]byte{
				[]byte("r\x00\x00\x00k0000000000001000"),
				[]byte("r\x00\x00\x00k0000000000001010"),
				[]byte("r\x00\x00\x00k0000000000001020"),
				[]byte("r\x00\x00\x00k0000000000001030"),
				[]byte("r\x00\x00\x00k0000000000001040"),
			}))

			keys, _, _ = GenerateTestData(nil, nil, nil, RAW_PREFIX[:], "k", 100, 2, []byte("0"), 1, 5, 5)
			Expect(len(keys)).To(Equal(5))
			Expect(keys[0]).To(Equal([]byte("r\x00\x00\x00k0000000000001050")))
			Expect(keys[4]).To(Equal([]byte("r\x00\x00\x00k0000000000001090")))

			keys, _, _ = GenerateTestData(nil, nil, nil, RAW_PREFIX[:], "k", 100, 2, []byte("0"), 1, 5, 95)
			Expect(len(keys)).To(Equal(5))
			Expect(keys[0]).To(Equal([]byte("r\x00\x00\x00k0000000000001950")))
			Expect(keys[4]).To(Equal([]byte("r\x00\x00\x00k0000000000001990")))

			keys, _, _ = GenerateTestData(nil, nil, nil, RAW_PREFIX[:], "k", 99, 2, []byte("0"), 0, 5, 95)
			Expect(len(keys)).To(Equal(4))
			Expect(keys[0]).To(Equal([]byte("r\x00\x00\x00k0000000000000950")))
			Expect(keys[3]).To(Equal([]byte("r\x00\x00\x00k0000000000000980")))

			keys, _, _ = GenerateTestData(nil, nil, nil, RAW_PREFIX[:], "k", 99, 2, []byte("0"), 1, 5, 0)
			Expect(len(keys)).To(Equal(5))
			Expect(keys[0]).To(Equal([]byte("r\x00\x00\x00k0000000000000990")))
			Expect(keys[4]).To(Equal([]byte("r\x00\x00\x00k0000000000001030")))
		})

		TestCase("Self test region change", "self_2", func() { // verify implementation of `RegionChangeRoutine`
			pdURL, err := tc.ServiceURL(resource.PDAddr)
			Expect(err).Should(Succeed())

			eg := new(errgroup.Group)
			closeCh := make(chan struct{})
			scheduler.RegionChangeRoutine(pdURL.String(), eg, closeCh)

			eg.Go(func() error {
				defer GinkgoRecover()
				durationRoutine()
				close(closeCh)
				return nil
			})
			eg.Wait()
		})

		TestCase("Self test workloads", "self_3", func() {
			w0 := func(_ context.Context, _ client.Client, _ *pdutil.Client, _ [][]byte, _ [][]byte, _ [][]byte, _ bool) {
				// log.Warn("workload 1")
			}
			w1 := func(_ context.Context, _ client.Client, _ *pdutil.Client, _ [][]byte, _ [][]byte, _ [][]byte, _ bool) {
				// log.Warn("workload 2")
			}
			w2 := func(_ context.Context, _ client.Client, _ *pdutil.Client, _ [][]byte, _ [][]byte, _ [][]byte, _ bool) {
				// log.Warn("workload 3")
			}

			workloads := NewWorkloads([]WorkloadWeight{
				{w0, false, 10},
				{w1, false, 10},
				{w2, false, 5},
			})

			for i := 0; i < 1000; i++ {
				fn, _ := workloads.Pick()
				fn(nil, nil, nil, nil, nil, nil, false)
			}

			log.Warn("Result", zap.Any("counter", workloads.Counter()))
		})

		TestCase("Basic test of raw_get & raw_put", "basic_1", func() {
			key := []byte("rKEY")
			value0 := []byte("VALUE")
			value1 := []byte("VALUE1")

			_ = PointWorkload(ctx, tikvCli, pdCli, key, value0, value1)
		})

		TestCase("Basic test of raw_delete_range & raw_scan", "basic_2", func() {
			keys, values0, values1 := generateTestData(nil, nil, nil, "b2", 0, 0)
			RangePointWorkload(ctx, tikvCli, pdCli, keys, values0, values1, true)
		})

		TestCase("Basic test of raw_batch_{put, get, delete} & raw_scan & raw_reverse_scan", "basic_3", func() {
			keys, values0, values1 := generateTestData(nil, nil, nil, "b3", 0, 0)
			BatchWorkload(ctx, tikvCli, pdCli, keys, values0, values1, true)
		})

		TestCase("Basic test of raw_compare_and_swap", "basic_4", func() {
			key := []byte("rKEY")
			CASAcumulateWorkload(ctx, tikvCliAtomic, key)
			CASLockingWorkload(ctx, tikvCliAtomic, key)
		})

		TestCase("Basic test of raw_xxx_ttl", "basic_5", func() {
			key := []byte("rKEY")
			value := []byte("VALUE")
			TTLWorkload(ctx, tikvCli, key, value, 1)

			keys, values0, values1 := generateTestData(nil, nil, nil, "b5", 0, 0)
			RangeTTLWorkload(ctx, tikvCli, pdCli, keys, values0, values1, true)
		})

		TestCase("Basic test of chaos", "chaos_1", func() {
			chaos := Try(chaos.NewAnsibleChaos(flagAnsibleHosts, flagAnsibleKey, flagAnsibleUser, false)).(*chaos.AnsibleChaos)
			chaos.SetDryRun(true)
			log.Warn("INFO: NewAnsibleChaos", zap.Any("chaos", chaos))

			workloads := NewChaosWorkloads([]ChaosWorkload{
				{chaos.KillOneTiKV, 5},
				{chaos.KillAllTiKV, 1},
				{chaos.KillOnePD, 5},
				{chaos.KillAllPD, 1},
			})

			closeCh := make(chan struct{})
			eg := new(errgroup.Group)

			ChaosRoutine(workloads, flagChaosInterval, eg, closeCh)

			eg.Go(func() error {
				defer GinkgoRecover()
				durationRoutine()
				close(closeCh)
				return nil
			})

			Try(eg.Wait())
			log.Warn("INFO: ChaosRoutine end", zap.Any("counter", workloads.Counter()))
		})

		// WARNING! Will KILL service.
		TestCase("Basic test of chaos", "chaos_2", func() {
			chaos := Try(chaos.NewAnsibleChaos(flagAnsibleHosts, flagAnsibleKey, flagAnsibleUser, false)).(chaos.Chaos)
			log.Warn("INFO: NewAnsibleChaos", zap.Any("chaos", chaos))

			workloads := NewChaosWorkloads([]ChaosWorkload{
				{chaos.KillOneTiKV, 20},
				{chaos.KillAllTiKV, 1},
				{chaos.KillOnePD, 10},
				{chaos.KillAllPD, 1},
			})

			closeCh := make(chan struct{})
			eg := new(errgroup.Group)

			ChaosRoutine(workloads, flagChaosInterval, eg, closeCh)

			eg.Go(func() error {
				defer GinkgoRecover()
				durationRoutine()
				close(closeCh)
				return nil
			})

			Try(eg.Wait())
			log.Warn("INFO: ChaosRoutine end", zap.Any("counter", workloads.Counter()))
		})

		newTypicalWorkloads := func() Workloads {
			return NewWorkloads([]WorkloadWeight{
				{RangePointWorkload, false, 20},
				{BatchWorkload, false, 5},
				{RangeTTLWorkload, false, 1}, // TTL workload would be 5x pressure.
				{RangeCASWorkload, true, 1},  // CAS workload would be 5x pressure.
			})
		}

		runWorkloads := func(ctx context.Context, keyPrefix string, workloads Workloads, scheduler bool, injectChaos bool) {
			ctx, cancel := context.WithCancel(ctx)
			defer func() {
				cancel()
			}()

			egWorker := new(errgroup.Group)
			closeCh := make(chan struct{})

			limit := flagKeysPerThread
			if limit > flagBatchSize {
				limit = flagBatchSize
			}

			for idx := int64(0); idx < int64(flagThreads); idx++ {
				idx := idx
				egWorker.Go(func() (err error) {
					defer GinkgoRecover()
					err = errors.New("fast error")
					defer func() {
						if err != nil {
							cancel()
						}
					}()

					log.Info("INFO: worker loop", zap.Int64("thread", idx), zap.Int64("limit", limit))

					var keys, values0, values1 [][]byte
					isFirstLoop := true
					for {
						for offset := int64(0); offset < flagKeysPerThread; offset += limit {
							keys, values0, values1 = generateTestData(keys, values0, values1, keyPrefix, idx, offset)
							startKey := keys[0]
							endKey := keys[len(keys)-1]

							log.Info("INFO: runWorkloads.subBatch",
								zap.Int64("thread", idx), zap.Int64("limit", limit), zap.Int64("offset", offset),
								zap.String("start", string(startKey)), zap.String("end", string(endKey)))

							fn, atomic := workloads.Pick()
							Expect(fn).ToNot(BeNil())

							if atomic {
								fn(ctx, tikvCliAtomic, pdCli, keys, values0, values1, isFirstLoop && flagStartupClear)
							} else {
								fn(ctx, tikvCli, pdCli, keys, values0, values1, isFirstLoop && flagStartupClear)
							}

							select {
							case <-closeCh:
								err = nil
								return
							case <-ctx.Done():
								err = nil
								return
							default:
							}
						}

						isFirstLoop = false
					}
				})
			}

			eg := new(errgroup.Group)

			if scheduler {
				schedulerRoutine(eg, closeCh)
			} else {
				clearSchedulers()
			}

			if injectChaos {
				chaosRoutine(eg, closeCh)
			}

			eg.Go(func() error { // Detect stop of all workers for fast fail.
				egWorker.Wait()
				cancel()
				return nil
			})

			timer := time.NewTimer(flagDuration)
			select {
			case <-timer.C:
				log.Warn("INFO: duration timeout")
				close(closeCh)
			case <-ctx.Done():
				close(closeCh)
			}

			Try(eg.Wait())
			log.Warn("INFO: runWorkloads finished", zap.Any("counter", workloads.Counter()))
		}

		// Correctness testing
		// Usage: ginkgo -p -focus '#correctness_1#' -focus '#correctness_2#' -focus '#correctness_3#' -focus '#correctness_4#' -- RawKV -duration=1m
		TestCase("Correctness test of put/get", "correctness_1", func() {
			workloads := NewWorkloads([]WorkloadWeight{
				{RangePointWorkload, false, 10},
			})

			ctx := context.Background()
			runWorkloads(ctx, "ks", workloads, false, false)
		})

		TestCase("Correctness test of batch put/get", "correctness_2", func() {
			workloads := NewWorkloads([]WorkloadWeight{
				{BatchWorkload, false, 10},
			})

			ctx := context.Background()
			runWorkloads(ctx, "kb", workloads, false, false)
		})

		TestCase("Correctness test of raw_compare_and_swap", "correctness_3", func() {
			workloads := NewWorkloads([]WorkloadWeight{
				{RangeCASWorkload, true, 10},
			})

			ctx := context.Background()
			runWorkloads(ctx, "ka", workloads, false, false)
		})

		TestCase("Correctness test of raw_xxx_ttl", "correctness_4", func() {
			workloads := NewWorkloads([]WorkloadWeight{
				{RangeTTLWorkload, false, 10},
			})

			ctx := context.Background()
			runWorkloads(ctx, "kt", workloads, false, false)
		})

		// Correctness testing for all
		// Usage: ginkgo -focus '#correctness_100#' -- RawKV -duration=5m
		TestCase("Correctness test of all", "correctness_100", func() {
			workloads := newTypicalWorkloads()
			ctx := context.Background()
			runWorkloads(ctx, "kk", workloads, false, false)
		})

		// Causality testing
		// Usage: ginkgo -focus '#causality_1#' -- RawKV -duration=1h
		TestCase("Causality test by writing repeatedly", "causality_1", func() {
			workloads := NewWorkloads([]WorkloadWeight{
				{RangePointWorkload, false, 10},
			})

			ctx := context.Background()
			runWorkloads(ctx, "c1", workloads, true, false)
		})

		TestCase("Causality test for point & cas operations", "causality_2", func() {
			workloads := NewWorkloads([]WorkloadWeight{
				{RangePointWorkload, false, 20},
				{RangeCASWorkload, true, 1}, // CAS workload would be 5x pressure.
			})

			ctx := context.Background()
			runWorkloads(ctx, "c2", workloads, true, false)
		})

		TestCase("Causality test by all workloads", "causality_100", func() {
			workloads := newTypicalWorkloads()
			ctx := context.Background()
			runWorkloads(ctx, "cc", workloads, true, false)
		})

		// WARNING! Will KILL service.
		TestCase("High-available test with chaos injected", "available_1", func() {
			workloads := newTypicalWorkloads()
			ctx := context.Background()
			runWorkloads(ctx, "a1", workloads, false, true)
		})
	})
})