Skip to content

Commit c4f0d9f

Browse files
authored
Update system test example_emr to have logs (#30715)
1 parent bb5f63a commit c4f0d9f

File tree

1 file changed

+16
-0
lines changed

1 file changed

+16
-0
lines changed

tests/system/providers/amazon/aws/example_emr.py

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@
3333
EmrModifyClusterOperator,
3434
EmrTerminateJobFlowOperator,
3535
)
36+
from airflow.providers.amazon.aws.operators.s3 import S3CreateBucketOperator, S3DeleteBucketOperator
3637
from airflow.providers.amazon.aws.sensors.emr import EmrJobFlowSensor
3738
from airflow.utils.trigger_rule import TriggerRule
3839
from tests.system.providers.amazon.aws.utils import ENV_ID_KEY, SystemTestContextBuilder
@@ -126,12 +127,18 @@ def delete_security_config(config_name: str):
126127
tags=["example"],
127128
) as dag:
128129
test_context = sys_test_context_task()
130+
129131
env_id = test_context[ENV_ID_KEY]
130132
config_name = f"{CONFIG_NAME}-{env_id}"
131133
execution_role_arn = test_context[EXECUTION_ROLE_ARN_KEY]
134+
s3_bucket = f"{env_id}-emr-bucket"
135+
136+
JOB_FLOW_OVERRIDES["LogUri"] = f"s3://{s3_bucket}/"
132137
JOB_FLOW_OVERRIDES["SecurityConfiguration"] = config_name
133138
JOB_FLOW_OVERRIDES["Instances"]["InstanceGroups"][0]["CustomAmiId"] = get_ami_id()
134139

140+
create_s3_bucket = S3CreateBucketOperator(task_id="create_s3_bucket", bucket_name=s3_bucket)
141+
135142
create_security_configuration = configure_security_config(config_name)
136143

137144
# [START howto_operator_emr_create_job_flow]
@@ -172,9 +179,17 @@ def delete_security_config(config_name: str):
172179

173180
delete_security_configuration = delete_security_config(config_name)
174181

182+
delete_s3_bucket = S3DeleteBucketOperator(
183+
task_id="delete_s3_bucket",
184+
bucket_name=s3_bucket,
185+
force_delete=True,
186+
trigger_rule=TriggerRule.ALL_DONE,
187+
)
188+
175189
chain(
176190
# TEST SETUP
177191
test_context,
192+
create_s3_bucket,
178193
create_security_configuration,
179194
# TEST BODY
180195
create_job_flow,
@@ -184,6 +199,7 @@ def delete_security_config(config_name: str):
184199
remove_cluster,
185200
check_job_flow,
186201
delete_security_configuration,
202+
delete_s3_bucket,
187203
)
188204

189205
from tests.system.utils.watcher import watcher

0 commit comments

Comments
 (0)