MistralQuerierTest.test_query_get_workflow_retry_exhausted() - Code Metrics - Inspection of "Merge pull request #3420 from StackStorm/fix-mistr..." - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 0891a6...20e5ca )

unknown

created 2017-05-26 10:00 UTC

test_query_get_workflow_retry_exhausted() A

↳ Parent: MistralQuerierTest

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
dl	0
loc	15
rs	9.4285
c	0
b	0
f	0

# Licensed to the StackStorm, Inc ('StackStorm') under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import copy
import datetime
import json
import requests
import time
import uuid

import mock
from mock import call

from mistralclient.api import base as mistralclient_base
from mistralclient.api.v2 import executions
from mistralclient.api.v2 import tasks
from oslo_config import cfg

import st2tests.config as tests_config
tests_config.parse_args()

from st2common.constants import action as action_constants
from st2common.exceptions import db as db_exc
from st2common.models.db.liveaction import LiveActionDB
from st2common.util import action_db as action_utils
from st2common.util import loader
from st2tests import DbTestCase


MOCK_WF_TASKS_SUCCEEDED = [
    {'name': 'task1', 'state': 'SUCCESS'},
    {'name': 'task2', 'state': 'SUCCESS'}
]

MOCK_WF_TASKS_ERRORED = [
    {'name': 'task1', 'state': 'SUCCESS'},
    {'name': 'task2', 'state': 'ERROR'}
]

MOCK_WF_TASKS_RUNNING = [
    {'name': 'task1', 'state': 'SUCCESS'},
    {'name': 'task2', 'state': 'RUNNING'}
]

MOCK_WF_TASKS_WAITING = [
    {'name': 'task1', 'state': 'SUCCESS'},
    {'name': 'task2', 'state': 'WAITING'}
]

MOCK_WF_EX_DATA = {
    'id': uuid.uuid4().hex,
    'name': 'main',
    'output': '{"k1": "v1"}',
    'state': 'SUCCESS',
    'state_info': None
}

MOCK_WF_EX = executions.Execution(None, MOCK_WF_EX_DATA)

MOCK_WF_EX_TASKS_DATA = [
    {
        'id': uuid.uuid4().hex,
        'name': 'task1',
        'workflow_execution_id': MOCK_WF_EX_DATA['id'],
        'workflow_name': MOCK_WF_EX_DATA['name'],
        'created_at': str(datetime.datetime.utcnow()),
        'updated_at': str(datetime.datetime.utcnow()),
        'state': 'SUCCESS',
        'state_info': None,
        'input': '{"a": "b"}',
        'result': '{"c": "d"}',
        'published': '{"c": "d"}'
    },
    {
        'id': uuid.uuid4().hex,
        'name': 'task2',
        'workflow_execution_id': MOCK_WF_EX_DATA['id'],
        'workflow_name': MOCK_WF_EX_DATA['name'],
        'created_at': str(datetime.datetime.utcnow()),
        'updated_at': str(datetime.datetime.utcnow()),
        'state': 'SUCCESS',
        'state_info': None,
        'input': '{"e": "f", "g": "h"}',
        'result': '{"i": "j", "k": "l"}',
        'published': '{"k": "l"}'
    }
]

MOCK_WF_EX_TASKS = [
    tasks.Task(None, MOCK_WF_EX_TASKS_DATA[0]),
    tasks.Task(None, MOCK_WF_EX_TASKS_DATA[1])
]

MOCK_QRY_CONTEXT = {
    'mistral': {
        'execution_id': uuid.uuid4().hex
    }
}

MOCK_LIVEACTION_RESULT = {
    'tasks': [
        {
            'id': MOCK_WF_EX_TASKS_DATA[0]['id'],
            'name': MOCK_WF_EX_TASKS_DATA[0]['name'],
            'workflow_execution_id': MOCK_WF_EX_TASKS_DATA[0]['workflow_execution_id'],
            'workflow_name': MOCK_WF_EX_TASKS_DATA[0]['workflow_name'],
            'created_at': MOCK_WF_EX_TASKS_DATA[0]['created_at'],
            'updated_at': MOCK_WF_EX_TASKS_DATA[0]['updated_at'],
            'state': MOCK_WF_EX_TASKS_DATA[0]['state'],
            'state_info': MOCK_WF_EX_TASKS_DATA[0]['state_info'],
            'input': json.loads(MOCK_WF_EX_TASKS_DATA[0]['input']),
            'result': json.loads(MOCK_WF_EX_TASKS_DATA[0]['result']),
            'published': json.loads(MOCK_WF_EX_TASKS_DATA[0]['published'])
        }
    ]
}

MOCK_LIVEACTION_RUNNING = LiveActionDB(
    action='mock.workflow',
    status=action_constants.LIVEACTION_STATUS_RUNNING
)

MOCK_LIVEACTION_RUNNING_WITH_STREAMING_RESULT = LiveActionDB(
    action='mock.workflow',
    status=action_constants.LIVEACTION_STATUS_RUNNING,
    result=MOCK_LIVEACTION_RESULT
)

MOCK_LIVEACTION_CANCELING = LiveActionDB(
    action='mock.workflow',
    status=action_constants.LIVEACTION_STATUS_CANCELING
)


class MistralQuerierTest(DbTestCase):

    @classmethod
    def setUpClass(cls):
        super(MistralQuerierTest, cls).setUpClass()

        # Override the retry configuration here otherwise st2tests.config.parse_args
        # in DbTestCase.setUpClass will reset these overrides.
        cfg.CONF.set_override('retry_exp_msec', 100, group='mistral')
        cfg.CONF.set_override('retry_exp_max_msec', 200, group='mistral')
        cfg.CONF.set_override('retry_stop_max_msec', 200, group='mistral')

        # Register query module.
        cls.query_module = loader.register_query_module('mistral_v2')

    def setUp(self):
        super(MistralQuerierTest, self).setUp()
        self.querier = self.query_module.get_instance()

    def test_determine_status_wf_running_tasks_running(self):
        status = self.querier._determine_execution_status(
            MOCK_LIVEACTION_RUNNING,
            'RUNNING',
            MOCK_WF_TASKS_RUNNING
        )

        self.assertEqual(action_constants.LIVEACTION_STATUS_RUNNING, status)

    def test_determine_status_wf_running_tasks_completed(self):
        status = self.querier._determine_execution_status(
            MOCK_LIVEACTION_RUNNING,
            'RUNNING',
            MOCK_WF_TASKS_SUCCEEDED
        )

        self.assertEqual(action_constants.LIVEACTION_STATUS_RUNNING, status)

    def test_determine_status_wf_succeeded_tasks_completed(self):
        status = self.querier._determine_execution_status(
            MOCK_LIVEACTION_RUNNING,
            'SUCCESS',
            MOCK_WF_TASKS_SUCCEEDED
        )

        self.assertEqual(action_constants.LIVEACTION_STATUS_SUCCEEDED, status)

    def test_determine_status_wf_succeeded_tasks_running(self):
        status = self.querier._determine_execution_status(
            MOCK_LIVEACTION_RUNNING,
            'SUCCESS',
            MOCK_WF_TASKS_RUNNING
        )

        self.assertEqual(action_constants.LIVEACTION_STATUS_RUNNING, status)

    def test_determine_status_wf_errored_tasks_completed(self):
        status = self.querier._determine_execution_status(
            MOCK_LIVEACTION_RUNNING,
            'ERROR',
            MOCK_WF_TASKS_SUCCEEDED
        )

        self.assertEqual(action_constants.LIVEACTION_STATUS_FAILED, status)

    def test_determine_status_wf_errored_tasks_running(self):
        status = self.querier._determine_execution_status(
            MOCK_LIVEACTION_RUNNING,
            'ERROR',
            MOCK_WF_TASKS_RUNNING
        )

        self.assertEqual(action_constants.LIVEACTION_STATUS_RUNNING, status)

    def test_determine_status_wf_canceled_tasks_completed(self):
        status = self.querier._determine_execution_status(
            MOCK_LIVEACTION_CANCELING,
            'CANCELLED',
            MOCK_WF_TASKS_SUCCEEDED
        )

        self.assertEqual(action_constants.LIVEACTION_STATUS_CANCELED, status)

    def test_determine_status_wf_canceled_tasks_running(self):
        status = self.querier._determine_execution_status(
            MOCK_LIVEACTION_CANCELING,
            'CANCELLED',
            MOCK_WF_TASKS_RUNNING
        )

        self.assertEqual(action_constants.LIVEACTION_STATUS_CANCELING, status)

    def test_determine_status_wf_canceled_tasks_waiting(self):
        status = self.querier._determine_execution_status(
            MOCK_LIVEACTION_CANCELING,
            'CANCELLED',
            MOCK_WF_TASKS_WAITING
        )

        self.assertEqual(action_constants.LIVEACTION_STATUS_CANCELED, status)

    def test_determine_status_wf_canceled_exec_running_tasks_completed(self):
        status = self.querier._determine_execution_status(
            MOCK_LIVEACTION_CANCELING,
            'RUNNING',
            MOCK_WF_TASKS_SUCCEEDED
        )

        self.assertEqual(action_constants.LIVEACTION_STATUS_CANCELING, status)

    def test_determine_status_wf_canceled_exec_running_tasks_running(self):
        status = self.querier._determine_execution_status(
            MOCK_LIVEACTION_CANCELING,
            'RUNNING',
            MOCK_WF_TASKS_RUNNING
        )

        self.assertEqual(action_constants.LIVEACTION_STATUS_CANCELING, status)

    def test_determine_status_wf_canceled_exec_running_tasks_waiting(self):
        status = self.querier._determine_execution_status(
            MOCK_LIVEACTION_CANCELING,
            'RUNNING',
            MOCK_WF_TASKS_WAITING
        )

        self.assertEqual(action_constants.LIVEACTION_STATUS_CANCELING, status)

    def test_determine_status_wf_running_exec_paused_tasks_completed(self):
        status = self.querier._determine_execution_status(
            MOCK_LIVEACTION_RUNNING,
            'PAUSED',
            MOCK_WF_TASKS_SUCCEEDED
        )

        self.assertEqual(action_constants.LIVEACTION_STATUS_RUNNING, status)

    def test_determine_status_wf_running_exec_cancelled_tasks_running(self):
        status = self.querier._determine_execution_status(
            MOCK_LIVEACTION_RUNNING,
            'CANCELLED',
            MOCK_WF_TASKS_RUNNING
        )

        self.assertEqual(action_constants.LIVEACTION_STATUS_CANCELING, status)

    @mock.patch.object(
        executions.ExecutionManager, 'get',
        mock.MagicMock(return_value=MOCK_WF_EX))
    def test_get_workflow_result(self):
        result = self.querier._get_workflow_result(uuid.uuid4().hex)

        expected = {
            'k1': 'v1',
            'extra': {
                'state': MOCK_WF_EX.state,
                'state_info': MOCK_WF_EX.state_info
            }
        }

        self.assertDictEqual(expected, result)

    @mock.patch.object(
        tasks.TaskManager, 'list',
        mock.MagicMock(return_value=MOCK_WF_EX_TASKS))
    @mock.patch.object(
        tasks.TaskManager, 'get',
        mock.MagicMock(side_effect=[
            MOCK_WF_EX_TASKS[0],
            MOCK_WF_EX_TASKS[1]]))
    def test_get_workflow_tasks(self):
        tasks = self.querier._get_workflow_tasks(uuid.uuid4().hex)

        expected = copy.deepcopy(MOCK_WF_EX_TASKS_DATA)
        for task in expected:
            task['input'] = json.loads(task['input'])
            task['result'] = json.loads(task['result'])
            task['published'] = json.loads(task['published'])

        for i in range(0, len(tasks)):
            self.assertDictEqual(expected[i], tasks[i])

    @mock.patch.object(
        action_utils, 'get_liveaction_by_id',
        mock.MagicMock(return_value=MOCK_LIVEACTION_RUNNING))
    @mock.patch.object(
        executions.ExecutionManager, 'get',
        mock.MagicMock(return_value=MOCK_WF_EX))
    @mock.patch.object(
        tasks.TaskManager, 'list',
        mock.MagicMock(return_value=MOCK_WF_EX_TASKS))
    @mock.patch.object(
        tasks.TaskManager, 'get',
        mock.MagicMock(side_effect=[
            MOCK_WF_EX_TASKS[0],
            MOCK_WF_EX_TASKS[1]]))
    def test_query(self):
        (status, result) = self.querier.query(uuid.uuid4().hex, MOCK_QRY_CONTEXT)

        expected = {
            'k1': 'v1',
            'tasks': copy.deepcopy(MOCK_WF_EX_TASKS_DATA),
            'extra': {
                'state': MOCK_WF_EX.state,
                'state_info': MOCK_WF_EX.state_info
            }
        }

        for task in expected['tasks']:
            task['input'] = json.loads(task['input'])
            task['result'] = json.loads(task['result'])
            task['published'] = json.loads(task['published'])

        self.assertEqual(action_constants.LIVEACTION_STATUS_SUCCEEDED, status)
        self.assertDictEqual(expected, result)

    @mock.patch.object(
        action_utils, 'get_liveaction_by_id',
        mock.MagicMock(return_value=MOCK_LIVEACTION_RUNNING_WITH_STREAMING_RESULT))
    @mock.patch.object(
        executions.ExecutionManager, 'get',
        mock.MagicMock(return_value=MOCK_WF_EX))
    @mock.patch.object(
        tasks.TaskManager, 'list',
        mock.MagicMock(return_value=[MOCK_WF_EX_TASKS[1]]))
    @mock.patch.object(
        tasks.TaskManager, 'get',
        mock.MagicMock(return_value=MOCK_WF_EX_TASKS[1]))
    def test_query_with_last_query_time(self):
        last_query_time = time.time() - 3

        (status, result) = self.querier.query(
            uuid.uuid4().hex,
            MOCK_QRY_CONTEXT,
            last_query_time=last_query_time
        )

        expected = {
            'k1': 'v1',
            'tasks': copy.deepcopy(MOCK_WF_EX_TASKS_DATA),
            'extra': {
                'state': MOCK_WF_EX.state,
                'state_info': MOCK_WF_EX.state_info
            }
        }

        for task in expected['tasks']:
            task['input'] = json.loads(task['input'])
            task['result'] = json.loads(task['result'])
            task['published'] = json.loads(task['published'])

        self.assertEqual(action_constants.LIVEACTION_STATUS_SUCCEEDED, status)
        self.assertDictEqual(expected, result)

    @mock.patch.object(
        action_utils, 'get_liveaction_by_id',
        mock.MagicMock(side_effect=db_exc.StackStormDBObjectNotFoundError()))
    def test_query_liveaction_not_found(self):
        self.assertRaises(
            db_exc.StackStormDBObjectNotFoundError,
            self.querier.query,
            uuid.uuid4().hex,
            MOCK_QRY_CONTEXT
        )

    @mock.patch.object(
        action_utils, 'get_liveaction_by_id',
        mock.MagicMock(return_value=MOCK_LIVEACTION_RUNNING))
    @mock.patch.object(
        executions.ExecutionManager, 'get',
        mock.MagicMock(side_effect=[
            requests.exceptions.ConnectionError(),
            MOCK_WF_EX]))
    @mock.patch.object(
        tasks.TaskManager, 'list',
        mock.MagicMock(return_value=MOCK_WF_EX_TASKS))
    @mock.patch.object(
        tasks.TaskManager, 'get',
        mock.MagicMock(side_effect=[
            MOCK_WF_EX_TASKS[0],
            MOCK_WF_EX_TASKS[1]]))
    def test_query_get_workflow_retry(self):
        (status, result) = self.querier.query(uuid.uuid4().hex, MOCK_QRY_CONTEXT)

        expected = {
            'k1': 'v1',
            'tasks': copy.deepcopy(MOCK_WF_EX_TASKS_DATA),
            'extra': {
                'state': MOCK_WF_EX.state,
                'state_info': MOCK_WF_EX.state_info
            }
        }

        for task in expected['tasks']:
            task['input'] = json.loads(task['input'])
            task['result'] = json.loads(task['result'])
            task['published'] = json.loads(task['published'])

        self.assertEqual(action_constants.LIVEACTION_STATUS_SUCCEEDED, status)
        self.assertDictEqual(expected, result)

        calls = [call(MOCK_QRY_CONTEXT['mistral']['execution_id']) for i in range(0, 2)]
        executions.ExecutionManager.get.assert_has_calls(calls)

    @mock.patch.object(
        action_utils, 'get_liveaction_by_id',
        mock.MagicMock(return_value=MOCK_LIVEACTION_RUNNING))
    @mock.patch.object(
        executions.ExecutionManager, 'get',
        mock.MagicMock(side_effect=[requests.exceptions.ConnectionError()] * 4))
    def test_query_get_workflow_retry_exhausted(self):
        self.assertRaises(
            requests.exceptions.ConnectionError,
            self.querier.query,
            uuid.uuid4().hex,
            MOCK_QRY_CONTEXT)

        calls = [call(MOCK_QRY_CONTEXT['mistral']['execution_id']) for i in range(0, 2)]
        executions.ExecutionManager.get.assert_has_calls(calls)

    @mock.patch.object(
        action_utils, 'get_liveaction_by_id',
        mock.MagicMock(return_value=MOCK_LIVEACTION_RUNNING))
    @mock.patch.object(
        executions.ExecutionManager, 'get',
        mock.MagicMock(
            side_effect=mistralclient_base.APIException(
                error_code=404, error_message='Workflow not found.')))
    def test_query_get_workflow_not_found(self):
        (status, result) = self.querier.query(uuid.uuid4().hex, MOCK_QRY_CONTEXT)

        self.assertEqual(action_constants.LIVEACTION_STATUS_FAILED, status)
        self.assertEqual('Workflow not found.', result)

    @mock.patch.object(
        action_utils, 'get_liveaction_by_id',
        mock.MagicMock(return_value=MOCK_LIVEACTION_RUNNING))
    @mock.patch.object(
        executions.ExecutionManager, 'get',
        mock.MagicMock(return_value=MOCK_WF_EX))
    @mock.patch.object(
        tasks.TaskManager, 'list',
        mock.MagicMock(side_effect=[
            requests.exceptions.ConnectionError(),
            MOCK_WF_EX_TASKS]))
    @mock.patch.object(
        tasks.TaskManager, 'get',
        mock.MagicMock(side_effect=[
            MOCK_WF_EX_TASKS[0],
            MOCK_WF_EX_TASKS[1]]))
    def test_query_list_workflow_tasks_retry(self):
        (status, result) = self.querier.query(uuid.uuid4().hex, MOCK_QRY_CONTEXT)

        expected = {
            'k1': 'v1',
            'tasks': copy.deepcopy(MOCK_WF_EX_TASKS_DATA),
            'extra': {
                'state': MOCK_WF_EX.state,
                'state_info': MOCK_WF_EX.state_info
            }
        }

        for task in expected['tasks']:
            task['input'] = json.loads(task['input'])
            task['result'] = json.loads(task['result'])
            task['published'] = json.loads(task['published'])

        self.assertEqual(action_constants.LIVEACTION_STATUS_SUCCEEDED, status)
        self.assertDictEqual(expected, result)

        mock_call = call(workflow_execution_id=MOCK_QRY_CONTEXT['mistral']['execution_id'])
        calls = [mock_call for i in range(0, 2)]
        tasks.TaskManager.list.assert_has_calls(calls)

        calls = [call(MOCK_WF_EX_TASKS[0].id), call(MOCK_WF_EX_TASKS[1].id)]
        tasks.TaskManager.get.assert_has_calls(calls)

    @mock.patch.object(
        action_utils, 'get_liveaction_by_id',
        mock.MagicMock(return_value=MOCK_LIVEACTION_RUNNING))
    @mock.patch.object(
        executions.ExecutionManager, 'get',
        mock.MagicMock(return_value=MOCK_WF_EX))
    @mock.patch.object(
        tasks.TaskManager, 'list',
        mock.MagicMock(return_value=MOCK_WF_EX_TASKS))
    @mock.patch.object(
        tasks.TaskManager, 'get',
        mock.MagicMock(side_effect=[
            requests.exceptions.ConnectionError(),
            MOCK_WF_EX_TASKS[0],
            MOCK_WF_EX_TASKS[1]]))
    def test_query_get_workflow_tasks_retry(self):
        (status, result) = self.querier.query(uuid.uuid4().hex, MOCK_QRY_CONTEXT)

        expected = {
            'k1': 'v1',
            'tasks': copy.deepcopy(MOCK_WF_EX_TASKS_DATA),
            'extra': {
                'state': MOCK_WF_EX.state,
                'state_info': MOCK_WF_EX.state_info
            }
        }

        for task in expected['tasks']:
            task['input'] = json.loads(task['input'])
            task['result'] = json.loads(task['result'])
            task['published'] = json.loads(task['published'])

        self.assertEqual(action_constants.LIVEACTION_STATUS_SUCCEEDED, status)
        self.assertDictEqual(expected, result)

        calls = [
            call(MOCK_WF_EX_TASKS[0].id),
            call(MOCK_WF_EX_TASKS[0].id),
            call(MOCK_WF_EX_TASKS[1].id)
        ]

        tasks.TaskManager.get.assert_has_calls(calls)

    @mock.patch.object(
        action_utils, 'get_liveaction_by_id',
        mock.MagicMock(return_value=MOCK_LIVEACTION_RUNNING))
    @mock.patch.object(
        executions.ExecutionManager, 'get',
        mock.MagicMock(return_value=MOCK_WF_EX))
    @mock.patch.object(
        tasks.TaskManager, 'list',
        mock.MagicMock(side_effect=[requests.exceptions.ConnectionError()] * 4))
    def test_query_list_workflow_tasks_retry_exhausted(self):
        self.assertRaises(
            requests.exceptions.ConnectionError,
            self.querier.query,
            uuid.uuid4().hex,
            MOCK_QRY_CONTEXT)

        mock_call = call(workflow_execution_id=MOCK_QRY_CONTEXT['mistral']['execution_id'])
        calls = [mock_call for i in range(0, 2)]
        tasks.TaskManager.list.assert_has_calls(calls)

    @mock.patch.object(
        action_utils, 'get_liveaction_by_id',
        mock.MagicMock(return_value=MOCK_LIVEACTION_RUNNING))
    @mock.patch.object(
        executions.ExecutionManager, 'get',
        mock.MagicMock(return_value=MOCK_WF_EX))
    @mock.patch.object(
        tasks.TaskManager, 'list',
        mock.MagicMock(return_value=MOCK_WF_EX_TASKS))
    @mock.patch.object(
        tasks.TaskManager, 'get',
        mock.MagicMock(side_effect=[requests.exceptions.ConnectionError()] * 4))
    def test_query_get_workflow_tasks_retry_exhausted(self):
        self.assertRaises(
            requests.exceptions.ConnectionError,
            self.querier.query,
            uuid.uuid4().hex,
            MOCK_QRY_CONTEXT)

        calls = [
            call(MOCK_WF_EX_TASKS[0].id),
            call(MOCK_WF_EX_TASKS[0].id)
        ]

        tasks.TaskManager.get.assert_has_calls(calls)

    @mock.patch.object(
        action_utils, 'get_liveaction_by_id',
        mock.MagicMock(return_value=MOCK_LIVEACTION_RUNNING))
    @mock.patch.object(
        executions.ExecutionManager, 'get',
        mock.MagicMock(return_value=MOCK_WF_EX))
    @mock.patch.object(
        tasks.TaskManager, 'list',
        mock.MagicMock(return_value=MOCK_WF_EX_TASKS))
    @mock.patch.object(
        tasks.TaskManager, 'get',
        mock.MagicMock(
            side_effect=mistralclient_base.APIException(
                error_code=404, error_message='Task not found.')))
    def test_query_get_workflow_tasks_not_found(self):
        (status, result) = self.querier.query(uuid.uuid4().hex, MOCK_QRY_CONTEXT)

        self.assertEqual(action_constants.LIVEACTION_STATUS_FAILED, status)
        self.assertEqual('Task not found.', result)

    def test_query_missing_context(self):
        self.assertRaises(Exception, self.querier.query, uuid.uuid4().hex, {})

    def test_query_missing_mistral_execution_id(self):
        self.assertRaises(Exception, self.querier.query, uuid.uuid4().hex, {'mistral': {}})


1			# Licensed to the StackStorm, Inc ('StackStorm') under one or more
2			# contributor license agreements. See the NOTICE file distributed with
3			# this work for additional information regarding copyright ownership.
4			# The ASF licenses this file to You under the Apache License, Version 2.0
5			# (the "License"); you may not use this file except in compliance with
6			# the License. You may obtain a copy of the License at
7			#
8			# http://www.apache.org/licenses/LICENSE-2.0
9			#
10			# Unless required by applicable law or agreed to in writing, software
11			# distributed under the License is distributed on an "AS IS" BASIS,
12			# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13			# See the License for the specific language governing permissions and
14			# limitations under the License.
15
16			import copy
17			import datetime
18			import json
19			import requests
20			import time
21			import uuid
22
23			import mock
24			from mock import call
25
26			from mistralclient.api import base as mistralclient_base
27			from mistralclient.api.v2 import executions
28			from mistralclient.api.v2 import tasks
29			from oslo_config import cfg
30
31			import st2tests.config as tests_config
32			tests_config.parse_args()
33
34			from st2common.constants import action as action_constants
35			from st2common.exceptions import db as db_exc
36			from st2common.models.db.liveaction import LiveActionDB
37			from st2common.util import action_db as action_utils
38			from st2common.util import loader
39			from st2tests import DbTestCase
40
41
42			MOCK_WF_TASKS_SUCCEEDED = [
43			{'name': 'task1', 'state': 'SUCCESS'},
44			{'name': 'task2', 'state': 'SUCCESS'}
45			]
46
47			MOCK_WF_TASKS_ERRORED = [
48			{'name': 'task1', 'state': 'SUCCESS'},
49			{'name': 'task2', 'state': 'ERROR'}
50			]
51
52			MOCK_WF_TASKS_RUNNING = [
53			{'name': 'task1', 'state': 'SUCCESS'},
54			{'name': 'task2', 'state': 'RUNNING'}
55			]
56
57			MOCK_WF_TASKS_WAITING = [
58			{'name': 'task1', 'state': 'SUCCESS'},
59			{'name': 'task2', 'state': 'WAITING'}
60			]
61
62			MOCK_WF_EX_DATA = {
63			'id': uuid.uuid4().hex,
64			'name': 'main',
65			'output': '{"k1": "v1"}',
66			'state': 'SUCCESS',
67			'state_info': None
68			}
69
70			MOCK_WF_EX = executions.Execution(None, MOCK_WF_EX_DATA)
71
72			MOCK_WF_EX_TASKS_DATA = [
73			{
74			'id': uuid.uuid4().hex,
75			'name': 'task1',
76			'workflow_execution_id': MOCK_WF_EX_DATA['id'],
77			'workflow_name': MOCK_WF_EX_DATA['name'],
78			'created_at': str(datetime.datetime.utcnow()),
79			'updated_at': str(datetime.datetime.utcnow()),
80			'state': 'SUCCESS',
81			'state_info': None,
82			'input': '{"a": "b"}',
83			'result': '{"c": "d"}',
84			'published': '{"c": "d"}'
85			},
86			{
87			'id': uuid.uuid4().hex,
88			'name': 'task2',
89			'workflow_execution_id': MOCK_WF_EX_DATA['id'],
90			'workflow_name': MOCK_WF_EX_DATA['name'],
91			'created_at': str(datetime.datetime.utcnow()),
92			'updated_at': str(datetime.datetime.utcnow()),
93			'state': 'SUCCESS',
94			'state_info': None,
95			'input': '{"e": "f", "g": "h"}',
96			'result': '{"i": "j", "k": "l"}',
97			'published': '{"k": "l"}'
98			}
99			]
100
101			MOCK_WF_EX_TASKS = [
102			tasks.Task(None, MOCK_WF_EX_TASKS_DATA[0]),
103			tasks.Task(None, MOCK_WF_EX_TASKS_DATA[1])
104			]
105
106			MOCK_QRY_CONTEXT = {
107			'mistral': {
108			'execution_id': uuid.uuid4().hex
109			}
110			}
111
112			MOCK_LIVEACTION_RESULT = {
113			'tasks': [
114			{
115			'id': MOCK_WF_EX_TASKS_DATA[0]['id'],
116			'name': MOCK_WF_EX_TASKS_DATA[0]['name'],
117			'workflow_execution_id': MOCK_WF_EX_TASKS_DATA[0]['workflow_execution_id'],
118			'workflow_name': MOCK_WF_EX_TASKS_DATA[0]['workflow_name'],
119			'created_at': MOCK_WF_EX_TASKS_DATA[0]['created_at'],
120			'updated_at': MOCK_WF_EX_TASKS_DATA[0]['updated_at'],
121			'state': MOCK_WF_EX_TASKS_DATA[0]['state'],
122			'state_info': MOCK_WF_EX_TASKS_DATA[0]['state_info'],
123			'input': json.loads(MOCK_WF_EX_TASKS_DATA[0]['input']),
124			'result': json.loads(MOCK_WF_EX_TASKS_DATA[0]['result']),
125			'published': json.loads(MOCK_WF_EX_TASKS_DATA[0]['published'])
126			}
127			]
128			}
129
130			MOCK_LIVEACTION_RUNNING = LiveActionDB(
131			action='mock.workflow',
132			status=action_constants.LIVEACTION_STATUS_RUNNING
133			)
134
135			MOCK_LIVEACTION_RUNNING_WITH_STREAMING_RESULT = LiveActionDB(
136			action='mock.workflow',
137			status=action_constants.LIVEACTION_STATUS_RUNNING,
138			result=MOCK_LIVEACTION_RESULT
139			)
140
141			MOCK_LIVEACTION_CANCELING = LiveActionDB(
142			action='mock.workflow',
143			status=action_constants.LIVEACTION_STATUS_CANCELING
144			)
145
146
147			class MistralQuerierTest(DbTestCase):
148
149			@classmethod
150			def setUpClass(cls):
151			super(MistralQuerierTest, cls).setUpClass()
152
153			# Override the retry configuration here otherwise st2tests.config.parse_args
154			# in DbTestCase.setUpClass will reset these overrides.
155			cfg.CONF.set_override('retry_exp_msec', 100, group='mistral')
156			cfg.CONF.set_override('retry_exp_max_msec', 200, group='mistral')
157			cfg.CONF.set_override('retry_stop_max_msec', 200, group='mistral')
158
159			# Register query module.
160			cls.query_module = loader.register_query_module('mistral_v2')
161
162			def setUp(self):
163			super(MistralQuerierTest, self).setUp()
164			self.querier = self.query_module.get_instance()
165
166			def test_determine_status_wf_running_tasks_running(self):
167			status = self.querier._determine_execution_status(
168			MOCK_LIVEACTION_RUNNING,
169			'RUNNING',
170			MOCK_WF_TASKS_RUNNING
171			)
172
173			self.assertEqual(action_constants.LIVEACTION_STATUS_RUNNING, status)
174
175			def test_determine_status_wf_running_tasks_completed(self):
176			status = self.querier._determine_execution_status(
177			MOCK_LIVEACTION_RUNNING,
178			'RUNNING',
179			MOCK_WF_TASKS_SUCCEEDED
180			)
181
182			self.assertEqual(action_constants.LIVEACTION_STATUS_RUNNING, status)
183
184			def test_determine_status_wf_succeeded_tasks_completed(self):
185			status = self.querier._determine_execution_status(
186			MOCK_LIVEACTION_RUNNING,
187			'SUCCESS',
188			MOCK_WF_TASKS_SUCCEEDED
189			)
190
191			self.assertEqual(action_constants.LIVEACTION_STATUS_SUCCEEDED, status)
192
193			def test_determine_status_wf_succeeded_tasks_running(self):
194			status = self.querier._determine_execution_status(
195			MOCK_LIVEACTION_RUNNING,
196			'SUCCESS',
197			MOCK_WF_TASKS_RUNNING
198			)
199
200			self.assertEqual(action_constants.LIVEACTION_STATUS_RUNNING, status)
201
202			def test_determine_status_wf_errored_tasks_completed(self):
203			status = self.querier._determine_execution_status(
204			MOCK_LIVEACTION_RUNNING,
205			'ERROR',
206			MOCK_WF_TASKS_SUCCEEDED
207			)
208
209			self.assertEqual(action_constants.LIVEACTION_STATUS_FAILED, status)
210
211			def test_determine_status_wf_errored_tasks_running(self):
212			status = self.querier._determine_execution_status(
213			MOCK_LIVEACTION_RUNNING,
214			'ERROR',
215			MOCK_WF_TASKS_RUNNING
216			)
217
218			self.assertEqual(action_constants.LIVEACTION_STATUS_RUNNING, status)
219
220			def test_determine_status_wf_canceled_tasks_completed(self):
221			status = self.querier._determine_execution_status(
222			MOCK_LIVEACTION_CANCELING,
223			'CANCELLED',
224			MOCK_WF_TASKS_SUCCEEDED
225			)
226
227			self.assertEqual(action_constants.LIVEACTION_STATUS_CANCELED, status)
228
229			def test_determine_status_wf_canceled_tasks_running(self):
230			status = self.querier._determine_execution_status(
231			MOCK_LIVEACTION_CANCELING,
232			'CANCELLED',
233			MOCK_WF_TASKS_RUNNING
234			)
235
236			self.assertEqual(action_constants.LIVEACTION_STATUS_CANCELING, status)
237
238			def test_determine_status_wf_canceled_tasks_waiting(self):
239			status = self.querier._determine_execution_status(
240			MOCK_LIVEACTION_CANCELING,
241			'CANCELLED',
242			MOCK_WF_TASKS_WAITING
243			)
244
245			self.assertEqual(action_constants.LIVEACTION_STATUS_CANCELED, status)
246
247			def test_determine_status_wf_canceled_exec_running_tasks_completed(self):
248			status = self.querier._determine_execution_status(
249			MOCK_LIVEACTION_CANCELING,
250			'RUNNING',
251			MOCK_WF_TASKS_SUCCEEDED
252			)
253
254			self.assertEqual(action_constants.LIVEACTION_STATUS_CANCELING, status)
255
256			def test_determine_status_wf_canceled_exec_running_tasks_running(self):
257			status = self.querier._determine_execution_status(
258			MOCK_LIVEACTION_CANCELING,
259			'RUNNING',
260			MOCK_WF_TASKS_RUNNING
261			)
262
263			self.assertEqual(action_constants.LIVEACTION_STATUS_CANCELING, status)
264
265			def test_determine_status_wf_canceled_exec_running_tasks_waiting(self):
266			status = self.querier._determine_execution_status(
267			MOCK_LIVEACTION_CANCELING,
268			'RUNNING',
269			MOCK_WF_TASKS_WAITING
270			)
271
272			self.assertEqual(action_constants.LIVEACTION_STATUS_CANCELING, status)
273
274			def test_determine_status_wf_running_exec_paused_tasks_completed(self):
275			status = self.querier._determine_execution_status(
276			MOCK_LIVEACTION_RUNNING,
277			'PAUSED',
278			MOCK_WF_TASKS_SUCCEEDED
279			)
280
281			self.assertEqual(action_constants.LIVEACTION_STATUS_RUNNING, status)
282
283			def test_determine_status_wf_running_exec_cancelled_tasks_running(self):
284			status = self.querier._determine_execution_status(
285			MOCK_LIVEACTION_RUNNING,
286			'CANCELLED',
287			MOCK_WF_TASKS_RUNNING
288			)
289
290			self.assertEqual(action_constants.LIVEACTION_STATUS_CANCELING, status)
291
292			@mock.patch.object(
293			executions.ExecutionManager, 'get',
294			mock.MagicMock(return_value=MOCK_WF_EX))
295			def test_get_workflow_result(self):
296			result = self.querier._get_workflow_result(uuid.uuid4().hex)
297
298			expected = {
299			'k1': 'v1',
300			'extra': {
301			'state': MOCK_WF_EX.state,
302			'state_info': MOCK_WF_EX.state_info
303			}
304			}
305
306			self.assertDictEqual(expected, result)
307
308			@mock.patch.object(
309			tasks.TaskManager, 'list',
310			mock.MagicMock(return_value=MOCK_WF_EX_TASKS))
311			@mock.patch.object(
312			tasks.TaskManager, 'get',
313			mock.MagicMock(side_effect=[
314			MOCK_WF_EX_TASKS[0],
315			MOCK_WF_EX_TASKS[1]]))
316			def test_get_workflow_tasks(self):
317			tasks = self.querier._get_workflow_tasks(uuid.uuid4().hex)
318
319			expected = copy.deepcopy(MOCK_WF_EX_TASKS_DATA)
320			for task in expected:
321			task['input'] = json.loads(task['input'])
322			task['result'] = json.loads(task['result'])
323			task['published'] = json.loads(task['published'])
324
325			for i in range(0, len(tasks)):
326			self.assertDictEqual(expected[i], tasks[i])
327
328			@mock.patch.object(
329			action_utils, 'get_liveaction_by_id',
330			mock.MagicMock(return_value=MOCK_LIVEACTION_RUNNING))
331			@mock.patch.object(
332			executions.ExecutionManager, 'get',
333			mock.MagicMock(return_value=MOCK_WF_EX))
334			@mock.patch.object(
335			tasks.TaskManager, 'list',
336			mock.MagicMock(return_value=MOCK_WF_EX_TASKS))
337			@mock.patch.object(
338			tasks.TaskManager, 'get',
339			mock.MagicMock(side_effect=[
340			MOCK_WF_EX_TASKS[0],
341			MOCK_WF_EX_TASKS[1]]))
342			def test_query(self):
343			(status, result) = self.querier.query(uuid.uuid4().hex, MOCK_QRY_CONTEXT)
344
345			expected = {
346			'k1': 'v1',
347			'tasks': copy.deepcopy(MOCK_WF_EX_TASKS_DATA),
348			'extra': {
349			'state': MOCK_WF_EX.state,
350			'state_info': MOCK_WF_EX.state_info
351			}
352			}
353
354			for task in expected['tasks']:
355			task['input'] = json.loads(task['input'])
356			task['result'] = json.loads(task['result'])
357			task['published'] = json.loads(task['published'])
358
359			self.assertEqual(action_constants.LIVEACTION_STATUS_SUCCEEDED, status)
360			self.assertDictEqual(expected, result)
361
362			@mock.patch.object(
363			action_utils, 'get_liveaction_by_id',
364			mock.MagicMock(return_value=MOCK_LIVEACTION_RUNNING_WITH_STREAMING_RESULT))
365			@mock.patch.object(
366			executions.ExecutionManager, 'get',
367			mock.MagicMock(return_value=MOCK_WF_EX))
368			@mock.patch.object(
369			tasks.TaskManager, 'list',
370			mock.MagicMock(return_value=[MOCK_WF_EX_TASKS[1]]))
371			@mock.patch.object(
372			tasks.TaskManager, 'get',
373			mock.MagicMock(return_value=MOCK_WF_EX_TASKS[1]))
374			def test_query_with_last_query_time(self):
375			last_query_time = time.time() - 3
376
377			(status, result) = self.querier.query(
378			uuid.uuid4().hex,
379			MOCK_QRY_CONTEXT,
380			last_query_time=last_query_time
381			)
382
383			expected = {
384			'k1': 'v1',
385			'tasks': copy.deepcopy(MOCK_WF_EX_TASKS_DATA),
386			'extra': {
387			'state': MOCK_WF_EX.state,
388			'state_info': MOCK_WF_EX.state_info
389			}
390			}
391
392			for task in expected['tasks']:
393			task['input'] = json.loads(task['input'])
394			task['result'] = json.loads(task['result'])
395			task['published'] = json.loads(task['published'])
396
397			self.assertEqual(action_constants.LIVEACTION_STATUS_SUCCEEDED, status)
398			self.assertDictEqual(expected, result)
399
400			@mock.patch.object(
401			action_utils, 'get_liveaction_by_id',
402			mock.MagicMock(side_effect=db_exc.StackStormDBObjectNotFoundError()))
403			def test_query_liveaction_not_found(self):
404			self.assertRaises(
405			db_exc.StackStormDBObjectNotFoundError,
406			self.querier.query,
407			uuid.uuid4().hex,
408			MOCK_QRY_CONTEXT
409			)
410
411			@mock.patch.object(
412			action_utils, 'get_liveaction_by_id',
413			mock.MagicMock(return_value=MOCK_LIVEACTION_RUNNING))
414			@mock.patch.object(
415			executions.ExecutionManager, 'get',
416			mock.MagicMock(side_effect=[
417			requests.exceptions.ConnectionError(),
418			MOCK_WF_EX]))
419			@mock.patch.object(
420			tasks.TaskManager, 'list',
421			mock.MagicMock(return_value=MOCK_WF_EX_TASKS))
422			@mock.patch.object(
423			tasks.TaskManager, 'get',
424			mock.MagicMock(side_effect=[
425			MOCK_WF_EX_TASKS[0],
426			MOCK_WF_EX_TASKS[1]]))
427			def test_query_get_workflow_retry(self):
428			(status, result) = self.querier.query(uuid.uuid4().hex, MOCK_QRY_CONTEXT)
429
430			expected = {
431			'k1': 'v1',
432			'tasks': copy.deepcopy(MOCK_WF_EX_TASKS_DATA),
433			'extra': {
434			'state': MOCK_WF_EX.state,
435			'state_info': MOCK_WF_EX.state_info
436			}
437			}
438
439			for task in expected['tasks']:
440			task['input'] = json.loads(task['input'])
441			task['result'] = json.loads(task['result'])
442			task['published'] = json.loads(task['published'])
443
444			self.assertEqual(action_constants.LIVEACTION_STATUS_SUCCEEDED, status)
445			self.assertDictEqual(expected, result)
446
447			calls = [call(MOCK_QRY_CONTEXT['mistral']['execution_id']) for i in range(0, 2)]
448			executions.ExecutionManager.get.assert_has_calls(calls)
449
450			@mock.patch.object(
451			action_utils, 'get_liveaction_by_id',
452			mock.MagicMock(return_value=MOCK_LIVEACTION_RUNNING))
453			@mock.patch.object(
454			executions.ExecutionManager, 'get',
455			mock.MagicMock(side_effect=[requests.exceptions.ConnectionError()] * 4))
456			def test_query_get_workflow_retry_exhausted(self):
457			self.assertRaises(
458			requests.exceptions.ConnectionError,
459			self.querier.query,
460			uuid.uuid4().hex,
461			MOCK_QRY_CONTEXT)
462
463			calls = [call(MOCK_QRY_CONTEXT['mistral']['execution_id']) for i in range(0, 2)]
464			executions.ExecutionManager.get.assert_has_calls(calls)
465
466			@mock.patch.object(
467			action_utils, 'get_liveaction_by_id',
468			mock.MagicMock(return_value=MOCK_LIVEACTION_RUNNING))
469			@mock.patch.object(
470			executions.ExecutionManager, 'get',
471			mock.MagicMock(
472			side_effect=mistralclient_base.APIException(
473			error_code=404, error_message='Workflow not found.')))
474			def test_query_get_workflow_not_found(self):
475			(status, result) = self.querier.query(uuid.uuid4().hex, MOCK_QRY_CONTEXT)
476
477			self.assertEqual(action_constants.LIVEACTION_STATUS_FAILED, status)
478			self.assertEqual('Workflow not found.', result)
479
480			@mock.patch.object(
481			action_utils, 'get_liveaction_by_id',
482			mock.MagicMock(return_value=MOCK_LIVEACTION_RUNNING))
483			@mock.patch.object(
484			executions.ExecutionManager, 'get',
485			mock.MagicMock(return_value=MOCK_WF_EX))
486			@mock.patch.object(
487			tasks.TaskManager, 'list',
488			mock.MagicMock(side_effect=[
489			requests.exceptions.ConnectionError(),
490			MOCK_WF_EX_TASKS]))
491			@mock.patch.object(
492			tasks.TaskManager, 'get',
493			mock.MagicMock(side_effect=[
494			MOCK_WF_EX_TASKS[0],
495			MOCK_WF_EX_TASKS[1]]))
496			def test_query_list_workflow_tasks_retry(self):
497			(status, result) = self.querier.query(uuid.uuid4().hex, MOCK_QRY_CONTEXT)
498
499			expected = {
500			'k1': 'v1',
501			'tasks': copy.deepcopy(MOCK_WF_EX_TASKS_DATA),
502			'extra': {
503			'state': MOCK_WF_EX.state,
504			'state_info': MOCK_WF_EX.state_info
505			}
506			}
507
508			for task in expected['tasks']:
509			task['input'] = json.loads(task['input'])
510			task['result'] = json.loads(task['result'])
511			task['published'] = json.loads(task['published'])
512
513			self.assertEqual(action_constants.LIVEACTION_STATUS_SUCCEEDED, status)
514			self.assertDictEqual(expected, result)
515
516			mock_call = call(workflow_execution_id=MOCK_QRY_CONTEXT['mistral']['execution_id'])
517			calls = [mock_call for i in range(0, 2)]
518			tasks.TaskManager.list.assert_has_calls(calls)
519
520			calls = [call(MOCK_WF_EX_TASKS[0].id), call(MOCK_WF_EX_TASKS[1].id)]
521			tasks.TaskManager.get.assert_has_calls(calls)
522
523			@mock.patch.object(
524			action_utils, 'get_liveaction_by_id',
525			mock.MagicMock(return_value=MOCK_LIVEACTION_RUNNING))
526			@mock.patch.object(
527			executions.ExecutionManager, 'get',
528			mock.MagicMock(return_value=MOCK_WF_EX))
529			@mock.patch.object(
530			tasks.TaskManager, 'list',
531			mock.MagicMock(return_value=MOCK_WF_EX_TASKS))
532			@mock.patch.object(
533			tasks.TaskManager, 'get',
534			mock.MagicMock(side_effect=[
535			requests.exceptions.ConnectionError(),
536			MOCK_WF_EX_TASKS[0],
537			MOCK_WF_EX_TASKS[1]]))
538			def test_query_get_workflow_tasks_retry(self):
539			(status, result) = self.querier.query(uuid.uuid4().hex, MOCK_QRY_CONTEXT)
540
541			expected = {
542			'k1': 'v1',
543			'tasks': copy.deepcopy(MOCK_WF_EX_TASKS_DATA),
544			'extra': {
545			'state': MOCK_WF_EX.state,
546			'state_info': MOCK_WF_EX.state_info
547			}
548			}
549
550			for task in expected['tasks']:
551			task['input'] = json.loads(task['input'])
552			task['result'] = json.loads(task['result'])
553			task['published'] = json.loads(task['published'])
554
555			self.assertEqual(action_constants.LIVEACTION_STATUS_SUCCEEDED, status)
556			self.assertDictEqual(expected, result)
557
558			calls = [
559			call(MOCK_WF_EX_TASKS[0].id),
560			call(MOCK_WF_EX_TASKS[0].id),
561			call(MOCK_WF_EX_TASKS[1].id)
562			]
563
564			tasks.TaskManager.get.assert_has_calls(calls)
565
566			@mock.patch.object(
567			action_utils, 'get_liveaction_by_id',
568			mock.MagicMock(return_value=MOCK_LIVEACTION_RUNNING))
569			@mock.patch.object(
570			executions.ExecutionManager, 'get',
571			mock.MagicMock(return_value=MOCK_WF_EX))
572			@mock.patch.object(
573			tasks.TaskManager, 'list',
574			mock.MagicMock(side_effect=[requests.exceptions.ConnectionError()] * 4))
575			def test_query_list_workflow_tasks_retry_exhausted(self):
576			self.assertRaises(
577			requests.exceptions.ConnectionError,
578			self.querier.query,
579			uuid.uuid4().hex,
580			MOCK_QRY_CONTEXT)
581
582			mock_call = call(workflow_execution_id=MOCK_QRY_CONTEXT['mistral']['execution_id'])
583			calls = [mock_call for i in range(0, 2)]
584			tasks.TaskManager.list.assert_has_calls(calls)
585
586			@mock.patch.object(
587			action_utils, 'get_liveaction_by_id',
588			mock.MagicMock(return_value=MOCK_LIVEACTION_RUNNING))
589			@mock.patch.object(
590			executions.ExecutionManager, 'get',
591			mock.MagicMock(return_value=MOCK_WF_EX))
592			@mock.patch.object(
593			tasks.TaskManager, 'list',
594			mock.MagicMock(return_value=MOCK_WF_EX_TASKS))
595			@mock.patch.object(
596			tasks.TaskManager, 'get',
597			mock.MagicMock(side_effect=[requests.exceptions.ConnectionError()] * 4))
598			def test_query_get_workflow_tasks_retry_exhausted(self):
599			self.assertRaises(
600			requests.exceptions.ConnectionError,
601			self.querier.query,
602			uuid.uuid4().hex,
603			MOCK_QRY_CONTEXT)
604
605			calls = [
606			call(MOCK_WF_EX_TASKS[0].id),
607			call(MOCK_WF_EX_TASKS[0].id)
608			]
609
610			tasks.TaskManager.get.assert_has_calls(calls)
611
612			@mock.patch.object(
613			action_utils, 'get_liveaction_by_id',
614			mock.MagicMock(return_value=MOCK_LIVEACTION_RUNNING))
615			@mock.patch.object(
616			executions.ExecutionManager, 'get',
617			mock.MagicMock(return_value=MOCK_WF_EX))
618			@mock.patch.object(
619			tasks.TaskManager, 'list',
620			mock.MagicMock(return_value=MOCK_WF_EX_TASKS))
621			@mock.patch.object(
622			tasks.TaskManager, 'get',
623			mock.MagicMock(
624			side_effect=mistralclient_base.APIException(
625			error_code=404, error_message='Task not found.')))
626			def test_query_get_workflow_tasks_not_found(self):
627			(status, result) = self.querier.query(uuid.uuid4().hex, MOCK_QRY_CONTEXT)
628
629			self.assertEqual(action_constants.LIVEACTION_STATUS_FAILED, status)
630			self.assertEqual('Task not found.', result)
631
632			def test_query_missing_context(self):
633			self.assertRaises(Exception, self.querier.query, uuid.uuid4().hex, {})
634
635			def test_query_missing_mistral_execution_id(self):
636			self.assertRaises(Exception, self.querier.query, uuid.uuid4().hex, {'mistral': {}})
637

StackStorm / st2

Push — master ( 0891a6...20e5ca )

test_query_get_workflow_retry_exhausted() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like