Remove mistral from the list_available_roles deploy workflow

This change removes all of mistral from the list_available_roles function
by calling the required functions directly.

Story: 2007212
Task: 38433

Change-Id: I78030ffc75398f520d40ac56061cf1d6c35d3332
Signed-off-by: Kevin Carter <kecarter@redhat.com>
This commit is contained in:
Kevin Carter 2020-02-17 17:41:56 -06:00
parent 6fa0e3d1be
commit b49610cea2
No known key found for this signature in database
GPG Key ID: CE94BD890A47B20A
3 changed files with 126 additions and 211 deletions

View File

@ -18,67 +18,85 @@ from osc_lib.tests import utils
from tripleoclient.v1 import overcloud_plan_roles
class TestOvercloudListCurrentRoles(utils.TestCommand):
class BaseTestCommand(utils.TestCommand):
def setUp(self):
super(BaseTestCommand, self).setUp()
tc = self.app.client_manager.tripleoclient = mock.Mock()
tc.object_store.get_object.return_value = (
{},
'{"result": [{"name":"Controller","description":"Test desc",'
'"random": "abcd"},{"name":"Test"}]}'
)
tc.object_store.get_container.return_value = (
'container',
[
{
"name": "Controller",
"description": "Test desc",
"random": "abcd",
"efg": "123",
"ServicesDefault": [
"b",
"c",
"a"
]
}
]
)
self.tripleoclient = tc
class TestOvercloudListCurrentRoles(BaseTestCommand):
def setUp(self):
super(TestOvercloudListCurrentRoles, self).setUp()
self.cmd = overcloud_plan_roles.ListRoles(self.app, None)
self.app.client_manager.workflow_engine = mock.Mock()
self.workflow = self.app.client_manager.workflow_engine
def test_list_empty_on_non_default_plan(self):
self.workflow.action_executions.create.return_value = (
mock.Mock(output='{"result": []}'))
@mock.patch(
'tripleo_common.actions.plan.ListRolesAction.run',
autospec=True,
return_value=[]
)
def test_list_empty_on_non_default_plan(self, mock_list):
arglist = ['--name', 'overcast', '--current']
verifylist = [('name', 'overcast'), ('current', True)]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.workflow.action_executions.create.assert_called_once_with(
'tripleo.role.list',
{'container': 'overcast', 'detail': False},
run_sync=True, save_result=True
)
self.assertEqual(0, len(result[1]))
def test_list(self):
self.workflow.action_executions.create.return_value = (
mock.MagicMock(
output='{"result": ["ObjectStorage", "Controller"]}'))
@mock.patch(
'tripleo_common.actions.plan.ListRolesAction.run',
autospec=True,
return_value=["ObjectStorage", "Controller"]
)
def test_list(self, mock_list):
arglist = ['--current']
verifylist = [('name', 'overcloud'), ('current', True)]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.workflow.action_executions.create.assert_called_once_with(
'tripleo.role.list',
{'container': 'overcloud', 'detail': False},
run_sync=True, save_result=True
)
self.assertEqual(2, len(result[1]))
self.assertEqual([('Controller',), ('ObjectStorage',)], result[1])
def test_list_with_details(self):
self.workflow.action_executions.create.return_value = (
mock.MagicMock(output=(
'{"result": [{"name":"Controller","description":"Test desc",'
'"random": "abcd"},{"name":"Test"}]}')))
@mock.patch(
'tripleo_common.actions.plan.ListRolesAction.run',
autospec=True,
return_value=[
{
"name": "Controller",
"description": "Test desc",
"random": "abcd"
},
{"name": "Test"}
]
)
def test_list_with_details(self, mock_list):
parsed_args = self.check_parser(self.cmd,
['--current', '--detail'],
[])
result = self.cmd.take_action(parsed_args)
self.workflow.action_executions.create.assert_called_once_with(
'tripleo.role.list',
{'container': 'overcloud', 'detail': True},
run_sync=True, save_result=True
)
data = result[1]
self.assertEqual(2, len(data))
@ -87,39 +105,35 @@ class TestOvercloudListCurrentRoles(utils.TestCommand):
self.assertEqual(data[1][0], "Test")
self.assertEqual(data[1][3], "")
def test_list_with_details_empty(self):
self.workflow.action_executions.create.return_value = (
mock.Mock(output='{"result": []}'))
@mock.patch(
'tripleo_common.actions.plan.ListRolesAction.run',
autospec=True,
return_value=[]
)
def test_list_with_details_empty(self, mock_list):
parsed_args = self.check_parser(self.cmd,
['--current', '--detail'],
[])
result = self.cmd.take_action(parsed_args)
self.workflow.action_executions.create.assert_called_once_with(
'tripleo.role.list',
{'container': 'overcloud', 'detail': True},
run_sync=True, save_result=True
)
self.assertEqual(0, len(result[1]))
def test_list_with_details_sorted(self):
self.workflow.action_executions.create.return_value = (
mock.MagicMock(output=(
'{"result": [{"name":"Compute"},{"name":"Random"},'
'{"name": "BlockStorage","ServicesDefault":["c","b","a"]}]}')))
@mock.patch(
'tripleo_common.actions.plan.ListRolesAction.run',
autospec=True,
return_value=[
{"name": "Compute"},
{"name": "Random"},
{"name": "BlockStorage", "ServicesDefault": ["c", "b", "a"]}
]
)
def test_list_with_details_sorted(self, mock_list):
parsed_args = self.check_parser(self.cmd,
['--current', '--detail'],
[])
result = self.cmd.take_action(parsed_args)
self.workflow.action_executions.create.assert_called_once_with(
'tripleo.role.list',
{'container': 'overcloud', 'detail': True},
run_sync=True, save_result=True
)
self.assertEqual(3, len(result[1]))
# Test main list sorted
@ -131,113 +145,21 @@ class TestOvercloudListCurrentRoles(utils.TestCommand):
self.assertEqual(result[1][0][2], "a\nb\nc")
class TestOvercloudListRole(utils.TestCommand):
def setUp(self):
super(TestOvercloudListRole, self).setUp()
self.cmd = overcloud_plan_roles.ListRoles(self.app, None)
self.workflow = self.app.client_manager.workflow_engine = mock.Mock()
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
execution = mock.Mock()
execution.id = "IDID"
self.workflow.executions.create.return_value = execution
def test_list_empty(self):
self.websocket.wait_for_messages.return_value = [{
'execution_id': 'IDID',
'status': 'SUCCESS',
'available_roles': []
}]
arglist = ['--name', 'overcast']
verifylist = [('name', 'overcast')]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.workflow.executions.create.assert_called_once_with(
'tripleo.plan_management.v1.list_available_roles',
workflow_input={'container': 'overcast'},
)
self.assertEqual(0, len(result[1]))
def test_list(self):
self.websocket.wait_for_messages.return_value = [{
'execution_id': 'IDID',
'status': 'SUCCESS',
'available_roles': [{'name': 'ObjectStorage'},
{'name': 'Compute'}]
}]
parsed_args = self.check_parser(self.cmd, [], [('name', 'overcloud')])
result = self.cmd.take_action(parsed_args)
self.workflow.executions.create.assert_called_once_with(
'tripleo.plan_management.v1.list_available_roles',
workflow_input={'container': 'overcloud'},
)
self.assertEqual(2, len(result[1]))
self.assertEqual([('Compute',), ('ObjectStorage',)], result[1])
def test_list_with_details(self):
self.websocket.wait_for_messages.return_value = [{
'execution_id': 'IDID',
'status': 'SUCCESS',
'available_roles': [
{'name': 'Controller', 'description': 'Test description',
'random': 'abcd'},
{'name': 'Test'}]
}]
parsed_args = self.check_parser(self.cmd, ['--detail'], [])
result = self.cmd.take_action(parsed_args)
self.workflow.executions.create.assert_called_once_with(
'tripleo.plan_management.v1.list_available_roles',
workflow_input={'container': 'overcloud'},
)
data = result[1]
self.assertEqual(2, len(data))
self.assertEqual(data[0][0], "Controller")
self.assertEqual(data[0][3], "random: abcd")
self.assertEqual(data[1][0], "Test")
self.assertEqual(data[1][3], "")
class TestOvercloudShowRole(utils.TestCommand):
class TestOvercloudShowRole(BaseTestCommand):
def setUp(self):
super(TestOvercloudShowRole, self).setUp()
self.cmd = overcloud_plan_roles.ShowRole(self.app, None)
self.workflow = self.app.client_manager.workflow_engine = mock.Mock()
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
execution = mock.Mock()
execution.id = "IDID"
self.workflow.executions.create.return_value = execution
def test_role_not_found(self):
self.websocket.wait_for_messages.return_value = [{
'execution_id': 'IDID',
'status': 'SUCCESS',
'available_roles': []
}]
@mock.patch(
'tripleo_common.actions.plan.ListRolesAction.run',
autospec=True,
return_value=[]
)
def test_role_not_found(self, mock_list):
arglist = ['--name', 'overcast', 'doesntexist']
verifylist = [('name', 'overcast'), ('role', 'doesntexist')]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
@ -245,35 +167,3 @@ class TestOvercloudShowRole(utils.TestCommand):
self.assertRaises(exceptions.CommandError,
self.cmd.take_action,
parsed_args)
def test_role(self):
self.websocket.wait_for_messages.return_value = [{
'execution_id': 'IDID',
'status': 'SUCCESS',
'available_roles': [
{"name": "Test", "a": "b"},
{"name": "Controller", "description": "Test desc",
"random": "abcd", "efg": "123",
"ServicesDefault": ["b", "c", "a"]}]}]
arglist = ['Controller']
verifylist = [('name', 'overcloud'), ('role', 'Controller')]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.workflow.executions.create.assert_called_once_with(
'tripleo.plan_management.v1.list_available_roles',
workflow_input={'container': 'overcloud'},
)
self.assertEqual(len(result), 2)
# Check that all the columns are picked up correctly
expected = ['Name', 'Description', 'Services Default', 'efg', 'random']
actual = result[0]
self.assertEqual(expected, actual)
# Check the content
expected = ['Controller', 'Test desc', "a\nb\nc", '123', 'abcd']
actual = result[1]
self.assertEqual(expected, actual)

View File

@ -50,7 +50,7 @@ class ListRoles(command.Lister):
if parsed_args.current:
result = roles.list_roles(
self.app.client_manager.workflow_engine,
self.app.client_manager,
container=parsed_args.name,
detail=parsed_args.detail)
else:

View File

@ -12,34 +12,59 @@
import logging
from tripleoclient import exceptions
from tripleoclient.workflows import base
import yaml
from tripleo_common.actions import plan
LOG = logging.getLogger(__name__)
def list_roles(workflow_client, **input_):
return base.call_action(workflow_client, 'tripleo.role.list', **input_)
def list_available_roles(clients, container='overcloud'):
"""Return a list of available roles.
:param clients: openstack clients
:type clients: Object
:param container: Name of swift object container
:type container: String
:returns: List
"""
LOG.info('Pulling role list from: {}'.format(container))
obj_client = clients.tripleoclient.object_store
available_yaml_roles = list()
LOG.info('Indexing roles from: {}'.format(container))
for obj in obj_client.get_container(container)[-1]:
name = obj['name']
if name.startswith('roles/') and name.endswith(('yml', 'yaml')):
role_data = yaml.safe_load(
obj_client.get_object(container, name)[-1]
)
available_yaml_roles.append(role_data[0])
return available_yaml_roles
def list_available_roles(clients, **workflow_input):
workflow_client = clients.workflow_engine
tripleoclients = clients.tripleoclient
def list_roles(clients, container, detail=False):
"""Return a list of roles.
available_roles = []
with tripleoclients.messaging_websocket() as ws:
execution = base.start_workflow(
workflow_client,
'tripleo.plan_management.v1.list_available_roles',
workflow_input=workflow_input
)
:param clients: openstack clients
:type clients: Object
for payload in base.wait_for_messages(workflow_client, ws, execution):
if payload['status'] == 'SUCCESS':
available_roles = payload['available_roles']
else:
raise exceptions.WorkflowServiceError(
'Error retrieving available roles: {}'.format(
payload.get('message')))
:param container: Name of swift object container
:type container: String
return available_roles
:param detail: Enable or disable extra detail
:type detail: Boolean
:returns: List
"""
context = clients.tripleoclient.create_mistral_context()
LOG.info('Pulling roles from: {}'.format(container))
return plan.ListRolesAction(
container=container,
detail=detail
).run(context=context)