Test Case ID | Test Objective | Test Steps | Expected Outcome | Test Type |
|---|
TC-001 | Verify DAG is scheduled based on external event | - Create a DAG scheduled based on asset. Associate a watcher to the asset. Example:
| Code Block |
|---|
| 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())
trigger = FileTrigger(filepath=file_path, poke_interval=10)
asset = Asset("example_asset", watchers=[AssetWatcher(name="file_trigger", trigger=trigger)])
with DAG(
dag_id="test_asset_with_watchers",
schedule=[asset],
catchup=False,
):
@task
def delete_file():
if os.path.exists(file_path):
os.remove(file_path)
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 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.
Do not - Associate
a - no watcher to the asset. Example:
| Code Block |
|---|
| 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())
asset = Asset("example_asset")
with DAG(
dag_id="test_asset_with_watchers",
schedule=[asset],
catchup=False,
):
@task
def delete_file():
if os.path.exists(file_path):
os.remove(file_path)
chain(delete_file()) |
Enable DAGs test_create_file and test_asset_with_watchers - Trigger DAG
test_create_file - Wait
1 minute - 30 seconds and verify DAG
test_asset_with_watchers has not been triggered
| The DAG was not executed successfully | Negative |
TC-003 | Verify DAG is scheduled based on two different external events | - Create a DAG scheduled based on asset. Associate a two watchers to the asset. Example:
| Code Block |
|---|
| file_path_1 = "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=" |
testexample_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)
asset = Asset("example_asset", watchers=[
AssetWatcher(name="file_trigger", trigger=trigger_1),
AssetWatcher(name="file_trigger", trigger=trigger_2)
])
with DAG(
dag_id=" |
testexample_asset_with_watchers",
schedule=[asset],
catchup=False,
):
@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_ |
fileEnable DAGs test 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 API processes the request successfully, and the CLI outputs the correct resultDAG was executed successfully twice (once per event).
| Positive |
TC-004 | Verify that API integration enforces token-based authentication. (Automated Flow, Unauthenticated) | - Ensure there isn't any authentication from CLI ~/.airflow/config
Execute the CLI command (e.g., airflow dags list --token=<invalid_token> OR AIRFLOW_CLI_TOKEN=<TOKEN> airflow dags list). Capture the API request sent by the CLI. 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. | Assign a user to a role with limited permissions (e.g., don't have permission can view DAGs ). - Ensure CLI is authenticated.
Execute CLI command airflow dags list Verify CLI displays an appropriate error message.
| Restricted actions fail with an authorization error. | Negative |
TC-006 | Verify that API integration respects RBAC policies. | Assign a user to a role with limited permissions (e.g., don't have permission can view DAGs ). - Ensure CLI is authenticated.
Execute CLI command airflow dags list 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. | Simulate API downtime (e.g., stop the API service or block network access). - Ensure CLI is authenticated.
Execute a CLI command (e.g., airflow dags list). 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. | - Ensure CLI is authenticated.
Execute a CLI command that interacts with sensitive data (e.g., airflow connections list). Review the CLI output for any sensitive information (e.g., passwords, tokens). 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 |
TC-009 | Ensure that CLI commands trigger the correct API calls and logs reflect the endpoint usage. | - Ensure CLI is authenticated.
Execute a CLI command (e.g., airflow dags list). Access the API logs to identify the endpoint called. Verify that the logged API call matches the expected behaviour for the CLI command.
| The API logs show the correct endpoint is triggered. | Positive |