update e2e test, create keyspaces, verify replication changes

test/e2e/suite_test.go

@@ -623,28 +623,47 @@ func createMultiDatacenterCluster(t *testing.T, ctx context.Context, namespace s
 
 func addDcToCluster(t *testing.T, ctx context.Context, namespace string, f *framework.E2eFramework) {
 	require := require.New(t)
+	assert := assert.New(t)
 
 	t.Log("check that the K8ssandraCluster was created")
 	kcKey := client.ObjectKey{Namespace: namespace, Name: "test"}
 	kc := &api.K8ssandraCluster{}
 	err := f.Client.Get(ctx, kcKey, kc)
 	require.NoError(err, "failed to get K8ssandraCluster in namespace %s", namespace)
 
-	dc1Key := framework.ClusterKey{K8sContext: "kind-k8ssandra-0", NamespacedName: types.NamespacedName{Namespace: namespace, Name: "dc1"}}
+	k8sCtx0 := "kind-k8ssandra-0"
+	k8sCtx1 := "kind-k8ssandra-1"
+
+	dc1Key := framework.ClusterKey{K8sContext: k8sCtx0, NamespacedName: types.NamespacedName{Namespace: namespace, Name: "dc1"}}
 	checkDatacenterReady(t, ctx, dc1Key, f)
 
+	t.Log("create keyspaces")
+	_, err = f.ExecuteCql(ctx, k8sCtx0, namespace, "test", "test-dc1-default-sts-0",
+		"CREATE KEYSPACE ks1 WITH REPLICATION = {'class' : 'NetworkTopologyStrategy', 'dc1' : 1}")
+	require.NoError(err, "failed to create keyspace")
+
+	_, err = f.ExecuteCql(ctx, k8sCtx0, namespace, "test", "test-dc1-default-sts-0",
+		"CREATE KEYSPACE ks2 WITH REPLICATION = {'class' : 'NetworkTopologyStrategy', 'dc1' : 1}")
+	require.NoError(err, "failed to create keyspace")
+
 	t.Log("add dc2 to cluster")
 	err = f.Client.Get(ctx, kcKey, kc)
 	require.NoError(err, "failed to get K8ssandraCluster %s", kcKey)
 
 	kc.Spec.Cassandra.Datacenters = append(kc.Spec.Cassandra.Datacenters, api.CassandraDatacenterTemplate{
 		Meta: api.EmbeddedObjectMeta{
 			Name: "dc2",
+			Annotations: map[string]string{
+				api.DcReplicationAnnotation: `{"dc2": {"ks1": 1, "ks2": 1}}`,
+			},
 		},
-		K8sContext: "kind-k8ssandra-1",
+		K8sContext: k8sCtx1,
 		Size: 1,
 	})
 
+	err = f.Client.Update(ctx, kc)
+	require.NoError(err, "failed to update K8ssandraCluster")
+
 	dc2Key := framework.ClusterKey{K8sContext: "kind-k8ssandra-1", NamespacedName: types.NamespacedName{Namespace: namespace, Name: "dc2"}}
 	checkDatacenterReady(t, ctx, dc2Key, f)
 
@@ -654,16 +673,29 @@ func addDcToCluster(t *testing.T, ctx context.Context, namespace string, f *fram
 
 	t.Log("check that nodes in dc1 see nodes in dc2")
 	pod := "test-dc1-default-sts-0"
-	count := 6
+	count := 2
 	checkNodeToolStatusUN(t, f, "kind-k8ssandra-0", namespace, pod, count, "-u", username, "-pw", password)
 
-	assert.NoError(t, err, "timed out waiting for nodetool status check against "+pod)
+	assert.NoError(err, "timed out waiting for nodetool status check against "+pod)
 
 	t.Log("check nodes in dc2 see nodes in dc1")
 	pod = "test-dc2-default-sts-0"
 	checkNodeToolStatusUN(t, f, "kind-k8ssandra-1", namespace, pod, count, "-u", username, "-pw", password)
 
-	assert.NoError(t, err, "timed out waiting for nodetool status check against "+pod)
+	assert.NoError(err, "timed out waiting for nodetool status check against "+pod)
+
+	keyspaces := []string{"system_auth", "ks1", "ks2"}
+	for _, ks := range keyspaces {
+		assert.Eventually(func() bool {
+			output, err := f.ExecuteCql(ctx, k8sCtx0, namespace, "test", "test-dc1-default-sts-0",
+				fmt.Sprintf("SELECT replication FROM system_schema.keyspaces WHERE keyspace_name = '%s'", ks))
+			if err != nil {
+				t.Logf("replication check for keyspace %s failed: %v", ks, err)
+				return false
+			}
+			return strings.Contains(output, "'dc1': '1'") && strings.Contains(output, "'dc2': '1'")
+		}, 1 * time.Minute, 5 * time.Second, "failed to verify replication updated for keyspace %s", ks)
+	}
 }
 
 func checkStargateApisWithMultiDcCluster(t *testing.T, ctx context.Context, namespace string, f *framework.E2eFramework) {