package clusterctl

import (
	"encoding/base64"
	"encoding/json"
	"fmt"
	"strings"
	"testing"
	"time"

	containerAPI "github.com/GoogleCloudPlatform/k8s-config-connector/pkg/clients/generated/apis/container/v1beta1"
	"github.com/GoogleCloudPlatform/k8s-config-connector/pkg/clients/generated/apis/k8s/v1alpha1"
	kustomizeApi "github.com/fluxcd/kustomize-controller/api/v1beta2"
	sourceApi "github.com/fluxcd/source-controller/api/v1beta2"
	"github.com/google/uuid"
	"github.com/stretchr/testify/assert"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"edge-infra.dev/pkg/edge/api/graph/mapper"
	sqlquery "edge-infra.dev/pkg/edge/api/sql"
	clusterApi "edge-infra.dev/pkg/edge/apis/cluster/v1alpha1"
	gkeClusterApi "edge-infra.dev/pkg/edge/apis/gkecluster/v1alpha1"
	syncedobjectApi "edge-infra.dev/pkg/edge/apis/syncedobject/apis/v1alpha1"
	"edge-infra.dev/pkg/edge/compatibility"
	"edge-infra.dev/pkg/edge/constants"
	clusterConstant "edge-infra.dev/pkg/edge/constants/api/cluster"
	"edge-infra.dev/pkg/edge/constants/api/fleet"
	"edge-infra.dev/pkg/edge/controllers/util/edgedb"
	"edge-infra.dev/pkg/edge/k8objectsutils"
	whv1 "edge-infra.dev/pkg/f8n/warehouse/k8s/apis/v1alpha2"
	"edge-infra.dev/pkg/k8s/runtime/conditions"
	euuid "edge-infra.dev/pkg/lib/uuid"
	"edge-infra.dev/pkg/sds/clustersecrets/breakglass"
	"edge-infra.dev/pkg/sds/clustersecrets/grub"
	raconstants "edge-infra.dev/pkg/sds/remoteaccess/constants"
	"edge-infra.dev/test/framework/integration"
)

func (s *Suite) TestClusterControllerDelete() {
	// sos & infraShip must be deleted when the cluster is deleted
	var infraShip whv1.Shipment
	var sos syncedobjectApi.SyncedObjectList

	cluster := clusterApi.NewCluster(uuid.New().String(),
		s.ProjectID,
		s.Organization,
		fleet.Store,
		clusterConstant.GKE,
		s.Location, s.NodeVersion, s.MachineType, uuid.NewString(), s.NumNodes, s.Banner)

	s.createCluster(cluster)
	defer func() {
		s.deleteCluster(cluster)

		s.Eventually(func() bool {
			err := s.Client.Get(s.ctx, types.NamespacedName{Name: cluster.Name}, &infraShip)
			return errors.IsNotFound(err)
		}, s.timeout, s.tick, "shipment not cleaned up")

		s.Eventually(func() bool {
			s.NoError(s.Client.List(s.ctx, &sos))
			return len(sos.Items) == 0
		}, s.timeout, s.tick, "syncedobjects not cleaned up")
	}()

	// check infra_status is READY
	s.checkInfraStatusDatabaseValue(cluster, edgedb.InfraStatusReady)

	// ensure infraShip was created & sos was populated, before checking if they were deleted.
	s.NoError(s.Client.Get(s.ctx, types.NamespacedName{Name: cluster.Name}, &infraShip))
	s.NoError(s.Client.List(s.ctx, &sos))
	s.NotZero(len(sos.Items))
}

func (s *Suite) TestClusterControllerSetsInfraStatusError() {
	cluster := clusterApi.NewCluster(uuid.New().String(), s.ProjectID, s.Organization,
		"", // Set the fleet to "" so the reconcile loop produces an error
		clusterConstant.GKE, s.Location, s.NodeVersion, s.MachineType, uuid.NewString(), s.NumNodes, s.Banner)

	s.createCluster(cluster)
	defer s.deleteCluster(cluster)

	// Ensure the cluster reconciler properly set the infra_status to ERROR
	s.checkInfraStatusDatabaseValue(cluster, edgedb.InfraStatusError)
}

func (s *Suite) TestClusterControllerGKE() {
	cluster := clusterApi.NewCluster(uuid.New().String(),
		s.ProjectID,
		s.Organization,
		fleet.Store,
		clusterConstant.GKE,
		s.Location, s.NodeVersion, s.MachineType, uuid.NewString(), s.NumNodes, s.Banner)

	s.createCluster(cluster)
	defer s.deleteCluster(cluster)

	// check for Namespace creation
	ns := &corev1.Namespace{}
	s.Eventually(func() bool {
		err := s.Client.Get(s.ctx, types.NamespacedName{Name: cluster.Name}, ns)
		return !errors.IsNotFound(err)
	}, s.timeout, s.tick, "expected namespace was never created")

	// check shipment(s) created
	s.checkShipment(cluster)

	// check infra_status is READY
	s.checkInfraStatusDatabaseValue(cluster, edgedb.InfraStatusReady)

	// syncedobjects containing cluster's kustomizations
	var sos syncedobjectApi.SyncedObjectList
	s.Eventually(func() bool {
		if err := s.Client.List(s.ctx, &sos); err != nil {
			return false
		}
		s.NotZero(sos.Items)
		var filtered []syncedobjectApi.SyncedObject
		for _, so := range sos.Items {
			// skip non-kustomize syncedobjects
			if *so.Spec.Directory != constants.KustomizationsDir {
				continue
			}
			obj, err := base64.StdEncoding.DecodeString(so.Spec.Object)
			if err != nil {
				s.NoError(err, "failed to decode syncedobject data")
			}

			// special checks for banner kustomization
			if strings.Contains(so.Name, "banner-sync") {
				kusto := &kustomizeApi.Kustomization{}
				s.NoError(json.Unmarshal(obj, kusto), "failed to unmarshal kustomization from syncedobject")
				s.NotEmpty(kusto.Spec)
				s.Equal(
					kusto.Spec.SourceRef.Name,
					constants.EdgeBannerBucketName, // ensure the kustomization sourceRef Name is cluster_edge_id
					fmt.Sprintf("SO '%v' has incorrect or missing sourceRef name", so.Name),
				)
				filtered = append(filtered, so)
				continue
			}
			if strings.Contains(so.Name, "edge-banner-bucket") {
				bucko := &sourceApi.Bucket{}
				s.NoError(json.Unmarshal(obj, bucko), "failed to unmarshal bucket from syncedobject")
				s.NotEmpty(bucko.Spec)
				s.Contains(
					*bucko.Spec.Ignore,
					"chariot", // ensure the kustomization sourceRef Name is cluster_edge_id
					fmt.Sprintf("SO '%v' has incorrect or missing ignore", so.Name),
				)
				filtered = append(filtered, so)
				continue
			}
			kusto := &kustomizeApi.Kustomization{}
			s.NoError(json.Unmarshal(obj, kusto), "failed to unmarshal kustomization from syncedobject")
			s.NotEmpty(kusto.Spec)
			s.Equal(
				kusto.Spec.SourceRef.Name,
				constants.EdgeBucketName, // ensure the kustomization sourceRef Name is cluster_edge_id
				fmt.Sprintf("SO '%v' has incorrect or missing sourceRef name", so.Name),
			)
			s.True(
				strings.HasPrefix(kusto.Spec.Path, fmt.Sprintf("./%s/", cluster.Name)), // ensure the kustomization path is cluster_edge_id
				"expected component manifest kustomizations to point to a path prefixed by the cluser edge id",
			)
			filtered = append(filtered, so)
		}
		return len(filtered) == 4
	}, s.timeout, s.tick, "expected syncedobjects were never created")

	s.Eventually(func() bool {
		s.NoError(s.Client.Get(s.ctx, client.ObjectKeyFromObject(cluster), cluster))
		if cluster.Status.Inventory == nil {
			return false
		}
		return len(cluster.Status.Inventory.Entries) == 9
	}, s.timeout, s.tick, "expected inventory not created")

	// check that GKECluster is not created for store cluster
	gkeCluster := &gkeClusterApi.GKECluster{}
	s.Never(func() bool {
		err := s.Client.Get(s.ctx, types.NamespacedName{Name: cluster.Name, Namespace: cluster.Name}, gkeCluster)
		return !errors.IsNotFound(err)
	}, time.Second, 100*time.Millisecond, "GKECluster should not exist for store clusters")
}

func (s *Suite) TestClusterControllerGKENonStore() {
	cluster := clusterApi.NewCluster(uuid.New().String(),
		s.ProjectID,
		s.Organization,
		fleet.Cluster,
		clusterConstant.GKE,
		s.Location, s.NodeVersion, s.MachineType, uuid.NewString(), s.NumNodes, s.Banner)

	s.createCluster(cluster)
	defer s.deleteCluster(cluster)

	// check shipment(s) created
	s.checkShipment(cluster)

	// check infra_status is READY
	s.checkInfraStatusDatabaseValue(cluster, edgedb.InfraStatusReady)

	// check for GKECluster creation
	s.checkGKECluster(cluster, fleet.Cluster)

	// check for ns creation
	ns := &corev1.Namespace{}
	s.Eventually(func() bool {
		err := s.Client.Get(s.ctx, types.NamespacedName{Name: cluster.Name}, ns)
		return !errors.IsNotFound(err)
	}, s.timeout, s.tick, "expected namespace was never created")

	s.Eventually(func() bool {
		return checkNSOwnerReference(cluster, ns.GetOwnerReferences())
	}, s.timeout, s.tick, "expected namespace owner reference was not found")

	// check gke cluster resources at the end, they take longer to create
	s.checkContainerCluster(cluster)

	s.Eventually(func() bool {
		s.NoError(s.Client.Get(s.ctx, client.ObjectKeyFromObject(cluster), cluster))
		if cluster.Status.Inventory == nil {
			return false
		}
		return len(cluster.Status.Inventory.Entries) == 7
	}, s.timeout, s.tick, "expected inventory not created")
}

func (s *Suite) TestClusterControllerSDSDistributed() {
	integration.SkipIf(s.Framework) // this is for DSDS clusters not GKE
	cluster := clusterApi.NewCluster(uuid.New().String(),
		s.ProjectID,
		s.Organization,
		fleet.Store,
		clusterConstant.DSDS,
		// for non gke clusters below fields are ignored
		s.Location, "", "", uuid.NewString(), 0, s.Banner)

	s.createCluster(cluster)
	defer s.deleteCluster(cluster)

	// check shipment(s) created
	s.checkShipment(cluster)

	// check infra_status is READY
	s.checkInfraStatusDatabaseValue(cluster, edgedb.InfraStatusReady)

	// check for cluster ns creation
	ns := &corev1.Namespace{}
	s.Eventually(func() bool {
		err := s.Client.Get(s.ctx, types.NamespacedName{Name: cluster.Name}, ns)
		return !errors.IsNotFound(err)
	}, s.timeout, s.tick, "expected namespace was never created")

	s.Eventually(func() bool {
		return checkNSOwnerReference(cluster, ns.GetOwnerReferences())
	}, s.timeout, s.tick, "expected namespace owner reference was not found")

	// check store-credentials plugin: grub2 resources
	s.checkHashedGrubSecret(cluster)

	// check store-credentials plugin: IEN host OS user resources
	s.checkHashedBreakGlassSecret(cluster)

	// check that GKECluster was never created
	gkeCluster := &gkeClusterApi.GKECluster{}
	s.Never(func() bool {
		err := s.Client.Get(s.ctx, types.NamespacedName{Name: cluster.Name, Namespace: cluster.Name}, gkeCluster)
		return !errors.IsNotFound(err)
	}, time.Second, 100*time.Millisecond, "GKECluster should not exits for non-gke clusters")
}

func (s *Suite) TestExplicitClusterArtifacts_WithDigest() {
	cluster := s.genStoreCluster()
	s.createCluster(cluster)
	defer s.deleteCluster(cluster)
	withDigest := "sha256:d00a39e1a5dce242e7928cb8377df2540197f9d8710f2e6ccd620cb2e324a64a" //nolint
	s.updateClusterArtifactVersion(cluster, "store", withDigest)

	// check that specified cluster artifacts are added to shipment
	shipment := s.getRuntimeShipmentForCluster(cluster, constants.ClusterShipment)
	s.Require().True(shipment.Spec.Runtime, "expected Shipment in cluster's SyncedObject to be runtime only")
	s.Require().False(shipment.Spec.Infra, "expected Shipment in cluster's SyncedObject to be runtime only")
	s.Require().Len(shipment.Spec.Pallets, 1, "expected 1 pallet for store")
	storePallet := shipment.Spec.Pallets[0]
	s.Require().Equal("store", storePallet.Name, "expected store pallet to match cluster fleet name")
	s.Require().Equal("", storePallet.Tag, "expected store pallet tag to be empty if digest is set")
	s.Require().Equal(withDigest, storePallet.Digest, "expected store pallet version to be set to a non-default value")
}

func (s *Suite) TestExplicitClusterArtifacts_WithTag() {
	cluster := s.genStoreCluster()
	s.createCluster(cluster)
	defer s.deleteCluster(cluster)
	withTag := "0.14.0-test"
	s.updateClusterArtifactVersion(cluster, "store", withTag)

	// check that specified cluster artifacts are added to shipment
	shipment := s.getRuntimeShipmentForCluster(cluster, constants.ClusterShipment)
	s.Require().True(shipment.Spec.Runtime, "expected Shipment in cluster's SyncedObject to be runtime only")
	s.Require().False(shipment.Spec.Infra, "expected Shipment in cluster's SyncedObject to be runtime only")
	s.Require().Len(shipment.Spec.Pallets, 1, "expected 1 pallet for store")
	storePallet := shipment.Spec.Pallets[0]
	s.Require().Equal("store", storePallet.Name, "expected store pallet to match cluster fleet name")
	s.Require().Equal("", storePallet.Digest, "expected store pallet digest to be empty if tag is set")
	s.Require().Equal(withTag, storePallet.Tag, "expected store pallet version to be set to a non-default value")
}

func (s *Suite) TestRequeueInterval() {
	cluster := s.genStoreCluster()
	// configure this Cluster so that it re-reconciles within 1-2 ticks
	cluster.Spec.Interval = &v1.Duration{Duration: s.tick}
	startingVersion := "0.16-test"
	s.createCluster(cluster)
	s.updateClusterArtifactVersion(cluster, "store", startingVersion)
	defer s.deleteCluster(cluster)

	// get shipments, make sure pallet is store:latest
	shipment := s.getRuntimeShipmentForCluster(cluster, constants.ClusterShipment)
	s.Require().Equal(startingVersion, shipment.Spec.Pallets[0].Tag, "expected store pallet version to be fetched from database")

	// update db
	expectUpdateTo := "0.17-updated"
	go s.updateClusterArtifactVersion(cluster, "store", expectUpdateTo)

	s.Eventually(func() bool {
		// shipment, pallets should eventually (~requeue interal ish), get updated
		updatedShipment := s.getRuntimeShipmentForCluster(cluster, constants.ClusterShipment)
		return updatedShipment.Spec.Pallets[0].Tag == expectUpdateTo
	}, s.timeout, s.tick, "expected store pallet version to be updated asynchronously and corrected on requeue")
}

// create fake wireguard service and cluster, then check cluster becomes ready and the remote_access_ip value is set in the database
func (s *Suite) TestRemoteAccessIPRecorded() {
	ns := &corev1.Namespace{ObjectMeta: v1.ObjectMeta{Name: raconstants.VPNNamespace}}
	s.NoError(s.Client.Create(s.ctx, ns))
	loadBalancerIP := "34.123.45.67"
	wireguardRelayService := &corev1.Service{
		ObjectMeta: v1.ObjectMeta{Namespace: raconstants.VPNNamespace, Name: raconstants.RelayName},
		Spec: corev1.ServiceSpec{
			Ports: []corev1.ServicePort{
				{
					Port: 8080,
				},
			},
		},
	}
	s.NoError(s.Client.Create(s.ctx, wireguardRelayService))
	wireguardRelayService.Status.LoadBalancer.Ingress = []corev1.LoadBalancerIngress{
		{IP: loadBalancerIP},
	}
	s.NoError(s.Client.Status().Update(s.ctx, wireguardRelayService))
	s.Eventually(func() bool {
		service := &corev1.Service{}
		if err := s.Client.Get(s.ctx, client.ObjectKeyFromObject(wireguardRelayService), service); err != nil {
			return false
		}
		if len(service.Status.LoadBalancer.Ingress) != 1 {
			return false
		}
		return service.Status.LoadBalancer.Ingress[0].IP == loadBalancerIP
	}, s.timeout, s.tick, "no ingress IP")
	clusterObj := clusterApi.NewCluster(uuid.NewString(),
		s.ProjectID,
		s.Organization,
		fleet.Store,
		clusterConstant.DSDS,
		s.Location,
		"",
		"",
		uuid.NewString(),
		0,
		s.Banner)
	s.createCluster(clusterObj)
	defer s.deleteCluster(clusterObj)
	s.Eventually(func() bool {
		cluster := &clusterApi.Cluster{}
		if err := s.Client.Get(s.ctx, client.ObjectKeyFromObject(clusterObj), cluster); err != nil {
			return false
		}
		return conditions.IsReady(cluster)
	}, s.timeout, s.tick, "cluster did not become ready")
	s.checkRemoteAccessIPDatabaseValue(clusterObj, loadBalancerIP)
}

func (s *Suite) TestClusterShipmentLocationRegression() {
	testCases := []struct {
		title   string
		tag     string
		cluster *clusterApi.Cluster
		legacy  bool
	}{
		{
			title:   "Latest Tag",
			tag:     "latest",
			cluster: s.genStoreCluster(),
			legacy:  false,
		},
		{
			title:   "Greater than 0.18",
			tag:     "0.19",
			cluster: s.genStoreCluster(),
			legacy:  false,
		},
		{
			title:   "Less than 0.18",
			tag:     "0.17.2",
			cluster: s.genStoreCluster(),
			legacy:  true,
		},
		{
			title:   "Equal to 0.18",
			tag:     "0.18",
			cluster: s.genStoreCluster(),
			legacy:  false,
		},
	}

	for _, testCase := range testCases {
		s.Run(testCase.title, func() {
			s.createCluster(testCase.cluster)
			defer s.deleteCluster(testCase.cluster)
			s.updateClusterArtifactVersion(testCase.cluster, "store", testCase.tag)
			// check that specified cluster artifacts are added to shipment
			shipment := s.getRuntimeShipmentForCluster(testCase.cluster, constants.ClusterShipment)
			s.Require().True(shipment.Spec.Runtime, "expected Shipment in cluster's SyncedObject to be runtime only")
			s.Require().False(shipment.Spec.Infra, "expected Shipment in cluster's SyncedObject to be runtime only")
			s.Require().Len(shipment.Spec.Pallets, 1, "expected 1 pallet for store")
			storePallet := shipment.Spec.Pallets[0]
			s.Require().Equal("store", storePallet.Name, "expected store pallet to match cluster fleet name")
			s.Require().Equal("", storePallet.Digest, "expected store pallet digest to be empty if tag is set")
			s.Require().Equal(testCase.tag, storePallet.Tag, "expected store pallet version to be set to a non-default value")
			if testCase.legacy {
				s.Require().NotEmpty(shipment.Spec.Rendering[0].Variables["cluster_location"])
			} else {
				s.Require().Equal(shipment.Spec.Rendering[0].Variables["gcp_zone"], "c")
				s.Require().Equal(shipment.Spec.Rendering[0].Variables["gcp_region"], "us-east1")
			}
		})
	}
}

func (s *Suite) checkRemoteAccessIPDatabaseValue(cluster *clusterApi.Cluster, remoteAccessIP string) {
	s.Require().Eventually(func() bool {
		var v string
		err := s.DB.QueryRowContext(s.ctx, sqlquery.GetRemoteAccessIPByBannerEdgeID, cluster.Spec.BannerEdgeID).Scan(&v)
		if err != nil {
			return false
		}
		return v == remoteAccessIP
	}, s.timeout, s.tick, "expected remote_access_ip value %s not set in database", remoteAccessIP)
}

func (s *Suite) genStoreCluster() *clusterApi.Cluster {
	clusterEdgeID := uuid.NewString()
	return clusterApi.NewCluster(clusterEdgeID,
		s.ProjectID,
		s.Organization,
		fleet.Store,
		clusterConstant.Generic,
		s.Location,
		"",
		"",
		clusterEdgeID,
		0,
		s.Banner,
	)
}

func (s *Suite) checkGKECluster(cluster *clusterApi.Cluster, fleetType string) {
	// check for GKECluster creation
	gkeCluster := &gkeClusterApi.GKECluster{}
	s.Eventually(func() bool {
		err := s.Client.Get(s.ctx, types.NamespacedName{Name: cluster.Name, Namespace: cluster.Name}, gkeCluster)
		return !errors.IsNotFound(err)
	}, s.timeout, s.tick, "expected GKECluster was never created")

	s.Equal(s.ProjectID, gkeCluster.Spec.ProjectID)
	s.Equal(s.Banner.Name, gkeCluster.Spec.Banner)
	s.Equal(s.Organization, gkeCluster.Spec.Organization)
	s.Equal(cluster.Spec.Name, gkeCluster.Spec.Name)
	s.Equal(fleetType, string(gkeCluster.Spec.Fleet))
	s.Equal(s.Location, gkeCluster.Spec.Location)
	s.Equal(s.NodeVersion, gkeCluster.Spec.NodeVersion)
	s.Equal(s.NumNodes, gkeCluster.Spec.NumNode)
}

func (s *Suite) getRuntimeShipmentForCluster(cluster *clusterApi.Cluster, soPrefix string) *whv1.Shipment {
	so := &syncedobjectApi.SyncedObject{}
	soName := k8objectsutils.NameWithPrefix(soPrefix, cluster.Name)
	s.Eventually(func() bool {
		err := s.Client.Get(s.ctx, types.NamespacedName{Name: soName, Namespace: cluster.Name}, so)

		return !errors.IsNotFound(err)
	}, s.timeout, s.tick, "expected synced object was never created for: %s", soName)
	obj, err := base64.StdEncoding.DecodeString(so.Spec.Object)
	s.Require().NoError(err)
	clusterShip := &whv1.Shipment{}
	s.Require().NoError(json.Unmarshal(obj, clusterShip))
	return clusterShip
}

func (s *Suite) checkShipment(cluster *clusterApi.Cluster) {
	clusterShip := s.getRuntimeShipmentForCluster(cluster, constants.ClusterShipment)
	s.Equal(cluster.Name, clusterShip.Name)

	oldClusterShip := s.getRuntimeShipmentForCluster(cluster, constants.ClusterShipmentOld)
	s.Equal(cluster.Name, oldClusterShip.Name)

	infraShip := &whv1.Shipment{}
	s.Eventually(func() bool {
		err := s.Client.Get(s.ctx, types.NamespacedName{Name: cluster.Name}, infraShip)
		return !errors.IsNotFound(err)
	}, s.timeout, s.tick, "expected infraShipment was never created for: %s", cluster.Name)
	s.Equal(cluster.Name, infraShip.Name)
}

func (s *Suite) checkHashedBreakGlassSecret(cluster *clusterApi.Cluster) {
	breakGlassSecret := &syncedobjectApi.SyncedObject{}
	secretName := k8objectsutils.NameWithPrefix(breakglass.HashedSecretName, cluster.Name)
	s.Eventually(func() bool {
		err := s.Client.Get(s.ctx, types.NamespacedName{Name: secretName, Namespace: cluster.Name}, breakGlassSecret)
		return !errors.IsNotFound(err)
	}, s.timeout, s.tick, "expected synced object was never created for: %s", secretName)

	s.Equal(breakGlassSecret.Spec.Banner, cluster.Spec.ProjectID)
	s.Equal(breakGlassSecret.Spec.Cluster, cluster.Name)
}

func (s *Suite) checkContainerCluster(cluster *clusterApi.Cluster) {
	key := client.ObjectKey{Namespace: cluster.Name, Name: euuid.FromUUID(cluster.Name).Hash()}
	cc := &containerAPI.ContainerCluster{}
	s.Eventually(func() bool {
		err := s.Client.Get(s.ctx, key, cc)
		return !errors.IsNotFound(err)
	}, s.timeout, s.tick, "expected container cluster was never created")

	if !integration.IsIntegrationTest() {
		// update container cluster to be ready
		status := containerAPI.ContainerClusterStatus{Conditions: []v1alpha1.Condition{{Status: "True", Type: "Ready"}}}
		cc.Status = status
		s.NoError(s.Client.Update(s.ctx, cc))
	}

	nodePool := &containerAPI.ContainerNodePool{}
	s.Eventually(func() bool {
		err := s.Client.Get(s.ctx, key, nodePool)
		return !errors.IsNotFound(err)
	}, s.timeout, s.tick, "expected container node pool was never created")
}

// checkNSOwnerReference check that namespace has cluster owner reference.
func checkNSOwnerReference(cluster *clusterApi.Cluster, ownerRef []v1.OwnerReference) bool {
	if len(ownerRef) < 1 {
		return false
	}
	if ownerRef[0].Name != mapper.ConvertK8sName(cluster.Name) {
		return false
	}
	if ownerRef[0].UID != cluster.GetUID() {
		return false
	}
	return true
}

func (s *Suite) checkHashedGrubSecret(cluster *clusterApi.Cluster) {
	grubSecret := &syncedobjectApi.SyncedObject{}
	secretName := k8objectsutils.NameWithPrefix(grub.HashedSecretName, cluster.Name)
	s.Eventually(func() bool {
		err := s.Client.Get(s.ctx, types.NamespacedName{Name: secretName, Namespace: cluster.Name}, grubSecret)
		return !errors.IsNotFound(err)
	}, s.timeout, s.tick, "expected synced object was never created for: %s", secretName)

	s.Equal(grubSecret.Spec.Banner, cluster.Spec.ProjectID)
	s.Equal(grubSecret.Spec.Cluster, cluster.Name)
}

func (s *Suite) checkInfraStatusDatabaseValue(cluster *clusterApi.Cluster, infraStatus edgedb.InfraStatus) {
	const stmt = "SELECT infra_status FROM clusters WHERE cluster_edge_id=$1"
	s.Require().Eventually(func() bool {
		var v string
		err := s.DB.QueryRow(stmt, cluster.ObjectMeta.Name).Scan(&v)
		if err != nil {
			return false
		}
		return v == string(infraStatus)
	}, s.timeout, s.tick, "expected infra_status value %q not set in database", infraStatus)
}

func (s *Suite) createCluster(cluster *clusterApi.Cluster) {
	// insert the test cluster into the database so that setting infra_status works
	const stmt = "INSERT INTO clusters (cluster_edge_id, cluster_name, project_id, registered, active, banner_edge_id) VALUES ($1, $2, $3, $4, $5, $6)"
	_, err := s.DB.Exec(stmt, cluster.ObjectMeta.Name, cluster.Spec.Name, cluster.Spec.ProjectID, true, true, cluster.Spec.BannerEdgeID)
	s.Require().NoError(err)

	// store clusters must have an explicit version assigned to them
	if fleet.IsStoreCluster(cluster.Spec.Fleet) {
		const stmt2 = `INSERT INTO cluster_artifact_versions (cluster_edge_id, artifact_name, artifact_version) VALUES ($1, $2, $3)`
		_, err := s.DB.Exec(stmt2, cluster.ObjectMeta.Name, "store", "0.0.0-test")
		s.Require().NoError(err)
	}
	s.NoError(s.Client.Create(s.ctx, cluster))
}

func (s *Suite) deleteCluster(cluster *clusterApi.Cluster) {
	s.NoError(s.Client.Delete(s.ctx, cluster))
	s.Eventually(func() bool {
		return errors.IsNotFound(s.Client.Get(s.ctx, client.ObjectKeyFromObject(cluster), cluster))
	}, s.timeout, s.tick, "cluster resource was never deleted %v", cluster)
}

func (s *Suite) updateClusterArtifactVersion(cluster *clusterApi.Cluster, artifactName, artifactVersion string) { // nolint:unparam
	const stmt = `UPDATE cluster_artifact_versions SET artifact_version = $3 WHERE cluster_edge_id = $1 AND artifact_name = $2;`
	_, err := s.DB.Exec(stmt, cluster.ObjectMeta.Name, artifactName, artifactVersion)
	s.Require().NoError(err)
}

func TestSupportsOptionalDatasync(t *testing.T) {
	testCases := []struct {
		title    string
		version  string
		expected bool
	}{
		{
			title:    "Version 0.19",
			version:  "0.19",
			expected: false,
		},
		{
			title:    "Version 0.20",
			version:  "0.20",
			expected: true,
		},
		{
			title:    "Version 0.21",
			version:  "0.21",
			expected: true,
		},
		{
			title:    "Latest Version",
			version:  "latest",
			expected: true,
		},
		{
			title:    "Version 0.18",
			version:  "0.18",
			expected: false,
		},
	}
	for _, testCase := range testCases {
		t.Run(testCase.title, func(t *testing.T) {
			supported, err := compatibility.Compare(compatibility.LessThanOrEqual, "0.20", testCase.version)
			assert.NoError(t, err)
			assert.Equal(t, testCase.expected, supported)
		})
	}
}