diff options
Diffstat (limited to 'openstackclient')
| -rw-r--r-- | openstackclient/compute/v2/server.py | 66 | ||||
| -rw-r--r-- | openstackclient/tests/unit/compute/v2/test_server.py | 96 |
2 files changed, 162 insertions, 0 deletions
diff --git a/openstackclient/compute/v2/server.py b/openstackclient/compute/v2/server.py index 3ae75167..13c4c2d8 100644 --- a/openstackclient/compute/v2/server.py +++ b/openstackclient/compute/v2/server.py @@ -253,6 +253,39 @@ class AddFloatingIP(command.Command): parsed_args.fixed_ip_address) +class AddPort(command.Command): + _description = _("Add port to server") + + def get_parser(self, prog_name): + parser = super(AddPort, self).get_parser(prog_name) + parser.add_argument( + "server", + metavar="<server>", + help=_("Server to add the port to (name or ID)"), + ) + parser.add_argument( + "port", + metavar="<port>", + help=_("Port to add to the server (name or ID)"), + ) + return parser + + def take_action(self, parsed_args): + compute_client = self.app.client_manager.compute + + server = utils.find_resource( + compute_client.servers, parsed_args.server) + + if self.app.client_manager.is_network_endpoint_enabled(): + network_client = self.app.client_manager.network + port_id = network_client.find_port( + parsed_args.port, ignore_missing=False).id + else: + port_id = parsed_args.port + + server.interface_attach(port_id=port_id, net_id=None, fixed_ip=None) + + class AddServerSecurityGroup(command.Command): _description = _("Add security group to server") @@ -1342,6 +1375,39 @@ class RemoveFloatingIP(command.Command): server.remove_floating_ip(parsed_args.ip_address) +class RemovePort(command.Command): + _description = _("Remove port from server") + + def get_parser(self, prog_name): + parser = super(RemovePort, self).get_parser(prog_name) + parser.add_argument( + "server", + metavar="<server>", + help=_("Server to remove the port from (name or ID)"), + ) + parser.add_argument( + "port", + metavar="<port>", + help=_("Port to remove from the server (name or ID)"), + ) + return parser + + def take_action(self, parsed_args): + compute_client = self.app.client_manager.compute + + server = utils.find_resource( + compute_client.servers, parsed_args.server) + + if self.app.client_manager.is_network_endpoint_enabled(): + network_client = self.app.client_manager.network + port_id = network_client.find_port( + parsed_args.port, ignore_missing=False).id + else: + port_id = parsed_args.port + + server.interface_detach(port_id) + + class RemoveServerSecurityGroup(command.Command): _description = _("Remove security group from server") diff --git a/openstackclient/tests/unit/compute/v2/test_server.py b/openstackclient/tests/unit/compute/v2/test_server.py index a0716e4c..0e3bb28f 100644 --- a/openstackclient/tests/unit/compute/v2/test_server.py +++ b/openstackclient/tests/unit/compute/v2/test_server.py @@ -146,6 +146,9 @@ class TestServerAddFloatingIP(TestServer): 'add_floating_ip': None, } + self.find_port = mock.Mock() + self.app.client_manager.network.find_port = self.find_port + def _test_server_add_floating_ip(self, extralist, fixed_ip_address): servers = self.setup_servers_mock(count=1) @@ -174,6 +177,53 @@ class TestServerAddFloatingIP(TestServer): self._test_server_add_floating_ip(extralist, '5.6.7.8') +class TestServerAddPort(TestServer): + + def setUp(self): + super(TestServerAddPort, self).setUp() + + # Get the command object to test + self.cmd = server.AddPort(self.app, None) + + # Set add_fixed_ip method to be tested. + self.methods = { + 'interface_attach': None, + } + + self.find_port = mock.Mock() + self.app.client_manager.network.find_port = self.find_port + + def _test_server_add_port(self, port_id): + servers = self.setup_servers_mock(count=1) + port = 'fake-port' + + arglist = [ + servers[0].id, + port, + ] + verifylist = [ + ('server', servers[0].id), + ('port', port) + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + + servers[0].interface_attach.assert_called_once_with( + port_id=port_id, net_id=None, fixed_ip=None) + self.assertIsNone(result) + + def test_server_add_port(self): + self._test_server_add_port(self.find_port.return_value.id) + self.find_port.assert_called_once_with( + 'fake-port', ignore_missing=False) + + def test_server_add_port_no_neutron(self): + self.app.client_manager.network_endpoint_enabled = False + self._test_server_add_port('fake-port') + self.find_port.assert_not_called() + + class TestServerAddSecurityGroup(TestServer): def setUp(self): @@ -1612,6 +1662,52 @@ class TestServerRemoveFloatingIP(TestServer): self.assertIsNone(result) +class TestServerRemovePort(TestServer): + + def setUp(self): + super(TestServerRemovePort, self).setUp() + + # Get the command object to test + self.cmd = server.RemovePort(self.app, None) + + # Set method to be tested. + self.methods = { + 'interface_detach': None, + } + + self.find_port = mock.Mock() + self.app.client_manager.network.find_port = self.find_port + + def _test_server_remove_port(self, port_id): + servers = self.setup_servers_mock(count=1) + port = 'fake-port' + + arglist = [ + servers[0].id, + port, + ] + verifylist = [ + ('server', servers[0].id), + ('port', port), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + + servers[0].interface_detach.assert_called_once_with(port_id) + self.assertIsNone(result) + + def test_server_remove_port(self): + self._test_server_remove_port(self.find_port.return_value.id) + self.find_port.assert_called_once_with( + 'fake-port', ignore_missing=False) + + def test_server_remove_port_no_neutron(self): + self.app.client_manager.network_endpoint_enabled = False + self._test_server_remove_port('fake-port') + self.find_port.assert_not_called() + + class TestServerRemoveSecurityGroup(TestServer): def setUp(self): |