Page History

...

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:

Code Block

language	py

file_pathtrigger = "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)
SqsSensorTrigger(sqs_queue="https://sqs.us-east-1.amazonaws.com/0123456789/MyQueue")
asset = Asset("examplesqs_asset", watchers=[
    AssetWatcher(name="filesqs_asset_triggerwatcher", trigger=trigger)
])

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

    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)

task)

Update the value of sqs_queue on line 1 to use one of your SQS queue. If do not have a SQS queue defined in your AWS account, you need to create one.
Enable DAG example_sqs_watcher
Send a message to the SQS queue referenced by sqs_queue . You can do it through the AWS console or with the CLI: aws sqs send-message --queue-url https://sqs.us-east-1.amazonaws.com/0123456789/MyQueue --message-body test
Verify DAG example_sqs_watcher has been triggered (it can take up to one minute to be triggered)

The DAG example_sqs_watcher was 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 Create a watcher without associating it to the asset. Example:

Code Block

language	py

file_pathtrigger = "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())
 AssetWatcher(name="sqs_asset_watcher", SqsSensorTrigger(sqs_queue="https://sqs.us-east-1.amazonaws.com/0123456789/MyQueue"))
asset = Asset("examplesqs_asset")

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

    chain(delete_file())task)

Update the value of sqs_queue on line 1 to use one of your SQS queue. If do not have a SQS queue defined in your AWS account, you need to create one.
Enable DAG example_sqs_watcher
Send a message to the SQS queue referenced by sqs_queue . You can do it through the AWS console or with the CLI: aws sqs send-message --queue-url https://sqs.us-east-1.amazonaws.com/0123456789/MyQueue --message-body test
Enable DAGs test_create_file and test_asset_with_watchers
Trigger DAG test_create_file
Wait 1 minute and verify DAG example_sqs_watcher
Wait 30 seconds and verify DAG test_asset_with_watchers has not been triggered

The DAG testexample_asset_with_watchers was sqs_watcher 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:

Code Block

language	py

file_path_1trigger1 = "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)
SqsSensorTrigger(sqs_queue="https://sqs.us-east-1.amazonaws.com/0123456789/MyQueue1")
trigger2 = SqsSensorTrigger(sqs_queue="https://sqs.us-east-1.amazonaws.com/0123456789/MyQueue2")

asset = Asset("examplesqs_asset", watchers=[
    AssetWatcher(name="filesqs_asset_triggerwatcher1", trigger=trigger_1trigger1),
    AssetWatcher(name="filesqs_asset_triggerwatcher2", trigger=trigger_2trigger2)
])

with DAG(
    dag_id="example_assetsqs_with_watcherswatcher",
    schedule=[asset],
    catchup=False,
):
    @task
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)

= EmptyOperator(task_id="task")

    chain(task)

Update the values of sqs_queue on line 1 and 2 to use two different SQS queues. If do not have two SQS queues defined in your AWS account, you need to create them.
Enable DAG example_sqs_watcher
Send a message to the first SQS queue referenced by sqs_queue . You can do it through the AWS console or with the CLI: aws sqs send-message --queue-url https://sqs.us-east-1.amazonaws.com/0123456789/MyQueue1 --message-body test
Verify DAG example_sqs_watcher has been triggered (it can take up to one minute to be triggered)
Send a message to the second SQS queue referenced by sqs_queue . You can do it through the AWS console or with the CLI: aws sqs send-message --queue-url https://sqs.us-east-1.amazonaws.com/0123456789/MyQueue2 --message-body test
Verify DAG example_sqs_watcher has been triggered (it can take up to one minute to be triggered)

The DAG example_sqs_watcher was 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:

Code Block

language	py

file_pathtrigger = "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=10SqsSensorTrigger(sqs_queue="https://sqs.us-east-1.amazonaws.com/0123456789/MyQueue")
asset = Asset("examplesqs_asset", watchers=[
    AssetWatcher(name="filesqs_asset_triggerwatcher", trigger=trigger)
])

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

= EmptyOperator(task_id="task")

    chain(delete_file(task))

Update the value of sqs_queue on line 1 to use one of your SQS queue. If do not have a SQS queue defined in your AWS account, you need to create one.
Pause example_sqs_watcher
Send a message to the SQS queue referenced by sqs_queue . You can do it through the AWS console or with the CLI: aws sqs send-message --queue-url https://sqs.us-east-1.amazonaws.com/0123456789/MyQueue --message-body test
Enable DAGs test_create_file
Disable/pause DAG test_asset_with_watchers
Trigger DAG test_create_file
Wait 1 minute and verify DAG example_sqs_watcher
Wait 30 seconds and verify DAG test_asset_with_watchers has not been triggered

The DAG testexample_asset_with_watchers was sqs_watcher was not executed

Negative

TC-005

Verify DAG is not scheduled when watcher is not associated to the assetthere is a parsing error when a wrong trigger is used

Create a DAG scheduled based on asset. Create a watcher without associating it to the assetusing a trigger that does not extend BaseEventTrigger. Example:

Code Block

language	py

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   @task
    def delete_file():
        if os.path.exists(file_path):
            os.remove(file_path)= EmptyOperator(task_id="task")	

    chain(delete_file(task))

Enable DAGs test_create_file and the 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 was not executed

Negative

TC-006

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

Verify you get an import error "The trigger used to watch an asset must inherit ``BaseEventTrigger``"

Parsing error when the DAG was parsed

NegativePositive

Space shortcuts

Page tree

Versions Compared

Old Version 5

New Version 6

Key