ado_artifact_download: add extra artifact field/property filtering

The method/helper will now get a new filtering mechanism, beyond the

  if run.result == "succeeded" and run.state == "completed"

hardcoded one.

Imagine one needs to check for the presence of the following
attributes, with the following values:

    additional_build_checks = {
        "additional_properties" : {
            "templateParameters": {
                "build_tools_image": True,
                "build_base_os": True
    	    }
        }
    }

With this commit, it is now possible. Without it, it would blindly
pick the last succesful build in the pipeline, that could still miss
artifacts we are interested in.

Author: glima

lisa/advanced_tools/ado_artifact_download.py

@@ -3,7 +3,7 @@
 import re
 from pathlib import Path
 from time import time
-from typing import Any, List
+from typing import Any, Dict, List, Union
 
 import requests
 from assertpy import assert_that
@@ -28,6 +28,47 @@ def _check_exists(self) -> bool:
     def can_install(self) -> bool:
         return False
 
+    def _check_attrs_recursive(
+        self, obj: Any, checks: Union[Dict[Any, Any], None]
+    ) -> bool:
+        if not checks:
+            return True
+
+        for key, expected_value in checks.items():
+            if isinstance(obj, dict) and key in obj:
+                current_value = obj[key]
+            elif hasattr(obj, key):
+                current_value = getattr(obj, key)
+            else:
+                found = False
+                if isinstance(obj, dict):
+                    for v in obj.values():
+                        if isinstance(
+                            v, (dict, object)
+                        ) and self._check_attrs_recursive(v, {key: expected_value}):
+                            found = True
+                            break
+                elif hasattr(obj, "__dict__"):
+                    for attr_val in vars(obj).values():
+                        if isinstance(
+                            attr_val, (dict, object)
+                        ) and self._check_attrs_recursive(
+                            attr_val, {key: expected_value}
+                        ):
+                            found = True
+                            break
+                if not found:
+                    return False
+                continue  # skip rest of loop since match was deep
+
+            if isinstance(expected_value, dict):
+                if not self._check_attrs_recursive(current_value, expected_value):
+                    return False
+            elif current_value != expected_value:
+                return False
+
+        return True
+
     def download(
         self,
         personal_access_token: str,
@@ -39,6 +80,7 @@ def download(
         build_name: str = "",
         timeout: int = 600,
         output_path: Path = constants.RUN_LOCAL_WORKING_PATH,
+        additional_build_checks: Union[Dict[Any, Any], None] = None,
     ) -> List[Path]:
         credentials = BasicAuthentication("", personal_access_token)
         connection = Connection(base_url=organization_url, creds=credentials)
@@ -50,7 +92,9 @@ def download(
             pipeline_run = [
                 run
                 for run in pipeline_runs
-                if run.result == "succeeded" and run.state == "completed"
+                if run.result == "succeeded"
+                and run.state == "completed"
+                and self._check_attrs_recursive(run, additional_build_checks)
             ]
             assert_that(len(pipeline_run)).described_as(
                 f"no succeeded and completed run found for pipeline {pipeline_name}"
@@ -61,7 +105,12 @@ def download(
             pipeline_runs = self._get_pipeline_runs(
                 connection, pipeline_name, project_name
             )
-            pipeline_run = [run for run in pipeline_runs if run.name == build_name]
+            pipeline_run = [
+                run
+                for run in pipeline_runs
+                if run.name == build_name
+                and self._check_attrs_recursive(run, additional_build_checks)
+            ]
             assert_that(len(pipeline_run)).described_as(
                 f"no succeeded and completed run found for pipeline {pipeline_name}"
             ).is_not_zero()