Repository: devlace/pytest-adf (language: python)

Issue:
Support testing storage event triggered ADF pipelines 
As we discussed previously, it's value-added to also support testing storage event triggered ADF pipelines, therefore I create this issue to track this new feature.

And below is the rough ideas.
- For testing blob created event triggered pipelines, we'll add a new fixture to accept trigger file path, trigger name, blob event type and storage account relevant arguments to do the test with below steps:
  1. upload declared event trigger file to target container of specified storage account.
  2. get _eTag_ and _last modified time_ from the just uploaded event triggered file.
  3. query trigger-runs by _last modified time_ filter and look for the one with exactly same _eTag_ from event-payload of trigger-run.
  4. get triggered pipeline run id from the hit trigger-run, then poll the pipeline run status with the given run id, just like what the normal fixture does.
  
- For testing blob deleted event triggered pipelines, reuse the same new added fixture, but pass BlobDeleted to blob event type argument, then
  1. delete declared event trigger file from target storage account.
  2. since there's no eTag for file deletion, instead we query trigger-runs only by the file deletion time filter and pick the latest one with _DeleteBlob_ data-api action from event-payload of trigger-run.
  3.  get triggered pipeline run id from the hit trigger-run, then same as above, polling the pipeline run status with the given run id.

BTW, for scenarios like one ADF pipeline triggers another ones, since it could be quite different from case to case like below for instance, it's not feasible to cover all of them in _pytest-adf_ plugin.
- pipeline A -> pipeline B
- pipeline A -> pipeline B -> pipeline C
- pipeline A -> pipeline B
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;-> Pipeline C -> pipeline D
- ...

Instead, users could take use of the storage-event-trigger pipeline testing feature to do the test on separated stages one by one, and if each stage of the whole chained pipelines works as expected, then the whole one should work well too.  


Discussion / hints:

Fix patch:
diff --git a/README.md b/README.md
index 13022fa..e3697e3 100644
--- a/README.md
+++ b/README.md
@@ -37,6 +37,65 @@ For an example of how to use this in an overall Modern Data Warehouse solution a
 
 For additional usage information, see [caching pipeline runs](#Caching-pipeline-runs).
 
+### Testing Storage Event Triggered Pipelines
+For testing storage event triggered pipelines, please use another fixture `adf_storage_event_triggered_pipeline_run`.
+
+**Storage Event: BlobCreated**
+
+For testing blob created event triggered pipelines, unlike `adf_pipeline_run` fixture, below extra arguments are required.
+- trigger name, the configured trigger of the test target pipeline
+- trigger file path, where to store the event trigger file in container
+- storage account container name, together with trigger file path, the fixture knows where to upload the event trigger file
+- local trigger file path, tells the fixture which even trigger file to upload
+- blob event type, `BlobCreated` is the default value
+
+The fixture firstly helps uploading declared storage event trigger file to target container of storage account to trigger the storage-event-triggered pipeline. Then it will block and poll related triggered pipeline run till completion* before returning. Pipeline run completion definition is same as above.
+
+```python
+PIPELINE_NAME = "my_blobcreated_event_triggered_pipeline"
+TRIGGER_NAME = "storage_blob_created_event_trigger"
+TRIGGER_FILE_PATH = "event-check/TR_BLOB_CREATED_STORAGE_EVENT.txt"
+CONTAINER_NAME = "event-trigger"
+LOCAL_TRIGGER_FILE_PATH = "./resources/TR_BLOB_CREATED_STORAGE_EVENT.txt"
+
+def test_pipeline_succeeded(adf_storage_event_triggered_pipeline_run):
+    this_run = adf_storage_event_triggered_pipeline_run(PIPELINE_NAME,
+                                                        TRIGGER_NAME,
+                                                        TRIGGER_FILE_PATH,
+                                                        CONTAINER_NAME,
+                                                        LOCAL_TRIGGER_FILE_PATH,
+                                                        blob_event_type="BlobCreated") # Default value
+    
+    assert this_run.status == "Succeeded"
+```
+
+**Storage Event: BlobDeleted**
+
+For testing blob deleted event triggered pipelines, please refer to below sample, which just skips the local-trigger-file-path argument and assigns `BlobDeleted` value to *blob_event_type* explicitly, then the fixture will help to delete declared event trigger file from the target container to trigger the specified pipeline run.
+
+```python
+PIPELINE_NAME = "my_blobdeleted_event_triggered_pipeline"
+TRIGGER_NAME = "storage_blob_deleted_event_trigger"
+TRIGGER_FILE_PATH = "event-check/TR_BLOB_DELETED_STORAGE_EVENT.txt"
+CONTAINER_NAME = "event-trigger"
+
+def test_pipeline_succeeded(adf_storage_event_triggered_pipeline_run):
+    this_run = adf_storage_event_triggered_pipeline_run(PIPELINE_NAME,
+                                                        TRIGGER_NAME,
+                                                        TRIGGER_FILE_PATH,
+                                                        CONTAINER_NAME,
+                                                        blob_event_type="BlobDeleted")
+    
+    assert this_run.status == "Succeeded"
+```
+
+Note that 
+- since pipeline parameters are required to set default values when adding/editing pipeline triggers, therefore there's no *run_inputs* argument for this fixture.
+- since this fixture also helps to upload/delete event trigger file to/from storage account, we need additional storage account related environment variables, more details please refers to below *Environment Variables* section.
+- please also provide the target event trigger file as test resources when testing `BlobCreated` event trigger pipelines.
+- this fixture also supports caching pipeline runs, just append those two required arguments to enable it, for more details, please see [caching pipeline runs](#Caching-pipeline-runs).
+
+
 ## Configuration
 
 You need to provide pytest-adf with the necessary configuration to connect to your Azure Data Factory. You can provide it via Environment Variables or as pytest command line variables. Command line variables take precedence over Environment Variables.
@@ -50,6 +109,8 @@ You need to provide pytest-adf with the necessary configuration to connect to yo
 - **AZ_RESOURCE_GROUP_NAME** - Azure Resource Group name where Azure Data Factory is hosted.
 - **AZ_DATAFACTORY_NAME** - Name of the Azure Data Factory.
 - **AZ_DATAFACTORY_POLL_INTERVAL_SEC** - Optional. Seconds between poll intervals to check for status of the triggered run.
+- **AZ_STORAGE_ACCOUNT_NAME** - Optional. Only required if testing storage event triggered Azure Data Factory pipelines.
+- **AZ_STORAGE_ACCOUNT_KEY** - Optional. Only required if testing storage event triggered Azure Data Factory pipelines.
 
 For more information on how to create an Azure AD service principal, see [here](https://docs.microsoft.com/en-us/azure/active-directory/develop/howto-create-service-principal-portal).
 
@@ -58,6 +119,7 @@ For more information on how to create an Azure AD service principal, see [here](
 Alternatively, you can pass these like so:
 
 ```
+# Last two parameters are required only for testing storage event triggered pipelines
 pytest
     --sp_id=my_sp_id \
     --sp_password=my_sp_pass \
@@ -65,7 +127,9 @@ pytest
     --sub_id=my_s_id \
     --rg_name=my_rg \
     --adf_name=my_adf \
-    --poll_interval=20
+    --poll_interval=20 \
+    --storage_account_name=my_blob_storage_account \
+    --storage_account_key=my_blob_storage_account_key
 ```
 
 ## Caching pipeline runs
diff --git a/requirements.txt b/requirements.txt
index 71a912d..5469a6b 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -2,4 +2,5 @@ pytest
 azure-identity
 azure-mgmt-resource
 azure-mgmt-datafactory
+azure-storage-blob
 msrestazure
\ No newline at end of file
diff --git a/setup.py b/setup.py
index 690b265..545243f 100644
--- a/setup.py
+++ b/setup.py
@@ -27,7 +27,7 @@ def read(fname):
     packages=find_packages(where="src"),
     package_dir={"": "src"},
     python_requires='>=3.5',
-    install_requires=['pytest>=3.5.0', 'azure-identity', 'azure-mgmt-datafactory'],
+    install_requires=['pytest>=3.5.0', 'azure-identity', 'azure-mgmt-datafactory', 'azure-storage-blob'],
     classifiers=[
         'Development Status :: 4 - Beta',
         'Framework :: Pytest',


Test patch:
diff --git a/src/pytest_adf/pytest_adf.py b/src/pytest_adf/pytest_adf.py
index 042389a..97e97e0 100644
--- a/src/pytest_adf/pytest_adf.py
+++ b/src/pytest_adf/pytest_adf.py
@@ -4,12 +4,39 @@
 import os
 import time
 import logging
+import json
+from datetime import timedelta
+from datetime import datetime
+from pytest_adf.pytest_adf_enums import BlobEventType
+from pytest_adf.pytest_adf_enums import TriggerRunDataAPIAction
 from azure.identity import ClientSecretCredential, AzureCliCredential
 from azure.mgmt.datafactory import DataFactoryManagementClient
+from azure.mgmt.datafactory.models._models import RunQueryFilter
+from azure.mgmt.datafactory.models._models import RunQueryOrderBy
+from azure.mgmt.datafactory.models._models import RunFilterParameters
+from azure.mgmt.datafactory.models._data_factory_management_client_enums import RunQueryOrderByField
+from azure.mgmt.datafactory.models._data_factory_management_client_enums import RunQueryOrder
+from azure.mgmt.datafactory.models._data_factory_management_client_enums import RunQueryFilterOperand
+from azure.mgmt.datafactory.models._data_factory_management_client_enums import RunQueryFilterOperator
+from azure.storage.blob import BlobServiceClient
 
 LOG = logging.getLogger(__name__)
 
-OPTIONAL_ARGS = ["AZ_SERVICE_PRINCIPAL_ID", "AZ_SERVICE_PRINCIPAL_SECRET"]
+OPTIONAL_ARGS = ["AZ_SERVICE_PRINCIPAL_ID",
+                 "AZ_SERVICE_PRINCIPAL_SECRET",
+                 "AZ_STORAGE_ACCOUNT_NAME",
+                 "AZ_STORAGE_ACCOUNT_KEY"]
+OPTIONAL_STORAGE_ACCOUNT_ARGS = ["AZ_STORAGE_ACCOUNT_NAME", "AZ_STORAGE_ACCOUNT_KEY"]
+
+# Determines the maximum amount of time pytest-adf will look for the triggered pipeline run
+# within the FETCHING_TRIGGER_RUNS_MAX_TIME_IN_SECONDS
+FETCHING_TRIGGER_RUNS_MAX_TIME_IN_SECONDS = 180
+
+# Determines the ADF query time range to search for the corresponding trigger ADF pipeline
+# starting from when the blob file is uploaded/deleted.
+# For example, if file is uploaded/deleted at 01:00, this will search for the trigger-runs
+# between 01:00 and 01:03.
+FETCHING_TRIGGER_RUNS_TIME_RANGE_IN_MINUTES = 3
 
 
 def pytest_addoption(parser):
@@ -56,6 +83,18 @@ def pytest_addoption(parser):
         dest='AZ_DATAFACTORY_POLL_INTERVAL_SEC',
         help='Poll interval to query Data Factory status in seconds'
     )
+    group.addoption(
+        '--storage_account_name',
+        action='store',
+        dest='AZ_STORAGE_ACCOUNT_NAME',
+        help='Azure Storage Account name; optional, only required if testing storage event triggered ADF pipelines'
+    )
+    group.addoption(
+        '--storage_account_key',
+        action='store',
+        dest='AZ_STORAGE_ACCOUNT_KEY',
+        help='Azure Storage Account key; optional, only required if testing storage event triggered ADF pipelines'
+    )
 
 
 @pytest.fixture(scope="session")
@@ -78,7 +117,9 @@ def _default(val, default):
         "AZ_DATAFACTORY_NAME": _default(r.AZ_DATAFACTORY_NAME, os.getenv("AZ_DATAFACTORY_NAME")),
         "AZ_DATAFACTORY_POLL_INTERVAL_SEC": int(_default(r.AZ_DATAFACTORY_POLL_INTERVAL_SEC,
                                                          os.getenv("AZ_DATAFACTORY_POLL_INTERVAL_SEC",
-                                                                   DEFAULT_AZ_DATAFACTORY_POLL_INTERVAL_SEC)))
+                                                                   DEFAULT_AZ_DATAFACTORY_POLL_INTERVAL_SEC))),
+        "AZ_STORAGE_ACCOUNT_NAME": _default(r.AZ_STORAGE_ACCOUNT_NAME, os.getenv("AZ_STORAGE_ACCOUNT_NAME")),
+        "AZ_STORAGE_ACCOUNT_KEY": _default(r.AZ_STORAGE_ACCOUNT_KEY, os.getenv("AZ_STORAGE_ACCOUNT_KEY"))
     }
     # Ensure all required config is set.
     for config_key, value in config.items():
@@ -95,11 +136,29 @@ def adf_client(adf_config):
         credentials = AzureCliCredential()
     else:
         credentials = ClientSecretCredential(client_id=adf_config["AZ_SERVICE_PRINCIPAL_ID"],
-                                         client_secret=adf_config["AZ_SERVICE_PRINCIPAL_SECRET"],
-                                         tenant_id=adf_config["AZ_SERVICE_PRINCIPAL_TENANT_ID"])
+                                             client_secret=adf_config["AZ_SERVICE_PRINCIPAL_SECRET"],
+                                             tenant_id=adf_config["AZ_SERVICE_PRINCIPAL_TENANT_ID"])
     return DataFactoryManagementClient(credentials, adf_config["AZ_SUBSCRIPTION_ID"])
 
 
+@pytest.fixture(scope="session")
+def blob_service_client(adf_config):
+    """Creates an BlobServiceClient object"""
+    for arg in OPTIONAL_STORAGE_ACCOUNT_ARGS:
+        if adf_config[arg] is None or adf_config[arg].isspace():
+            raise ValueError("Required config for testing storage-event-triggered "
+                             "ADF pipelines: {} is not set properly.".format(arg))
+
+    account_name = adf_config["AZ_STORAGE_ACCOUNT_NAME"]
+    account_key = adf_config["AZ_STORAGE_ACCOUNT_KEY"]
+
+    service_client = BlobServiceClient(
+        account_url="https://" + account_name + ".blob.core.windows.net",
+        credential=account_key
+    )
+    return service_client
+
+
 @pytest.fixture(scope="session")
 def adf_pipeline_run(adf_client, adf_config):
     """Factory function for triggering an ADF Pipeline run given run_inputs, and polls for results."""
@@ -161,3 +220,195 @@ def _poll_adf_until(adf_client, az_resource_group, adf_name, pipeline_run_id,
         pipeline_run_status = pipeline_run.status  # InProgress, Succeeded, Queued, TimedOut, Failed, Cancelled
         time.sleep(poll_interval)
     return pipeline_run
+
+
+@pytest.fixture(scope="session")
+def adf_storage_event_triggered_pipeline_run(adf_client, adf_config, blob_service_client):
+    """Factory function for triggering an storage-event-triggered ADF Pipeline run
+    by uploading target trigger file to storage account, and polls for results.
+    """
+    # Because ADF pipeline can be expensive to execute, you may want to cache pipeline_runs.
+    # Cache pipeline runs are identified by pipeline_name and cached_run_name
+    cached_pipeline_runs = {}
+
+    def make_adf_storage_event_triggered_pipeline_run(pipeline_name,
+                                                      trigger_name,
+                                                      trigger_file_path,
+                                                      container_name,
+                                                      local_trigger_file_path="",
+                                                      blob_event_type=BlobEventType.BLOB_CREATED,
+                                                      cached_run_name: str = "",
+                                                      rerun=False):
+
+        # Check if pipeline run was already previous executed, simply return cached run
+        if (cached_run_name != ""
+            and not rerun
+            and pipeline_name in cached_pipeline_runs.keys()
+                and cached_run_name in cached_pipeline_runs[pipeline_name].keys()):
+            LOG.info("""Previously executed cached pipeline run found for pipeline: {pipeline}
+                     with run_name: {run_name}""".format(pipeline=pipeline_name, run_name=cached_run_name))
+            return cached_pipeline_runs[pipeline_name][cached_run_name]
+
+        az_resource_group = adf_config["AZ_RESOURCE_GROUP_NAME"]
+        adf_name = adf_config["AZ_DATAFACTORY_NAME"]
+
+        # Upload/delete event trigger file to/from target storage account, depends on given blob event type
+        if blob_event_type == BlobEventType.BLOB_CREATED:
+            blob_properties = _upload_event_trigger_file(blob_service_client,
+                                                         trigger_file_path,
+                                                         container_name,
+                                                         local_trigger_file_path)
+        elif blob_event_type == BlobEventType.BLOB_DELETED:
+            blob_properties = _delete_event_trigger_file(blob_service_client,
+                                                         trigger_file_path,
+                                                         container_name)
+
+        # Fetch triggered pipeline run id from trigger run
+        triggered_pipeline_run_id = _get_triggered_pipeline_run_id(adf_client,
+                                                                   az_resource_group,
+                                                                   adf_name,
+                                                                   pipeline_name,
+                                                                   trigger_name,
+                                                                   blob_event_type,
+                                                                   blob_properties["last_modified"],
+                                                                   blob_properties["etag"])
+
+        pipeline_run = _poll_adf_until(adf_client,
+                                       az_resource_group,
+                                       adf_name,
+                                       triggered_pipeline_run_id,
+                                       poll_interval=int(adf_config["AZ_DATAFACTORY_POLL_INTERVAL_SEC"]))
+
+        # Store run in cache, if run_name is specified
+        if cached_run_name != "":
+            LOG.info("Caching pipeline: {pipeline} with run_name: {run_name}".format(
+                pipeline=pipeline_name, run_name=cached_run_name))
+            cached_pipeline_runs[pipeline_name] = {}
+            cached_pipeline_runs[pipeline_name][cached_run_name] = pipeline_run
+        return pipeline_run
+
+    return make_adf_storage_event_triggered_pipeline_run
+
+
+def _upload_event_trigger_file(blob_service_client, trigger_file_path, container_name, local_trigger_file_path):
+    """Helper function that upload event trigger file to target container of storage account.
+
+    Args:
+        blob_service_client (BlobServiceClient): Azure Storage Account service client
+        trigger_file_path (str): Path of trigger file in storage account container
+        container_name (str): Name of target container to upload the trigger file
+        local_trigger_file_path (str): Local file path of the target trigger file.
+
+    Returns:
+        uploaded_blob_properties, including last modified time and eTag
+    """
+    blob_client = blob_service_client.get_blob_client(container_name, trigger_file_path)
+
+    # Upload the event trigger file to starage account
+    with open(local_trigger_file_path, "rb") as data:
+        blob_client.upload_blob(data, overwrite=True)
+
+    # Fetch eTag and last_modiefied from the uploaded event trigger file
+    etag = blob_client.get_blob_properties().etag
+    last_modified = blob_client.get_blob_properties().last_modified
+    blob_properties = {
+        "etag": etag.replace("\"", ""),
+        "last_modified": last_modified
+    }
+
+    return blob_properties
+
+
+def _delete_event_trigger_file(blob_service_client, trigger_file_path, container_name):
+    """Helper function that delete event trigger file from target container of storage account.
+    Warning! If target deleting file is not there, BlobNotFound exception will be raised.
+
+    Args:
+        blob_service_client (BlobServiceClient): Azure Storage Account service client
+        trigger_file_path (str): Path of trigger file in storage account container
+        container_name (str): Name of target container to upload the trigger file.
+
+    Returns:
+        blob_properties, including last modified time
+    """
+    container_client = blob_service_client.get_container_client(container_name)
+
+    # Delete target event trigger file from storage account
+    container_client.delete_blob(trigger_file_path)
+    deleted_time = datetime.utcnow()
+    blob_properties = {
+        "etag": "",
+        "last_modified": deleted_time
+    }
+
+    return blob_properties
+
+
+def _get_triggered_pipeline_run_id(adf_client,
+                                   az_resource_group,
+                                   adf_name,
+                                   pipeline_name,
+                                   trigger_name,
+                                   blob_event_type,
+                                   trigger_file_last_modified_time,
+                                   trigger_file_etag,
+                                   poll_interval=5):
+    """Helper functioin to get triggered pipeline run id by looking for expected trigger-run with
+    last modified time and eTag of target event trigger file.
+    Warning! TimeoutError exception may be raised if it still fails to find the expected trigger-run
+    after the configured max time.
+    """
+    max_iterations = int(FETCHING_TRIGGER_RUNS_MAX_TIME_IN_SECONDS / poll_interval)
+    iteration = 0
+    found_trigger_run = False
+    triggered_pipeline_run_id = ""
+    while (found_trigger_run is False):
+        LOG.info("Polling trigger-run by trigger name {}, last modified time and eTag "
+                 "of target trigger file.".format(trigger_name))
+
+        if not found_trigger_run:
+            run_query_filter = RunQueryFilter(operand=RunQueryFilterOperand.TRIGGER_NAME,
+                                              operator=RunQueryFilterOperator.EQUALS,
+                                              values=[trigger_name])
+            run_query_order_by = RunQueryOrderBy(order_by=RunQueryOrderByField.TRIGGER_RUN_TIMESTAMP,
+                                                 order=RunQueryOrder.DESC)
+            filter_params = RunFilterParameters(last_updated_after=trigger_file_last_modified_time,
+                                                last_updated_before=trigger_file_last_modified_time + timedelta(
+                                                    minutes=FETCHING_TRIGGER_RUNS_TIME_RANGE_IN_MINUTES
+                                                    ),
+                                                filters=[run_query_filter],
+                                                order_by=[run_query_order_by])
+
+            trigger_runs_query_response = adf_client.trigger_runs.query_by_factory(az_resource_group,
+                                                                                   adf_name,
+                                                                                   filter_params)
+            trigger_runs = trigger_runs_query_response.value
+            if (len(trigger_runs) > 0):
+                for trigger_run in trigger_runs:
+                    event_payload = trigger_run.properties.get("EventPayload")
+                    event_payload_json = json.loads(event_payload.replace("\r\n", ""))
+                    etag = event_payload_json["data"]["eTag"]
+                    api = event_payload_json["data"]["api"]
+
+                    if ((blob_event_type == BlobEventType.BLOB_CREATED
+                         and api == TriggerRunDataAPIAction.PUT_BLOB
+                         and etag == trigger_file_etag)
+                        or
+                        (blob_event_type == BlobEventType.BLOB_DELETED
+                         and api == TriggerRunDataAPIAction.DELETE_BLOB)):
+                        triggered_pipeline_run_id = trigger_run.triggered_pipelines.get(pipeline_name)
+                        found_trigger_run = True
+                        LOG.info("Found expected triggered pipeline with run id {}".format(triggered_pipeline_run_id))
+                        break
+                    else:
+                        LOG.info("Not found expected triggered pipeline run, will try again in next poll.")
+            else:
+                LOG.info("Not found any trigger run in this query, will try again in next poll.")
+
+        iteration += 1
+        if iteration > max_iterations:
+            raise TimeoutError("Time out when fetching expected trigger-run, "
+                               "please check relevant trigger configurations!")
+
+        time.sleep(poll_interval)
+    return triggered_pipeline_run_id
diff --git a/src/pytest_adf/pytest_adf_enums.py b/src/pytest_adf/pytest_adf_enums.py
new file mode 100644
index 0000000..86208a5
--- /dev/null
+++ b/src/pytest_adf/pytest_adf_enums.py
@@ -0,0 +1,17 @@
+from enum import Enum
+
+
+class BlobEventType(str, Enum):
+    """Specify supported blob event types of Azure Data Factory Trigger
+    """
+
+    BLOB_CREATED = "BlobCreated"
+    BLOB_DELETED = "BlobDeleted"
+
+
+class TriggerRunDataAPIAction(str, Enum):
+    """Specify available api actions from event payload of a trigger-run
+    """
+
+    PUT_BLOB = "PutBlob"
+    DELETE_BLOB = "DeleteBlob"