feluelle commented on a change in pull request #8877: URL: https://github.com/apache/airflow/pull/8877#discussion_r437309397
########## File path: tests/providers/amazon/aws/operators/test_s3_to_redshift_system.py ########## @@ -0,0 +1,44 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License 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. +import os + +import pytest + +from airflow.models import Connection +from airflow.utils import db +from tests.test_utils import AIRFLOW_MAIN_FOLDER +from tests.test_utils.amazon_system_helpers import AWS_DAG_FOLDER, AmazonSystemTest +from tests.test_utils.terraform import Terraform + + +@pytest.mark.backend("mysql", "postgres") +class TestS3ToRedshiftExampleDags(AmazonSystemTest, Terraform): + TERRAFORM_DIR = os.path.join( + AIRFLOW_MAIN_FOLDER, "tests", "providers", "amazon", "aws", "infrastructure", "example_s3_to_redshift" + ) + + def setUp(self) -> None: + super().setUp() + host, port = self.get_tf_output("redshift_endpoint").split(':') + schema = self.get_tf_output("redshift_database_name"), + login = self.get_tf_output("redshift_master_username"), + password = self.get_tf_output("redshift_master_password") + db.merge_conn(Connection("redshift_default", "postgres", host, login, password, schema, port)) Review comment: Nope. That's a TODO. Thanks 👍 ########## File path: airflow/providers/amazon/aws/example_dags/example_s3_to_redshift.py ########## @@ -41,13 +42,26 @@ schedule_interval=None, tags=['example'] ) as dag: + preparation__task_load_sample_data_to_s3 = PythonOperator( + python_callable=lambda: S3Hook().load_string("0,Airflow", f'{S3_KEY}/{REDSHIFT_TABLE}', S3_BUCKET), + task_id='preparation__load_sample_data_to_s3' + ) + preparation__task_create_table = PostgresOperator( + sql=f'CREATE TABLE {REDSHIFT_TABLE}(Id int, Name varchar)', + postgres_conn_id='redshift_default', + task_id='preparation__create_table' + ) Review comment: You are correct. But since I am `terraform destroy`ing the whole infra being used in `tearDown`. I don't need to remove the s3 object and the redshift table, do I? ########## File path: airflow/providers/amazon/aws/example_dags/example_s3_to_redshift.py ########## @@ -41,13 +42,26 @@ schedule_interval=None, tags=['example'] ) as dag: + preparation__task_load_sample_data_to_s3 = PythonOperator( + python_callable=lambda: S3Hook().load_string("0,Airflow", f'{S3_KEY}/{REDSHIFT_TABLE}', S3_BUCKET), + task_id='preparation__load_sample_data_to_s3' + ) + preparation__task_create_table = PostgresOperator( + sql=f'CREATE TABLE {REDSHIFT_TABLE}(Id int, Name varchar)', + postgres_conn_id='redshift_default', + task_id='preparation__create_table' + ) Review comment: You are correct. But since I am `terraform destroy`ing the whole infra being used in `tearDown` I don't need to remove the s3 object and the redshift table, do I? ########## File path: tests/test_utils/terraform.py ########## @@ -0,0 +1,36 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License 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. + +from tests.test_utils.system_tests_class import SystemTest + + +class Terraform(SystemTest): + TERRAFORM_DIR: str + + def setUp(self) -> None: + self.execute_cmd(["terraform", "init", "-input=false", self.TERRAFORM_DIR]) + self.execute_cmd(["terraform", "plan", "-input=false", self.TERRAFORM_DIR]) + self.execute_cmd(["terraform", "apply", "-input=false", "-auto-approve", self.TERRAFORM_DIR]) + + def get_tf_output(self, name): + output = self.check_output(["terraform", "output", name]).decode('utf-8').replace("\r\n", "") + self.log.info(output) + return output + + def tearDown(self) -> None: + self.execute_cmd(["terraform", "plan", "-destroy", "-input=false", self.TERRAFORM_DIR]) + self.execute_cmd(["terraform", "destroy", "-input=false", "-auto-approve", self.TERRAFORM_DIR]) Review comment: I checked this out, but didn't find that really useful to use an **extra package** only to wrap bash inside python. And it also [seems](https://github.com/beelit94/python-terraform/issues/85) to not be compatible to the latest terraform 0.12.x releases. ########## File path: airflow/providers/amazon/aws/example_dags/example_s3_to_redshift.py ########## @@ -41,13 +42,26 @@ schedule_interval=None, tags=['example'] ) as dag: + preparation__task_load_sample_data_to_s3 = PythonOperator( + python_callable=lambda: S3Hook().load_string("0,Airflow", f'{S3_KEY}/{REDSHIFT_TABLE}', S3_BUCKET), + task_id='preparation__load_sample_data_to_s3' + ) + preparation__task_create_table = PostgresOperator( + sql=f'CREATE TABLE {REDSHIFT_TABLE}(Id int, Name varchar)', + postgres_conn_id='redshift_default', + task_id='preparation__create_table' + ) Review comment: I do need to remove it, you are right. 👍 It should not depend on the `tearDown`. And I will also make the setup steps idempotent. 👍 ########## File path: airflow/providers/amazon/aws/example_dags/example_s3_to_redshift.py ########## @@ -0,0 +1,53 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License 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. + +""" +This is an example dag for using `S3ToRedshiftTransfer` to copy a S3 key into a Redshift table. +""" + +from os import getenv + +from airflow import DAG +from airflow.providers.amazon.aws.operators.s3_to_redshift import S3ToRedshiftTransfer +from airflow.utils.dates import days_ago + +# [START howto_operator_s3_to_redshift_env_variables] +S3_BUCKET = getenv("S3_BUCKET", "bucket") +S3_KEY = getenv("S3_KEY", "key") Review comment: But how do we then define defaults in case these `var.`s are not set? Or should we make all vars required then? ########## File path: tests/test_utils/terraform.py ########## @@ -0,0 +1,36 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License 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. + +from tests.test_utils.system_tests_class import SystemTest + + +class Terraform(SystemTest): + TERRAFORM_DIR: str + + def setUp(self) -> None: + self.execute_cmd(["terraform", "init", "-input=false", self.TERRAFORM_DIR]) + self.execute_cmd(["terraform", "plan", "-input=false", self.TERRAFORM_DIR]) + self.execute_cmd(["terraform", "apply", "-input=false", "-auto-approve", self.TERRAFORM_DIR]) + + def get_tf_output(self, name): + output = self.check_output(["terraform", "output", name]).decode('utf-8').replace("\r\n", "") + self.log.info(output) + return output + + def tearDown(self) -> None: + self.execute_cmd(["terraform", "plan", "-destroy", "-input=false", self.TERRAFORM_DIR]) + self.execute_cmd(["terraform", "destroy", "-input=false", "-auto-approve", self.TERRAFORM_DIR]) Review comment: I could do that. The real issue I am having with this, is also that `super().setUp()` and `super().tearDown()` are not explicit enough. Terraform is very much hidden behind them. You can easily forget to call those when you overwrite those which for `super().tearDown()` could be dramatically. But maybe we could add a static check that checks that?! ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
