fix: Add get_asset_names function

§

Author: Alan Christie

decoder/decoder.py

@@ -7,7 +7,7 @@
 import enum
 import os
 import re
-from typing import Any, Dict, List, Optional, Pattern, Tuple
+from typing import Any, Dict, List, Optional, Pattern, Set, Tuple
 
 import jsonschema
 import yaml
@@ -211,6 +211,29 @@ def get_job_doc_url(
     return doc_url
 
 
+def get_asset_names(job_definition: Dict[str, Any]) -> Set[str]:
+    """ "Given a Job definition this function returns the unique list of all the
+    asset names declared. Asset names can be used in image environment
+    variables and files.
+    """
+    asset_names: Set[str] = set()
+
+    # Check the environment block...
+    environment: List[Dict[str, Any]] = job_definition.get("image", {}).get(
+        "environment", []
+    )
+    for env in environment:
+        if "account-server-asset" in env["value-from"]:
+            asset_names.add(env["value-from"]["account-server-asset"]["name"])
+    # Check the file block...
+    file_block: List[Dict[str, Any]] = job_definition.get("image", {}).get("file", [])
+    for item in file_block:
+        if "account-server-asset" in item["content-from"]:
+            asset_names.add(item["content-from"]["account-server-asset"]["name"])
+
+    return asset_names
+
+
 def decode(
     template_text: str,
     variable_map: Optional[Dict[str, str]],

tests/test_get_asset_names.py

@@ -0,0 +1,40 @@
+# Tests for the decoder's get_job_doc_url() function.
+from typing import Dict
+
+import pytest
+
+pytestmark = pytest.mark.unit
+
+from decoder import decoder
+
+
+def test_get_asset_names():
+    # Arrange
+    job_definition: Dict = {
+        "image": {
+            "environment": [
+                {
+                    "name": "BLOB",
+                    "value-from": {"account-server-asset": {"name": "asset-a"}},
+                }
+            ],
+            "file": [
+                {
+                    "name": "/tmp/blob.txt",
+                    "content-from": {"account-server-asset": {"name": "asset-a"}},
+                },
+                {
+                    "name": "/tmp/blob.txt",
+                    "content-from": {"account-server-asset": {"name": "asset-b"}},
+                },
+            ],
+        },
+    }
+
+    # Act
+    asset_names = decoder.get_asset_names(job_definition)
+
+    # Assert
+    assert len(asset_names) == 2
+    assert "asset-a" in asset_names
+    assert "asset-b" in asset_names