for testing and deploying your application
for finding and fixing issues
for empowering human code reviews
# 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.
from __future__ import absolute_import
import eventlet
import mock
import os
import tempfile
from st2common.bootstrap import actionsregistrar
from st2common.bootstrap import runnersregistrar
from st2common.constants import action as action_constants
from st2common.models.db.liveaction import LiveActionDB
from st2common.persistence.execution import ActionExecution
from st2common.persistence.liveaction import LiveAction
from st2common.services import action as action_service
from st2common.transport.liveaction import LiveActionPublisher
from st2common.transport.publishers import CUDPublisher
from st2common.util import action_db as action_utils
from st2common.util import date as date_utils
from st2tests import DbTestCase
from st2tests import fixturesloader
from st2tests.mocks.liveaction import MockLiveActionPublisherNonBlocking
from six.moves import range
range
It is generally discouraged to redefine built-ins as this makes code very hard to read.
TEST_FIXTURES = {
'chains': [
'test_pause_resume.yaml',
'test_pause_resume_context_result',
'test_pause_resume_with_published_vars.yaml',
'test_pause_resume_with_error.yaml',
'test_pause_resume_with_subworkflow.yaml',
'test_pause_resume_with_context_access.yaml',
'test_pause_resume_with_init_vars.yaml',
'test_pause_resume_with_no_more_task.yaml',
'test_pause_resume_last_task_failed_with_no_next_task.yaml'
],
'actions': [
]
}
TEST_PACK = 'action_chain_tests'
TEST_PACK_PATH = fixturesloader.get_fixtures_packs_base_path() + '/' + TEST_PACK
PACKS = [
TEST_PACK_PATH,
fixturesloader.get_fixtures_packs_base_path() + '/core'
USERNAME = 'stanley'
@mock.patch.object(
CUDPublisher,
'publish_update',
mock.MagicMock(return_value=None))
'publish_create',
LiveActionPublisher,
'publish_state',
mock.MagicMock(side_effect=MockLiveActionPublisherNonBlocking.publish_state))
class ActionChainRunnerPauseResumeTest(DbTestCase):
temp_file_path = None
@classmethod
def setUpClass(cls):
super(ActionChainRunnerPauseResumeTest, cls).setUpClass()
# Register runners.
runnersregistrar.register_runners()
# Register test pack(s).
actions_registrar = actionsregistrar.ActionsRegistrar(
use_pack_cache=False,
fail_on_failure=True
)
for pack in PACKS:
actions_registrar.register_from_pack(pack)
def setUp(self):
super(ActionChainRunnerPauseResumeTest, self).setUp()
# Create temporary directory used by the tests
_, self.temp_file_path = tempfile.mkstemp()
os.chmod(self.temp_file_path, 0o755) # nosec
def tearDown(self):
if self.temp_file_path and os.path.exists(self.temp_file_path):
os.remove(self.temp_file_path)
super(ActionChainRunnerPauseResumeTest, self).tearDown()
def _wait_for_status(self, liveaction, status, interval=0.1, retries=100):
# Wait until the liveaction reaches status.
for i in range(0, retries):
liveaction = LiveAction.get_by_id(str(liveaction.id))
if liveaction.status != status:
eventlet.sleep(interval)
continue
else:
break
return liveaction
def _wait_for_children(self, execution, interval=0.1, retries=100):
# Wait until the execution has children.
execution = ActionExecution.get_by_id(str(execution.id))
if len(getattr(execution, 'children', [])) <= 0:
return execution
def test_chain_pause_resume(self):
# A temp file is created during test setup. Ensure the temp file exists.
# The test action chain will stall until this file is deleted. This gives
# the unit test a moment to run any test related logic.
path = self.temp_file_path
self.assertTrue(os.path.exists(path))
action = TEST_PACK + '.' + 'test_pause_resume'
params = {'tempfile': path, 'message': 'foobar'}
liveaction = LiveActionDB(action=action, parameters=params)
liveaction, execution = action_service.request(liveaction)
# Wait until the liveaction is running.
liveaction = self._wait_for_status(liveaction, action_constants.LIVEACTION_STATUS_RUNNING)
self.assertEqual(liveaction.status, action_constants.LIVEACTION_STATUS_RUNNING)
# Request action chain to pause.
liveaction, execution = action_service.request_pause(liveaction, USERNAME)
# Wait until the liveaction is pausing.
liveaction = self._wait_for_status(liveaction, action_constants.LIVEACTION_STATUS_PAUSING)
extra_info = str(liveaction)
self.assertEqual(liveaction.status, action_constants.LIVEACTION_STATUS_PAUSING, extra_info)
# Delete the temporary file that the action chain is waiting on.
os.remove(path)
self.assertFalse(os.path.exists(path))
# Wait until the liveaction is paused.
liveaction = self._wait_for_status(liveaction, action_constants.LIVEACTION_STATUS_PAUSED)
self.assertEqual(liveaction.status, action_constants.LIVEACTION_STATUS_PAUSED, extra_info)
# Wait for non-blocking threads to complete. Ensure runner is not running.
MockLiveActionPublisherNonBlocking.wait_all()
# Request action chain to resume.
liveaction, execution = action_service.request_resume(liveaction, USERNAME)
# Wait until the liveaction is completed.
liveaction = self._wait_for_status(liveaction, action_constants.LIVEACTION_STATUS_SUCCEEDED)
self.assertEqual(liveaction.status, action_constants.LIVEACTION_STATUS_SUCCEEDED)
# Wait for non-blocking threads to complete.
# Check liveaction result.
self.assertIn('tasks', liveaction.result)
self.assertEqual(len(liveaction.result['tasks']), 2)
def test_chain_pause_resume_with_published_vars(self):
action = TEST_PACK + '.' + 'test_pause_resume_with_published_vars'
self.assertIn('published', liveaction.result)
self.assertDictEqual({'var1': 'foobar', 'var2': 'fubar'}, liveaction.result['published'])
def test_chain_pause_resume_with_published_vars_display_false(self):
params = {'tempfile': path, 'message': 'foobar', 'display_published': False}
self.assertNotIn('published', liveaction.result)
def test_chain_pause_resume_with_error(self):
action = TEST_PACK + '.' + 'test_pause_resume_with_error'
self.assertTrue(liveaction.result['tasks'][0]['result']['failed'])
self.assertEqual(1, liveaction.result['tasks'][0]['result']['return_code'])
self.assertTrue(liveaction.result['tasks'][1]['result']['succeeded'])
self.assertEqual(0, liveaction.result['tasks'][1]['result']['return_code'])
def test_chain_pause_resume_cascade_to_subworkflow(self):
action = TEST_PACK + '.' + 'test_pause_resume_with_subworkflow'
# Wait for subworkflow to register.
execution = self._wait_for_children(execution)
self.assertEqual(len(execution.children), 1)
# Wait until the subworkflow is running.
task1_exec = ActionExecution.get_by_id(execution.children[0])
task1_live = LiveAction.get_by_id(task1_exec.liveaction['id'])
task1_live = self._wait_for_status(task1_live, action_constants.LIVEACTION_STATUS_RUNNING)
self.assertEqual(task1_live.status, action_constants.LIVEACTION_STATUS_RUNNING)
# Wait until the subworkflow is pausing.
task1_live = self._wait_for_status(task1_live, action_constants.LIVEACTION_STATUS_PAUSING)
extra_info = str(task1_live)
self.assertEqual(task1_live.status, action_constants.LIVEACTION_STATUS_PAUSING, extra_info)
# Wait until the subworkflow is paused.
task1_live = self._wait_for_status(task1_live, action_constants.LIVEACTION_STATUS_PAUSED)
self.assertEqual(task1_live.status, action_constants.LIVEACTION_STATUS_PAUSED, extra_info)
self.assertEqual(len(liveaction.result['tasks']), 1)
subworkflow = liveaction.result['tasks'][0]
self.assertEqual(len(subworkflow['result']['tasks']), 1)
self.assertEqual(subworkflow['state'], action_constants.LIVEACTION_STATUS_PAUSED)
self.assertEqual(len(subworkflow['result']['tasks']), 2)
self.assertEqual(subworkflow['state'], action_constants.LIVEACTION_STATUS_SUCCEEDED)
def test_chain_pause_resume_cascade_to_parent_workflow(self):
# Request subworkflow to pause.
task1_live, task1_exec = action_service.request_pause(task1_live, USERNAME)
# Wait until the parent liveaction is paused.
# Request subworkflow to resume.
task1_live, task1_exec = action_service.request_resume(task1_live, USERNAME)
task1_live = self._wait_for_status(task1_live, action_constants.LIVEACTION_STATUS_SUCCEEDED)
self.assertEqual(task1_live.status, action_constants.LIVEACTION_STATUS_SUCCEEDED)
# The parent workflow will stay paused.
self.assertEqual(liveaction.status, action_constants.LIVEACTION_STATUS_PAUSED)
# Check liveaction result of the parent, which should stay the same
# because only the subworkflow was resumed.
# Request parent workflow to resume.
def test_chain_pause_resume_with_context_access(self):
action = TEST_PACK + '.' + 'test_pause_resume_with_context_access'
self.assertEqual(len(liveaction.result['tasks']), 3)
self.assertEqual(liveaction.result['tasks'][2]['result']['stdout'], 'foobar')
def test_chain_pause_resume_with_init_vars(self):
action = TEST_PACK + '.' + 'test_pause_resume_with_init_vars'
self.assertEqual(liveaction.result['tasks'][1]['result']['stdout'], 'FOOBAR')
def test_chain_pause_resume_with_no_more_task(self):
action = TEST_PACK + '.' + 'test_pause_resume_with_no_more_task'
def test_chain_pause_resume_last_task_failed_with_no_next_task(self):
action = TEST_PACK + '.' + 'test_pause_resume_last_task_failed_with_no_next_task'
liveaction = self._wait_for_status(liveaction, action_constants.LIVEACTION_STATUS_FAILED)
self.assertEqual(liveaction.status, action_constants.LIVEACTION_STATUS_FAILED)
self.assertEqual(
liveaction.result['tasks'][0]['state'],
action_constants.LIVEACTION_STATUS_FAILED
def test_chain_pause_resume_status_change(self):
# Tests context_result is updated when last task's status changes between pause and resume
action = TEST_PACK + '.' + 'test_pause_resume_context_result'
liveaction = LiveActionDB(action=action)
last_task_liveaction_id = liveaction.result['tasks'][-1]['liveaction_id']
action_utils.update_liveaction_status(
status=action_constants.LIVEACTION_STATUS_SUCCEEDED,
end_timestamp=date_utils.get_datetime_utc_now(),
result={'foo': 'bar'},
liveaction_id=last_task_liveaction_id
liveaction.status,
action_constants.LIVEACTION_STATUS_SUCCEEDED,
str(liveaction)
self.assertEqual(liveaction.result['tasks'][0]['result']['foo'], 'bar')
StackStorm /
st2
Anthony Shaw
It is generally discouraged to redefine built-ins as this makes code very hard to read.