MistralRunnerPolicyTest._drop_all_other_policies() - Code Metrics - Inspection of "Fix the mistral unit tests for concurrency cancell..." - Plexxi/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — plexxi-v2.1.1 ( 2317c4...ed2af2 )

unknown

created 2016-12-20 13:57 UTC

MistralRunnerPolicyTest._drop_all_other_policies() A

↳ Parent: MistralRunnerPolicyTest

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	4
dl	0
loc	5
rs	9.2
c	1
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 uuid

import mock
import yaml

from mistralclient.api.v2 import action_executions
from mistralclient.api.v2 import executions
from mistralclient.api.v2 import workflows
from oslo_config import cfg

# XXX: actionsensor import depends on config being setup.
import st2tests.config as tests_config
tests_config.parse_args()

from mistral_v2 import MistralRunner
import st2common
from st2common.bootstrap import actionsregistrar
from st2common.bootstrap import policiesregistrar
from st2common.bootstrap import runnersregistrar
from st2common.constants import action as action_constants
from st2common.models.db.liveaction import LiveActionDB
from st2common.persistence.liveaction import LiveAction
from st2common.persistence.policy import Policy
from st2common.runners import base as runners
from st2common.services import action as action_service
from st2common.transport.liveaction import LiveActionPublisher
from st2common.transport.publishers import CUDPublisher
from st2common.util import loader
from st2tests import DbTestCase
from st2tests import fixturesloader
from st2tests.mocks.liveaction import MockLiveActionPublisher


TEST_PACK = 'mistral_tests'
TEST_PACK_PATH = fixturesloader.get_fixtures_packs_base_path() + '/' + TEST_PACK

PACKS = [
    TEST_PACK_PATH,
    fixturesloader.get_fixtures_packs_base_path() + '/core'
]

# Non-workbook with a single workflow
WF1_META_FILE_NAME = 'workflow_v2.yaml'
WF1_META_FILE_PATH = TEST_PACK_PATH + '/actions/' + WF1_META_FILE_NAME
WF1_META_CONTENT = loader.load_meta_file(WF1_META_FILE_PATH)
WF1_NAME = WF1_META_CONTENT['pack'] + '.' + WF1_META_CONTENT['name']
WF1_ENTRY_POINT = TEST_PACK_PATH + '/actions/' + WF1_META_CONTENT['entry_point']
WF1_ENTRY_POINT_X = WF1_ENTRY_POINT.replace(WF1_META_FILE_NAME, 'xformed_' + WF1_META_FILE_NAME)
WF1_SPEC = yaml.safe_load(MistralRunner.get_workflow_definition(WF1_ENTRY_POINT_X))
WF1_YAML = yaml.safe_dump(WF1_SPEC, default_flow_style=False)
WF1 = workflows.Workflow(None, {'name': WF1_NAME, 'definition': WF1_YAML})
MISTRAL_EXECUTION = {'id': str(uuid.uuid4()), 'state': 'RUNNING', 'workflow_name': WF1_NAME}
WF1_EXEC = copy.deepcopy(MISTRAL_EXECUTION)


@mock.patch.object(
    CUDPublisher,
    'publish_update',
    mock.MagicMock(return_value=None))
@mock.patch.object(
    CUDPublisher,
    'publish_create',
    mock.MagicMock(side_effect=MockLiveActionPublisher.publish_create))
@mock.patch.object(
    LiveActionPublisher,
    'publish_state',
    mock.MagicMock(side_effect=MockLiveActionPublisher.publish_state))
class MistralRunnerPolicyTest(DbTestCase):

    @classmethod
    def setUpClass(cls):
        super(MistralRunnerPolicyTest, 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')
        cfg.CONF.set_override('api_url', 'http://0.0.0.0:9101', group='auth')

    def setUp(self):
        super(MistralRunnerPolicyTest, self).setUp()

        # Start with a clean database for each test.
        self._establish_connection_and_re_create_db()

        # Register runners.
        runnersregistrar.register_runners()

        actions_registrar = actionsregistrar.ActionsRegistrar(
            use_pack_cache=False,
            fail_on_failure=True
        )

        for pack in PACKS:
            actions_registrar.register_from_pack(pack)

        # Register policies required for the tests.
        policiesregistrar.register_policy_types(st2common)

        policies_registrar = policiesregistrar.PolicyRegistrar(
            use_pack_cache=False,
            fail_on_failure=True
        )

        for pack in PACKS:
            policies_registrar.register_from_pack(pack)

    @classmethod
    def get_runner_class(cls, runner_name):
        return runners.get_runner(runner_name).__class__

    def _drop_all_other_policies(self, test_policy):
        policy_dbs = [policy_db for policy_db in Policy.get_all() if policy_db.ref != test_policy]

        for policy_db in policy_dbs:
            Policy.delete(policy_db, publish=False)

    @mock.patch.object(
        workflows.WorkflowManager, 'list',
        mock.MagicMock(return_value=[]))
    @mock.patch.object(
        workflows.WorkflowManager, 'get',
        mock.MagicMock(return_value=WF1))
    @mock.patch.object(
        workflows.WorkflowManager, 'create',
        mock.MagicMock(return_value=[WF1]))
    @mock.patch.object(
        executions.ExecutionManager, 'create',
        mock.MagicMock(return_value=executions.Execution(None, WF1_EXEC)))
    @mock.patch.object(
        action_executions.ActionExecutionManager, 'update',
        mock.MagicMock(return_value=None))
    def test_cancel_on_task_action_concurrency(self):
        # Delete other policies in the test pack to avoid conflicts.
        required_policy = 'mistral_tests.cancel_on_concurrency'
        self._drop_all_other_policies(required_policy)

        # Get threshold from the policy.
        policy = Policy.get_by_ref(required_policy)
        threshold = policy.parameters.get('threshold', 0)
        self.assertGreater(threshold, 0)

        # Launch instances of the workflow up to threshold.
        for i in range(0, threshold):
            liveaction = LiveActionDB(action=WF1_NAME, parameters={'friend': 'friend' + str(i)})
            liveaction, execution1 = action_service.request(liveaction)
            liveaction = LiveAction.get_by_id(str(liveaction.id))
            self.assertEqual(liveaction.status, action_constants.LIVEACTION_STATUS_RUNNING)

        # Check number of running instances
        running = LiveAction.count(
            action=WF1_NAME, status=action_constants.LIVEACTION_STATUS_RUNNING)

        self.assertEqual(running, threshold)

        # Mock the mistral runner cancel method to assert cancel is called.
        mistral_runner_cls = self.get_runner_class('mistral_v2')

        with mock.patch.object(mistral_runner_cls, 'cancel', mock.MagicMock(return_value=None)):
            # Launch another instance of the workflow with mistral callback defined
            # to indicate that this is executed under a workflow.
            callback = {
                'source': 'mistral',
                'url': 'http://127.0.0.1:8989/v2/action_executions/12345'
            }

            params = {'friend': 'grande animalerie'}

            liveaction2 = LiveActionDB(action=WF1_NAME, parameters=params, callback=callback)
            liveaction2, execution2 = action_service.request(liveaction2)

            action_executions.ActionExecutionManager.update.assert_called_once_with(
                '12345',
                output='{"error": "Execution canceled by user."}',
                state='ERROR'
            )

            liveaction2 = LiveAction.get_by_id(str(liveaction2.id))
            self.assertEqual(liveaction2.status, action_constants.LIVEACTION_STATUS_CANCELED)

            # Assert cancel has been called.
            mistral_runner_cls.cancel.assert_called_once_with()

    @mock.patch.object(
        workflows.WorkflowManager, 'list',
        mock.MagicMock(return_value=[]))
    @mock.patch.object(
        workflows.WorkflowManager, 'get',
        mock.MagicMock(return_value=WF1))
    @mock.patch.object(
        workflows.WorkflowManager, 'create',
        mock.MagicMock(return_value=[WF1]))
    @mock.patch.object(
        executions.ExecutionManager, 'create',
        mock.MagicMock(return_value=executions.Execution(None, WF1_EXEC)))
    @mock.patch.object(
        action_executions.ActionExecutionManager, 'update',
        mock.MagicMock(return_value=None))
    def test_cancel_on_task_action_concurrency_by_attr(self):
        # Delete other policies in the test pack to avoid conflicts.
        required_policy = 'mistral_tests.cancel_on_concurrency_by_attr'
        self._drop_all_other_policies(required_policy)

        # Get threshold from the policy.
        policy = Policy.get_by_ref(required_policy)
        threshold = policy.parameters.get('threshold', 0)
        self.assertGreater(threshold, 0)

        params = {'friend': 'grande animalerie'}

        # Launch instances of the workflow up to threshold.
        for i in range(0, threshold):
            liveaction = LiveActionDB(action=WF1_NAME, parameters=params)
            liveaction, execution1 = action_service.request(liveaction)
            liveaction = LiveAction.get_by_id(str(liveaction.id))
            self.assertEqual(liveaction.status, action_constants.LIVEACTION_STATUS_RUNNING)

        # Check number of running instances
        running = LiveAction.count(
            action=WF1_NAME, status=action_constants.LIVEACTION_STATUS_RUNNING,
            parameters__friend=params['friend'])

        self.assertEqual(running, threshold)

        # Mock the mistral runner cancel method to assert cancel is called.
        mistral_runner_cls = self.get_runner_class('mistral_v2')

        with mock.patch.object(mistral_runner_cls, 'cancel', mock.MagicMock(return_value=None)):
            # Launch another instance of the workflow with mistral callback defined
            # to indicate that this is executed under a workflow.
            callback = {
                'source': 'mistral',
                'url': 'http://127.0.0.1:8989/v2/action_executions/12345'
            }

            liveaction2 = LiveActionDB(action=WF1_NAME, parameters=params, callback=callback)
            liveaction2, execution2 = action_service.request(liveaction2)

            action_executions.ActionExecutionManager.update.assert_called_once_with(
                '12345',
                output='{"error": "Execution canceled by user."}',
                state='ERROR'
            )

            liveaction2 = LiveAction.get_by_id(str(liveaction2.id))
            self.assertEqual(liveaction2.status, action_constants.LIVEACTION_STATUS_CANCELED)

            # Assert cancel has been called.
            mistral_runner_cls.cancel.assert_called_once_with()


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 uuid
18
19			import mock
20			import yaml
21
22			from mistralclient.api.v2 import action_executions
23			from mistralclient.api.v2 import executions
24			from mistralclient.api.v2 import workflows
25			from oslo_config import cfg
26
27			# XXX: actionsensor import depends on config being setup.
28			import st2tests.config as tests_config
29			tests_config.parse_args()
30
31			from mistral_v2 import MistralRunner
32			import st2common
33			from st2common.bootstrap import actionsregistrar
34			from st2common.bootstrap import policiesregistrar
35			from st2common.bootstrap import runnersregistrar
36			from st2common.constants import action as action_constants
37			from st2common.models.db.liveaction import LiveActionDB
38			from st2common.persistence.liveaction import LiveAction
39			from st2common.persistence.policy import Policy
40			from st2common.runners import base as runners
41			from st2common.services import action as action_service
42			from st2common.transport.liveaction import LiveActionPublisher
43			from st2common.transport.publishers import CUDPublisher
44			from st2common.util import loader
45			from st2tests import DbTestCase
46			from st2tests import fixturesloader
47			from st2tests.mocks.liveaction import MockLiveActionPublisher
48
49
50			TEST_PACK = 'mistral_tests'
51			TEST_PACK_PATH = fixturesloader.get_fixtures_packs_base_path() + '/' + TEST_PACK
52
53			PACKS = [
54			TEST_PACK_PATH,
55			fixturesloader.get_fixtures_packs_base_path() + '/core'
56			]
57
58			# Non-workbook with a single workflow
59			WF1_META_FILE_NAME = 'workflow_v2.yaml'
60			WF1_META_FILE_PATH = TEST_PACK_PATH + '/actions/' + WF1_META_FILE_NAME
61			WF1_META_CONTENT = loader.load_meta_file(WF1_META_FILE_PATH)
62			WF1_NAME = WF1_META_CONTENT['pack'] + '.' + WF1_META_CONTENT['name']
63			WF1_ENTRY_POINT = TEST_PACK_PATH + '/actions/' + WF1_META_CONTENT['entry_point']
64			WF1_ENTRY_POINT_X = WF1_ENTRY_POINT.replace(WF1_META_FILE_NAME, 'xformed_' + WF1_META_FILE_NAME)
65			WF1_SPEC = yaml.safe_load(MistralRunner.get_workflow_definition(WF1_ENTRY_POINT_X))
66			WF1_YAML = yaml.safe_dump(WF1_SPEC, default_flow_style=False)
67			WF1 = workflows.Workflow(None, {'name': WF1_NAME, 'definition': WF1_YAML})
68			MISTRAL_EXECUTION = {'id': str(uuid.uuid4()), 'state': 'RUNNING', 'workflow_name': WF1_NAME}
69			WF1_EXEC = copy.deepcopy(MISTRAL_EXECUTION)
70
71
72			@mock.patch.object(
73			CUDPublisher,
74			'publish_update',
75			mock.MagicMock(return_value=None))
76			@mock.patch.object(
77			CUDPublisher,
78			'publish_create',
79			mock.MagicMock(side_effect=MockLiveActionPublisher.publish_create))
80			@mock.patch.object(
81			LiveActionPublisher,
82			'publish_state',
83			mock.MagicMock(side_effect=MockLiveActionPublisher.publish_state))
84			class MistralRunnerPolicyTest(DbTestCase):
85
86			@classmethod
87			def setUpClass(cls):
88			super(MistralRunnerPolicyTest, cls).setUpClass()
89
90			# Override the retry configuration here otherwise st2tests.config.parse_args
91			# in DbTestCase.setUpClass will reset these overrides.
92			cfg.CONF.set_override('retry_exp_msec', 100, group='mistral')
93			cfg.CONF.set_override('retry_exp_max_msec', 200, group='mistral')
94			cfg.CONF.set_override('retry_stop_max_msec', 200, group='mistral')
95			cfg.CONF.set_override('api_url', 'http://0.0.0.0:9101', group='auth')
96
97			def setUp(self):
98			super(MistralRunnerPolicyTest, self).setUp()
99
100			# Start with a clean database for each test.
101			self._establish_connection_and_re_create_db()
102
103			# Register runners.
104			runnersregistrar.register_runners()
105
106			actions_registrar = actionsregistrar.ActionsRegistrar(
107			use_pack_cache=False,
108			fail_on_failure=True
109			)
110
111			for pack in PACKS:
112			actions_registrar.register_from_pack(pack)
113
114			# Register policies required for the tests.
115			policiesregistrar.register_policy_types(st2common)
116
117			policies_registrar = policiesregistrar.PolicyRegistrar(
118			use_pack_cache=False,
119			fail_on_failure=True
120			)
121
122			for pack in PACKS:
123			policies_registrar.register_from_pack(pack)
124
125			@classmethod
126			def get_runner_class(cls, runner_name):
127			return runners.get_runner(runner_name).__class__
128
129			def _drop_all_other_policies(self, test_policy):
130			policy_dbs = [policy_db for policy_db in Policy.get_all() if policy_db.ref != test_policy]
131
132			for policy_db in policy_dbs:
133			Policy.delete(policy_db, publish=False)
134
135			@mock.patch.object(
136			workflows.WorkflowManager, 'list',
137			mock.MagicMock(return_value=[]))
138			@mock.patch.object(
139			workflows.WorkflowManager, 'get',
140			mock.MagicMock(return_value=WF1))
141			@mock.patch.object(
142			workflows.WorkflowManager, 'create',
143			mock.MagicMock(return_value=[WF1]))
144			@mock.patch.object(
145			executions.ExecutionManager, 'create',
146			mock.MagicMock(return_value=executions.Execution(None, WF1_EXEC)))
147			@mock.patch.object(
148			action_executions.ActionExecutionManager, 'update',
149			mock.MagicMock(return_value=None))
150			def test_cancel_on_task_action_concurrency(self):
151			# Delete other policies in the test pack to avoid conflicts.
152			required_policy = 'mistral_tests.cancel_on_concurrency'
153			self._drop_all_other_policies(required_policy)
154
155			# Get threshold from the policy.
156			policy = Policy.get_by_ref(required_policy)
157			threshold = policy.parameters.get('threshold', 0)
158			self.assertGreater(threshold, 0)
159
160			# Launch instances of the workflow up to threshold.
161			for i in range(0, threshold):
162			liveaction = LiveActionDB(action=WF1_NAME, parameters={'friend': 'friend' + str(i)})
163			liveaction, execution1 = action_service.request(liveaction)
164			liveaction = LiveAction.get_by_id(str(liveaction.id))
165			self.assertEqual(liveaction.status, action_constants.LIVEACTION_STATUS_RUNNING)
166
167			# Check number of running instances
168			running = LiveAction.count(
169			action=WF1_NAME, status=action_constants.LIVEACTION_STATUS_RUNNING)
170
171			self.assertEqual(running, threshold)
172
173			# Mock the mistral runner cancel method to assert cancel is called.
174			mistral_runner_cls = self.get_runner_class('mistral_v2')
175
176			with mock.patch.object(mistral_runner_cls, 'cancel', mock.MagicMock(return_value=None)):
177			# Launch another instance of the workflow with mistral callback defined
178			# to indicate that this is executed under a workflow.
179			callback = {
180			'source': 'mistral',
181			'url': 'http://127.0.0.1:8989/v2/action_executions/12345'
182			}
183
184			params = {'friend': 'grande animalerie'}
185
186			liveaction2 = LiveActionDB(action=WF1_NAME, parameters=params, callback=callback)
187			liveaction2, execution2 = action_service.request(liveaction2)
188
189			action_executions.ActionExecutionManager.update.assert_called_once_with(
190			'12345',
191			output='{"error": "Execution canceled by user."}',
192			state='ERROR'
193			)
194
195			liveaction2 = LiveAction.get_by_id(str(liveaction2.id))
196			self.assertEqual(liveaction2.status, action_constants.LIVEACTION_STATUS_CANCELED)
197
198			# Assert cancel has been called.
199			mistral_runner_cls.cancel.assert_called_once_with()
200
201			@mock.patch.object(
202			workflows.WorkflowManager, 'list',
203			mock.MagicMock(return_value=[]))
204			@mock.patch.object(
205			workflows.WorkflowManager, 'get',
206			mock.MagicMock(return_value=WF1))
207			@mock.patch.object(
208			workflows.WorkflowManager, 'create',
209			mock.MagicMock(return_value=[WF1]))
210			@mock.patch.object(
211			executions.ExecutionManager, 'create',
212			mock.MagicMock(return_value=executions.Execution(None, WF1_EXEC)))
213			@mock.patch.object(
214			action_executions.ActionExecutionManager, 'update',
215			mock.MagicMock(return_value=None))
216			def test_cancel_on_task_action_concurrency_by_attr(self):
217			# Delete other policies in the test pack to avoid conflicts.
218			required_policy = 'mistral_tests.cancel_on_concurrency_by_attr'
219			self._drop_all_other_policies(required_policy)
220
221			# Get threshold from the policy.
222			policy = Policy.get_by_ref(required_policy)
223			threshold = policy.parameters.get('threshold', 0)
224			self.assertGreater(threshold, 0)
225
226			params = {'friend': 'grande animalerie'}
227
228			# Launch instances of the workflow up to threshold.
229			for i in range(0, threshold):
230			liveaction = LiveActionDB(action=WF1_NAME, parameters=params)
231			liveaction, execution1 = action_service.request(liveaction)
232			liveaction = LiveAction.get_by_id(str(liveaction.id))
233			self.assertEqual(liveaction.status, action_constants.LIVEACTION_STATUS_RUNNING)
234
235			# Check number of running instances
236			running = LiveAction.count(
237			action=WF1_NAME, status=action_constants.LIVEACTION_STATUS_RUNNING,
238			parameters__friend=params['friend'])
239
240			self.assertEqual(running, threshold)
241
242			# Mock the mistral runner cancel method to assert cancel is called.
243			mistral_runner_cls = self.get_runner_class('mistral_v2')
244
245			with mock.patch.object(mistral_runner_cls, 'cancel', mock.MagicMock(return_value=None)):
246			# Launch another instance of the workflow with mistral callback defined
247			# to indicate that this is executed under a workflow.
248			callback = {
249			'source': 'mistral',
250			'url': 'http://127.0.0.1:8989/v2/action_executions/12345'
251			}
252
253			liveaction2 = LiveActionDB(action=WF1_NAME, parameters=params, callback=callback)
254			liveaction2, execution2 = action_service.request(liveaction2)
255
256			action_executions.ActionExecutionManager.update.assert_called_once_with(
257			'12345',
258			output='{"error": "Execution canceled by user."}',
259			state='ERROR'
260			)
261
262			liveaction2 = LiveAction.get_by_id(str(liveaction2.id))
263			self.assertEqual(liveaction2.status, action_constants.LIVEACTION_STATUS_CANCELED)
264
265			# Assert cancel has been called.
266			mistral_runner_cls.cancel.assert_called_once_with()
267

Plexxi / st2

GitHub Access Token became invalid

Push — plexxi-v2.1.1 ( 2317c4...ed2af2 )

MistralRunnerPolicyTest._drop_all_other_policies() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like