Page History

Test Case ID	Test Objective	Test Steps	Expected Outcome	Test Type	Execution
TC-001	Verify DAG is scheduled based on

external event

asset watcher

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

Code Block

language	py

file_path

trigger =

"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("

example

sqs_asset", watchers=[
    AssetWatcher(name="

file

sqs_asset_

trigger

watcher", trigger=trigger)
])

with DAG(
    dag_id="

test

example_

asset

sqs_

with_watchers

watcher",
    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)

The DAG test_asset_with_watchers was

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 executed successfully.

Positive

Pass

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_path

trigger =

"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("

example

sqs_asset")

with DAG(
    dag_id="

test

example_

asset

sqs_

with_watchers

watcher",
    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

Wait 30 seconds and verify DAG test_asset_with_watchers

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
Wait 1 minute and verify DAG example_sqs_watcher has not been triggered

The DAG

test

example_

asset_with_watchers was

sqs_watcher was not executed

Negative

Pass

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_1

trigger1 =

"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("

example

sqs_asset", watchers=[
    AssetWatcher(name="

file

sqs_asset_

trigger

watcher1", trigger=

trigger_1

trigger1),
    AssetWatcher(name="

file

sqs_asset_

trigger

watcher2", trigger=

trigger_2

trigger2)
])

with DAG(
    dag_id="example_

asset

sqs_

with_watchers

watcher",
    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)

The DAG test_asset_with_watchers was

= 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 executed successfully twice (once per event).

Positive

Pass

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_path

trigger =

"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("

example

sqs_asset", watchers=[
    AssetWatcher(name="

file

sqs_asset_

trigger

watcher", trigger=trigger)
])

with DAG(
    dag_id="

test

example_

asset

sqs_

with_watchers

watcher",
    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()Wait 30 seconds and verify DAG test_asset_with_watchers

task)

Enable DAGs test_create_file

Disable/pause DAG test_asset_with_watchers

Trigger DAG test_create_file

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
Wait 1 minute and verify DAG example_sqs_watcher

has not been triggered

The DAG

test

example_

asset_with_watchers was

`sqs_watcher` was not executed	Negative	Pass
TC-005	Verify

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

there 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 asset

using 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 =

asset = Asset("example_asset", watchers=[
 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

the DAG 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

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

Parsing error when the DAG was parsed

Negative

Pass

The DAG was not executed

Negative

Space shortcuts

Page tree

Versions Compared

Old Version 3

New Version Current

Key