feat: now offers get_file_assets()

Author: Alan Christie

decoder/decoder.py

@@ -212,7 +212,7 @@ def get_job_doc_url(
 
 
 def get_pull_secret_names(job_definition: Dict[str, Any]) -> Set[str]:
-    """ "Given a Job definition this function returns the unique list of the
+    """Given a Job definition this function returns the unique list of the
     pull-secrets declared.
     """
     names: Set[str] = set()
@@ -226,7 +226,7 @@ def get_pull_secret_names(job_definition: Dict[str, Any]) -> Set[str]:
 
 
 def get_asset_names(job_definition: Dict[str, Any]) -> Set[str]:
-    """ "Given a Job definition this function returns the unique list of all the
+    """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.
     """
@@ -248,6 +248,27 @@ def get_asset_names(job_definition: Dict[str, Any]) -> Set[str]:
     return asset_names
 
 
+def get_file_assets(job_definition: Dict[str, Any]) -> List[Dict[str, str]]:
+    """Given a Job definition this function returns the list of all the
+    file-based assets declared. What's returned is the asset 'name' and the
+    'image file' the asset is expected to mapped to.
+    """
+    file_assets: List[Dict[str, str]] = []
+
+    # Iterate through 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"]:
+            file_assets.append(
+                {
+                    "asset": item["content-from"]["account-server-asset"]["name"],
+                    "image-file": item["name"],
+                }
+            )
+
+    return file_assets
+
+
 def decode(
     template_text: str,
     variable_map: Optional[Dict[str, str]],

tests/test_get_asset_names.py

@@ -20,11 +20,11 @@ def test_get_asset_names():
             ],
             "file": [
                 {
-                    "name": "/tmp/blob.txt",
+                    "name": "/tmp/blob-1.txt",
                     "content-from": {"account-server-asset": {"name": "asset-a"}},
                 },
                 {
-                    "name": "/tmp/blob.txt",
+                    "name": "/tmp/blob-2.txt",
                     "content-from": {"account-server-asset": {"name": "asset-b"}},
                 },
             ],

tests/test_get_file_assets.py

@@ -0,0 +1,47 @@
+# 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_file_assets():
+    # Arrange
+    job_definition: Dict = {
+        "image": {
+            "environment": [
+                {
+                    "name": "BLOB",
+                    "value-from": {"account-server-asset": {"name": "asset-a"}},
+                }
+            ],
+            "file": [
+                {
+                    "name": "/tmp/blob-1.txt",
+                    "content-from": {"account-server-asset": {"name": "asset-a"}},
+                },
+                {
+                    "name": "/tmp/blob-2.txt",
+                    "content-from": {"account-server-asset": {"name": "asset-b"}},
+                },
+            ],
+        },
+    }
+
+    # Act
+    file_assets = decoder.get_file_assets(job_definition)
+
+    # Assert
+    assert len(file_assets) == 2
+    print(file_assets)
+    for file_asset in file_assets:
+        if file_asset["image-file"] == "/tmp/blob-1.txt":
+            assert file_asset["asset"] == "asset-a"
+        elif file_asset["image-file"] == "/tmp/blob-2.txt":
+            assert file_asset["asset"] == "asset-b"
+        else:
+            # How did we get here?
+            assert False