Fix Cloud SQL system tests (#19014)

- Add creation of Fine-grained bucket for ACLs
- Add patching of environment variables
- Add unique postfix to instances names

Author: deedmitrij

airflow/providers/google/cloud/example_dags/example_cloud_sql_query.py

@@ -48,10 +48,12 @@
 GCP_PROJECT_ID = os.environ.get('GCP_PROJECT_ID', 'example-project')
 GCP_REGION = os.environ.get('GCP_REGION', 'europe-west1')
 
-GCSQL_POSTGRES_INSTANCE_NAME_QUERY = os.environ.get('GCSQL_POSTGRES_INSTANCE_NAME_QUERY', 'testpostgres')
+GCSQL_POSTGRES_INSTANCE_NAME_QUERY = os.environ.get(
+    'GCSQL_POSTGRES_INSTANCE_NAME_QUERY', 'test-postgres-query'
+)
 GCSQL_POSTGRES_DATABASE_NAME = os.environ.get('GCSQL_POSTGRES_DATABASE_NAME', 'postgresdb')
 GCSQL_POSTGRES_USER = os.environ.get('GCSQL_POSTGRES_USER', 'postgres_user')
-GCSQL_POSTGRES_PASSWORD = os.environ.get('GCSQL_POSTGRES_PASSWORD', 'password')
+GCSQL_POSTGRES_PASSWORD = os.environ.get('GCSQL_POSTGRES_PASSWORD', 'JoxHlwrPzwch0gz9')
 GCSQL_POSTGRES_PUBLIC_IP = os.environ.get('GCSQL_POSTGRES_PUBLIC_IP', '0.0.0.0')
 GCSQL_POSTGRES_PUBLIC_PORT = os.environ.get('GCSQL_POSTGRES_PUBLIC_PORT', 5432)
 GCSQL_POSTGRES_CLIENT_CERT_FILE = os.environ.get(
@@ -62,10 +64,10 @@
 )
 GCSQL_POSTGRES_SERVER_CA_FILE = os.environ.get('GCSQL_POSTGRES_SERVER_CA_FILE', ".key/postgres-server-ca.pem")
 
-GCSQL_MYSQL_INSTANCE_NAME_QUERY = os.environ.get('GCSQL_MYSQL_INSTANCE_NAME_QUERY', 'testmysql')
+GCSQL_MYSQL_INSTANCE_NAME_QUERY = os.environ.get('GCSQL_MYSQL_INSTANCE_NAME_QUERY', 'test-mysql-query')
 GCSQL_MYSQL_DATABASE_NAME = os.environ.get('GCSQL_MYSQL_DATABASE_NAME', 'mysqldb')
 GCSQL_MYSQL_USER = os.environ.get('GCSQL_MYSQL_USER', 'mysql_user')
-GCSQL_MYSQL_PASSWORD = os.environ.get('GCSQL_MYSQL_PASSWORD', 'password')
+GCSQL_MYSQL_PASSWORD = os.environ.get('GCSQL_MYSQL_PASSWORD', 'JoxHlwrPzwch0gz9')
 GCSQL_MYSQL_PUBLIC_IP = os.environ.get('GCSQL_MYSQL_PUBLIC_IP', '0.0.0.0')
 GCSQL_MYSQL_PUBLIC_PORT = os.environ.get('GCSQL_MYSQL_PUBLIC_PORT', 3306)
 GCSQL_MYSQL_CLIENT_CERT_FILE = os.environ.get('GCSQL_MYSQL_CLIENT_CERT_FILE', ".key/mysql-client-cert.pem")

tests/providers/google/cloud/operators/test_cloud_sql_system.py

@@ -19,6 +19,7 @@
 import random
 import string
 import time
+import uuid
 
 import pytest
 
@@ -34,27 +35,99 @@
 from tests.test_utils.gcp_system_helpers import CLOUD_DAG_FOLDER, GoogleSystemTest, provide_gcp_context
 
 GCP_PROJECT_ID = os.environ.get('GCP_PROJECT_ID', 'project-id')
+CLOUD_SQL_BUCKET_NAME = os.environ.get('CLOUD_SQL_BUCKET_NAME', 'INVALID BUCKET NAME')
 
 SQL_QUERY_TEST_HELPER = CloudSqlQueryTestHelper()
 
 
+@pytest.fixture(scope='class')
+def env_patch():
+    """
+    A convenient fixture for environment variables patching.
+    All modifications will be undone after the requesting test class has finished.
+    """
+    from _pytest.monkeypatch import MonkeyPatch
+
+    mpatch = MonkeyPatch()
+    yield mpatch
+    mpatch.undo()
+
+
+@pytest.fixture(scope='module')
+def default_instances():
+    """
+    Collect list of environment variables default values
+    """
+    mysql_instance_name = os.environ.get('GCSQL_MYSQL_INSTANCE_NAME', 'test-mysql') + '%s'
+    mysql_instance_name2 = os.environ.get('GCSQL_MYSQL_INSTANCE_NAME2', 'test-mysql2') + '%s'
+    mysql_instance_name_query = mysql_instance_name + QUERY_SUFFIX
+    postgres_instance_name = os.environ.get('GCSQL_POSTGRES_INSTANCE_NAME', 'test-postgres') + '%s'
+    postgres_instance_name2 = os.environ.get('GCSQL_POSTGRES_INSTANCE_NAME2', 'test-postgres2') + '%s'
+    postgres_instance_name_query = postgres_instance_name + QUERY_SUFFIX
+    instances = [
+        ('GCSQL_MYSQL_INSTANCE_NAME', mysql_instance_name),
+        ('GCSQL_MYSQL_INSTANCE_NAME2', mysql_instance_name2),
+        ('GCSQL_MYSQL_INSTANCE_NAME_QUERY', mysql_instance_name_query),
+        ('GCSQL_POSTGRES_INSTANCE_NAME', postgres_instance_name),
+        ('GCSQL_POSTGRES_INSTANCE_NAME2', postgres_instance_name2),
+        ('GCSQL_POSTGRES_INSTANCE_NAME_QUERY', postgres_instance_name_query),
+    ]
+    return instances
+
+
+@pytest.fixture(scope='class', autouse=True)
+def set_unique_postfix(env_patch, default_instances):
+    """
+    Generate a unique postfix and add it to an instance name to avoid 409 HTTP error
+    in case of the instance name was already used during last week
+    """
+    unique_postfix = f'-{uuid.uuid4().hex[:5]}'  # generate 5 digits unique postfix
+    for instance in default_instances:
+        env_variable, value = instance
+        env_patch.setenv(env_variable, value % unique_postfix)
+
+
+@pytest.fixture
+def set_mysql_ip(monkeypatch):
+    """Set ip address of MySQL instance to GCSQL_MYSQL_PUBLIC_IP env variable"""
+    with open(os.environ["GCSQL_MYSQL_PUBLIC_IP_FILE"]) as file:
+        env, ip_address = file.read().split("=")
+        monkeypatch.setenv(env, ip_address)
+
+
+@pytest.fixture
+def set_postgres_ip(monkeypatch):
+    """Set ip address of Postgres instance to GCSQL_POSTGRES_PUBLIC_IP env variable"""
+    with open(os.environ["GCSQL_POSTGRES_PUBLIC_IP_FILE"]) as file:
+        env, ip_address = file.read().split("=")
+        monkeypatch.setenv(env, ip_address)
+
+
 @pytest.mark.backend("mysql", "postgres")
 @pytest.mark.credential_file(GCP_CLOUDSQL_KEY)
 class CloudSqlExampleDagsIntegrationTest(GoogleSystemTest):
+    @provide_gcp_context(GCP_CLOUDSQL_KEY)
     def setUp(self):
         super().setUp()
+        # 1. Create Fine-grained bucket to provide ACLs
+        self.log.info(f'Creating {CLOUD_SQL_BUCKET_NAME} bucket...')
+        self.execute_cmd(["gsutil", "mb", "-b", "off", f"gs://{CLOUD_SQL_BUCKET_NAME}"])
 
     @provide_gcp_context(GCP_CLOUDSQL_KEY)
     def tearDown(self):
         if os.path.exists(TEARDOWN_LOCK_FILE):
             self.log.info("Skip deleting instances as they were created manually (helps to iterate on tests)")
         else:
-            # Delete instances just in case the test failed and did not cleanup after itself
+            # 1. Delete instances just in case the test failed and did not cleanup after itself
             SQL_QUERY_TEST_HELPER.delete_instances(instance_suffix="-failover-replica")
             SQL_QUERY_TEST_HELPER.delete_instances(instance_suffix="-read-replica")
             SQL_QUERY_TEST_HELPER.delete_instances()
             SQL_QUERY_TEST_HELPER.delete_instances(instance_suffix="2")
             SQL_QUERY_TEST_HELPER.delete_service_account_acls()
+
+            # 2. Delete bucket
+            self.log.info(f'Deleting {CLOUD_SQL_BUCKET_NAME} bucket...')
+            self.execute_cmd(["gsutil", "rm", "-r", f"gs://{CLOUD_SQL_BUCKET_NAME}"])
         super().tearDown()
 
     @provide_gcp_context(GCP_CLOUDSQL_KEY)
@@ -79,7 +152,6 @@ class CloudSqlProxySystemTest(GoogleSystemTest):
     @classmethod
     @provide_gcp_context(GCP_CLOUDSQL_KEY)
     def setUpClass(cls):
-        SQL_QUERY_TEST_HELPER.set_ip_addresses_in_env()
         if os.path.exists(TEARDOWN_LOCK_FILE_QUERY):
             print(
                 "Skip creating and setting up instances as they were created manually "
@@ -176,5 +248,6 @@ def test_start_proxy_with_all_instances_specific_version(self):
         assert runner.get_proxy_version() == "1.13"
 
     @provide_gcp_context(GCP_CLOUDSQL_KEY)
+    @pytest.mark.usefixtures('set_mysql_ip', 'set_postgres_ip')
     def test_run_example_dag_cloudsql_query(self):
         self.run_dag('example_gcp_sql_query', CLOUD_DAG_FOLDER)

tests/providers/google/cloud/operators/test_cloud_sql_system_helper.py

@@ -41,12 +41,14 @@
 )
 GCSQL_POSTGRES_PUBLIC_IP_FILE = os.environ.get('GCSQL_POSTGRES_PUBLIC_IP_FILE', ".key/postgres-ip.env")
 GCSQL_POSTGRES_USER = os.environ.get('GCSQL_POSTGRES_USER', 'postgres_user')
+GCSQL_POSTGRES_PASSWORD = os.environ.get('GCSQL_POSTGRES_PASSWORD', 'JoxHlwrPzwch0gz9')
 GCSQL_POSTGRES_DATABASE_NAME = os.environ.get('GCSQL_POSTGRES_DATABASE_NAME', 'postgresdb')
 GCSQL_MYSQL_CLIENT_CERT_FILE = os.environ.get('GCSQL_MYSQL_CLIENT_CERT_FILE', ".key/mysql-client-cert.pem")
 GCSQL_MYSQL_CLIENT_KEY_FILE = os.environ.get('GCSQL_MYSQL_CLIENT_KEY_FILE', ".key/mysql-client-key.pem")
 GCSQL_MYSQL_SERVER_CA_FILE = os.environ.get('GCSQL_MYSQL_SERVER_CA_FILE', ".key/mysql-server-ca.pem")
 GCSQL_MYSQL_PUBLIC_IP_FILE = os.environ.get('GCSQL_MYSQL_PUBLIC_IP_FILE', ".key/mysql-ip.env")
 GCSQL_MYSQL_USER = os.environ.get('GCSQL_MYSQL_USER', 'mysql_user')
+GCSQL_MYSQL_PASSWORD = os.environ.get('GCSQL_MYSQL_PASSWORD', 'JoxHlwrPzwch0gz9')
 GCSQL_MYSQL_DATABASE_NAME = os.environ.get('GCSQL_MYSQL_DATABASE_NAME', 'mysqldb')
 
 GCSQL_MYSQL_EXPORT_URI = os.environ.get('GCSQL_MYSQL_EXPORT_URI', 'gs://bucketName/fileName')
@@ -80,13 +82,13 @@ def get_absolute_path(path):
 def get_postgres_instance_name(instance_suffix=''):
     if instance_suffix is None:
         return None
-    return os.environ.get('GCSQL_POSTGRES_INSTANCE_NAME', 'testpostgres') + instance_suffix
+    return os.environ.get('GCSQL_POSTGRES_INSTANCE_NAME', 'test-postgres') + instance_suffix
 
 
 def get_mysql_instance_name(instance_suffix=''):
     if instance_suffix is None:
         return None
-    return os.environ.get('GCSQL_MYSQL_INSTANCE_NAME', 'testmysql') + instance_suffix
+    return os.environ.get('GCSQL_MYSQL_INSTANCE_NAME', 'test-mysql') + instance_suffix
 
 
 class CloudSqlQueryTestHelper(CommandExecutor):
@@ -216,6 +218,7 @@ def setup_instances(self, instance_suffix=''):
                 client_cert_file_mysql,
                 GCSQL_MYSQL_DATABASE_NAME,
                 GCSQL_MYSQL_USER,
+                GCSQL_MYSQL_PASSWORD,
             )
         )
         postgres_thread = Thread(
@@ -227,6 +230,7 @@ def setup_instances(self, instance_suffix=''):
                 client_cert_file_postgres,
                 GCSQL_POSTGRES_DATABASE_NAME,
                 GCSQL_POSTGRES_USER,
+                GCSQL_POSTGRES_PASSWORD,
             )
         )
         mysql_thread.start()
@@ -320,6 +324,7 @@ def __setup_instance_and_certs(
         client_cert_file,
         db_name,
         db_username,
+        db_userpassword,
     ):
         self.log.info('Setting up a test %s instance "%s"...', db_version, instance_name)
         try:
@@ -336,7 +341,7 @@ def __setup_instance_and_certs(
             self.__write_to_file(client_cert_file, self.__get_client_cert(instance_name, client_cert_name))
             self.__wait_for_operations(instance_name)
             self.__wait_for_operations(instance_name)
-            self.__create_user(instance_name, db_username)
+            self.__create_user(instance_name, db_username, db_userpassword)
             self.__wait_for_operations(instance_name)
             self.__delete_db(instance_name, db_name)
             self.__create_db(instance_name, db_name)
@@ -414,7 +419,7 @@ def __get_client_cert(self, instance_name: str, client_cert_name: str) -> bytes:
         self.log.info('... Done.')
         return output
 
-    def __create_user(self, instance_name: str, username: str) -> None:
+    def __create_user(self, instance_name: str, username: str, userpassword: str) -> None:
         self.log.info('Creating user "%s" in Cloud SQL instance "%s"...', username, instance_name)
         self.execute_cmd(
             [
@@ -428,7 +433,7 @@ def __create_user(self, instance_name: str, username: str) -> None:
                 '--host',
                 '%',
                 '--password',
-                'JoxHlwrPzwch0gz9',
+                userpassword,
                 '--quiet',
             ]
         )