Test Case ID | Test Objective | Test Steps | Expected Outcome | Test Type | Execution |
|---|
TC-001 | Verify DAG is scheduled based on asset watcher | - Create a DAG scheduled based on asset. Associate a watcher to the asset. Example:
trigger = SqsSensorTrigger(sqs_queue="https://sqs.us-east-1.amazonaws.com/0123456789/MyQueue")
asset = Asset("sqs_asset", watchers=[
AssetWatcher(name="sqs_asset_watcher", trigger=trigger)
])
with DAG(
dag_id="example_sqs_watcher",
schedule=[asset],
catchup=False,
):
task = EmptyOperator(task_id="task")
chain(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. Create a watcher without associating it to the asset. Example:
trigger = AssetWatcher(name="sqs_asset_watcher", SqsSensorTrigger(sqs_queue="https://sqs.us-east-1.amazonaws.com/0123456789/MyQueue"))
asset = Asset("sqs_asset")
with DAG(
dag_id="example_sqs_watcher",
schedule=[asset],
catchup=False,
):
task = EmptyOperator(task_id="task")
chain(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 example_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:
trigger1 = 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("sqs_asset", watchers=[
AssetWatcher(name="sqs_asset_watcher1", trigger=trigger1),
AssetWatcher(name="sqs_asset_watcher2", trigger=trigger2)
])
with DAG(
dag_id="example_sqs_watcher",
schedule=[asset],
catchup=False,
):
task = 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 | Fail |
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:
trigger = SqsSensorTrigger(sqs_queue="https://sqs.us-east-1.amazonaws.com/0123456789/MyQueue")
asset = Asset("sqs_asset", watchers=[
AssetWatcher(name="sqs_asset_watcher", trigger=trigger)
])
with DAG(
dag_id="example_sqs_watcher",
schedule=[asset],
catchup=False,
):
task = EmptyOperator(task_id="task")
chain(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 - Wait 1 minute and verify DAG
example_sqs_watcher has not been triggered
| The DAG example_sqs_watcher was not executed | Negative | Pass |
TC-005 | Verify there is a parsing error when a wrong trigger is used | - Create a DAG scheduled based on asset. Create a watcher using a trigger that does not extend
BaseEventTrigger. Example:
file_path = "test_file"
asset = Asset("example_asset", watchers=[
AssetWatcher(name="file_trigger", trigger=FileTrigger(filepath=file_path, poke_interval=10))
])
with DAG(
dag_id="test_asset_with_watchers",
schedule=[asset],
catchup=False,
):
task = EmptyOperator(task_id="task")
chain(task)
- Enable the DAG
test_asset_with_watchers - 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 |