openstack
/
python-tripleoclient
Code Issues Proposed changes

Merge "Don't use the delete_deployment_plan workflow, call the action directly"

changes/97/695397/33
Zuul 2020-02-26 08:18:21 +00:00 committed by Gerrit Code Review
parent cdc32888a4 5bcf139b30
commit d6e526a9de
4 changed files with 20 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
tripleoclient/tests/fakes.py
View File

@ -54,6 +54,7 @@ class FakeClientManager(object):
self.auth_ref = None
self.tripleoclient = FakeClientWrapper()
self.workflow_engine = mock.Mock()
self.create_mistral_context = mock.Mock()
class FakeHandle(object):

46
tripleoclient/tests/v1/test_overcloud_plan.py
View File

@ -57,8 +57,7 @@ class TestOvercloudPlanList(utils.TestCommand):
"list_deployment_plans",
autospec=True)
def test_list(self, mock_list_plans):
mock_list_plans.return_value = (
['test-plan-1', 'test-plan-2'])
mock_list_plans.return_value = (['test-plan-1', 'test-plan-2'])
result = self.cmd.take_action(None)
@ -72,51 +71,30 @@ class TestOvercloudDeletePlan(utils.TestCommand):
super(TestOvercloudDeletePlan, self).setUp()
self.cmd = overcloud_plan.DeletePlan(self.app, None)
self.app.client_manager.workflow_engine = mock.Mock()
self.tripleoclient = mock.Mock()
self.context = mock.Mock()
self.app.client_manager.create_mistral_context = self.context
self.websocket = mock.Mock()
self.websocket.__enter__ = lambda s: self.websocket
self.websocket.__exit__ = lambda s, *exc: None
self.tripleoclient = mock.Mock()
self.tripleoclient.messaging_websocket.return_value = self.websocket
self.app.client_manager.tripleoclient = self.tripleoclient
self.workflow = self.app.client_manager.workflow_engine
def test_delete_plan(self):
@mock.patch("tripleo_common.actions.plan.DeletePlanAction.run",
return_value=None)
def test_delete_plan(self, mock_run):
parsed_args = self.check_parser(self.cmd, ['test-plan'],
[('plans', ['test-plan'])])
self.websocket.wait_for_messages.return_value = iter([{
"execution_id": "IDID",
"status": "SUCCESS"
}])
self.cmd.take_action(parsed_args)
self.workflow.executions.create.assert_called_once_with(
'tripleo.plan_management.v1.delete_deployment_plan',
workflow_input={'container': 'test-plan'})
mock_run.assert_called_once_with(self.context())
def test_delete_multiple_plans(self):
@mock.patch("tripleo_common.actions.plan.DeletePlanAction.run",
return_value=None)
def test_delete_multiple_plans(self, mock_run):
argslist = ['test-plan1', 'test-plan2']
verifylist = [('plans', ['test-plan1', 'test-plan2'])]
parsed_args = self.check_parser(self.cmd, argslist, verifylist)
self.websocket.wait_for_messages.return_value = iter([{
"execution_id": "IDID",
"status": "SUCCESS"
}])
self.cmd.take_action(parsed_args)
self.workflow.executions.create.assert_has_calls([
mock.call('tripleo.plan_management.v1.delete_deployment_plan',
workflow_input={'container': 'test-plan1'}),
mock.call('tripleo.plan_management.v1.delete_deployment_plan',
workflow_input={'container': 'test-plan2'}),
], any_order=True)
mock_run.assert_has_calls([
mock.call(self.context()), mock.call(self.context())])
class TestOvercloudCreatePlan(utils.TestCommand):

15
tripleoclient/tests/workflows/test_plan_management.py
View File

@ -182,21 +182,6 @@ class TestPlanCreationWorkflows(utils.TestCommand):
self.tripleoclient.object_store.put_object.assert_called_once_with(
'test-overcloud', 'network_data.yaml', mock_open_context())
def test_delete_plan(self):
output = mock.Mock(output='{"result": ""}')
output.id = "IDID"
self.workflow.action_executions.create.return_value = output
self.workflow.executions.create.return_value = output
self.websocket.wait_for_messages.return_value = self.message_success
plan_management.delete_deployment_plan(
self.app.client_manager,
container='test-overcloud')
self.workflow.executions.create.assert_called_once_with(
'tripleo.plan_management.v1.delete_deployment_plan',
workflow_input={'container': 'test-overcloud'})
@mock.patch('tripleoclient.workflows.plan_management.tarball',
autospec=True)
def test_create_plan_with_password_gen_disabled(self, mock_tarball):

26
tripleoclient/workflows/plan_management.py
View File

@ -92,25 +92,13 @@ def create_deployment_plan(clients, **workflow_input):
'Exception creating plan: {}'.format(payload['message']))
def delete_deployment_plan(clients, **workflow_input):
workflow_client = clients.workflow_engine
tripleoclients = clients.tripleoclient
with tripleoclients.messaging_websocket() as ws:
execution = base.start_workflow(
workflow_client,
'tripleo.plan_management.v1.delete_deployment_plan',
workflow_input=workflow_input
)
for payload in base.wait_for_messages(workflow_client, ws, execution,
_WORKFLOW_TIMEOUT):
if payload.get('message'):
print(payload['message'])
if payload['status'] != 'SUCCESS':
raise exceptions.WorkflowServiceError(
'Exception deleting plan: {}'.format(payload['message']))
def delete_deployment_plan(clients, container):
mistral_context = clients.create_mistral_context()
result = plan.DeletePlanAction(container=container).run(mistral_context)
# The action returns None if there are no errors.
if result and result.error:
raise exceptions.WorkflowServiceError(
'Exception deleting plan: {}'.format(result.error))
def update_deployment_plan(clients, **workflow_input):