// Copyright 2019 Orange
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// 	You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// 	See the License for the specific language governing permissions and
// limitations under the License.

package v1alpha1

import (
	"io/ioutil"
	v1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/resource"
	"log"
	"path/filepath"
	"sort"
	"strings"
	"testing"

	"github.com/ghodss/yaml"
	"github.com/stretchr/testify/assert"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

func helperLoadBytes(t *testing.T, name string) []byte {
	path := filepath.Join("testdata", name) // relative path
	bytes, err := ioutil.ReadFile(path)
	if err != nil {
		t.Fatal(err)
	}
	return bytes
}

func helperInitCluster(t *testing.T, name string) *CassandraCluster {
	var cc CassandraCluster
	err := yaml.Unmarshal(helperLoadBytes(t, name), &cc)
	if err != nil {
		//log.Fatal("error: %v", err)
		log.Fatal("error: helpInitCluster")
	}
	cc.InitCassandraRackList()
	return &cc
}

func TestGetNodesPerRacks_NoTopo(t *testing.T) {
	assert := assert.New(t)

	cc := helperInitCluster(t, "cassandracluster-NoTopo.yaml")

	nodesPerRack := cc.GetNodesPerRacks("dc-rack1")

	assert.Equal(int32(8), nodesPerRack)
}

func TestGetNodesPerRacks_1DC(t *testing.T) {
	assert := assert.New(t)

	cc := helperInitCluster(t, "cassandracluster-1DC.yaml")

	nodesPerRack := cc.GetNodesPerRacks("online-rack1")
	assert.Equal(int32(7), nodesPerRack)

	nodesPerRack = cc.GetNodesPerRacks("online-rack2")
	assert.Equal(int32(7), nodesPerRack)
}

func TestGetNodesPerRacks_2DC(t *testing.T) {
	assert := assert.New(t)

	cc := helperInitCluster(t, "cassandracluster-2DC.yaml")

	nodesPerRack := cc.GetNodesPerRacks("online-rack1")
	assert.Equal(int32(6), nodesPerRack)

	nodesPerRack = cc.GetNodesPerRacks("online-rack2")
	assert.Equal(int32(6), nodesPerRack)

	nodesPerRack = cc.GetNodesPerRacks("stats-rack1")
	assert.Equal(int32(2), nodesPerRack)

	nodesPerRack = cc.GetNodesPerRacks("toto-toto")
	assert.Equal(int32(6), nodesPerRack)

}

func TestGetNumTokensPerRacks_NoTopo(t *testing.T) {
	assert := assert.New(t)

	cc := helperInitCluster(t, "cassandracluster-NoTopo.yaml")

	nodesPerRack := cc.NumTokensPerRacks("dc-rack1")

	assert.Equal(int32(256), nodesPerRack)

}
func TestGetNumTokensPerRacks_2DC(t *testing.T) {
	assert := assert.New(t)

	cc := helperInitCluster(t, "cassandracluster-2DC.yaml")

	numTokens := cc.NumTokensPerRacks("online-rack1")
	assert.Equal(int32(200), numTokens)

	numTokens = cc.NumTokensPerRacks("online-rack2")
	assert.Equal(int32(200), numTokens)

	numTokens = cc.NumTokensPerRacks("stats-rack1")
	assert.Equal(int32(32), numTokens)

	numTokens = cc.NumTokensPerRacks("toto-toto")
	assert.Equal(int32(256), numTokens)

}

func TestGetDCSize(t *testing.T) {
	assert := assert.New(t)

	cc := helperInitCluster(t, "cassandracluster-2DC.yaml")

	nb := cc.GetDCSize()
	assert.Equal(int(2), nb)

}

func TestGetDCName(t *testing.T) {
	assert := assert.New(t)

	cc := helperInitCluster(t, "cassandracluster-2DC.yaml")

	name := cc.GetDCName(0)
	assert.Equal("online", name)

	name = cc.GetDCName(1)
	assert.Equal("stats", name)

}

func TestGetRackSize(t *testing.T) {
	assert := assert.New(t)

	cc := helperInitCluster(t, "cassandracluster-2DC.yaml")

	nb := cc.GetRackSize(0)
	assert.Equal(int(2), nb)

	nb = cc.GetRackSize(1)
	assert.Equal(int(2), nb)

}

func TestGetRackName(t *testing.T) {
	assert := assert.New(t)

	cc := helperInitCluster(t, "cassandracluster-2DC.yaml")

	name := cc.GetRackName(0, 0)
	assert.Equal("rack1", name)

	name = cc.GetRackName(0, 1)
	assert.Equal("rack2", name)

	name = cc.GetRackName(1, 0)
	assert.Equal("rack1", name)

}

func TestGetDCRackName(t *testing.T) {
	assert := assert.New(t)

	var cc CassandraCluster
	assert.Equal("online-rack1", cc.GetDCRackName("online", "rack1"))

	//must log error on matchname DNS-1035
	assert.Equal("", cc.GetDCRackName("online", "Rack1"))
}

func TestInitCassandraRackList_NoTopo(t *testing.T) {
	assert := assert.New(t)

	cc := helperInitCluster(t, "cassandracluster-NoTopo.yaml")

	cc.InitCassandraRackList()

	assert.Equal(ClusterPhaseInitial.Name, cc.Status.CassandraRackStatus["dc1-rack1"].CassandraLastAction.Name)
	assert.Equal(1, len(cc.Status.CassandraRackStatus))

	assert.Equal(1, cc.GetDCSize())
	assert.Equal(1, cc.GetRackSize(0))

}

func TestInitCassandraRackList_TopoDCNoRack(t *testing.T) {
	assert := assert.New(t)

	cc := helperInitCluster(t, "cassandracluster-TopoDCNoRack.yaml")

	cc.InitCassandraRackList()

	assert.Equal(ClusterPhaseInitial.Name, cc.Status.CassandraRackStatus["online-rack1"].CassandraLastAction.Name)
	assert.Equal(1, len(cc.Status.CassandraRackStatus))

	assert.Equal(1, cc.GetDCSize())
	assert.Equal(1, cc.GetRackSize(0))

}

func TestInitCassandraRackList_2DC(t *testing.T) {
	assert := assert.New(t)

	cc := helperInitCluster(t, "cassandracluster-2DC.yaml")

	cc.InitCassandraRackList()

	assert.Equal(ClusterPhaseInitial.Name, cc.Status.CassandraRackStatus["online-rack1"].CassandraLastAction.Name)
	assert.Equal(ClusterPhaseInitial.Name, cc.Status.CassandraRackStatus["online-rack2"].CassandraLastAction.Name)
	assert.Equal(ClusterPhaseInitial.Name, cc.Status.CassandraRackStatus["stats-rack1"].CassandraLastAction.Name)
	assert.Equal(ClusterPhaseInitial.Name, cc.Status.CassandraRackStatus["stats-rack2"].CassandraLastAction.Name)
	assert.Equal(4, len(cc.Status.CassandraRackStatus))

}

func TestInitCassandraRackinStatus(t *testing.T) {
	assert := assert.New(t)

	cc := helperInitCluster(t, "cassandracluster-2DC.yaml")

	cc.InitCassandraRackList()

	assert.Equal(ClusterPhaseInitial.Name, cc.Status.CassandraRackStatus["online-rack1"].CassandraLastAction.Name)
	assert.Equal(ClusterPhaseInitial.Name, cc.Status.CassandraRackStatus["online-rack2"].CassandraLastAction.Name)
	assert.Equal(ClusterPhaseInitial.Name, cc.Status.CassandraRackStatus["stats-rack1"].CassandraLastAction.Name)
	assert.Equal(ClusterPhaseInitial.Name, cc.Status.CassandraRackStatus["stats-rack2"].CassandraLastAction.Name)
	assert.Equal(4, len(cc.Status.CassandraRackStatus))
	//Add new DC from existing RackStatus
	cc.InitCassandraRackinStatus(&cc.Status, "foo", "bar")

	assert.Equal(ClusterPhaseInitial.Name, cc.Status.CassandraRackStatus["foo-bar"].CassandraLastAction.Name)
	assert.Equal(5, len(cc.Status.CassandraRackStatus))
}

func TestGetStatusDCRackSize(t *testing.T) {
	assert := assert.New(t)

	cc := helperInitCluster(t, "cassandracluster-2DC.yaml")

	cc.InitCassandraRackList()

	assert.Equal(ClusterPhaseInitial.Name, cc.Status.CassandraRackStatus["online-rack1"].CassandraLastAction.Name)

	assert.Equal(2, cc.GetDCSize())
	assert.Equal(2, cc.GetRackSize(0))
	assert.Equal(2, cc.GetRackSize(1))

	nb := cc.GetStatusDCRackSize()
	assert.Equal(int(4), nb)

	nb = cc.GetDCRackSize()
	assert.Equal(int(4), nb)

	//Remove 1 rack/dc
	cc.Spec.Topology.DC.Remove(1)

	cc.InitCassandraRackList()

	assert.Equal(1, cc.GetDCSize())
	assert.Equal(2, cc.GetRackSize(0))

	nb = cc.GetStatusDCRackSize()
	assert.Equal(int(2), nb)
	nb = cc.GetDCRackSize()
	assert.Equal(int(2), nb)
}

//Test that a reinit keep history of changes in the status
func TestGetStatusDCRackSize_KeepChanges(t *testing.T) {
	assert := assert.New(t)

	cc := helperInitCluster(t, "cassandracluster-2DC.yaml")
	status := cc.Status.DeepCopy()

	nb := cc.GetStatusDCRackSize()
	assert.Equal(int(4), nb)
	assert.Equal(nb, cc.GetDCRackSize())

	//add info in status
	cc.Status.CassandraRackStatus["online-rack1"].CassandraLastAction.Name = ActionUpdateSeedList.Name
	assert.Equal(cc.Status.CassandraRackStatus["online-rack1"].CassandraLastAction.Name, ActionUpdateSeedList.Name)

	//Remove 1 dc
	cc.Spec.Topology.DC.Remove(1)

	deleteRackList := cc.FixCassandraRackList(status)

	sort.Strings(deleteRackList)
	assert.Equal("stats-rack1", deleteRackList[0])
	assert.Equal("stats-rack2", deleteRackList[1])

	nb = len(status.CassandraRackStatus)
	assert.Equal(2, nb)
	assert.Equal(nb, cc.GetDCRackSize())
	assert.Equal(cc.Status.CassandraRackStatus["online-rack1"].CassandraLastAction.Name, ActionUpdateSeedList.Name)
	assert.Equal(2, len(status.CassandraRackStatus))
}

func TestInitSeedList_2DC(t *testing.T) {
	assert := assert.New(t)

	cc := helperInitCluster(t, "cassandracluster-2DC.yaml")

	//cc.InitCassandraRackList()

	assert.Equal(ClusterPhaseInitial.Name, cc.Status.CassandraRackStatus["online-rack1"].CassandraLastAction.Name)
	assert.Equal(ClusterPhaseInitial.Name, cc.Status.CassandraRackStatus["online-rack2"].CassandraLastAction.Name)
	assert.Equal(ClusterPhaseInitial.Name, cc.Status.CassandraRackStatus["stats-rack1"].CassandraLastAction.Name)
	assert.Equal(ClusterPhaseInitial.Name, cc.Status.CassandraRackStatus["stats-rack2"].CassandraLastAction.Name)

	cc.Status.SeedList = cc.InitSeedList()

	assert.Equal([]string{
		"cassandra-demo-online-rack1-0.cassandra-demo.ns",
		"cassandra-demo-online-rack1-1.cassandra-demo.ns",
		"cassandra-demo-online-rack2-0.cassandra-demo.ns",
		"cassandra-demo-stats-rack1-0.cassandra-demo.ns",
		"cassandra-demo-stats-rack1-1.cassandra-demo.ns",
		"cassandra-demo-stats-rack2-0.cassandra-demo.ns",
	}, cc.Status.SeedList)
}

func TestInitSeedList_NoTopo(t *testing.T) {
	assert := assert.New(t)

	cc := helperInitCluster(t, "cassandracluster-NoTopo.yaml")

	assert.Equal(ClusterPhaseInitial.Name, cc.Status.CassandraRackStatus["dc1-rack1"].CassandraLastAction.Name)
	assert.Equal(1, len(cc.Status.CassandraRackStatus))

	cc.Status.SeedList = cc.InitSeedList()

	assert.Equal([]string{
		"cassandra-demo-dc1-rack1-0.cassandra-demo.ns",
		"cassandra-demo-dc1-rack1-1.cassandra-demo.ns",
		"cassandra-demo-dc1-rack1-2.cassandra-demo.ns",
	}, cc.Status.SeedList)

}

func TestInitSeedList_1DC(t *testing.T) {
	assert := assert.New(t)

	cc := helperInitCluster(t, "cassandracluster-1DC.yaml")

	//cc.InitCassandraRackList()

	assert.Equal(ClusterPhaseInitial.Name, cc.Status.CassandraRackStatus["online-rack1"].CassandraLastAction.Name)
	assert.Equal(ClusterPhaseInitial.Name, cc.Status.CassandraRackStatus["online-rack2"].CassandraLastAction.Name)
	assert.Equal(2, len(cc.Status.CassandraRackStatus))

	cc.Status.SeedList = cc.InitSeedList()

	assert.Equal(strings.Join([]string{
		"cassandra-demo-online-rack1-0.cassandra-demo.ns",
		"cassandra-demo-online-rack1-1.cassandra-demo.ns",
		"cassandra-demo-online-rack2-0.cassandra-demo.ns",
	}, ","), cc.SeedList(&cc.Status.SeedList))

}

func TestInitSeedList_2DC5R(t *testing.T) {
	assert := assert.New(t)

	cc := helperInitCluster(t, "cassandracluster-2DC5R.yaml")

	//cc.InitCassandraRackList()

	assert.Equal(ClusterPhaseInitial.Name, cc.Status.CassandraRackStatus["online-rack1"].CassandraLastAction.Name)
	assert.Equal(ClusterPhaseInitial.Name, cc.Status.CassandraRackStatus["online-rack2"].CassandraLastAction.Name)
	assert.Equal(ClusterPhaseInitial.Name, cc.Status.CassandraRackStatus["online-rack3"].CassandraLastAction.Name)
	assert.Equal(ClusterPhaseInitial.Name, cc.Status.CassandraRackStatus["online-rack4"].CassandraLastAction.Name)
	assert.Equal(ClusterPhaseInitial.Name, cc.Status.CassandraRackStatus["online-rack5"].CassandraLastAction.Name)
	assert.Equal(ClusterPhaseInitial.Name, cc.Status.CassandraRackStatus["stats-rack1"].CassandraLastAction.Name)
	assert.Equal(ClusterPhaseInitial.Name, cc.Status.CassandraRackStatus["stats-rack2"].CassandraLastAction.Name)
	assert.Equal(ClusterPhaseInitial.Name, cc.Status.CassandraRackStatus["stats-rack3"].CassandraLastAction.Name)
	assert.Equal(ClusterPhaseInitial.Name, cc.Status.CassandraRackStatus["stats-rack4"].CassandraLastAction.Name)
	assert.Equal(9, len(cc.Status.CassandraRackStatus))

	cc.Status.SeedList = cc.InitSeedList()

	seeds := []string{
		"cassandra-demo-online-rack1-0.cassandra-demo.ns",
		"cassandra-demo-online-rack2-0.cassandra-demo.ns",
		"cassandra-demo-online-rack3-0.cassandra-demo.ns",
		"cassandra-demo-stats-rack1-0.cassandra-demo.ns",
		"cassandra-demo-stats-rack2-0.cassandra-demo.ns",
		"cassandra-demo-stats-rack3-0.cassandra-demo.ns",
	}

	assert.Equal(seeds, cc.Status.SeedList)

	assert.Equal(strings.Join(seeds, ","), cc.SeedList(&cc.Status.SeedList))
}

func TestInitSeedList_1DC1R1P(t *testing.T) {
	assert := assert.New(t)

	cc := helperInitCluster(t, "cassandracluster-1DC1R1P.yaml")

	//cc.InitCassandraRackList()

	assert.Equal(ClusterPhaseInitial.Name, cc.Status.CassandraRackStatus["online-rack1"].CassandraLastAction.Name)
	assert.Equal(ClusterPhaseInitial.Name, cc.Status.CassandraRackStatus["online-rack2"].CassandraLastAction.Name)
	assert.Equal(2, len(cc.Status.CassandraRackStatus))

	cc.Status.SeedList = cc.InitSeedList()

	assert.Equal(
		strings.Join([]string{
			"cassandra-demo-online-rack1-0.cassandra-demo.ns",
			"cassandra-demo-online-rack2-0.cassandra-demo.ns",
		}, ","),
		cc.SeedList(&cc.Status.SeedList))
}

func TestIsPodInSeedList(t *testing.T) {
	assert := assert.New(t)

	cc := helperInitCluster(t, "cassandracluster-1DC1R1P.yaml")

	assert.Equal(ClusterPhaseInitial.Name, cc.Status.CassandraRackStatus["online-rack1"].CassandraLastAction.Name)
	assert.Equal(ClusterPhaseInitial.Name, cc.Status.CassandraRackStatus["online-rack2"].CassandraLastAction.Name)
	assert.Equal(2, len(cc.Status.CassandraRackStatus))

	cc.Status.SeedList = cc.InitSeedList()

	assert.Equal(
		strings.Join([]string{
			"cassandra-demo-online-rack1-0.cassandra-demo.ns",
			"cassandra-demo-online-rack2-0.cassandra-demo.ns",
		}, ","),
		cc.SeedList(&cc.Status.SeedList))

	assert.Equal(true, cc.IsPodInSeedList("cassandra-demo-online-rack1-0.cassandra-demo.ns"))
	assert.Equal(true, cc.IsPodInSeedList("cassandra-demo-online-rack2-0.cassandra-demo.ns"))

	assert.Equal(false, cc.IsPodInSeedList("cassandra-demo-online-rack3-0.cassandra-demo.ns"))

}

//Test that a reinit keep history of changes in the status
func TestComputeLastAppliedConfiguration(t *testing.T) {
	assert := assert.New(t)

	cc := helperInitCluster(t, "cassandracluster-2DC.yaml")

	lastAppliedConfiguration, _ := cc.ComputeLastAppliedConfiguration()
	result := `{"kind":"CassandraCluster","apiVersion":"db.orange.com/v1alpha1","metadata":{"name":"cassandra-demo","namespace":"ns","creationTimestamp":null,"labels":{"cluster":"k8s.pic"}},"spec":{"nodesPerRacks":6,"cassandraImage":"cassandra:latest","resources":{"limits":{"cpu":"1","memory":"2Gi"},"requests":{"cpu":"1","memory":"2Gi"}},"deletePVC":true,"autoPilot":true,"dataCapacity":"3Gi","dataStorageClass":"local-storage","imagePullSecret":{},"imageJolokiaSecret":{},"topology":{"dc":[{"name":"online","labels":{"location.dfy.orange.com/site":"mts"},"rack":[{"name":"rack1","labels":{"location.dfy.orange.com/street":"street1"}},{"name":"rack2","labels":{"location.dfy.orange.com/street":"street2"}}],"numTokens":200,"resources":{}},{"name":"stats","labels":{"location.dfy.orange.com/site":"mts"},"rack":[{"name":"rack1","labels":{"location.dfy.orange.com/street":"street3"}},{"name":"rack2","labels":{"location.dfy.orange.com/street":"street4"}}],"nodesPerRacks":2,"numTokens":32,"resources":{}}]}},"status":{}}`

	//add info in status
	assert.Equal(result, string(lastAppliedConfiguration))
}

func TestSetDefaults(t *testing.T) {
	assert := assert.New(t)

	cluster := CassandraCluster{
		TypeMeta: metav1.TypeMeta{
			Kind:       "CassandraCluster",
			APIVersion: "db.orange.com/v1alpha1",
		},
		ObjectMeta: metav1.ObjectMeta{
			Name:      "defaults-test",
			Namespace: "default",
		},
		Spec: CassandraClusterSpec{
			Resources: v1.ResourceRequirements{
				Requests: v1.ResourceList{
					"cpu":    resource.MustParse("500m"),
					"memory": resource.MustParse("1Gi"),
				},
			},
		},
	}

	assert.True(cluster.SetDefaults())
	cluster.CheckDefaults()

	assert.Equal(int32(1), cluster.Spec.NodesPerRacks)
	assert.Equal(defaultCassandraImage, cluster.Spec.CassandraImage)
	assert.Equal(defaultImagePullPolicy, cluster.Spec.ImagePullPolicy)

	assert.Equal(defaultBootstrapImage, cluster.Spec.BootstrapImage)
	assert.Equal(cluster.Spec.CassandraImage, cluster.Spec.InitContainerImage)
	assert.Equal(InitContainerCmd, cluster.Spec.InitContainerCmd)

	assert.Equal(resource.MustParse("500m"), *cluster.Spec.Resources.Limits.Cpu())
	assert.Equal(resource.MustParse("1Gi"), *cluster.Spec.Resources.Limits.Memory())

	assert.Equal(DefaultUserID, *cluster.Spec.RunAsUser)
	assert.Equal(ClusterPhaseInitial.Name, cluster.Status.Phase)
	assert.Equal(int32(defaultMaxPodUnavailable), cluster.Spec.MaxPodUnavailable)
	assert.Equal([]string{"defaults-test-dc1-rack1-0.defaults-test.default"}, cluster.Status.SeedList)
}
