Add orquesta test which verifies "{{ config_context }}" notation for

default parameter values works for executions scheduled via Orquesta
workflow.

Author: Kami

contrib/runners/orquesta_runner/tests/unit/test_basic.py

@@ -298,6 +298,40 @@ def test_run_workflow_with_unicode_input(self):
         self.assertDictEqual(lv_ac_db.result, expected_result)
         self.assertDictEqual(ac_ex_db.result, expected_result)
 
+    def test_run_workflow_action_config_context(self):
+        wf_meta = base.get_wf_fixture_meta_data(TEST_PACK_PATH, 'config-context.yaml')
+        wf_input = {}
+        lv_ac_db = lv_db_models.LiveActionDB(action=wf_meta['name'], parameters=wf_input)
+        lv_ac_db, ac_ex_db = ac_svc.request(lv_ac_db)
+
+        # Assert action execution is running.
+        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
+        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_RUNNING, lv_ac_db.result)
+        wf_ex_db = wf_db_access.WorkflowExecution.query(action_execution=str(ac_ex_db.id))[0]
+        self.assertEqual(wf_ex_db.status, ac_const.LIVEACTION_STATUS_RUNNING)
+
+        # Assert task1 is already completed.
+        query_filters = {'workflow_execution': str(wf_ex_db.id), 'task_id': 'task1'}
+        tk1_ex_db = wf_db_access.TaskExecution.query(**query_filters)[0]
+        tk1_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(tk1_ex_db.id))[0]
+        tk1_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk1_ac_ex_db.liveaction['id'])
+        self.assertEqual(tk1_lv_ac_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
+        self.assertTrue(wf_svc.is_action_execution_under_workflow_context(tk1_ac_ex_db))
+
+        # Manually handle action execution completion.
+        wf_svc.handle_action_execution_completion(tk1_ac_ex_db)
+
+        # Assert workflow is completed.
+        wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(wf_ex_db.id)
+        self.assertEqual(wf_ex_db.status, wf_states.SUCCEEDED)
+        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
+        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
+        ac_ex_db = ex_db_access.ActionExecution.get_by_id(str(ac_ex_db.id))
+        self.assertEqual(ac_ex_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
+
+        # Verify config_context works
+        self.assertEqual(wf_ex_db.output, {'msg': 'value of config key a'})
+
     def test_run_workflow_with_action_less_tasks(self):
         wf_meta = base.get_wf_fixture_meta_data(TEST_PACK_PATH, 'action-less-tasks.yaml')
         wf_input = {'name': 'Thanos'}

st2tests/st2tests/fixtures/packs/orquesta_tests/actions/config-context-action.yaml

@@ -0,0 +1,9 @@
+---
+  name: "config-context-action"
+  runner_type: "local-shell-cmd"
+  enabled: true
+  entry_point: ""
+  parameters:
+    cmd:
+      immutable: true
+      default: "echo \"{{ config_context.config_key_a }}\""

st2tests/st2tests/fixtures/packs/orquesta_tests/actions/config-context.yaml

@@ -0,0 +1,7 @@
+---
+name: config-context
+description: Workflow which tests {{ config_context.foo }} notation works default parameter values for workflow actions.
+pack: orquesta_tests
+runner_type: orquesta
+entry_point: workflows/config-context.yaml
+enabled: true

st2tests/st2tests/fixtures/packs/orquesta_tests/actions/workflows/config-context.yaml

@@ -0,0 +1,13 @@
+version: 1.0
+
+description: Workflow which tests {{ config_context }} functionality.
+
+output:
+  - msg: <% ctx().message %>
+
+tasks:
+  task1:
+    action: orquesta_tests.config-context-action
+    next:
+      - when: <% succeeded() %>
+        publish: message=<% result().stdout %>

st2tests/st2tests/fixtures/packs/orquesta_tests/config.schema.yaml

@@ -0,0 +1,6 @@
+---
+config_key_a:
+  description: "Sample config key."
+  type: "string"
+  default: "value of config key a"
+  required: true