Test Case ID

Test Objective

Test Steps

Expected Outcome

Test Type

Execution

TC-001

Verify DAG is scheduled based on asset watcher

  1. 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)

  1. 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.

  2. Enable DAG example_sqs_watcher 
  3. 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
  4. 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


  1. 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)

  1. 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.

  2. Enable DAG example_sqs_watcher 
  3. 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
  4. 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

  1. 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)

  1. 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.

  2. Enable DAG example_sqs_watcher 
  3. 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
  4. Verify DAG example_sqs_watcher   has been triggered (it can take up to one minute to be triggered)
  5. 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
  6. 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
  1. 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)

  1. 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.

  2. Pause example_sqs_watcher
  3. 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
  4. Wait 1 minute and verify DAG example_sqs_watcher  has not been triggered
The DAG example_sqs_watcher was not executedNegativePass

TC-005

Verify there is a parsing error when a wrong trigger is used

  1. 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)
  1. Enable the DAG test_asset_with_watchers 
  2. 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