Versions Compared

Old Version 1

changes.mady.by.user Vincent BECK

Saved on

compared with

New Version Current

changes.mady.by.user Rahul Vats

Saved on

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.


Test Case ID

Test Objective

Test Steps

Expected Outcome

Test Type

Execution

TC-001

Verify DAG is scheduled based on

external event

asset watcher

  1. Create a DAG scheduled based on asset. Associate a watcher to the asset. Example:
Code Block
languagepy
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())

  1. Enable DAGs test_create_file  and test_asset_with_watchers 

  2. Trigger DAG test_create_file
  3. Verify DAG test_asset_with_watchers  has been triggered (wait few seconds if not)

The DAG was executed successfully.

Positive
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.
Do not Associate
  1. Create a watcher without associating it to the asset. Example:
Code Block
languagepy
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 

    • 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
    Trigger DAG test_create_file
    2. Wait 1 minute and verify DAG
    test
    1. example_
    asset
    1. sqs_
    with_watchers
    1. watcher  has not been triggered

    The DAG

    was

    example_sqs_watcher was not executed

    successfully

    Negative

    Pass

    TC-003

    Verify DAG is scheduled based on two different external events

    1. Create a DAG scheduled based on asset. Associate
    a
    1. two watchers to the asset. Example:
    Code Block
    languagepy
    file_path = "test_file" with DAG(
    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=[
    dag_id
    AssetWatcher(name="
    test
    sqs_
    create
    asset_
    file
    watcher1", trigger=trigger1),
    catchup=False, ): @task
    AssetWatcher(name="sqs_asset_watcher2", trigger=trigger2)
])

with DAG(
    def create_file():
    dag_id="example_sqs_watcher",
    schedule=[asset],
    with open(file_path, "w") as file
    catchup=False,
):
    task
    file.write("This is an example file.\n
    = 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

    _file()) trigger = FileTrigger(filepath=file_path, poke_interval=10

    1. 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:
    Code Block
    languagepy
    trigger = 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()
    task)

    1. Enable DAGs test_create_file  and test_asset_with_watchers 

    2. Trigger DAG test_create_file
    3. Verify DAG test_asset_with_watchers  has been triggered (wait few seconds if not)

    The API processes the request successfully, and the CLI outputs the correct result.

    Positive

    TC-004

    Verify that API integration enforces token-based authentication. (Automated Flow, Unauthenticated)
    1. Ensure there isn't any authentication from CLI ~/.airflow/config

    2. Execute the CLI command (e.g., airflow dags list --token=<invalid_token> OR AIRFLOW_CLI_TOKEN=<TOKEN> airflow dags list).

    3. Capture the API request sent by the CLI.

    4. Verify CLI displays an appropriate error message.

    The API returns an authentication error, and the CLI displays an appropriate error message.Negative

    TC-005

    Verify that API integration respects RBAC policies.

    1. Assign a user to a role with limited permissions (e.g., don't have permission can view DAGs ).

    2. Ensure CLI is authenticated.

    3. Execute CLI command airflow dags list

    4. Verify CLI displays an appropriate error message.

    Restricted actions fail with an authorization error.

    Negative

    TC-006

    Verify that API integration respects RBAC policies.

    1. Assign a user to a role with limited permissions (e.g., don't have permission can view DAGs ).

    2. Ensure CLI is authenticated.

    3. Execute CLI command airflow dags list

    4. Verify response executed the requested command properly dags list .

    Allowed actions succeed and return the correct data.

    Positive

    TC-007

    Ensure that the CLI handles API downtime gracefully.

    1. Simulate API downtime (e.g., stop the API service or block network access).

    2. Ensure CLI is authenticated.

    3. Execute a CLI command (e.g., airflow dags list).

    4. Observe the CLI behaviour and error messages.

    The CLI displays a clear and user-friendly error message indicating that the API is unavailable.

    Edge

    TC-008

    Ensure that sensitive data is not exposed in CLI outputs or API responses.

    1. Ensure CLI is authenticated.

    2. Execute a CLI command that interacts with sensitive data (e.g., airflow connections list).

    3. Review the CLI output for any sensitive information (e.g., passwords, tokens).

    4. Capture the corresponding API response and check for sensitive data exposure.

    Neither the CLI output nor the API response exposes sensitive data. Placeholder values (e.g., ***) are used where applicable.

    Positive

    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:
    Code Block
    languagepy
    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

    TC-009

    Ensure that CLI commands trigger the correct API calls and logs reflect the endpoint usage.

    1. Ensure CLI is authenticated.

    2. Execute a CLI command (e.g., airflow dags list).

    3. Access the API logs to identify the endpoint called.

    4. Verify that the logged API call matches the expected behaviour for the CLI command.

    The API logs show the correct endpoint is triggered.

    Positive