fix: safely parse task state and execution info in get_task_status

README.md

@@ -32,6 +32,12 @@ View the documentation [here](docs/index.md) for more help getting started with
 
 To build documentation locally, clone this repo, navigate to the root of this repo, then run `poetry run mkdocs serve`.
 
+To run tests with `uv`, use the project test command:
+
+```bash
+uv run test
+```
+
 
 ## Public Domain Standard Notice
 This repository constitutes a work of the United States Government and is not

cfa/cloudops/batch_helpers.py

@@ -1701,17 +1701,17 @@ def get_task_status(
         return json.dumps(
             {
                 "id": task.id,
-                "state": str(task.state.value),
-                "exit_code": task.execution_info.exit_code,
+                "state": getattr(task.state, "value", str(task.state)),
+                "exit_code": getattr(task.execution_info, "exit_code", None),
             }
         )
     out_json = []
     for task in tasks:
         out_json.append(
             {
                 "id": task.id,
-                "state": str(task.state.value),
-                "exit_code": task.execution_info.exit_code,
+                "state": getattr(task.state, "value", str(task.state)),
+                "exit_code": getattr(task.execution_info, "exit_code", None),
             }
         )
 

cfa/cloudops/scripts.py

@@ -1,4 +1,5 @@
 import argparse
+import sys
 import textwrap
 
 from cfa.cloudops import CloudClient
@@ -1215,3 +1216,15 @@ def generate_sample_env():
         print("Sample .env file 'cloudops-sample.env' created successfully.")
     except Exception as e:
         print(f"Error creating sample .env file: {e}")
+
+
+def test():
+    try:
+        import pytest
+    except ImportError as exc:
+        raise RuntimeError(
+            "pytest is not installed. Run `uv run test` from the project root, "
+            "or install the dev dependencies first with `uv sync --group dev`."
+        ) from exc
+
+    raise SystemExit(pytest.main(sys.argv[1:]))

pyproject.toml

@@ -51,6 +51,7 @@ allow-direct-references = true
 
 [project.scripts]
 hello = "cfa.cloudops.scripts:hello"
+test = "cfa.cloudops.scripts:test"
 create_pool = "cfa.cloudops.scripts:create_pool"
 create_job = "cfa.cloudops.scripts:create_job"
 create_job_schedule = "cfa.cloudops.scripts:create_job_schedule"
@@ -80,10 +81,21 @@ suspend_schedule = "cfa.cloudops.scripts:suspend_job_schedule"
 [tool.pytest.ini_options]
 pythonpath = ["."]
 
+[dependency-groups]
+dev = [
+    "ruff>=0.9.9",
+    "pytest>=8.3.5",
+    "pytest-asyncio>=1.2.0",
+    "pytest-mock>=3.14.0",
+    "pytest-cov>=6.1.1",
+    "coverage-badge>=1.1.2"
+]
+
 [project.optional-dependencies]
 dev = [
     "ruff>=0.9.9",
     "pytest>=8.3.5",
+    "pytest-asyncio>=1.2.0",
     "pytest-mock>=3.14.0",
     "pytest-cov>=6.1.1",
     "coverage-badge>=1.1.2"

tests/test_batch_helpers.py

@@ -276,14 +276,20 @@ def test_get_task_status_all_tasks():
             state=models.TaskState.completed,
             execution_info=MagicMock(exit_code=0),
         ),
+        MagicMock(
+            id="t3",
+            state="active",
+            execution_info=None,
+        ),
     ]
 
     result = get_task_status(job_name="my-job", batch_client=mock_batch_client)
     payload = json.loads(result)
 
     assert isinstance(payload, list)
-    assert {item["id"] for item in payload} == {"t1", "t2"}
-    assert {item["state"] for item in payload} == {"running", "completed"}
+    assert {item["id"] for item in payload} == {"t1", "t2", "t3"}
+    assert {item["state"] for item in payload} == {"running", "completed", "active"}
+    assert next(item["exit_code"] for item in payload if item["id"] == "t3") is None
 
 
 def test_get_task_status_single_task():