feat: Now offers get_environment_assets()

Author: Alan Christie

decoder/decoder.py

@@ -251,7 +251,10 @@ def get_asset_names(job_definition: Dict[str, Any]) -> Set[str]:
 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.
+    'image file' the asset is expected to be mapped to.
+
+    In order to get the asset the caller will need to interact with the
+    Account Server (AS), what is returned here is the asset name, not the asset.
     """
     file_assets: List[Dict[str, str]] = []
 
@@ -269,6 +272,32 @@ def get_file_assets(job_definition: Dict[str, Any]) -> List[Dict[str, str]]:
     return file_assets
 
 
+def get_environment_assets(job_definition: Dict[str, Any]) -> List[Dict[str, str]]:
+    """Given a Job definition this function returns the list of all the
+    environment-based assets declared. What's returned is the asset 'name' and the
+    environment 'variable' the asset is expected to be mapped to.
+
+    In order to get the asset the caller will need to interact with the
+    Account Server (AS), what is returned here is the asset name, not the asset.
+    """
+    env_assets: List[Dict[str, str]] = []
+
+    # Iterate through the environment block...
+    environment: List[Dict[str, Any]] = job_definition.get("image", {}).get(
+        "environment", []
+    )
+    for item in environment:
+        if "account-server-asset" in item["value-from"]:
+            env_assets.append(
+                {
+                    "asset": item["value-from"]["account-server-asset"]["name"],
+                    "variable": item["name"],
+                }
+            )
+
+    return env_assets
+
+
 def decode(
     template_text: str,
     variable_map: Optional[Dict[str, str]],

tests/test_get_asset_names.py

@@ -1,4 +1,4 @@
-# Tests for the decoder's get_job_doc_url() function.
+# Tests for the decoder's get_asset_names() function.
 from typing import Dict
 
 import pytest

tests/test_get_environment_assets.py

@@ -0,0 +1,40 @@
+# Tests for the decoder's get_environment_assets() function.
+from typing import Dict
+
+import pytest
+
+pytestmark = pytest.mark.unit
+
+from decoder import decoder
+
+
+def test_get_environment_assets():
+    # Arrange
+    job_definition: Dict = {
+        "image": {
+            "environment": [
+                {
+                    "name": "BLOB",
+                    "value-from": {"account-server-asset": {"name": "asset-c"}},
+                }
+            ],
+            "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
+    env_assets = decoder.get_environment_assets(job_definition)
+
+    # Assert
+    assert len(env_assets) == 1
+    assert env_assets[0]["asset"] == "asset-c"
+    assert env_assets[0]["variable"] == "BLOB"

tests/test_get_file_assets.py

@@ -1,4 +1,4 @@
-# Tests for the decoder's get_job_doc_url() function.
+# Tests for the decoder's get_file_assets() function.
 from typing import Dict
 
 import pytest
@@ -36,7 +36,6 @@ def test_get_file_assets():
 
     # 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"

tests/test_get_pull_secret_names.py

@@ -1,4 +1,4 @@
-# Tests for the decoder's get_job_doc_url() function.
+# Tests for the decoder's get_pull_secret_names() function.
 from typing import Dict
 
 import pytest