Add source_channel to Orquesta context

CHANGELOG.rst

@@ -53,6 +53,8 @@ Added
 
   Contributed by Nick Maludy (Encore Technologies) #4547
 
+* Added ``source_channel`` to Orquesta ``st2()`` context for workflows called via ChatOps. (#4600)
+
 Changed
 ~~~~~~~
 

contrib/runners/orquesta_runner/orquesta_runner/orquesta_runner.py

@@ -77,6 +77,9 @@ def _construct_st2_context(self):
         if self.execution.context.get('api_user'):
             st2_ctx['st2']['api_user'] = self.execution.context.get('api_user')
 
+        if self.execution.context.get('source_channel'):
+            st2_ctx['st2']['source_channel'] = self.execution.context.get('source_channel')
+
         if self.execution.context:
             st2_ctx['parent'] = self.execution.context
 

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

@@ -244,3 +244,116 @@ def test_action_context_api_user(self):
         }
 
         self.assertDictEqual(lv_ac_db.result, expected_result)
+
+    def test_action_context_no_channel(self):
+        wf_name = 'subworkflow-source-channel-from-action-context'
+        wf_meta = base.get_wf_fixture_meta_data(TEST_PACK_PATH, wf_name + '.yaml')
+        lv_ac_db = lv_db_models.LiveActionDB(action=wf_meta['name'])
+        lv_ac_db, ac_ex_db = ac_svc.request(lv_ac_db)
+        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)
+
+        # Identify the records for the main workflow.
+        wf_ex_db = wf_db_access.WorkflowExecution.query(action_execution=str(ac_ex_db.id))[0]
+        t1_ex_db = wf_db_access.TaskExecution.query(workflow_execution=str(wf_ex_db.id))[0]
+        t1_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(t1_ex_db.id))[0]
+        t1_wf_ex_db = wf_db_access.WorkflowExecution.query(action_execution=str(t1_ac_ex_db.id))[0]
+        self.assertEqual(t1_ex_db.status, wf_statuses.RUNNING)
+        self.assertEqual(t1_ac_ex_db.status, ac_const.LIVEACTION_STATUS_RUNNING)
+        self.assertEqual(t1_wf_ex_db.status, wf_statuses.RUNNING)
+
+        # Complete subworkflow under task1.
+        query_filters = {'workflow_execution': str(t1_wf_ex_db.id), 'task_id': 'task1'}
+        t1_t1_ex_db = wf_db_access.TaskExecution.query(**query_filters)[0]
+        t1_t1_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(t1_t1_ex_db.id))[0]
+        wf_svc.handle_action_execution_completion(t1_t1_ac_ex_db)
+
+        query_filters = {'workflow_execution': str(t1_wf_ex_db.id), 'task_id': 'task2'}
+        t1_t2_ex_db = wf_db_access.TaskExecution.query(**query_filters)[0]
+        t1_t2_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(t1_t2_ex_db.id))[0]
+        wf_svc.handle_action_execution_completion(t1_t2_ac_ex_db)
+
+        query_filters = {'workflow_execution': str(t1_wf_ex_db.id), 'task_id': 'task3'}
+        t1_t3_ex_db = wf_db_access.TaskExecution.query(**query_filters)[0]
+        t1_t3_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(t1_t3_ex_db.id))[0]
+        wf_svc.handle_action_execution_completion(t1_t3_ac_ex_db)
+
+        t1_wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(str(t1_wf_ex_db.id))
+        t1_ac_ex_db = ex_db_access.ActionExecution.get_by_id(str(t1_ac_ex_db.id))
+        self.assertEqual(t1_wf_ex_db.status, wf_statuses.SUCCEEDED)
+        self.assertEqual(t1_ac_ex_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
+
+        # Complete task1 and main workflow.
+        wf_svc.handle_action_execution_completion(t1_ac_ex_db)
+        t1_ex_db = wf_db_access.TaskExecution.get_by_id(str(t1_ex_db.id))
+        wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(str(wf_ex_db.id))
+        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
+        self.assertEqual(t1_ex_db.status, wf_statuses.SUCCEEDED)
+        self.assertEqual(wf_ex_db.status, wf_statuses.SUCCEEDED)
+        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
+
+        # Check result.
+        expected_result = {
+            'output': {
+                'msg': 'no_channel, All your base are belong to us!'
+            }
+        }
+
+        self.assertDictEqual(lv_ac_db.result, expected_result)
+
+    def test_action_context_source_channel(self):
+        wf_name = 'subworkflow-source-channel-from-action-context'
+        wf_meta = base.get_wf_fixture_meta_data(TEST_PACK_PATH, wf_name + '.yaml')
+        lv_ac_db = lv_db_models.LiveActionDB(action=wf_meta['name'],
+                                             context={'source_channel': 'general'})
+        lv_ac_db, ac_ex_db = ac_svc.request(lv_ac_db)
+        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)
+
+        # Identify the records for the main workflow.
+        wf_ex_db = wf_db_access.WorkflowExecution.query(action_execution=str(ac_ex_db.id))[0]
+        t1_ex_db = wf_db_access.TaskExecution.query(workflow_execution=str(wf_ex_db.id))[0]
+        t1_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(t1_ex_db.id))[0]
+        t1_wf_ex_db = wf_db_access.WorkflowExecution.query(action_execution=str(t1_ac_ex_db.id))[0]
+        self.assertEqual(t1_ex_db.status, wf_statuses.RUNNING)
+        self.assertEqual(t1_ac_ex_db.status, ac_const.LIVEACTION_STATUS_RUNNING)
+        self.assertEqual(t1_wf_ex_db.status, wf_statuses.RUNNING)
+
+        # Complete subworkflow under task1.
+        query_filters = {'workflow_execution': str(t1_wf_ex_db.id), 'task_id': 'task1'}
+        t1_t1_ex_db = wf_db_access.TaskExecution.query(**query_filters)[0]
+        t1_t1_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(t1_t1_ex_db.id))[0]
+        wf_svc.handle_action_execution_completion(t1_t1_ac_ex_db)
+
+        query_filters = {'workflow_execution': str(t1_wf_ex_db.id), 'task_id': 'task2'}
+        t1_t2_ex_db = wf_db_access.TaskExecution.query(**query_filters)[0]
+        t1_t2_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(t1_t2_ex_db.id))[0]
+        wf_svc.handle_action_execution_completion(t1_t2_ac_ex_db)
+
+        query_filters = {'workflow_execution': str(t1_wf_ex_db.id), 'task_id': 'task3'}
+        t1_t3_ex_db = wf_db_access.TaskExecution.query(**query_filters)[0]
+        t1_t3_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(t1_t3_ex_db.id))[0]
+        wf_svc.handle_action_execution_completion(t1_t3_ac_ex_db)
+
+        t1_wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(str(t1_wf_ex_db.id))
+        t1_ac_ex_db = ex_db_access.ActionExecution.get_by_id(str(t1_ac_ex_db.id))
+        self.assertEqual(t1_wf_ex_db.status, wf_statuses.SUCCEEDED)
+        self.assertEqual(t1_ac_ex_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
+
+        # Complete task1 and main workflow.
+        wf_svc.handle_action_execution_completion(t1_ac_ex_db)
+        t1_ex_db = wf_db_access.TaskExecution.get_by_id(str(t1_ex_db.id))
+        wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(str(wf_ex_db.id))
+        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
+        self.assertEqual(t1_ex_db.status, wf_statuses.SUCCEEDED)
+        self.assertEqual(wf_ex_db.status, wf_statuses.SUCCEEDED)
+        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
+
+        # Check result.
+        expected_result = {
+            'output': {
+                'msg': 'general, All your base are belong to us!'
+            }
+        }
+
+        self.assertDictEqual(lv_ac_db.result, expected_result)

st2common/st2common/services/workflows.py

@@ -568,6 +568,9 @@ def request_action_execution(wf_ex_db, task_ex_db, st2_ctx, ac_ex_req, delay=Non
     if st2_ctx.get('api_user'):
         ac_ex_ctx['api_user'] = st2_ctx.get('api_user')
 
+    if st2_ctx.get('source_channel'):
+        ac_ex_ctx['source_channel'] = st2_ctx.get('source_channel')
+
     if item_id is not None:
         ac_ex_ctx['orquesta']['item_id'] = item_id
 
@@ -921,6 +924,9 @@ def request_next_tasks(wf_ex_db, task_ex_id=None):
                 if root_st2_ctx.get('api_user'):
                     st2_ctx['api_user'] = root_st2_ctx.get('api_user')
 
+                if root_st2_ctx.get('source_channel'):
+                    st2_ctx['source_channel'] = root_st2_ctx.get('source_channel')
+
                 # Request the task execution.
                 request_task_execution(wf_ex_db, st2_ctx, task)
             except Exception as e:

st2tests/st2tests/fixtures/packs/orquesta_tests/actions/source-channel-from-action-context.yaml

@@ -0,0 +1,12 @@
+---
+name: source-channel-from-action-context
+description: Test getting source_channel from context if provided
+pack: orquesta_tests
+runner_type: orquesta
+entry_point: workflows/sequential.yaml
+enabled: true
+parameters:
+  who:
+    required: true
+    type: string
+    default: "{{ 'source_channel' in action_context and action_context.source_channel or 'no_channel' }}"

st2tests/st2tests/fixtures/packs/orquesta_tests/actions/subworkflow-source-channel-from-action-context.yaml

@@ -0,0 +1,7 @@
+---
+name: subworkflow-source-channel-from-action-context.yaml
+description: A sample workflow that calls another subworkflow.
+pack: orquesta_tests
+runner_type: orquesta
+entry_point: workflows/subworkflow-source-channel-from-action-context.yaml
+enabled: true

st2tests/st2tests/fixtures/packs/orquesta_tests/actions/workflows/subworkflow-source-channel-from-action-context.yaml

@@ -0,0 +1,10 @@
+version: 1.0
+
+description: A sample workflow that calls another subworkflow.
+
+output:
+  - msg: <% task(task1).result.output.msg %>  
+
+tasks:
+  task1:
+    action: orquesta_tests.source-channel-from-action-context