test_delete_ephemeral_signals_.py

from turbineflow_pytest_airflow.turbine_bdd import bdd, TurbineBDD
from hamcrest import *

def test_task_delete_ephemeral_signals_is_rendered_correctly(bdd: TurbineBDD):
    """Ensure rendered fields of delete_ephemeral_signals are correctly rendered"""
    bdd.given_dag("daily_signals")
    bdd.given_task("load_files.delete_ephemeral_signals")
    bdd.given_execution_date("1976-08-13")
    bdd.when_I_render_the_task_template_fields()
    bdd.then_it_should_match(has_property("deletion_dataset_table", equal_to("daily-signals-project.07_day_retention.signals_in_19760813")))
    bdd.then_it_should_match(has_property("impersonation_chain", equal_to("sa-daily-signals-pipeline@daily-signals-project.iam.gserviceaccount.com")))
    bdd.then_it_should_match(has_property("downstream_task_ids", has_item("load_files.load_to_bigquery")))