View Source

Test Case ID

Test Objective

Test Steps

Expected Outcome

Test Type

TC-001

Verify DAG is scheduled based on asset watcher

Create a DAG scheduled based on asset. Associate a watcher to the asset. Example:

file_path = "test_file"

with DAG(
    dag_id="test_create_file",
    catchup=False,
):
    @task
    def create_file():
        with open(file_path, "w") as file:
            file.write("This is an example file.\n")

    chain(create_file())

trigger = FileTrigger(filepath=file_path, poke_interval=10)
asset = Asset("example_asset", watchers=[AssetWatcher(name="file_trigger", trigger=trigger)])
with DAG(
    dag_id="test_asset_with_watchers",
    schedule=[asset],
    catchup=False,
):
    @task
    def delete_file():
        if os.path.exists(file_path):
            os.remove(file_path)

    chain(delete_file())

Enable DAGs test_create_file and test_asset_with_watchers
Trigger DAG test_create_file
Verify DAG test_asset_with_watchers has been triggered (wait few seconds if not)

The DAG test_asset_with_watchers was executed successfully.

Positive

TC-002

Verify DAG is not scheduled when no watcher is associated to the asset

Create a DAG scheduled based on asset. Associate no watcher to the asset. Example:

file_path = "test_file"

with DAG(
    dag_id="test_create_file",
    catchup=False,
):
    @task
    def create_file():
        with open(file_path, "w") as file:
            file.write("This is an example file.\n")

    chain(create_file())

asset = Asset("example_asset")
with DAG(
    dag_id="test_asset_with_watchers",
    schedule=[asset],
    catchup=False,
):
    @task
    def delete_file():
        if os.path.exists(file_path):
            os.remove(file_path)

    chain(delete_file())

Enable DAGs test_create_file and test_asset_with_watchers
Trigger DAG test_create_file
Wait 30 seconds and verify DAG test_asset_with_watchers has not been triggered

The DAG test_asset_with_watchers was not executed

Negative

TC-003

Verify DAG is scheduled based on two different external events

Create a DAG scheduled based on asset. Associate two watchers to the asset. Example:

file_path_1 = "test_file_1"
file_path_2 = "test_file_2"

with DAG(
    dag_id="example_create_file_1",
    catchup=False,
):
    @task
    def create_file():
        with open(file_path_1, "w") as file:
            file.write("This is an example file.\n")

    chain(create_file())

with DAG(
    dag_id="example_create_file_2",
    catchup=False,
):
    @task
    def create_file():
        with open(file_path_2, "w") as file:
            file.write("This is an example file.\n")

    chain(create_file())

trigger_1 = FileTrigger(filepath=file_path_1, poke_interval=10)
trigger_2 = FileTrigger(filepath=file_path_2, poke_interval=10)
asset = Asset("example_asset", watchers=[
    AssetWatcher(name="file_trigger", trigger=trigger_1),
    AssetWatcher(name="file_trigger", trigger=trigger_2)
])

with DAG(
    dag_id="example_asset_with_watchers",
    schedule=[asset],
    catchup=False,
):
    @task
    def delete_files():
        if os.path.exists(file_path_1):
            os.remove(file_path_1)
        if os.path.exists(file_path_2):
            os.remove(file_path_2)

    chain(delete_files())

Enable DAGs example_create_file_1, example_create_file_2 and test_asset_with_watchers
Trigger DAG example_create_file_1
Verify DAG test_asset_with_watchers has been triggered (wait few seconds if not)
Trigger DAG example_create_file_2
Verify DAG test_asset_with_watchers has been triggered (wait few seconds if not)

The DAG test_asset_with_watchers was executed successfully twice (once per event).

Positive

TC-004

Verify DAG is not scheduled when DAG is not enabled/paused

Create a DAG scheduled based on asset. Associate a watcher to the asset. Example:

file_path = "test_file"

with DAG(
    dag_id="test_create_file",
    catchup=False,
):
    @task
    def create_file():
        with open(file_path, "w") as file:
            file.write("This is an example file.\n")

    chain(create_file())

trigger = FileTrigger(filepath=file_path, poke_interval=10)
asset = Asset("example_asset", watchers=[AssetWatcher(name="file_trigger", trigger=trigger)])
with DAG(
    dag_id="test_asset_with_watchers",
    schedule=[asset],
    catchup=False,
):
    @task
    def delete_file():
        if os.path.exists(file_path):
            os.remove(file_path)

    chain(delete_file())

Enable DAGs test_create_file
Disable/pause DAG test_asset_with_watchers
Trigger DAG test_create_file
Wait 30 seconds and verify DAG test_asset_with_watchers has not been triggered

The DAG test_asset_with_watchers was not executed

Negative

TC-005

Verify DAG is not scheduled when watcher is not associated to the asset

Create a DAG scheduled based on asset. Create a watcher without associating it to the asset. Example:

file_path = "test_file"

with DAG(
    dag_id="test_create_file",
    catchup=False,
):
    @task
    def create_file():
        with open(file_path, "w") as file:
            file.write("This is an example file.\n")

    chain(create_file())

watcher = AssetWatcher(name="file_trigger", trigger=FileTrigger(filepath=file_path, poke_interval=10))
asset = Asset("example_asset")
with DAG(
    dag_id="test_asset_with_watchers",
    schedule=[asset],
    catchup=False,
):
    @task
    def delete_file():
        if os.path.exists(file_path):
            os.remove(file_path)

    chain(delete_file())

Enable DAGs test_create_file and test_asset_with_watchers
Trigger DAG test_create_file
Wait 30 seconds and verify DAG test_asset_with_watchers has not been triggered

The DAG was not executed

Negative

TC-005

Verify DAG is scheduled based on external event

Create a DAG scheduled based on asset. Associate a watcher to the asset that monitors an external resource. Example: use SqsSensorTrigger to monitor an AWS SQS queue
Enable the DAG
Create the external resource. Example: if SqsSensorTrigger is used, create a message in the specific queue through AWS console or AWS CLI
Verify the DAG has been triggered (wait few seconds if not)

The DAG was executed

Positive