package configwatcher

import (
	"context"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/resourcemanager/testutils"
	rm "github.com/tidbcloud/aws-shared-provider/pkg/tidbworker/resourcemanager/types"
	"k8s.io/apimachinery/pkg/api/equality"
)

func mustNewConfigWatcher(re *require.Assertions) (rm.ConfigWatcher, testutils.MockManager) {
	mockConfig := testutils.DefaultConfig()
	mockManager, err := testutils.NewMockManager(mockConfig)
	re.NoError(err)

	return NewConfigWatcher(mockManager, &rm.ConfigWatcherConfig{
		PollInterval:    300 * time.Millisecond,
		WorkerConfigMap: mockConfig.WorkerConfig.Name,
		PoolsConfigMap:  mockConfig.PoolsConfig.Name,
		WorkerSecret:    mockConfig.SecretName,
	}), mockManager
}

func TestConfigWatcher(t *testing.T) {
	re := require.New(t)
	watcher, manager := mustNewConfigWatcher(re)

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	go func() {
		re.NoError(watcher.Start(ctx))
	}()

	var (
		pools map[rm.WorkerPool]*rm.PoolConfig
		err   error
	)

	// Should get the correct pool configurations.
	target, err := manager.GetPools()
	re.NoError(err)
	re.Eventually(func() bool {
		pools, err = watcher.GetPools()
		re.NoError(err)
		return equalPoolsConfig(pools, target)
	}, 3*time.Second, 500*time.Millisecond)

	// Remove a pool from the configmap, and check if the watcher can detect the change.
	currentConfig := manager.GetConfig()
	for pool := range currentConfig.PoolsConfig.Pools {
		delete(currentConfig.PoolsConfig.Pools, pool)
		break
	}
	re.NoError(manager.UpdateCluster(ctx, currentConfig))
	target, err = manager.GetPools()
	re.NoError(err)
	re.Eventually(func() bool {
		pools, err = watcher.GetPools()
		re.NoError(err)
		return equalPoolsConfig(pools, target)
	}, 3*time.Second, 500*time.Millisecond)

	// Change the pool size, and check if the watcher can detect the change.
	currentConfig = manager.GetConfig()
	for pool := range currentConfig.PoolsConfig.Pools {
		currentConfig.PoolsConfig.Pools[pool].Size++
	}
	re.NoError(manager.UpdateCluster(ctx, currentConfig))
	target, err = manager.GetPools()
	re.NoError(err)
	re.Eventually(func() bool {
		pools, err = watcher.GetPools()
		re.NoError(err)
		return equalPoolsConfig(pools, target)
	}, 3*time.Second, 500*time.Millisecond)
}

// equalPoolsConfig assert that two poolsConfigs are semantically equal.
func equalPoolsConfig(orig, target map[rm.WorkerPool]*rm.PoolConfig) bool {
	if len(orig) != len(target) {
		return false
	}
	for pool, origConfig := range orig {
		targetConfig, ok := target[pool]
		if !ok {
			return false
		}
		if origConfig.Size != targetConfig.Size {
			return false
		}
		if !equality.Semantic.DeepEqual(origConfig.WorkerPodSpec, targetConfig.WorkerPodSpec) {
			return false
		}
	}
	return true
}