diff options
Diffstat (limited to 'openstackclient')
| -rw-r--r-- | openstackclient/network/v2/network_agent.py | 119 | ||||
| -rw-r--r-- | openstackclient/tests/network/v2/fakes.py | 67 | ||||
| -rw-r--r-- | openstackclient/tests/network/v2/test_network_agent.py | 219 |
3 files changed, 405 insertions, 0 deletions
diff --git a/openstackclient/network/v2/network_agent.py b/openstackclient/network/v2/network_agent.py new file mode 100644 index 00000000..1fb70a50 --- /dev/null +++ b/openstackclient/network/v2/network_agent.py @@ -0,0 +1,119 @@ +# 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. +# + +"""Network agent action implementations""" + +import logging + +from osc_lib.command import command +from osc_lib import exceptions +from osc_lib import utils + +from openstackclient.i18n import _ + + +LOG = logging.getLogger(__name__) + + +def _format_admin_state(state): + return 'UP' if state else 'DOWN' + + +_formatters = { + 'admin_state_up': _format_admin_state, + 'configurations': utils.format_dict, +} + + +class DeleteNetworkAgent(command.Command): + """Delete network agent(s)""" + + def get_parser(self, prog_name): + parser = super(DeleteNetworkAgent, self).get_parser(prog_name) + parser.add_argument( + 'network_agent', + metavar="<network-agent>", + nargs='+', + help=(_("Network agent(s) to delete (ID only)")) + ) + return parser + + def take_action(self, parsed_args): + client = self.app.client_manager.network + result = 0 + + for agent in parsed_args.network_agent: + try: + obj = client.get_agent(agent, ignore_missing=False) + client.delete_agent(obj) + except Exception as e: + result += 1 + LOG.error(_("Failed to delete network agent with " + "ID '%(agent)s': %(e)s"), + {'agent': agent, 'e': e}) + + if result > 0: + total = len(parsed_args.network_agent) + msg = (_("%(result)s of %(total)s network agents failed " + "to delete.") % {'result': result, 'total': total}) + raise exceptions.CommandError(msg) + + +class ListNetworkAgent(command.Lister): + """List network agents""" + + def take_action(self, parsed_args): + client = self.app.client_manager.network + columns = ( + 'id', + 'agent_type', + 'host', + 'availability_zone', + 'alive', + 'admin_state_up', + 'binary' + ) + column_headers = ( + 'ID', + 'Agent Type', + 'Host', + 'Availability Zone', + 'Alive', + 'State', + 'Binary' + ) + data = client.agents() + return (column_headers, + (utils.get_item_properties( + s, columns, formatters=_formatters, + ) for s in data)) + + +class ShowNetworkAgent(command.ShowOne): + """Display network agent details""" + + def get_parser(self, prog_name): + parser = super(ShowNetworkAgent, self).get_parser(prog_name) + parser.add_argument( + 'network_agent', + metavar="<network-agent>", + help=(_("Network agent to display (ID only)")) + ) + return parser + + def take_action(self, parsed_args): + client = self.app.client_manager.network + obj = client.get_agent(parsed_args.network_agent, ignore_missing=False) + columns = tuple(sorted(list(obj.keys()))) + data = utils.get_item_properties(obj, columns, formatters=_formatters,) + return columns, data diff --git a/openstackclient/tests/network/v2/fakes.py b/openstackclient/tests/network/v2/fakes.py index 05752094..8c8f3a00 100644 --- a/openstackclient/tests/network/v2/fakes.py +++ b/openstackclient/tests/network/v2/fakes.py @@ -491,6 +491,73 @@ class FakePort(object): return mock.MagicMock(side_effect=ports) +class FakeNetworkAgent(object): + """Fake one or more network agents.""" + + @staticmethod + def create_one_network_agent(attrs=None): + """Create a fake network agent + + :param Dictionary attrs: + A dictionary with all attributes + :return: + A FakeResource object, with id, agent_type, and so on. + """ + attrs = attrs or {} + + # Set default attributes + agent_attrs = { + 'id': 'agent-id-' + uuid.uuid4().hex, + 'agent_type': 'agent-type-' + uuid.uuid4().hex, + 'host': 'host-' + uuid.uuid4().hex, + 'availability_zone': 'zone-' + uuid.uuid4().hex, + 'alive': True, + 'admin_state_up': True, + 'binary': 'binary-' + uuid.uuid4().hex, + 'configurations': {'subnet': 2, 'networks': 1}, + } + agent_attrs.update(attrs) + agent = fakes.FakeResource(info=copy.deepcopy(agent_attrs), + loaded=True) + return agent + + @staticmethod + def create_network_agents(attrs=None, count=2): + """Create multiple fake network agents. + + :param Dictionary attrs: + A dictionary with all attributes + :param int count: + The number of network agents to fake + :return: + A list of FakeResource objects faking the network agents + """ + agents = [] + for i in range(0, count): + agents.append(FakeNetworkAgent.create_one_network_agent(attrs)) + + return agents + + @staticmethod + def get_network_agents(agents=None, count=2): + """Get an iterable MagicMock object with a list of faked network agents. + + If network agents list is provided, then initialize the Mock object + with the list. Otherwise create one. + + :param List agents: + A list of FakeResource objects faking network agents + :param int count: + The number of network agents to fake + :return: + An iterable Mock object with side_effect set to a list of faked + network agents + """ + if agents is None: + agents = FakeNetworkAgent.create_network_agents(count) + return mock.MagicMock(side_effect=agents) + + class FakeNetworkRBAC(object): """Fake one or more network rbac policies.""" diff --git a/openstackclient/tests/network/v2/test_network_agent.py b/openstackclient/tests/network/v2/test_network_agent.py new file mode 100644 index 00000000..3cf9a530 --- /dev/null +++ b/openstackclient/tests/network/v2/test_network_agent.py @@ -0,0 +1,219 @@ +# 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. +# + +import mock +from mock import call + +from osc_lib import exceptions +from osc_lib import utils + +from openstackclient.network.v2 import network_agent +from openstackclient.tests.network.v2 import fakes as network_fakes +from openstackclient.tests import utils as tests_utils + + +class TestNetworkAgent(network_fakes.TestNetworkV2): + + def setUp(self): + super(TestNetworkAgent, self).setUp() + + # Get a shortcut to the network client + self.network = self.app.client_manager.network + + +class TestDeleteNetworkAgent(TestNetworkAgent): + + network_agents = ( + network_fakes.FakeNetworkAgent.create_network_agents(count=2)) + + def setUp(self): + super(TestDeleteNetworkAgent, self).setUp() + self.network.delete_agent = mock.Mock(return_value=None) + self.network.get_agent = ( + network_fakes.FakeNetworkAgent.get_network_agents( + agents=self.network_agents) + ) + + # Get the command object to test + self.cmd = network_agent.DeleteNetworkAgent(self.app, self.namespace) + + def test_network_agent_delete(self): + arglist = [ + self.network_agents[0].id, + ] + verifylist = [ + ('network_agent', [self.network_agents[0].id]), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + self.network.get_agent.assert_called_once_with( + self.network_agents[0].id, ignore_missing=False) + self.network.delete_agent.assert_called_once_with( + self.network_agents[0]) + self.assertIsNone(result) + + def test_multi_network_agents_delete(self): + arglist = [] + verifylist = [] + + for n in self.network_agents: + arglist.append(n.id) + verifylist = [ + ('network_agent', arglist), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + + calls = [] + for n in self.network_agents: + calls.append(call(n)) + self.network.delete_agent.assert_has_calls(calls) + self.assertIsNone(result) + + def test_multi_network_agents_delete_with_exception(self): + arglist = [ + self.network_agents[0].id, + 'unexist_network_agent', + ] + verifylist = [ + ('network_agent', + [self.network_agents[0].id, 'unexist_network_agent']), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + find_mock_result = [self.network_agents[0], exceptions.CommandError] + self.network.get_agent = ( + mock.MagicMock(side_effect=find_mock_result) + ) + + try: + self.cmd.take_action(parsed_args) + self.fail('CommandError should be raised.') + except exceptions.CommandError as e: + self.assertEqual('1 of 2 network agents failed to delete.', str(e)) + + self.network.get_agent.assert_any_call( + self.network_agents[0].id, ignore_missing=False) + self.network.get_agent.assert_any_call( + 'unexist_network_agent', ignore_missing=False) + self.network.delete_agent.assert_called_once_with( + self.network_agents[0] + ) + + +class TestListNetworkAgent(TestNetworkAgent): + + network_agents = ( + network_fakes.FakeNetworkAgent.create_network_agents(count=3)) + + columns = ( + 'ID', + 'Agent Type', + 'Host', + 'Availability Zone', + 'Alive', + 'State', + 'Binary' + ) + data = [] + for agent in network_agents: + data.append(( + agent.id, + agent.agent_type, + agent.host, + agent.availability_zone, + agent.alive, + network_agent._format_admin_state(agent.admin_state_up), + agent.binary, + )) + + def setUp(self): + super(TestListNetworkAgent, self).setUp() + self.network.agents = mock.Mock( + return_value=self.network_agents) + + # Get the command object to test + self.cmd = network_agent.ListNetworkAgent(self.app, self.namespace) + + def test_network_agents_list(self): + arglist = [] + verifylist = [] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + columns, data = self.cmd.take_action(parsed_args) + + self.network.agents.assert_called_once_with(**{}) + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, list(data)) + + +class TestShowNetworkAgent(TestNetworkAgent): + + _network_agent = ( + network_fakes.FakeNetworkAgent.create_one_network_agent()) + + columns = ( + 'admin_state_up', + 'agent_type', + 'alive', + 'availability_zone', + 'binary', + 'configurations', + 'host', + 'id', + ) + data = ( + network_agent._format_admin_state(_network_agent.admin_state_up), + _network_agent.agent_type, + _network_agent.alive, + _network_agent.availability_zone, + _network_agent.binary, + utils.format_dict(_network_agent.configurations), + _network_agent.host, + _network_agent.id, + ) + + def setUp(self): + super(TestShowNetworkAgent, self).setUp() + self.network.get_agent = mock.Mock( + return_value=self._network_agent) + + # Get the command object to test + self.cmd = network_agent.ShowNetworkAgent(self.app, self.namespace) + + def test_show_no_options(self): + arglist = [] + verifylist = [] + + # Missing required args should bail here + self.assertRaises(tests_utils.ParserException, self.check_parser, + self.cmd, arglist, verifylist) + + def test_show_all_options(self): + arglist = [ + self._network_agent.id, + ] + verifylist = [ + ('network_agent', self._network_agent.id), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + columns, data = self.cmd.take_action(parsed_args) + + self.network.get_agent.assert_called_once_with( + self._network_agent.id, ignore_missing=False) + self.assertEqual(self.columns, columns) + self.assertEqual(list(self.data), list(data)) |