817a237267
This will enable the mistral UI to go from task execution screen to the workflow execution screen of the workflow execution created by the task. Change-Id: Ia2ac765a26acc8dcd27134f5f467cddef03fea95 Implements: blueprint filter-workflow-executions-by-creating-task-id
1463 lines
52 KiB
Python
1463 lines
52 KiB
Python
# Copyright (c) 2014 Mirantis, Inc.
|
|
#
|
|
# Licensed 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 tempest.lib import exceptions
|
|
|
|
from mistralclient.tests.functional.cli import base
|
|
from mistralclient.tests.functional.cli.v2 import base_v2
|
|
|
|
|
|
MISTRAL_URL = "http://localhost:8989/v2"
|
|
|
|
|
|
class SimpleMistralCLITests(base.MistralCLIAuth):
|
|
"""Basic tests, check '-list', '-help' commands."""
|
|
|
|
_mistral_url = MISTRAL_URL
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
super(SimpleMistralCLITests, cls).setUpClass()
|
|
|
|
def test_workbooks_list(self):
|
|
workbooks = self.parser.listing(self.mistral('workbook-list'))
|
|
self.assertTableStruct(
|
|
workbooks,
|
|
['Name', 'Tags', 'Created at', 'Updated at']
|
|
)
|
|
|
|
def test_workflow_list(self):
|
|
workflows = self.parser.listing(
|
|
self.mistral('workflow-list'))
|
|
self.assertTableStruct(
|
|
workflows,
|
|
['ID', 'Name', 'Tags', 'Input', 'Created at', 'Updated at']
|
|
)
|
|
|
|
def test_executions_list(self):
|
|
executions = self.parser.listing(
|
|
self.mistral('execution-list'))
|
|
self.assertTableStruct(
|
|
executions,
|
|
['ID', 'Workflow name', 'Workflow ID', 'State', 'Created at',
|
|
'Updated at']
|
|
)
|
|
|
|
def test_tasks_list(self):
|
|
tasks = self.parser.listing(
|
|
self.mistral('task-list'))
|
|
self.assertTableStruct(
|
|
tasks,
|
|
['ID', 'Name', 'Workflow name', 'Execution ID', 'State']
|
|
)
|
|
|
|
def test_cron_trigger_list(self):
|
|
triggers = self.parser.listing(
|
|
self.mistral('cron-trigger-list'))
|
|
self.assertTableStruct(
|
|
triggers,
|
|
['Name', 'Workflow', 'Pattern', 'Next execution time',
|
|
'Remaining executions', 'Created at', 'Updated at']
|
|
)
|
|
|
|
def test_actions_list(self):
|
|
actions = self.parser.listing(
|
|
self.mistral('action-list'))
|
|
self.assertTableStruct(
|
|
actions,
|
|
['Name', 'Is system', 'Input', 'Description',
|
|
'Tags', 'Created at', 'Updated at']
|
|
)
|
|
|
|
def test_environments_list(self):
|
|
envs = self.parser.listing(
|
|
self.mistral('environment-list'))
|
|
self.assertTableStruct(
|
|
envs,
|
|
['Name', 'Description', 'Scope', 'Created at', 'Updated at']
|
|
)
|
|
|
|
def test_action_execution_list(self):
|
|
act_execs = self.parser.listing(
|
|
self.mistral('action-execution-list'))
|
|
self.assertTableStruct(
|
|
act_execs,
|
|
['ID', 'Name', 'Workflow name', 'State', 'Accepted']
|
|
)
|
|
|
|
|
|
class WorkbookCLITests(base_v2.MistralClientTestBase):
|
|
"""Test suite checks commands to work with workbooks."""
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
super(WorkbookCLITests, cls).setUpClass()
|
|
|
|
def test_workbook_create_delete(self):
|
|
wb = self.mistral_admin(
|
|
'workbook-create', params=self.wb_def)
|
|
wb_name = self.get_field_value(wb, "Name")
|
|
|
|
self.assertTableStruct(wb, ['Field', 'Value'])
|
|
|
|
wbs = self.mistral_admin('workbook-list')
|
|
self.assertIn(wb_name, [workbook['Name'] for workbook in wbs])
|
|
|
|
wbs = self.mistral_admin('workbook-list')
|
|
self.assertIn(wb_name, [workbook['Name'] for workbook in wbs])
|
|
|
|
self.mistral_admin('workbook-delete', params=wb_name)
|
|
|
|
wbs = self.mistral_admin('workbook-list')
|
|
self.assertNotIn(wb_name, [workbook['Name'] for workbook in wbs])
|
|
|
|
def test_workbook_create_with_tags(self):
|
|
wb = self.workbook_create(self.wb_with_tags_def)
|
|
|
|
tags = self.get_field_value(wb, 'Tags')
|
|
self.assertIn('tag', tags)
|
|
|
|
def test_workbook_update(self):
|
|
wb = self.workbook_create(self.wb_def)
|
|
wb_name = self.get_field_value(wb, "Name")
|
|
|
|
init_update_at = self.get_field_value(wb, "Updated at")
|
|
tags = self.get_field_value(wb, 'Tags')
|
|
self.assertNotIn('tag', tags)
|
|
|
|
wb = self.mistral_admin(
|
|
'workbook-update', params=self.wb_def)
|
|
update_at = self.get_field_value(wb, "Updated at")
|
|
name = self.get_field_value(wb, 'Name')
|
|
tags = self.get_field_value(wb, 'Tags')
|
|
|
|
self.assertEqual(wb_name, name)
|
|
self.assertNotIn('tag', tags)
|
|
self.assertEqual(init_update_at, update_at)
|
|
|
|
wb = self.mistral_admin(
|
|
'workbook-update', params=self.wb_with_tags_def)
|
|
self.assertTableStruct(wb, ['Field', 'Value'])
|
|
|
|
update_at = self.get_field_value(wb, "Updated at")
|
|
name = self.get_field_value(wb, 'Name')
|
|
tags = self.get_field_value(wb, 'Tags')
|
|
|
|
self.assertEqual(wb_name, name)
|
|
self.assertIn('tag', tags)
|
|
self.assertNotEqual(init_update_at, update_at)
|
|
|
|
def test_workbook_get(self):
|
|
created = self.workbook_create(self.wb_with_tags_def)
|
|
wb_name = self.get_field_value(created, "Name")
|
|
|
|
fetched = self.mistral_admin('workbook-get', params=wb_name)
|
|
|
|
created_wb_name = self.get_field_value(created, 'Name')
|
|
fetched_wb_name = self.get_field_value(fetched, 'Name')
|
|
|
|
self.assertEqual(created_wb_name, fetched_wb_name)
|
|
|
|
created_wb_tag = self.get_field_value(created, 'Tags')
|
|
fetched_wb_tag = self.get_field_value(fetched, 'Tags')
|
|
|
|
self.assertEqual(created_wb_tag, fetched_wb_tag)
|
|
|
|
def test_workbook_get_definition(self):
|
|
wb = self.workbook_create(self.wb_def)
|
|
wb_name = self.get_field_value(wb, "Name")
|
|
|
|
definition = self.mistral_admin(
|
|
'workbook-get-definition', params=wb_name)
|
|
self.assertNotIn('404 Not Found', definition)
|
|
|
|
def test_workbook_validate_with_valid_def(self):
|
|
wb = self.mistral_admin(
|
|
'workbook-validate', params=self.wb_def)
|
|
wb_valid = self.get_field_value(wb, 'Valid')
|
|
wb_error = self.get_field_value(wb, 'Error')
|
|
|
|
self.assertEqual('True', wb_valid)
|
|
self.assertEqual('None', wb_error)
|
|
|
|
def test_workbook_validate_with_invalid_def(self):
|
|
self.create_file('wb.yaml',
|
|
'name: wb\n')
|
|
wb = self.mistral_admin(
|
|
'workbook-validate', params='wb.yaml')
|
|
|
|
wb_valid = self.get_field_value(wb, 'Valid')
|
|
wb_error = self.get_field_value(wb, 'Error')
|
|
|
|
self.assertEqual('False', wb_valid)
|
|
self.assertNotEqual('None', wb_error)
|
|
|
|
|
|
class WorkflowCLITests(base_v2.MistralClientTestBase):
|
|
"""Test suite checks commands to work with workflows."""
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
super(WorkflowCLITests, cls).setUpClass()
|
|
|
|
def test_workflow_create_delete(self):
|
|
init_wfs = self.mistral_admin(
|
|
'workflow-create', params=self.wf_def)
|
|
wf_names = [wf['Name'] for wf in init_wfs]
|
|
self.assertTableStruct(init_wfs, ['Name', 'Created at', 'Updated at'])
|
|
|
|
wfs = self.mistral_admin('workflow-list')
|
|
self.assertIn(wf_names[0], [workflow['Name'] for workflow in wfs])
|
|
|
|
for wf_name in wf_names:
|
|
self.mistral_admin('workflow-delete', params=wf_name)
|
|
|
|
wfs = self.mistral_admin('workflow-list')
|
|
for wf in wf_names:
|
|
self.assertNotIn(wf, [workflow['Name'] for workflow in wfs])
|
|
|
|
def test_create_wf_with_tags(self):
|
|
init_wfs = self.workflow_create(self.wf_def)
|
|
wf_name = init_wfs[1]['Name']
|
|
self.assertTableStruct(init_wfs, ['Name', 'Created at',
|
|
'Updated at', 'Tags'])
|
|
|
|
created_wf_info = self.get_item_info(
|
|
get_from=init_wfs, get_by='Name', value=wf_name)
|
|
|
|
self.assertEqual('tag', created_wf_info['Tags'])
|
|
|
|
def test_workflow_update(self):
|
|
wf = self.workflow_create(self.wf_def)
|
|
wf_name = wf[0]['Name']
|
|
wf_id = wf[0]['ID']
|
|
|
|
created_wf_info = self.get_item_info(
|
|
get_from=wf,
|
|
get_by='Name',
|
|
value=wf_name
|
|
)
|
|
|
|
# Update a workflow with definition unchanged.
|
|
upd_wf = self.mistral_admin(
|
|
'workflow-update',
|
|
params='{0}'.format(self.wf_def)
|
|
)
|
|
|
|
self.assertTableStruct(upd_wf, ['Name', 'Created at', 'Updated at'])
|
|
|
|
updated_wf_info = self.get_item_info(
|
|
get_from=upd_wf,
|
|
get_by='Name',
|
|
value=wf_name
|
|
)
|
|
|
|
self.assertEqual(wf_name, upd_wf[0]['Name'])
|
|
self.assertEqual(
|
|
created_wf_info['Created at'].split(".")[0],
|
|
updated_wf_info['Created at']
|
|
)
|
|
self.assertEqual(
|
|
created_wf_info['Updated at'],
|
|
updated_wf_info['Updated at']
|
|
)
|
|
|
|
# Update a workflow with definition changed.
|
|
upd_wf = self.mistral_admin(
|
|
'workflow-update',
|
|
params='{0}'.format(self.wf_with_delay_def)
|
|
)
|
|
|
|
self.assertTableStruct(upd_wf, ['Name', 'Created at', 'Updated at'])
|
|
|
|
updated_wf_info = self.get_item_info(
|
|
get_from=upd_wf,
|
|
get_by='Name',
|
|
value=wf_name
|
|
)
|
|
|
|
self.assertEqual(wf_name, upd_wf[0]['Name'])
|
|
self.assertEqual(
|
|
created_wf_info['Created at'].split(".")[0],
|
|
updated_wf_info['Created at']
|
|
)
|
|
self.assertNotEqual(
|
|
created_wf_info['Updated at'],
|
|
updated_wf_info['Updated at']
|
|
)
|
|
|
|
# Update a workflow with uuid.
|
|
upd_wf = self.mistral_admin(
|
|
'workflow-update',
|
|
params='{0} --id {1}'.format(self.wf_with_delay_def, wf_id)
|
|
)
|
|
|
|
self.assertTableStruct(upd_wf, ['Name', 'Created at', 'Updated at'])
|
|
|
|
updated_wf_info = self.get_item_info(
|
|
get_from=upd_wf,
|
|
get_by='ID',
|
|
value=wf_id
|
|
)
|
|
|
|
self.assertEqual(wf_name, upd_wf[0]['Name'])
|
|
self.assertEqual(
|
|
created_wf_info['Created at'].split(".")[0],
|
|
updated_wf_info['Created at']
|
|
)
|
|
self.assertNotEqual(
|
|
created_wf_info['Updated at'],
|
|
updated_wf_info['Updated at']
|
|
)
|
|
|
|
def test_workflow_update_truncate_input(self):
|
|
input_value = "very_long_input_parameter_name_that_should_be_truncated"
|
|
wf_def = """
|
|
version: "2.0"
|
|
workflow1:
|
|
input:
|
|
- {0}
|
|
tasks:
|
|
task1:
|
|
action: std.noop
|
|
""".format(input_value)
|
|
self.create_file('wf.yaml', wf_def)
|
|
self.workflow_create('wf.yaml')
|
|
upd_wf = self.mistral_admin(
|
|
'workflow-update', params='wf.yaml')
|
|
upd_wf_info = self.get_item_info(
|
|
get_from=upd_wf, get_by='Name', value='workflow1')
|
|
|
|
self.assertEqual(upd_wf_info['Input'][:-3], input_value[:25])
|
|
|
|
def test_workflow_get(self):
|
|
created = self.workflow_create(self.wf_def)
|
|
wf_name = created[0]['Name']
|
|
|
|
fetched = self.mistral_admin('workflow-get', params=wf_name)
|
|
fetched_wf_name = self.get_field_value(fetched, 'Name')
|
|
self.assertEqual(wf_name, fetched_wf_name)
|
|
|
|
def test_workflow_get_with_id(self):
|
|
created = self.workflow_create(self.wf_def)
|
|
wf_name = created[0]['Name']
|
|
wf_id = created[0]['ID']
|
|
|
|
fetched = self.mistral_admin('workflow-get', params=wf_id)
|
|
fetched_wf_name = self.get_field_value(fetched, 'Name')
|
|
self.assertEqual(wf_name, fetched_wf_name)
|
|
|
|
def test_workflow_get_definition(self):
|
|
wf = self.workflow_create(self.wf_def)
|
|
wf_name = wf[0]['Name']
|
|
|
|
definition = self.mistral_admin(
|
|
'workflow-get-definition', params=wf_name)
|
|
self.assertNotIn('404 Not Found', definition)
|
|
|
|
def test_workflow_validate_with_valid_def(self):
|
|
wf = self.mistral_admin(
|
|
'workflow-validate', params=self.wf_def)
|
|
wf_valid = self.get_field_value(wf, 'Valid')
|
|
wf_error = self.get_field_value(wf, 'Error')
|
|
|
|
self.assertEqual('True', wf_valid)
|
|
self.assertEqual('None', wf_error)
|
|
|
|
def test_workflow_validate_with_invalid_def(self):
|
|
self.create_file('wf.yaml',
|
|
'name: wf\n')
|
|
wf = self.mistral_admin(
|
|
'workflow-validate', params='wf.yaml')
|
|
|
|
wf_valid = self.get_field_value(wf, 'Valid')
|
|
wf_error = self.get_field_value(wf, 'Error')
|
|
|
|
self.assertEqual('False', wf_valid)
|
|
self.assertNotEqual('None', wf_error)
|
|
|
|
|
|
class ExecutionCLITests(base_v2.MistralClientTestBase):
|
|
"""Test suite checks commands to work with executions."""
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
super(ExecutionCLITests, cls).setUpClass()
|
|
|
|
def setUp(self):
|
|
super(ExecutionCLITests, self).setUp()
|
|
|
|
wfs = self.workflow_create(self.wf_def)
|
|
|
|
self.direct_wf = wfs[0]
|
|
self.reverse_wf = wfs[1]
|
|
|
|
self.create_file('input', '{\n "farewell": "Bye"\n}\n')
|
|
self.create_file('task_name', '{\n "task_name": "goodbye"\n}\n')
|
|
|
|
def test_execution_create_delete(self):
|
|
execution = self.mistral_admin(
|
|
'execution-create',
|
|
params='{0} -d "execution test"'.format(self.direct_wf['Name'])
|
|
)
|
|
exec_id = self.get_field_value(execution, 'ID')
|
|
self.assertTableStruct(execution, ['Field', 'Value'])
|
|
|
|
wf_name = self.get_field_value(execution, 'Workflow name')
|
|
wf_id = self.get_field_value(execution, 'Workflow ID')
|
|
created_at = self.get_field_value(execution, 'Created at')
|
|
description = self.get_field_value(execution, 'Description')
|
|
|
|
self.assertEqual(self.direct_wf['Name'], wf_name)
|
|
self.assertIsNotNone(wf_id)
|
|
self.assertIsNotNone(created_at)
|
|
self.assertEqual("execution test", description)
|
|
|
|
execs = self.mistral_admin('execution-list')
|
|
self.assertIn(exec_id, [ex['ID'] for ex in execs])
|
|
self.assertIn(wf_name, [ex['Workflow name'] for ex in execs])
|
|
|
|
self.mistral_admin('execution-delete', params=exec_id)
|
|
|
|
def test_execution_create_with_input_and_start_task(self):
|
|
execution = self.execution_create(
|
|
"%s input task_name" % self.reverse_wf['Name'])
|
|
exec_id = self.get_field_value(execution, 'ID')
|
|
|
|
result = self.wait_execution_success(exec_id)
|
|
self.assertTrue(result)
|
|
|
|
def test_execution_update(self):
|
|
execution = self.execution_create(self.direct_wf['Name'])
|
|
|
|
exec_id = self.get_field_value(execution, 'ID')
|
|
status = self.get_field_value(execution, 'State')
|
|
|
|
self.assertEqual('RUNNING', status)
|
|
|
|
# update execution state
|
|
execution = self.mistral_admin(
|
|
'execution-update', params='{0} -s PAUSED'.format(exec_id))
|
|
|
|
updated_exec_id = self.get_field_value(execution, 'ID')
|
|
status = self.get_field_value(execution, 'State')
|
|
|
|
self.assertEqual(exec_id, updated_exec_id)
|
|
self.assertEqual('PAUSED', status)
|
|
|
|
# update execution description
|
|
execution = self.mistral_admin(
|
|
'execution-update',
|
|
params='{0} -d "execution update test"'.format(exec_id)
|
|
)
|
|
|
|
description = self.get_field_value(execution, 'Description')
|
|
|
|
self.assertEqual("execution update test", description)
|
|
|
|
def test_execution_get(self):
|
|
execution = self.execution_create(self.direct_wf['Name'])
|
|
exec_id = self.get_field_value(execution, 'ID')
|
|
|
|
execution = self.mistral_admin(
|
|
'execution-get',
|
|
params='{0}'.format(exec_id)
|
|
)
|
|
|
|
gotten_id = self.get_field_value(execution, 'ID')
|
|
wf_name = self.get_field_value(execution, 'Workflow name')
|
|
wf_id = self.get_field_value(execution, 'Workflow ID')
|
|
|
|
self.assertIsNotNone(wf_id)
|
|
self.assertEqual(exec_id, gotten_id)
|
|
self.assertEqual(self.direct_wf['Name'], wf_name)
|
|
|
|
def test_execution_get_input(self):
|
|
execution = self.execution_create(self.direct_wf['Name'])
|
|
exec_id = self.get_field_value(execution, 'ID')
|
|
|
|
ex_input = self.mistral_admin('execution-get-input', params=exec_id)
|
|
self.assertEqual([], ex_input)
|
|
|
|
def test_execution_get_output(self):
|
|
execution = self.execution_create(self.direct_wf['Name'])
|
|
exec_id = self.get_field_value(execution, 'ID')
|
|
|
|
ex_output = self.mistral_admin(
|
|
'execution-get-output',
|
|
params=exec_id
|
|
)
|
|
|
|
self.assertEqual([], ex_output)
|
|
|
|
def test_executions_list_with_task(self):
|
|
wrapping_wf = self.workflow_create(self.wf_wrapping_wf)
|
|
decoy = self.execution_create(wrapping_wf[-1]['Name'])
|
|
wrapping_wf_ex = self.execution_create(wrapping_wf[-1]['Name'])
|
|
|
|
wrapping_wf_ex_id = self.get_field_value(wrapping_wf_ex, 'ID')
|
|
|
|
self.assertIsNot(wrapping_wf_ex_id, self.get_field_value(decoy, 'ID'))
|
|
|
|
tasks = self.mistral_admin(
|
|
'task-list',
|
|
params=wrapping_wf_ex_id
|
|
)
|
|
|
|
wrapping_task_id = tasks[-1]['ID']
|
|
|
|
wf_execs = self.mistral_cli(
|
|
True,
|
|
'execution-list',
|
|
params="--task {}".format(wrapping_task_id)
|
|
)
|
|
|
|
self.assertEqual(1, len(wf_execs))
|
|
wf_exec = wf_execs[0]
|
|
self.assertEqual(wrapping_task_id, wf_exec['Task Execution ID'])
|
|
|
|
def test_executions_list_with_pagination(self):
|
|
wf_ex1 = self.execution_create(
|
|
params='{0} -d "a"'.format(self.direct_wf['Name'])
|
|
)
|
|
|
|
wf_ex2 = self.execution_create(
|
|
params='{0} -d "b"'.format(self.direct_wf['Name'])
|
|
)
|
|
|
|
wf_execs = self.mistral_cli(True, 'execution-list')
|
|
|
|
self.assertEqual(2, len(wf_execs))
|
|
|
|
wf_execs = self.mistral_cli(
|
|
True,
|
|
'execution-list',
|
|
params="--limit 1"
|
|
)
|
|
|
|
self.assertEqual(1, len(wf_execs))
|
|
|
|
wf_ex1_id = self.get_field_value(wf_ex1, 'ID')
|
|
wf_ex2_id = self.get_field_value(wf_ex2, 'ID')
|
|
|
|
wf_execs = self.mistral_cli(
|
|
True,
|
|
'execution-list',
|
|
params="--marker %s" % wf_ex1_id
|
|
)
|
|
|
|
self.assertNotIn(wf_ex1_id, [ex['ID'] for ex in wf_execs])
|
|
self.assertIn(wf_ex2_id, [ex['ID'] for ex in wf_execs])
|
|
|
|
wf_execs = self.mistral_cli(
|
|
True,
|
|
'execution-list',
|
|
params="--sort_keys Description"
|
|
)
|
|
|
|
self.assertIn(wf_ex1_id, [ex['ID'] for ex in wf_execs])
|
|
self.assertIn(wf_ex2_id, [ex['ID'] for ex in wf_execs])
|
|
|
|
wf_ex1_index = -1
|
|
wf_ex2_index = -1
|
|
|
|
for idx, ex in enumerate(wf_execs):
|
|
if ex['ID'] == wf_ex1_id:
|
|
wf_ex1_index = idx
|
|
elif ex['ID'] == wf_ex2_id:
|
|
wf_ex2_index = idx
|
|
|
|
self.assertTrue(wf_ex1_index < wf_ex2_index)
|
|
|
|
wf_execs = self.mistral_cli(
|
|
True,
|
|
'execution-list',
|
|
params="--sort_keys Description --sort_dirs=desc"
|
|
)
|
|
|
|
self.assertIn(wf_ex1_id, [ex['ID'] for ex in wf_execs])
|
|
self.assertIn(wf_ex2_id, [ex['ID'] for ex in wf_execs])
|
|
|
|
wf_ex1_index = -1
|
|
wf_ex2_index = -1
|
|
|
|
for idx, ex in enumerate(wf_execs):
|
|
if ex['ID'] == wf_ex1_id:
|
|
wf_ex1_index = idx
|
|
elif ex['ID'] == wf_ex2_id:
|
|
wf_ex2_index = idx
|
|
|
|
self.assertTrue(wf_ex1_index > wf_ex2_index)
|
|
|
|
|
|
class CronTriggerCLITests(base_v2.MistralClientTestBase):
|
|
"""Test suite checks commands to work with cron-triggers."""
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
super(CronTriggerCLITests, cls).setUpClass()
|
|
|
|
def setUp(self):
|
|
super(CronTriggerCLITests, self).setUp()
|
|
|
|
wf = self.workflow_create(self.wf_def)
|
|
self.wf_name = wf[0]['Name']
|
|
|
|
def test_cron_trigger_create_delete(self):
|
|
trigger = self.mistral_admin(
|
|
'cron-trigger-create',
|
|
params=('trigger %s {} --pattern "5 * * * *" --count 5'
|
|
' --first-time "4242-12-25 13:37"' % self.wf_name))
|
|
self.assertTableStruct(trigger, ['Field', 'Value'])
|
|
|
|
tr_name = self.get_field_value(trigger, 'Name')
|
|
wf_name = self.get_field_value(trigger, 'Workflow')
|
|
created_at = self.get_field_value(trigger, 'Created at')
|
|
remain = self.get_field_value(trigger, 'Remaining executions')
|
|
next_time = self.get_field_value(trigger, 'Next execution time')
|
|
|
|
self.assertEqual('trigger', tr_name)
|
|
self.assertEqual(self.wf_name, wf_name)
|
|
self.assertIsNotNone(created_at)
|
|
self.assertEqual("4242-12-25 13:37:00", next_time)
|
|
self.assertEqual("5", remain)
|
|
|
|
trgs = self.mistral_admin('cron-trigger-list')
|
|
self.assertIn(tr_name, [tr['Name'] for tr in trgs])
|
|
self.assertIn(wf_name, [tr['Workflow'] for tr in trgs])
|
|
|
|
self.mistral('cron-trigger-delete', params=tr_name)
|
|
|
|
trgs = self.mistral_admin('cron-trigger-list')
|
|
self.assertNotIn(tr_name, [tr['Name'] for tr in trgs])
|
|
|
|
def test_two_cron_triggers_for_one_wf(self):
|
|
self.cron_trigger_create(
|
|
'trigger1', self.wf_name, '{}', "5 * * * *")
|
|
self.cron_trigger_create(
|
|
'trigger2', self.wf_name, '{}', "15 * * * *")
|
|
|
|
trgs = self.mistral_admin('cron-trigger-list')
|
|
self.assertIn("trigger1", [tr['Name'] for tr in trgs])
|
|
self.assertIn("trigger2", [tr['Name'] for tr in trgs])
|
|
|
|
def test_cron_trigger_get(self):
|
|
trigger = self.cron_trigger_create(
|
|
'trigger', self.wf_name, '{}', "5 * * * *")
|
|
self.assertTableStruct(trigger, ['Field', 'Value'])
|
|
|
|
tr_name = self.get_field_value(trigger, 'Name')
|
|
|
|
fetched_tr = self.mistral_admin(
|
|
'cron-trigger-get', params='trigger')
|
|
|
|
self.assertTableStruct(trigger, ['Field', 'Value'])
|
|
|
|
tr_name = self.get_field_value(fetched_tr, 'Name')
|
|
wf_name = self.get_field_value(fetched_tr, 'Workflow')
|
|
created_at = self.get_field_value(fetched_tr, 'Created at')
|
|
|
|
self.assertEqual('trigger', tr_name)
|
|
self.assertEqual(self.wf_name, wf_name)
|
|
self.assertIsNotNone(created_at)
|
|
|
|
|
|
class TaskCLITests(base_v2.MistralClientTestBase):
|
|
"""Test suite checks commands to work with tasks."""
|
|
|
|
def setUp(self):
|
|
super(TaskCLITests, self).setUp()
|
|
|
|
wfs = self.workflow_create(self.wf_def)
|
|
|
|
self.direct_wf = wfs[0]
|
|
self.reverse_wf = wfs[1]
|
|
|
|
self.create_file('input', '{\n "farewell": "Bye"\n}\n')
|
|
self.create_file('task_name', '{\n "task_name": "goodbye"\n}\n')
|
|
|
|
def test_task_get(self):
|
|
wf_ex = self.execution_create(self.direct_wf['Name'])
|
|
wf_ex_id = self.get_field_value(wf_ex, 'ID')
|
|
|
|
tasks = self.mistral_admin('task-list', params=wf_ex_id)
|
|
|
|
created_task_id = tasks[-1]['ID']
|
|
|
|
fetched_task = self.mistral_admin('task-get', params=created_task_id)
|
|
fetched_task_id = self.get_field_value(fetched_task, 'ID')
|
|
task_execution_id = self.get_field_value(fetched_task, 'Execution ID')
|
|
|
|
self.assertEqual(created_task_id, fetched_task_id)
|
|
self.assertEqual(wf_ex_id, task_execution_id)
|
|
|
|
|
|
class ActionCLITests(base_v2.MistralClientTestBase):
|
|
"""Test suite checks commands to work with actions."""
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
super(ActionCLITests, cls).setUpClass()
|
|
|
|
def test_action_create_delete(self):
|
|
init_acts = self.mistral_admin(
|
|
'action-create', params=self.act_def)
|
|
self.assertTableStruct(init_acts, ['Name', 'Is system', 'Input',
|
|
'Description', 'Tags',
|
|
'Created at', 'Updated at'])
|
|
|
|
self.assertIn('greeting', [action['Name'] for action in init_acts])
|
|
self.assertIn('farewell', [action['Name'] for action in init_acts])
|
|
|
|
action_1 = self.get_item_info(
|
|
get_from=init_acts, get_by='Name', value='greeting')
|
|
action_2 = self.get_item_info(
|
|
get_from=init_acts, get_by='Name', value='farewell')
|
|
|
|
self.assertEqual('<none>', action_1['Tags'])
|
|
self.assertEqual('<none>', action_2['Tags'])
|
|
|
|
self.assertEqual('False', action_1['Is system'])
|
|
self.assertEqual('False', action_2['Is system'])
|
|
|
|
self.assertEqual('name', action_1['Input'])
|
|
self.assertEqual('None', action_2['Input'])
|
|
|
|
acts = self.mistral_admin('action-list')
|
|
self.assertIn(action_1['Name'], [action['Name'] for action in acts])
|
|
self.assertIn(action_2['Name'], [action['Name'] for action in acts])
|
|
|
|
self.mistral_admin(
|
|
'action-delete', params='{0}'.format(action_1['Name']))
|
|
self.mistral_admin(
|
|
'action-delete', params='{0}'.format(action_2['Name']))
|
|
|
|
acts = self.mistral_admin('action-list')
|
|
self.assertNotIn(action_1['Name'], [action['Name'] for action in acts])
|
|
self.assertNotIn(action_2['Name'], [action['Name'] for action in acts])
|
|
|
|
def test_action_update(self):
|
|
acts = self.action_create(self.act_def)
|
|
|
|
created_action = self.get_item_info(
|
|
get_from=acts, get_by='Name', value='greeting')
|
|
|
|
acts = self.mistral_admin('action-update', params=self.act_def)
|
|
|
|
updated_action = self.get_item_info(
|
|
get_from=acts, get_by='Name', value='greeting')
|
|
|
|
self.assertEqual(created_action['Created at'].split(".")[0],
|
|
updated_action['Created at'])
|
|
self.assertEqual(created_action['Name'], updated_action['Name'])
|
|
self.assertEqual(created_action['Updated at'],
|
|
updated_action['Updated at'])
|
|
|
|
acts = self.mistral_admin('action-update', params=self.act_tag_def)
|
|
|
|
updated_action = self.get_item_info(
|
|
get_from=acts, get_by='Name', value='greeting')
|
|
|
|
self.assertEqual('tag, tag1', updated_action['Tags'])
|
|
self.assertEqual(created_action['Created at'].split(".")[0],
|
|
updated_action['Created at'])
|
|
self.assertEqual(created_action['Name'], updated_action['Name'])
|
|
self.assertNotEqual(created_action['Updated at'],
|
|
updated_action['Updated at'])
|
|
|
|
def test_action_update_with_id(self):
|
|
acts = self.action_create(self.act_def)
|
|
|
|
created_action = self.get_item_info(
|
|
get_from=acts,
|
|
get_by='Name',
|
|
value='greeting'
|
|
)
|
|
|
|
action_id = created_action['ID']
|
|
|
|
params = '{0} --id {1}'.format(self.act_tag_def, action_id)
|
|
acts = self.mistral_admin('action-update', params=params)
|
|
|
|
updated_action = self.get_item_info(
|
|
get_from=acts,
|
|
get_by='ID',
|
|
value=action_id
|
|
)
|
|
|
|
self.assertEqual(
|
|
created_action['Created at'].split(".")[0],
|
|
updated_action['Created at']
|
|
)
|
|
self.assertEqual(created_action['Name'], updated_action['Name'])
|
|
self.assertNotEqual(
|
|
created_action['Updated at'],
|
|
updated_action['Updated at']
|
|
)
|
|
|
|
def test_action_update_truncate_input(self):
|
|
input_value = "very_long_input_parameter_name_that_should_be_truncated"
|
|
act_def = """
|
|
version: "2.0"
|
|
action1:
|
|
input:
|
|
- {0}
|
|
base: std.noop
|
|
""".format(input_value)
|
|
self.create_file('action.yaml', act_def)
|
|
self.action_create('action.yaml')
|
|
upd_act = self.mistral_admin(
|
|
'action-update', params='action.yaml')
|
|
upd_act_info = self.get_item_info(
|
|
get_from=upd_act, get_by='Name', value='action1')
|
|
|
|
self.assertEqual(upd_act_info['Input'][:-3], input_value[:25])
|
|
|
|
def test_action_get_definition(self):
|
|
self.action_create(self.act_def)
|
|
|
|
definition = self.mistral_admin(
|
|
'action-get-definition', params='greeting')
|
|
self.assertNotIn('404 Not Found', definition)
|
|
|
|
def test_action_get_with_id(self):
|
|
created = self.action_create(self.act_def)
|
|
action_name = created[0]['Name']
|
|
action_id = created[0]['ID']
|
|
|
|
fetched = self.mistral_admin('action-get', params=action_id)
|
|
fetched_action_name = self.get_field_value(fetched, 'Name')
|
|
self.assertEqual(action_name, fetched_action_name)
|
|
|
|
|
|
class EnvironmentCLITests(base_v2.MistralClientTestBase):
|
|
"""Test suite checks commands to work with environments."""
|
|
|
|
def setUp(self):
|
|
super(EnvironmentCLITests, self).setUp()
|
|
|
|
self.create_file('env.yaml',
|
|
'name: env\n'
|
|
'description: Test env\n'
|
|
'variables:\n'
|
|
' var: "value"')
|
|
|
|
def test_environment_create(self):
|
|
env = self.mistral_admin('environment-create', params='env.yaml')
|
|
env_name = self.get_field_value(env, 'Name')
|
|
env_desc = self.get_field_value(env, 'Description')
|
|
|
|
self.assertTableStruct(env, ['Field', 'Value'])
|
|
|
|
envs = self.mistral_admin('environment-list')
|
|
self.assertIn(env_name, [en['Name'] for en in envs])
|
|
self.assertIn(env_desc, [en['Description'] for en in envs])
|
|
|
|
self.mistral_admin('environment-delete', params=env_name)
|
|
|
|
envs = self.mistral_admin('environment-list')
|
|
self.assertNotIn(env_name, [en['Name'] for en in envs])
|
|
|
|
def test_environment_create_without_description(self):
|
|
self.create_file('env_without_des.yaml',
|
|
'name: env\n'
|
|
'variables:\n'
|
|
' var: "value"')
|
|
|
|
env = self.mistral_admin('environment-create',
|
|
params='env_without_des.yaml')
|
|
|
|
env_name = self.get_field_value(env, 'Name')
|
|
env_desc = self.get_field_value(env, 'Description')
|
|
|
|
self.assertTableStruct(env, ['Field', 'Value'])
|
|
|
|
envs = self.mistral_admin('environment-list')
|
|
self.assertIn(env_name, [en['Name'] for en in envs])
|
|
self.assertIn(env_desc, 'None')
|
|
|
|
self.mistral_admin('environment-delete', params='env')
|
|
|
|
envs = self.mistral_admin('environment-list')
|
|
self.assertNotIn(env_name, [en['Name'] for en in envs])
|
|
|
|
def test_environment_update(self):
|
|
env = self.environment_create('env.yaml')
|
|
env_name = self.get_field_value(env, 'Name')
|
|
env_desc = self.get_field_value(env, 'Description')
|
|
env_created_at = self.get_field_value(env, 'Created at')
|
|
env_updated_at = self.get_field_value(env, 'Updated at')
|
|
|
|
self.assertIsNotNone(env_created_at)
|
|
self.assertEqual('None', env_updated_at)
|
|
|
|
self.create_file('env_upd.yaml',
|
|
'name: env\n'
|
|
'description: Updated env\n'
|
|
'variables:\n'
|
|
' var: "value"')
|
|
|
|
env = self.mistral_admin('environment-update', params='env_upd.yaml')
|
|
self.assertTableStruct(env, ['Field', 'Value'])
|
|
|
|
updated_env_name = self.get_field_value(env, 'Name')
|
|
updated_env_desc = self.get_field_value(env, 'Description')
|
|
updated_env_created_at = self.get_field_value(env, 'Created at')
|
|
updated_env_updated_at = self.get_field_value(env, 'Updated at')
|
|
|
|
self.assertEqual(env_name, updated_env_name)
|
|
self.assertNotEqual(env_desc, updated_env_desc)
|
|
self.assertEqual('Updated env', updated_env_desc)
|
|
self.assertEqual(env_created_at.split('.')[0], updated_env_created_at)
|
|
self.assertIsNotNone(updated_env_updated_at)
|
|
|
|
def test_environment_get(self):
|
|
env = self.environment_create('env.yaml')
|
|
env_name = self.get_field_value(env, 'Name')
|
|
env_desc = self.get_field_value(env, 'Description')
|
|
|
|
env = self.mistral_admin('environment-get', params=env_name)
|
|
fetched_env_name = self.get_field_value(env, 'Name')
|
|
fetched_env_desc = self.get_field_value(env, 'Description')
|
|
|
|
self.assertTableStruct(env, ['Field', 'Value'])
|
|
self.assertEqual(env_name, fetched_env_name)
|
|
self.assertEqual(env_desc, fetched_env_desc)
|
|
|
|
|
|
class ActionExecutionCLITests(base_v2.MistralClientTestBase):
|
|
"""Test suite checks commands to work with action executions."""
|
|
|
|
def setUp(self):
|
|
super(ActionExecutionCLITests, self).setUp()
|
|
|
|
wfs = self.workflow_create(self.wf_def)
|
|
|
|
self.direct_wf = wfs[0]
|
|
|
|
direct_wf_exec = self.execution_create(self.direct_wf['Name'])
|
|
|
|
self.direct_ex_id = self.get_field_value(direct_wf_exec, 'ID')
|
|
|
|
def test_act_execution_get(self):
|
|
self.wait_execution_success(self.direct_ex_id)
|
|
|
|
task = self.mistral_admin('task-list', params=self.direct_ex_id)[0]
|
|
|
|
act_ex_from_list = self.mistral_admin(
|
|
'action-execution-list',
|
|
params=task['ID']
|
|
)[0]
|
|
|
|
act_ex = self.mistral_admin(
|
|
'action-execution-get',
|
|
params=act_ex_from_list['ID']
|
|
)
|
|
|
|
wf_name = self.get_field_value(act_ex, 'Workflow name')
|
|
status = self.get_field_value(act_ex, 'State')
|
|
|
|
self.assertEqual(
|
|
act_ex_from_list['ID'],
|
|
self.get_field_value(act_ex, 'ID')
|
|
)
|
|
self.assertEqual(self.direct_wf['Name'], wf_name)
|
|
self.assertEqual('SUCCESS', status)
|
|
|
|
def test_act_execution_create_delete(self):
|
|
action_ex = self.mistral_admin(
|
|
'run-action',
|
|
params="std.echo '{0}' --save-result".format(
|
|
'{"output": "Hello!"}')
|
|
)
|
|
action_ex_id = self.get_field_value(action_ex, 'ID')
|
|
|
|
self.assertTableStruct(action_ex, ['Field', 'Value'])
|
|
|
|
name = self.get_field_value(action_ex, 'Name')
|
|
wf_name = self.get_field_value(action_ex, 'Workflow name')
|
|
task_name = self.get_field_value(action_ex, 'Task name')
|
|
|
|
self.assertEqual('std.echo', name)
|
|
self.assertEqual('None', wf_name)
|
|
self.assertEqual('None', task_name)
|
|
|
|
action_exs = self.mistral_admin('action-execution-list')
|
|
|
|
self.assertIn(action_ex_id, [ex['ID'] for ex in action_exs])
|
|
|
|
self.mistral_admin('action-execution-delete', params=action_ex_id)
|
|
|
|
action_exs = self.mistral_admin('action-execution-list')
|
|
|
|
self.assertNotIn(action_ex_id, [ex['ID'] for ex in action_exs])
|
|
|
|
|
|
class NegativeCLITests(base_v2.MistralClientTestBase):
|
|
"""This class contains negative tests."""
|
|
|
|
def test_wb_list_extra_param(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'workbook-list',
|
|
params='param')
|
|
|
|
def test_wb_get_unexist_wb(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'workbook-get',
|
|
params='wb')
|
|
|
|
def test_wb_get_without_param(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'workbook-get')
|
|
|
|
def test_wb_create_same_name(self):
|
|
self.workbook_create(self.wb_def)
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.workbook_create,
|
|
self.wb_def)
|
|
|
|
def test_wb_create_with_wrong_path_to_definition(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'workbook_create',
|
|
'wb')
|
|
|
|
def test_wb_delete_unexist_wb(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'workbook-delete',
|
|
params='wb')
|
|
|
|
def test_wb_update_wrong_path_to_def(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'workbook-update',
|
|
params='wb')
|
|
|
|
def test_wb_update_nonexistant_wb(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'workbook-update',
|
|
params=self.wb_with_tags_def)
|
|
|
|
def test_wb_create_empty_def(self):
|
|
self.create_file('empty')
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'workbook-create',
|
|
params='empty')
|
|
|
|
def test_wb_update_empty_def(self):
|
|
self.create_file('empty')
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'workbook-update',
|
|
params='empty')
|
|
|
|
def test_wb_get_definition_unexist_wb(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'workbook-get-definition',
|
|
params='wb')
|
|
|
|
def test_wb_create_invalid_def(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'workbook-create',
|
|
params=self.wf_def)
|
|
|
|
def test_wb_update_invalid_def(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'workbook-update',
|
|
params=self.wf_def)
|
|
|
|
def test_wb_update_without_def(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'workbook-update')
|
|
|
|
def test_wf_list_extra_param(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'workflow-list',
|
|
params='param')
|
|
|
|
def test_wf_get_unexist_wf(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'workflow-get',
|
|
params='wf')
|
|
|
|
def test_wf_get_without_param(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'workflow-get')
|
|
|
|
def test_wf_create_without_definition(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'workflow-create',
|
|
params='')
|
|
|
|
def test_wf_create_with_wrong_path_to_definition(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'workflow-create',
|
|
params='wf')
|
|
|
|
def test_wf_delete_unexist_wf(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'workflow-delete',
|
|
params='wf')
|
|
|
|
def test_wf_update_unexist_wf(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'workflow-update',
|
|
params='wf')
|
|
|
|
def test_wf_get_definition_unexist_wf(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'workflow-get-definition',
|
|
params='wf')
|
|
|
|
def test_wf_get_definition_missed_param(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'workflow-get-definition')
|
|
|
|
def test_wf_create_invalid_def(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'workflow-create',
|
|
params=self.wb_def)
|
|
|
|
def test_wf_update_invalid_def(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'workflow-update',
|
|
params=self.wb_def)
|
|
|
|
def test_wf_create_empty_def(self):
|
|
self.create_file('empty')
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'workflow-create',
|
|
params='empty')
|
|
|
|
def test_wf_update_empty_def(self):
|
|
self.create_file('empty')
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'workflow-update',
|
|
params='empty')
|
|
|
|
def test_ex_list_extra_param(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'execution-list',
|
|
params='param')
|
|
|
|
def test_ex_create_unexist_wf(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'execution-create',
|
|
params='wf')
|
|
|
|
def test_ex_create_unexist_task(self):
|
|
wf = self.workflow_create(self.wf_def)
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'execution-create',
|
|
params='%s param {}' % wf[0]['Name'])
|
|
|
|
def test_ex_create_with_invalid_input(self):
|
|
wf = self.workflow_create(self.wf_def)
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'execution-create',
|
|
params="%s input" % wf[0]['Name'])
|
|
|
|
def test_ex_get_nonexist_execution(self):
|
|
wf = self.workflow_create(self.wf_def)
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'execution-get',
|
|
params='%s id' % wf[0]['Name'])
|
|
|
|
def test_ex_create_without_wf_name(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'execution-create')
|
|
|
|
def test_ex_create_reverse_wf_without_start_task(self):
|
|
wf = self.workflow_create(self.wf_def)
|
|
self.create_file('input', '{\n "farewell": "Bye"\n}\n')
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'execution-create ',
|
|
params=wf[1]['Name'])
|
|
|
|
def test_ex_create_missed_input(self):
|
|
self.create_file('empty')
|
|
wf = self.workflow_create(self.wf_def)
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'execution-create empty',
|
|
params=wf[1]['Name'])
|
|
|
|
def test_ex_update_both_state_and_description(self):
|
|
wf = self.workflow_create(self.wf_def)
|
|
execution = self.execution_create(params=wf[0]['Name'])
|
|
exec_id = self.get_field_value(execution, 'ID')
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'execution-update',
|
|
params='%s -s ERROR -d update' % exec_id)
|
|
|
|
def test_ex_delete_nonexistent_execution(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'execution-delete', params='1a2b3c')
|
|
|
|
def test_tr_create_without_pattern(self):
|
|
wf = self.workflow_create(self.wf_def)
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'cron-trigger-create',
|
|
params='tr %s {}' % wf[0]['Name'])
|
|
|
|
def test_tr_create_invalid_pattern(self):
|
|
wf = self.workflow_create(self.wf_def)
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'cron-trigger-create',
|
|
params='tr %s {} --pattern "q"' % wf[0]['Name'])
|
|
|
|
def test_tr_create_invalid_pattern_value_out_of_range(self):
|
|
wf = self.workflow_create(self.wf_def)
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'cron-trigger-create',
|
|
params='tr %s {}'
|
|
' --pattern "80 * * * *"' % wf[0]['Name'])
|
|
|
|
def test_tr_create_nonexistent_wf(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'cron-trigger-create',
|
|
params='tr wb.wf1 {} --pattern "* * * * *"')
|
|
|
|
def test_tr_delete_nonexistant_tr(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'cron-trigger-delete',
|
|
params='tr')
|
|
|
|
def test_tr_get_nonexistant_tr(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'cron-trigger-get',
|
|
params='tr')
|
|
|
|
def test_tr_create_invalid_count(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'cron-trigger-create',
|
|
params='tr wb.wf1 {}'
|
|
' --pattern "* * * * *" --count q')
|
|
|
|
def test_tr_create_negative_count(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'cron-trigger-create',
|
|
params='tr wb.wf1 {}'
|
|
' --pattern "* * * * *" --count -1')
|
|
|
|
def test_tr_create_invalid_first_date(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'cron-trigger-create',
|
|
params='tr wb.wf1 {}'
|
|
' --pattern "* * * * *"'
|
|
' --first-date "q"')
|
|
|
|
def test_tr_create_count_only(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'cron-trigger-create',
|
|
params='tr wb.wf1 {} --count 42')
|
|
|
|
def test_tr_create_date_and_count_without_pattern(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'cron-trigger-create',
|
|
params='tr wb.wf1 {} --count 42'
|
|
' --first-time "4242-12-25 13:37"')
|
|
|
|
def test_action_get_nonexistent(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral,
|
|
'action-get',
|
|
params='nonexist')
|
|
|
|
def test_action_double_creation(self):
|
|
self.action_create(self.act_def)
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'action-create',
|
|
params='{0}'.format(self.act_def))
|
|
|
|
def test_action_create_without_def(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'action-create',
|
|
params='')
|
|
|
|
def test_action_create_invalid_def(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'action-create',
|
|
params='{0}'.format(self.wb_def))
|
|
|
|
def test_action_delete_nonexistent_act(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'action-delete',
|
|
params='nonexist')
|
|
|
|
def test_action_delete_standard_action(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'action-delete',
|
|
params='heat.events_get')
|
|
|
|
def test_action_get_definition_nonexistent_action(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'action-get-definition',
|
|
params='nonexist')
|
|
|
|
def test_task_get_nonexistent_task(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'task-get',
|
|
params='nonexist')
|
|
|
|
def test_env_get_without_param(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'environment-get')
|
|
|
|
def test_env_get_nonexistent(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'environment-get',
|
|
params='nonexist')
|
|
|
|
def test_env_create_same_name(self):
|
|
self.create_file('env.yaml',
|
|
'name: env\n'
|
|
'description: Test env\n'
|
|
'variables:\n'
|
|
' var: "value"')
|
|
|
|
self.environment_create('env.yaml')
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.environment_create,
|
|
'env.yaml')
|
|
|
|
def test_env_create_empty(self):
|
|
self.create_file('env.yaml')
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.environment_create,
|
|
'env.yaml')
|
|
|
|
def test_env_create_with_wrong_path_to_definition(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'execution_create',
|
|
'env')
|
|
|
|
def test_env_delete_unexist_env(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'environment-delete',
|
|
params='env')
|
|
|
|
def test_env_update_wrong_path_to_def(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'environment-update',
|
|
params='env')
|
|
|
|
def test_env_update_empty(self):
|
|
self.create_file('env.yaml')
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'environment-update',
|
|
params='env')
|
|
|
|
def test_env_update_nonexistant_env(self):
|
|
self.create_file('env.yaml',
|
|
'name: env'
|
|
'variables:\n var: "value"')
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'environment-update',
|
|
params='env.yaml')
|
|
|
|
def test_env_create_without_name(self):
|
|
self.create_file('env.yaml',
|
|
'variables:\n var: "value"')
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'environment-create',
|
|
params='env.yaml')
|
|
|
|
def test_env_create_without_variables(self):
|
|
self.create_file('env.yaml',
|
|
'name: env')
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'environment-create',
|
|
params='env.yaml')
|
|
|
|
def test_action_execution_get_without_params(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'action-execution-get')
|
|
|
|
def test_action_execution_get_unexistent_obj(self):
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'action-execution-get',
|
|
params='123456')
|
|
|
|
def test_action_execution_update(self):
|
|
wfs = self.workflow_create(self.wf_def)
|
|
direct_wf_exec = self.execution_create(wfs[0]['Name'])
|
|
direct_ex_id = self.get_field_value(direct_wf_exec, 'ID')
|
|
|
|
self.assertRaises(exceptions.CommandFailed,
|
|
self.mistral_admin,
|
|
'action-execution-update',
|
|
params='%s ERROR' % direct_ex_id)
|