Merge pull request #6 from ircwaves/get-correct-sfn-exec

fix sfn execution retrieval

CHANGELOG.md

@@ -10,7 +10,10 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 ### Fixed
 - CLI `run-workflow` dropped an `s` from `json.loads`, and the help text on the
   timout argument was
-  incorrect. ([#5](https://github.com/cirrus-geo/cirrus-mgmt/pull/5)
+  incorrect. ([#5](https://github.com/cirrus-geo/cirrus-mgmt/pull/5))
+- Return correct execution(`[-1]`), as new Step Function executions are
+  appended to the `executions` list, from the StateDB
+  Item. ([#6](https://github.com/cirrus-geo/cirrus-mgmt/pull/6))
 
 ## [v0.1.0] - 2023-08-01
 

src/cirrus/plugins/management/deployment.py

@@ -302,7 +302,7 @@ def get_execution(self, arn):
     def get_execution_by_payload_id(self, payload_id):
         execs = self.get_payload_state(payload_id).get("executions", [])
         try:
-            exec_arn = execs[0]
+            exec_arn = execs[-1]
         except IndexError:
             raise exceptions.NoExecutionsError(payload_id)
 

tests/conftest.py

@@ -19,6 +19,7 @@
     EventDB = None
 
 from cirrus.core.project import Project
+from cirrus.lib2.process_payload import ProcessPayload, ProcessPayloads
 from cirrus.lib2.statedb import StateDB
 
 
@@ -225,3 +226,14 @@ def _invoke(cmd, **kwargs):
         return cli_runner.invoke(cli, shlex.split(cmd), **kwargs)
 
     return _invoke
+
+
+@pytest.fixture
+def basic_payloads(fixtures):
+    return ProcessPayloads(
+        process_payloads=[
+            ProcessPayload(
+                json.loads(fixtures.joinpath("basic_payload.json").read_text())
+            )
+        ]
+    )

tests/fixtures/basic_payload.json

@@ -0,0 +1,41 @@
+{
+    "type": "FeatureCollection",
+    "features": [
+        {
+            "type": "Feature",
+            "stac_version": "1.0.0",
+            "id": "dummy-item",
+            "properties": {
+                "datetime": "1970-01-01T00:00:00Z",
+                "created": "1970-01-01T00:00:00Z",
+                "updated": "1970-01-01T00:00:00Z"
+            },
+            "geometry": {
+                "type": "Polygon",
+                "coordinates": []
+            },
+            "links": [],
+            "assets": {},
+            "bbox": [],
+            "stac_extensions": [],
+            "collection": "dummy-collection"
+        }
+    ],
+    "process": {
+        "workflow": "workflow1",
+        "replace": true,
+        "upload_options": {
+            "path_template": "${collection}/${year}/${id}",
+            "collections": {
+                "mycollection": ".*"
+},
+            "s3_urls": true
+        },
+        "tasks": {
+            "task1": {
+                "param1": true
+            }
+        }
+    },
+    "id": "test/workflow-workflow1/test-item"
+}

tests/test_manage.py

@@ -1,4 +1,6 @@
 import json
+import os
+from copy import deepcopy
 
 import pytest
 
@@ -21,7 +23,7 @@ def _manage(cmd):
 
 
 @pytest.fixture
-def deployment_meta(queue, statedb, payloads, data):
+def deployment_meta(queue, statedb, payloads, data, workflow):
     return {
         "name": DEPLYOMENT_NAME,
         "created": "2022-11-07T04:42:26.666916+00:00",
@@ -30,7 +32,9 @@ def deployment_meta(queue, statedb, payloads, data):
         "profile": None,
         "environment": {
             "CIRRUS_STATE_DB": statedb.table_name,
-            # "CIRRUS_PUBLISH_TOPIC_ARN": ,
+            "CIRRUS_BASE_WORKFLOW_ARN": workflow["stateMachineArn"].replace(
+                "workflow1", ""
+            ),
             "CIRRUS_LOG_LEVEL": "DEBUG",
             "CIRRUS_STACK": STACK_NAME,
             "CIRRUS_DATA_BUCKET": data,
@@ -95,3 +99,17 @@ def test_manage_refresh(deployment, mock_lambda_get_conf, lambda_env):
     assert result.exit_code == 0
     new = json.loads(deployment("show").stdout)
     assert new["environment"] == lambda_env
+
+
+# @pytest.mark.parametrize("item", (
+def test_manage_get_execution_by_payload_id(deployment, basic_payloads, statedb):
+    current_env = deepcopy(os.environ)  # stash env
+    deployment.set_env()
+    basic_payloads.process()
+    pid = basic_payloads[0]["id"]
+    sfn_exe1 = deployment.get_execution_by_payload_id(pid)
+    statedb.set_aborted(pid, execution_arn=sfn_exe1["executionArn"])
+    basic_payloads.process()
+    sfn_exe2 = deployment.get_execution_by_payload_id(pid)
+    assert sfn_exe1["executionArn"] != sfn_exe2["executionArn"]
+    os.environ = current_env  # pop stash