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
apache · Nov 8, 2021 · c8dc031 · c8dc031
1 parent d1a2b47
commit c8dc031
Show file tree

Hide file tree

Showing 3 changed files with 91 additions and 11 deletions.
diff --git a/airflow/providers/google/cloud/example_dags/example_cloud_sql_query.py b/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")

diff --git a/tests/providers/google/cloud/operators/test_cloud_sql_system.py b/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)
diff --git a/tests/providers/google/cloud/operators/test_cloud_sql_system_helper.py b/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',
  ]
  )