Add E2e tests global cluster (#116)

Issue #, if available:

Description of changes:
Add simple e2e tests for global cluster CRD.
It's just testing creation and deletion of global cluster. currently RDS global cluster doesn't support tags, so not add tag part here. 

It is part of aws-controllers-k8s/community#1440 

Testing result:
```
scheduling tests via LoadScheduling

tests/test_global_cluster.py::TestGlobalCluster::test_crud_postgresql_globalcluster




[gw0] [100%] PASSED tests/test_global_cluster.py::TestGlobalCluster::test_crud_postgresql_globalcluster

=============================================================================================================== 1 passed in 154.64s (0:02:34) ================================================================================================================
2022-08-29T22:22:11+00:00 [INFO] Tests succeeded!
2022-08-29T22:22:11+00:00 [INFO] Running test cleanup ...
INFO:root:🧹 Cleaning up resources ...
INFO:root:Attempting cleanup Role
INFO:root:Successfully cleaned up Role
INFO:root:Attempting cleanup VPC
INFO:root:Attempting cleanup Subnets
INFO:root:Attempting cleanup RouteTable
INFO:root:Successfully cleaned up RouteTable
INFO:root:Successfully cleaned up Subnets
INFO:root:Attempting cleanup Subnets
INFO:root:Attempting cleanup RouteTable
INFO:root:Attempting cleanup InternetGateway
INFO:root:Successfully cleaned up InternetGateway
INFO:root:Successfully cleaned up RouteTable
INFO:root:Successfully cleaned up Subnets
INFO:root:Successfully cleaned up VPC
```

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.

Author: brucegucode

test/e2e/global_cluster.py

@@ -0,0 +1,140 @@
+# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"). You may
+# not use this file except in compliance with the License. A copy of the
+# License is located at
+#
+#	 http://aws.amazon.com/apache2.0/
+#
+# or in the "license" file accompanying this file. This file 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.
+
+"""Utilities for working with Global Cluster resources"""
+
+import datetime
+import time
+import typing
+
+import boto3
+import pytest
+
+DEFAULT_WAIT_UNTIL_TIMEOUT_SECONDS = 60*10
+DEFAULT_WAIT_UNTIL_INTERVAL_SECONDS = 15
+DEFAULT_WAIT_UNTIL_DELETED_TIMEOUT_SECONDS = 60*10
+DEFAULT_WAIT_UNTIL_DELETED_INTERVAL_SECONDS = 15
+
+GlobalClusterMatchFunc = typing.NewType(
+    'GlobalClusterMatchFunc',
+    typing.Callable[[dict], bool],
+)
+
+class StatusMatcher:
+    def __init__(self, status):
+        self.match_on = status
+
+    def __call__(self, record: dict) -> bool:
+        return 'Status' in record and record['Status'] == self.match_on
+
+
+def status_matches(status: str) -> GlobalClusterMatchFunc:
+    return StatusMatcher(status)
+
+
+def wait_until(
+        global_cluster_id: str,
+        match_fn: GlobalClusterMatchFunc,
+        timeout_seconds: int = DEFAULT_WAIT_UNTIL_TIMEOUT_SECONDS,
+        interval_seconds: int = DEFAULT_WAIT_UNTIL_INTERVAL_SECONDS,
+    ) -> None:
+    """Waits until a DB global cluster with a supplied ID is returned from the RDS API
+    and the matching functor returns True.
+
+    Usage:
+        from e2e.db_global_cluster import wait_until, status_matches
+
+        wait_until(
+            global_cluster_id,
+            status_matches("available"),
+        )
+
+    Raises:
+        pytest.fail upon timeout
+    """
+    now = datetime.datetime.now()
+    timeout = now + datetime.timedelta(seconds=timeout_seconds)
+
+    while not match_fn(get(global_cluster_id)):
+        if datetime.datetime.now() >= timeout:
+            pytest.fail("failed to match Global Cluster before timeout")
+        time.sleep(interval_seconds)
+
+
+def wait_until_deleted(
+        global_cluster_id: str,
+        timeout_seconds: int = DEFAULT_WAIT_UNTIL_DELETED_TIMEOUT_SECONDS,
+        interval_seconds: int = DEFAULT_WAIT_UNTIL_DELETED_INTERVAL_SECONDS,
+    ) -> None:
+    """Waits until a Global Cluster with a supplied ID is no longer returned from
+    the RDS API.
+
+    Usage:
+        from e2e.global_cluster import wait_until_deleted
+
+        wait_until_deleted(global_cluster_id)
+
+    Raises:
+        pytest.fail upon timeout or if the Global Cluster goes to any other status
+        other than 'deleting'
+    """
+    now = datetime.datetime.now()
+    timeout = now + datetime.timedelta(seconds=timeout_seconds)
+
+    while True:
+        if datetime.datetime.now() >= timeout:
+            pytest.fail(
+                "Timed out waiting for Global Cluster to be "
+                "deleted in RDS API"
+            )
+        time.sleep(interval_seconds)
+
+        latest = get(global_cluster_id)
+        if latest is None:
+            break
+
+        if latest['Status'] != "deleting":
+            pytest.fail(
+                "Status is not 'deleting' for global cluster that was "
+                "deleted. Status is " + latest['Status']
+            )
+
+
+def get(global_cluster_id):
+    """Returns a dict containing the Global Cluster record from the RDS API.
+
+    If no such Global Cluster exists, returns None.
+    """
+    c = boto3.client('rds')
+    try:
+        resp = c.describe_global_clusters(GlobalClusterIdentifier=global_cluster_id)
+        assert len(resp['GlobalClusters']) == 1
+        return resp['GlobalClusters'][0]
+    except c.exceptions.GlobalClusterNotFoundFault:
+        return None
+
+
+    
+def get_tags(global_cluster_arn):
+    """Returns a dict containing the Global Cluster's tag records from the RDS API.
+    Currently RDS doesn't support add tags to global cluster, so this is a noop for now
+    If no such global cluster exists, returns None.
+    """
+    c = boto3.client('rds')
+    try:
+        resp = c.list_tags_for_resource(
+            ResourceName=global_cluster_arn,
+        )
+        return resp['TagList']
+    except c.exceptions.GlobalClusterNotFoundFault:
+        return None

test/e2e/resources/global_cluster.yaml

@@ -0,0 +1,8 @@
+apiVersion: rds.services.k8s.aws/v1alpha1
+kind: GlobalCluster
+metadata:
+  name: $GLOBAL_CLUSTER_NAME
+spec:
+  globalClusterIdentifier: $GLOBAL_CLUSTER_NAME
+  engine: $GLOBAL_CLUSTER_ENGINE
+  databaseName: $GLOBAL_CLUSTER_DB_NAME

test/e2e/tests/test_global_cluster.py

@@ -0,0 +1,91 @@
+# Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"). You may
+# not use this file except in compliance with the License. A copy of the
+# License is located at
+#
+#	 http://aws.amazon.com/apache2.0/
+#
+# or in the "license" file accompanying this file. This file 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.
+
+"""Integration tests for the RDS API Global Cluster resource
+"""
+
+import time
+
+import pytest
+
+from acktest.k8s import resource as k8s
+from e2e import service_marker, CRD_GROUP, CRD_VERSION, load_rds_resource
+from e2e.replacement_values import REPLACEMENT_VALUES
+from e2e import condition
+from e2e import global_cluster
+from e2e.fixtures import k8s_secret
+from e2e import tag
+from e2e.bootstrap_resources import get_bootstrap_resources
+
+RESOURCE_PLURAL = 'globalclusters'
+
+DELETE_WAIT_AFTER_SECONDS = 120
+
+# Time we wait after resource becoming available in RDS and checking the CR's
+# Status has been updated.
+CHECK_STATUS_WAIT_SECONDS = 60*4
+
+MODIFY_WAIT_AFTER_SECONDS = 20
+
+
+@service_marker
+@pytest.mark.canary
+class TestGlobalCluster:
+
+    def test_crud_postgresql_globalcluster(
+            self,
+    ):
+        global_cluster_id = "my-test-global-cluster"
+        global_cluster_engine = "aurora-postgresql"
+        global_cluster_db_name = 'testdb'
+
+        replacements = REPLACEMENT_VALUES.copy()
+        replacements["GLOBAL_CLUSTER_NAME"] = global_cluster_id
+        replacements["GLOBAL_CLUSTER_ENGINE"] = global_cluster_engine
+        replacements["GLOBAL_CLUSTER_DB_NAME"] = global_cluster_db_name
+
+        resource_data = load_rds_resource(
+            "global_cluster",
+            additional_replacements=replacements,
+        )
+
+        ref = k8s.CustomResourceReference(
+            CRD_GROUP, CRD_VERSION, RESOURCE_PLURAL,
+            global_cluster_id, namespace="default",
+        )
+        # First try create global cluster 
+        k8s.create_custom_resource(ref, resource_data)
+        cr = k8s.wait_resource_consumed_by_controller(ref)
+
+        # global cluster is available immediately upon created
+        assert cr is not None
+        assert 'status' in cr
+        assert 'status' in cr['status']
+        assert cr['status']['status'] == 'available'
+
+        # assert global cluster is synced
+        cr = k8s.get_resource(ref)
+        assert cr is not None
+        assert 'status' in cr
+        assert 'status' in cr['status']
+        condition.assert_synced(ref)
+
+        latest = global_cluster.get(global_cluster_id)
+        arn = latest['GlobalClusterArn']
+
+        # now start delete global cluster
+        k8s.delete_custom_resource(ref)
+
+        time.sleep(DELETE_WAIT_AFTER_SECONDS)
+
+        global_cluster.wait_until_deleted(global_cluster_id)