diff options
Diffstat (limited to 'openstackclient')
| -rw-r--r-- | openstackclient/network/v2/port.py | 84 | ||||
| -rw-r--r-- | openstackclient/tests/unit/network/v2/test_port.py | 194 |
2 files changed, 274 insertions, 4 deletions
diff --git a/openstackclient/network/v2/port.py b/openstackclient/network/v2/port.py index 20f3ad75..9ce0ce63 100644 --- a/openstackclient/network/v2/port.py +++ b/openstackclient/network/v2/port.py @@ -250,6 +250,17 @@ def _add_updatable_args(parser): # TODO(abhiraut): Use the SDK resource mapped attribute names once the # OSC minimum requirements include SDK 1.0. +def _convert_address_pairs(parsed_args): + ops = [] + for opt in parsed_args.allowed_address_pairs: + addr = {} + addr['ip_address'] = opt['ip-address'] + if 'mac-address' in opt: + addr['mac_address'] = opt['mac-address'] + ops.append(addr) + return ops + + class CreatePort(command.ShowOne): _description = _("Create a new port") @@ -309,7 +320,7 @@ class CreatePort(command.ShowOne): help=_("Name of this port") ) # TODO(singhj): Add support for extended options: - # qos,dhcp, address pairs + # qos,dhcp secgroups = parser.add_mutually_exclusive_group() secgroups.add_argument( '--security-group', @@ -336,7 +347,17 @@ class CreatePort(command.ShowOne): action='store_true', help=_("Disable port security for this port") ) - + parser.add_argument( + '--allowed-address', + metavar='ip-address=<ip-address>[,mac-address=<mac-address>]', + action=parseractions.MultiKeyValueAction, + dest='allowed_address_pairs', + required_keys=['ip-address'], + optional_keys=['mac-address'], + help=_("Add allowed-address pair associated with this port: " + "ip-address=<ip-address>[,mac-address=<mac-address>] " + "(repeat option to set multiple allowed-address pairs)") + ) return parser def take_action(self, parsed_args): @@ -353,6 +374,9 @@ class CreatePort(command.ShowOne): for sg in parsed_args.security_groups] if parsed_args.no_security_group: attrs['security_groups'] = [] + if parsed_args.allowed_address_pairs: + attrs['allowed_address_pairs'] = ( + _convert_address_pairs(parsed_args)) obj = client.create_port(**attrs) display_columns, columns = _get_columns(obj) @@ -573,7 +597,26 @@ class SetPort(command.Command): action='store_true', help=_("Disable port security for this port") ) - + parser.add_argument( + '--allowed-address', + metavar='ip-address=<ip-address>[,mac-address=<mac-address>]', + action=parseractions.MultiKeyValueAction, + dest='allowed_address_pairs', + required_keys=['ip-address'], + optional_keys=['mac-address'], + help=_("Add allowed-address pair associated with this port: " + "ip-address=<ip-address>[,mac-address=<mac-address>] " + "(repeat option to set multiple allowed-address pairs)") + ) + parser.add_argument( + '--no-allowed-address', + dest='no_allowed_address_pair', + action='store_true', + help=_("Clear existing allowed-address pairs associated" + "with this port." + "(Specify both --allowed-address and --no-allowed-address" + "to overwrite the current allowed-address pairs)") + ) return parser def take_action(self, parsed_args): @@ -613,6 +656,19 @@ class SetPort(command.Command): elif parsed_args.no_security_group: attrs['security_groups'] = [] + if (parsed_args.allowed_address_pairs and + parsed_args.no_allowed_address_pair): + attrs['allowed_address_pairs'] = ( + _convert_address_pairs(parsed_args)) + + elif parsed_args.allowed_address_pairs: + attrs['allowed_address_pairs'] = ( + [addr for addr in obj.allowed_address_pairs if addr] + + _convert_address_pairs(parsed_args)) + + elif parsed_args.no_allowed_address_pair: + attrs['allowed_address_pairs'] = [] + client.update_port(obj, **attrs) @@ -673,6 +729,19 @@ class UnsetPort(command.Command): metavar="<port>", help=_("Port to modify (name or ID)") ) + parser.add_argument( + '--allowed-address', + metavar='ip-address=<ip-address>[,mac-address=<mac-address>]', + action=parseractions.MultiKeyValueAction, + dest='allowed_address_pairs', + required_keys=['ip-address'], + optional_keys=['mac-address'], + help=_("Desired allowed-address pair which should be removed " + "from this port: ip-address=<ip-address> " + "[,mac-address=<mac-address>] (repeat option to set " + "multiple allowed-address pairs)") + ) + return parser def take_action(self, parsed_args): @@ -684,6 +753,7 @@ class UnsetPort(command.Command): tmp_fixed_ips = copy.deepcopy(obj.fixed_ips) tmp_binding_profile = copy.deepcopy(obj.binding_profile) tmp_secgroups = copy.deepcopy(obj.security_groups) + tmp_addr_pairs = copy.deepcopy(obj.allowed_address_pairs) _prepare_fixed_ips(self.app.client_manager, parsed_args) attrs = {} if parsed_args.fixed_ip: @@ -712,6 +782,14 @@ class UnsetPort(command.Command): msg = _("Port does not contain security group %s") % sg raise exceptions.CommandError(msg) attrs['security_groups'] = tmp_secgroups + if parsed_args.allowed_address_pairs: + try: + for addr in _convert_address_pairs(parsed_args): + tmp_addr_pairs.remove(addr) + except ValueError: + msg = _("Port does not contain allowed-address-pair %s") % addr + raise exceptions.CommandError(msg) + attrs['allowed_address_pairs'] = tmp_addr_pairs if attrs: client.update_port(obj, **attrs) diff --git a/openstackclient/tests/unit/network/v2/test_port.py b/openstackclient/tests/unit/network/v2/test_port.py index 255e8116..9deb77ac 100644 --- a/openstackclient/tests/unit/network/v2/test_port.py +++ b/openstackclient/tests/unit/network/v2/test_port.py @@ -216,7 +216,7 @@ class TestCreatePort(TestPort): 'test-port', ] verifylist = [ - ('network', self._port.network_id,), + ('network', self._port.network_id), ('enable', True), ('binding_profile', {'parent_name': 'fake_parent', 'tag': 42}), ('name', 'test-port'), @@ -351,6 +351,74 @@ class TestCreatePort(TestPort): self.assertEqual(ref_columns, columns) self.assertEqual(ref_data, data) + def test_create_port_with_allowed_address_pair_ipaddr(self): + pairs = [{'ip_address': '192.168.1.123'}, + {'ip_address': '192.168.1.45'}] + arglist = [ + '--network', self._port.network_id, + '--allowed-address', 'ip-address=192.168.1.123', + '--allowed-address', 'ip-address=192.168.1.45', + 'test-port', + ] + verifylist = [ + ('network', self._port.network_id), + ('enable', True), + ('allowed_address_pairs', [{'ip-address': '192.168.1.123'}, + {'ip-address': '192.168.1.45'}]), + ('name', 'test-port'), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = (self.cmd.take_action(parsed_args)) + + self.network.create_port.assert_called_once_with(**{ + 'admin_state_up': True, + 'network_id': self._port.network_id, + 'allowed_address_pairs': pairs, + 'name': 'test-port', + }) + + ref_columns, ref_data = self._get_common_cols_data(self._port) + self.assertEqual(ref_columns, columns) + self.assertEqual(ref_data, data) + + def test_create_port_with_allowed_address_pair(self): + pairs = [{'ip_address': '192.168.1.123', + 'mac_address': 'aa:aa:aa:aa:aa:aa'}, + {'ip_address': '192.168.1.45', + 'mac_address': 'aa:aa:aa:aa:aa:b1'}] + arglist = [ + '--network', self._port.network_id, + '--allowed-address', + 'ip-address=192.168.1.123,mac-address=aa:aa:aa:aa:aa:aa', + '--allowed-address', + 'ip-address=192.168.1.45,mac-address=aa:aa:aa:aa:aa:b1', + 'test-port', + ] + verifylist = [ + ('network', self._port.network_id), + ('enable', True), + ('allowed_address_pairs', [{'ip-address': '192.168.1.123', + 'mac-address': 'aa:aa:aa:aa:aa:aa'}, + {'ip-address': '192.168.1.45', + 'mac-address': 'aa:aa:aa:aa:aa:b1'}]), + ('name', 'test-port'), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = (self.cmd.take_action(parsed_args)) + + self.network.create_port.assert_called_once_with(**{ + 'admin_state_up': True, + 'network_id': self._port.network_id, + 'allowed_address_pairs': pairs, + 'name': 'test-port', + }) + + ref_columns, ref_data = self._get_common_cols_data(self._port) + self.assertEqual(ref_columns, columns) + self.assertEqual(ref_data, data) + def test_create_port_security_enabled(self): arglist = [ '--network', self._port.network_id, @@ -996,6 +1064,91 @@ class TestSetPort(TestPort): self.network.update_port.assert_called_once_with(_testport, **attrs) self.assertIsNone(result) + def test_set_allowed_address_pair(self): + arglist = [ + '--allowed-address', 'ip-address=192.168.1.123', + self._port.name, + ] + verifylist = [ + ('allowed_address_pairs', [{'ip-address': '192.168.1.123'}]), + ('port', self._port.name), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + result = self.cmd.take_action(parsed_args) + + attrs = { + 'allowed_address_pairs': [{'ip_address': '192.168.1.123'}], + } + self.network.update_port.assert_called_once_with(self._port, **attrs) + self.assertIsNone(result) + + def test_append_allowed_address_pair(self): + _testport = network_fakes.FakePort.create_one_port( + {'allowed_address_pairs': [{'ip_address': '192.168.1.123'}]}) + self.network.find_port = mock.Mock(return_value=_testport) + arglist = [ + '--allowed-address', 'ip-address=192.168.1.45', + _testport.name, + ] + verifylist = [ + ('allowed_address_pairs', [{'ip-address': '192.168.1.45'}]), + ('port', _testport.name), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + result = self.cmd.take_action(parsed_args) + + attrs = { + 'allowed_address_pairs': [{'ip_address': '192.168.1.123'}, + {'ip_address': '192.168.1.45'}], + } + self.network.update_port.assert_called_once_with(_testport, **attrs) + self.assertIsNone(result) + + def test_overwrite_allowed_address_pair(self): + _testport = network_fakes.FakePort.create_one_port( + {'allowed_address_pairs': [{'ip_address': '192.168.1.123'}]}) + self.network.find_port = mock.Mock(return_value=_testport) + arglist = [ + '--allowed-address', 'ip-address=192.168.1.45', + '--no-allowed-address', + _testport.name, + ] + verifylist = [ + ('allowed_address_pairs', [{'ip-address': '192.168.1.45'}]), + ('no_allowed_address_pair', True), + ('port', _testport.name), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + result = self.cmd.take_action(parsed_args) + + attrs = { + 'allowed_address_pairs': [{'ip_address': '192.168.1.45'}], + } + self.network.update_port.assert_called_once_with(_testport, **attrs) + self.assertIsNone(result) + + def test_set_no_allowed_address_pairs(self): + arglist = [ + '--no-allowed-address', + self._port.name, + ] + verifylist = [ + ('no_allowed_address_pair', True), + ('port', self._port.name), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + result = self.cmd.take_action(parsed_args) + + attrs = { + 'allowed_address_pairs': [], + } + self.network.update_port.assert_called_once_with(self._port, **attrs) + self.assertIsNone(result) + def test_port_security_enabled(self): arglist = [ '--enable-port-security', @@ -1192,3 +1345,42 @@ class TestUnsetPort(TestPort): self.assertRaises(exceptions.CommandError, self.cmd.take_action, parsed_args) + + def test_unset_port_allowed_address_pair(self): + _fake_port = network_fakes.FakePort.create_one_port( + {'allowed_address_pairs': [{'ip_address': '192.168.1.123'}]}) + self.network.find_port = mock.Mock(return_value=_fake_port) + arglist = [ + '--allowed-address', 'ip-address=192.168.1.123', + _fake_port.name, + ] + verifylist = [ + ('allowed_address_pairs', [{'ip-address': '192.168.1.123'}]), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + result = self.cmd.take_action(parsed_args) + + attrs = { + 'allowed_address_pairs': [], + } + + self.network.update_port.assert_called_once_with(_fake_port, **attrs) + self.assertIsNone(result) + + def test_unset_port_allowed_address_pair_not_existent(self): + _fake_port = network_fakes.FakePort.create_one_port( + {'allowed_address_pairs': [{'ip_address': '192.168.1.123'}]}) + self.network.find_port = mock.Mock(return_value=_fake_port) + arglist = [ + '--allowed-address', 'ip-address=192.168.1.45', + _fake_port.name, + ] + verifylist = [ + ('allowed_address_pairs', [{'ip-address': '192.168.1.45'}]), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + self.assertRaises(exceptions.CommandError, + self.cmd.take_action, + parsed_args) |