Re-implement novaclient bits removed in 10.0

a)

/os-floating-ips was removed in Compute API 2.36 and from novaclient's
Python API in 10.0

Add to api.computev2:
floating_ip_add()
floating_ip_remove()

Convert add floating IP command to nova-net/neutron split:
"server add floating ip"
"server remove floating ip"

b)

/os-hosts was removed in Compute API 2.43 and from novaclient's
Python API in 10.0.

Add to api.computev2:
host_list()
host_set()
host_show()

Convert host commands to use intenal api:
"host list"
"host set"
"host show"

c)

The introduction of the Network-style commands into the server group
broke cliff's autoprogram directive as it executes the get_parser() methods
without fully initializing the Command object.

NOTE: This is really three reviews squashed to get through the gate in one pass.

Depends-on: Id6de87211d6c4ea8fd14aa9203d8d5b17e9e2f04
Change-Id: I5116086f9a9e4b2b31a744bf8f4558c79f0bfe59
This commit is contained in:
Dean Troyer 2018-03-05 14:18:41 -06:00
parent b10941ddf6
commit 53e7aab7ed
8 changed files with 622 additions and 114 deletions

View File

@ -93,7 +93,51 @@ class APIv2(api.BaseAPI):
return ret
# Flaoting IPs
# Floating IPs
def floating_ip_add(
self,
server,
address,
fixed_address=None,
):
"""Add a floating IP to a server
:param server:
The :class:`Server` (or its ID) to add an IP to.
:param address:
The FloatingIP or string floating address to add.
:param fixed_address:
The FixedIP the floatingIP should be associated with (optional)
"""
url = '/servers'
server = self.find(
url,
attr='name',
value=server,
)
address = address.ip if hasattr(address, 'ip') else address
if fixed_address:
if hasattr(fixed_address, 'ip'):
fixed_address = fixed_address.ip
body = {
'address': address,
'fixed_address': fixed_address,
}
else:
body = {
'address': address,
}
return self._request(
"POST",
"/%s/%s/action" % (url, server['id']),
json={'addFloatingIp': body},
)
def floating_ip_create(
self,
@ -175,6 +219,38 @@ class APIv2(api.BaseAPI):
return self.list(url)["floating_ips"]
def floating_ip_remove(
self,
server,
address,
):
"""Remove a floating IP from a server
:param server:
The :class:`Server` (or its ID) to add an IP to.
:param address:
The FloatingIP or string floating address to add.
"""
url = '/servers'
server = self.find(
url,
attr='name',
value=server,
)
address = address.ip if hasattr(address, 'ip') else address
body = {
'address': address,
}
return self._request(
"POST",
"/%s/%s/action" % (url, server['id']),
json={'removeFloatingIp': body},
)
# Floating IP Pools
def floating_ip_pool_list(
@ -192,6 +268,84 @@ class APIv2(api.BaseAPI):
return self.list(url)["floating_ip_pools"]
# Hosts
def host_list(
self,
zone=None,
):
"""Lists hypervisor Hosts
https://developer.openstack.org/api-ref/compute/#list-hosts
Valid for Compute 2.0 - 2.42
:param string zone:
Availability zone
:returns: A dict of the floating IP attributes
"""
url = "/os-hosts"
if zone:
url = '/os-hosts?zone=%s' % zone
return self.list(url)["hosts"]
def host_set(
self,
host=None,
status=None,
maintenance_mode=None,
**params
):
"""Modify host properties
https://developer.openstack.org/api-ref/compute/#update-host-status
Valid for Compute 2.0 - 2.42
status
maintenance_mode
"""
url = "/os-hosts"
params = {}
if status:
params['status'] = status
if maintenance_mode:
params['maintenance_mode'] = maintenance_mode
if params == {}:
# Don't bother calling if nothing given
return None
else:
return self._request(
"PUT",
"/%s/%s" % (url, host),
json=params,
).json()
def host_show(
self,
host=None,
):
"""Show host
https://developer.openstack.org/api-ref/compute/#show-host-details
Valid for Compute 2.0 - 2.42
"""
url = "/os-hosts"
r_host = self.find(
url,
attr='host_name',
value=host,
)
data = []
for h in r_host:
data.append(h['resource'])
return data
# Networks
def network_create(

View File

@ -40,9 +40,9 @@ class ListHost(command.Lister):
"Service",
"Zone"
)
data = compute_client.hosts.list_all(parsed_args.zone)
data = compute_client.api.host_list(parsed_args.zone)
return (columns,
(utils.get_item_properties(
(utils.get_dict_properties(
s, columns,
) for s in data))
@ -95,13 +95,7 @@ class SetHost(command.Command):
compute_client = self.app.client_manager.compute
# More than one hosts will be returned by using find_resource()
# so that the return value cannot be used in host update() method.
# find_resource() is just used for checking existence of host and
# keeping the exception message consistent with other commands.
utils.find_resource(compute_client.hosts, parsed_args.host)
compute_client.hosts.update(
compute_client.api.host_set(
parsed_args.host,
kwargs
)
@ -128,8 +122,10 @@ class ShowHost(command.Lister):
"Memory MB",
"Disk GB"
)
data = compute_client.hosts.get(parsed_args.host)
data = compute_client.api.host_show(parsed_args.host)
return (columns,
(utils.get_item_properties(
(utils.get_dict_properties(
s, columns,
) for s in data))

View File

@ -32,6 +32,7 @@ import six
from openstackclient.i18n import _
from openstackclient.identity import common as identity_common
from openstackclient.network import common as network_common
LOG = logging.getLogger(__name__)
@ -234,11 +235,10 @@ class AddFixedIP(command.Command):
)
class AddFloatingIP(command.Command):
class AddFloatingIP(network_common.NetworkAndComputeCommand):
_description = _("Add floating IP address to server")
def get_parser(self, prog_name):
parser = super(AddFloatingIP, self).get_parser(prog_name)
def update_parser_common(self, parser):
parser.add_argument(
"server",
metavar="<server>",
@ -252,19 +252,37 @@ class AddFloatingIP(command.Command):
parser.add_argument(
"--fixed-ip-address",
metavar="<ip-address>",
help=_("Fixed IP address to associate with this floating IP "
"address"),
help=_(
"Fixed IP address to associate with this floating IP address"
),
)
return parser
def take_action(self, parsed_args):
def take_action_network(self, client, parsed_args):
compute_client = self.app.client_manager.compute
attrs = {}
obj = client.find_ip(
parsed_args.ip_address,
ignore_missing=False,
)
server = utils.find_resource(
compute_client.servers, parsed_args.server)
compute_client.servers,
parsed_args.server,
)
port = list(client.ports(device_id=server.id))[0]
attrs['port_id'] = port.id
if parsed_args.fixed_ip_address:
attrs['fixed_ip_address'] = parsed_args.fixed_ip_address
server.add_floating_ip(parsed_args.ip_address,
parsed_args.fixed_ip_address)
client.update_ip(obj, **attrs)
def take_action_compute(self, client, parsed_args):
client.api.floating_ip_add(
parsed_args.server,
parsed_args.ip_address,
fixed_address=parsed_args.fixed_ip_address,
)
class AddPort(command.Command):
@ -1482,11 +1500,10 @@ class RemoveFixedIP(command.Command):
server.remove_fixed_ip(parsed_args.ip_address)
class RemoveFloatingIP(command.Command):
class RemoveFloatingIP(network_common.NetworkAndComputeCommand):
_description = _("Remove floating IP address from server")
def get_parser(self, prog_name):
parser = super(RemoveFloatingIP, self).get_parser(prog_name)
def update_parser_common(self, parser):
parser.add_argument(
"server",
metavar="<server>",
@ -1501,13 +1518,21 @@ class RemoveFloatingIP(command.Command):
)
return parser
def take_action(self, parsed_args):
compute_client = self.app.client_manager.compute
def take_action_network(self, client, parsed_args):
attrs = {}
obj = client.find_ip(
parsed_args.ip_address,
ignore_missing=False,
)
attrs['port_id'] = None
server = utils.find_resource(
compute_client.servers, parsed_args.server)
client.update_ip(obj, **attrs)
server.remove_floating_ip(parsed_args.ip_address)
def take_action_compute(self, client, parsed_args):
client.api.floating_ip_remove(
parsed_args.server,
parsed_args.ip_address,
)
class RemovePort(command.Command):

View File

@ -48,7 +48,10 @@ class NetworkAndComputeCommand(command.Command):
parser = super(NetworkAndComputeCommand, self).get_parser(prog_name)
parser = self.update_parser_common(parser)
LOG.debug('common parser: %s', parser)
if self.app.client_manager.is_network_endpoint_enabled():
if (
self.app is None or
self.app.client_manager.is_network_endpoint_enabled()
):
return self.update_parser_network(parser)
else:
return self.update_parser_compute(parser)

View File

@ -55,6 +55,43 @@ class TestFloatingIP(TestComputeAPIv2):
FAKE_FLOATING_IP_RESP_2,
]
FAKE_SERVER_RESP_1 = {
'id': 1,
'name': 'server1',
}
def test_floating_ip_add_id(self):
self.requests_mock.register_uri(
'POST',
FAKE_URL + '/servers/1/action',
json={'server': {}},
status_code=200,
)
self.requests_mock.register_uri(
'GET',
FAKE_URL + '/servers/1',
json={'server': self.FAKE_SERVER_RESP_1},
status_code=200,
)
ret = self.api.floating_ip_add('1', '1.0.1.0')
self.assertEqual(200, ret.status_code)
def test_floating_ip_add_name(self):
self.requests_mock.register_uri(
'POST',
FAKE_URL + '/servers/1/action',
json={'server': {}},
status_code=200,
)
self.requests_mock.register_uri(
'GET',
FAKE_URL + '/servers/server1',
json={'server': self.FAKE_SERVER_RESP_1},
status_code=200,
)
ret = self.api.floating_ip_add('server1', '1.0.1.0')
self.assertEqual(200, ret.status_code)
def test_floating_ip_create(self):
self.requests_mock.register_uri(
'POST',
@ -144,6 +181,36 @@ class TestFloatingIP(TestComputeAPIv2):
ret = self.api.floating_ip_list()
self.assertEqual(self.LIST_FLOATING_IP_RESP, ret)
def test_floating_ip_remove_id(self):
self.requests_mock.register_uri(
'POST',
FAKE_URL + '/servers/1/action',
status_code=200,
)
self.requests_mock.register_uri(
'GET',
FAKE_URL + '/servers/1',
json={'server': self.FAKE_SERVER_RESP_1},
status_code=200,
)
ret = self.api.floating_ip_remove('1', '1.0.1.0')
self.assertEqual(200, ret.status_code)
def test_floating_ip_remove_name(self):
self.requests_mock.register_uri(
'POST',
FAKE_URL + '/servers/1/action',
status_code=200,
)
self.requests_mock.register_uri(
'GET',
FAKE_URL + '/servers/server1',
json={'server': self.FAKE_SERVER_RESP_1},
status_code=200,
)
ret = self.api.floating_ip_remove('server1', '1.0.1.0')
self.assertEqual(200, ret.status_code)
class TestFloatingIPPool(TestComputeAPIv2):
@ -163,6 +230,115 @@ class TestFloatingIPPool(TestComputeAPIv2):
self.assertEqual(self.LIST_FLOATING_IP_POOL_RESP, ret)
class TestHost(TestComputeAPIv2):
FAKE_HOST_RESP_1 = {
"zone": "internal",
"host_name": "myhost",
"service": "conductor",
}
FAKE_HOST_RESP_2 = {
"zone": "internal",
"host_name": "myhost",
"service": "scheduler",
}
FAKE_HOST_RESP_3 = {
"zone": "nova",
"host_name": "myhost",
"service": "compute",
}
LIST_HOST_RESP = [
FAKE_HOST_RESP_1,
FAKE_HOST_RESP_2,
FAKE_HOST_RESP_3,
]
def test_host_list_no_options(self):
self.requests_mock.register_uri(
'GET',
FAKE_URL + '/os-hosts',
json={'hosts': self.LIST_HOST_RESP},
status_code=200,
)
ret = self.api.host_list()
self.assertEqual(self.LIST_HOST_RESP, ret)
def test_host_list_zone(self):
self.requests_mock.register_uri(
'GET',
FAKE_URL + '/os-hosts?zone=nova',
json={'hosts': [self.FAKE_HOST_RESP_3]},
status_code=200,
)
self.requests_mock.register_uri(
'GET',
FAKE_URL + '/os-hosts',
json={'hosts': [self.FAKE_HOST_RESP_3]},
status_code=200,
)
ret = self.api.host_list(zone='nova')
self.assertEqual([self.FAKE_HOST_RESP_3], ret)
def test_host_set_none(self):
ret = self.api.host_set(host='myhost')
self.assertIsNone(ret)
def test_host_set(self):
self.requests_mock.register_uri(
'PUT',
FAKE_URL + '/os-hosts/myhost',
json={},
status_code=200,
)
ret = self.api.host_set(host='myhost', status='enabled')
self.assertEqual({}, ret)
def test_host_show(self):
FAKE_RESOURCE_1 = {
"cpu": 2,
"disk_gb": 1028,
"host": "c1a7de0ac9d94e4baceae031d05caae3",
"memory_mb": 8192,
"project": "(total)",
}
FAKE_RESOURCE_2 = {
"cpu": 0,
"disk_gb": 0,
"host": "c1a7de0ac9d94e4baceae031d05caae3",
"memory_mb": 512,
"project": "(used_now)",
}
FAKE_RESOURCE_3 = {
"cpu": 0,
"disk_gb": 0,
"host": "c1a7de0ac9d94e4baceae031d05caae3",
"memory_mb": 0,
"project": "(used_max)",
}
FAKE_HOST_RESP = [
{'resource': FAKE_RESOURCE_1},
{'resource': FAKE_RESOURCE_2},
{'resource': FAKE_RESOURCE_3},
]
FAKE_HOST_LIST = [
FAKE_RESOURCE_1,
FAKE_RESOURCE_2,
FAKE_RESOURCE_3,
]
self.requests_mock.register_uri(
'GET',
FAKE_URL + '/os-hosts/myhost',
json={'host': FAKE_HOST_RESP},
status_code=200,
)
ret = self.api.host_show(host='myhost')
self.assertEqual(FAKE_HOST_LIST, ret)
class TestNetwork(TestComputeAPIv2):
FAKE_NETWORK_RESP = {

View File

@ -1230,10 +1230,7 @@ class FakeHost(object):
'project': 'project-' + uuid.uuid4().hex,
}
host_info.update(attrs)
host = fakes.FakeResource(
info=copy.deepcopy(host_info),
loaded=True)
return host
return host_info
class FakeServerGroup(object):

View File

@ -13,6 +13,8 @@
# under the License.
#
import mock
from openstackclient.compute.v2 import host
from openstackclient.tests.unit.compute.v2 import fakes as compute_fakes
from openstackclient.tests.unit import utils as tests_utils
@ -23,11 +25,13 @@ class TestHost(compute_fakes.TestComputev2):
def setUp(self):
super(TestHost, self).setUp()
# Get a shortcut to the FlavorManager Mock
self.host_mock = self.app.client_manager.compute.hosts
self.host_mock.reset_mock()
# Get a shortcut to the compute client
self.compute = self.app.client_manager.compute
@mock.patch(
'openstackclient.api.compute_v2.APIv2.host_list'
)
class TestHostList(TestHost):
host = compute_fakes.FakeHost.create_one_host()
@ -39,19 +43,18 @@ class TestHostList(TestHost):
)
data = [(
host.host_name,
host.service,
host.zone,
host['host_name'],
host['service'],
host['zone'],
)]
def setUp(self):
super(TestHostList, self).setUp()
self.host_mock.list_all.return_value = [self.host]
self.cmd = host.ListHost(self.app, None)
def test_host_list_no_option(self):
def test_host_list_no_option(self, h_mock):
h_mock.return_value = [self.host]
arglist = []
verifylist = []
@ -59,44 +62,48 @@ class TestHostList(TestHost):
columns, data = self.cmd.take_action(parsed_args)
self.host_mock.list_all.assert_called_with(None)
h_mock.assert_called_with(None)
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, list(data))
def test_host_list_with_option(self):
def test_host_list_with_option(self, h_mock):
h_mock.return_value = [self.host]
arglist = [
'--zone', self.host.zone,
'--zone', self.host['zone'],
]
verifylist = [
('zone', self.host.zone),
('zone', self.host['zone']),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
self.host_mock.list_all.assert_called_with(self.host.zone)
h_mock.assert_called_with(self.host['zone'])
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, list(data))
@mock.patch(
'openstackclient.api.compute_v2.APIv2.host_set'
)
class TestHostSet(TestHost):
def setUp(self):
super(TestHostSet, self).setUp()
self.host = compute_fakes.FakeHost.create_one_host()
self.host_mock.get.return_value = self.host
self.host_mock.update.return_value = None
self.cmd = host.SetHost(self.app, None)
def test_host_set_no_option(self):
def test_host_set_no_option(self, h_mock):
h_mock.return_value = self.host
h_mock.update.return_value = None
arglist = [
self.host.host
self.host['host'],
]
verifylist = [
('host', self.host.host)
('host', self.host['host']),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
@ -105,18 +112,20 @@ class TestHostSet(TestHost):
self.assertIsNone(result)
body = {}
self.host_mock.update.assert_called_with(self.host.host, body)
h_mock.assert_called_with(self.host['host'], body)
def test_host_set(self):
def test_host_set(self, h_mock):
h_mock.return_value = self.host
h_mock.update.return_value = None
arglist = [
'--enable',
'--disable-maintenance',
self.host.host
self.host['host'],
]
verifylist = [
('enable', True),
('enable_maintenance', False),
('host', self.host.host)
('host', self.host['host']),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
@ -125,9 +134,12 @@ class TestHostSet(TestHost):
self.assertIsNone(result)
body = {'status': 'enable', 'maintenance_mode': 'disable'}
self.host_mock.update.assert_called_with(self.host.host, body)
h_mock.assert_called_with(self.host['host'], body)
@mock.patch(
'openstackclient.api.compute_v2.APIv2.host_show'
)
class TestHostShow(TestHost):
host = compute_fakes.FakeHost.create_one_host()
@ -139,22 +151,22 @@ class TestHostShow(TestHost):
'Memory MB',
'Disk GB',
)
data = [(
host.host,
host.project,
host.cpu,
host.memory_mb,
host.disk_gb,
host['host'],
host['project'],
host['cpu'],
host['memory_mb'],
host['disk_gb'],
)]
def setUp(self):
super(TestHostShow, self).setUp()
self.host_mock.get.return_value = [self.host]
self.cmd = host.ShowHost(self.app, None)
def test_host_show_no_option(self):
def test_host_show_no_option(self, h_mock):
h_mock.host_show.return_value = [self.host]
arglist = []
verifylist = []
@ -162,18 +174,19 @@ class TestHostShow(TestHost):
self.assertRaises(tests_utils.ParserException, self.check_parser,
self.cmd, arglist, verifylist)
def test_host_show_with_option(self):
def test_host_show_with_option(self, h_mock):
h_mock.return_value = [self.host]
arglist = [
self.host.host_name,
self.host['host_name'],
]
verifylist = [
('host', self.host.host_name),
('host', self.host['host_name']),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = self.cmd.take_action(parsed_args)
self.host_mock.get.assert_called_with(self.host.host_name)
h_mock.assert_called_with(self.host['host_name'])
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, list(data))

View File

@ -141,48 +141,148 @@ class TestServerAddFixedIP(TestServer):
self._test_server_add_fixed_ip(extralist, '5.6.7.8')
class TestServerAddFloatingIP(TestServer):
@mock.patch(
'openstackclient.api.compute_v2.APIv2.floating_ip_add'
)
class TestServerAddFloatingIPCompute(compute_fakes.TestComputev2):
def setUp(self):
super(TestServerAddFloatingIP, self).setUp()
super(TestServerAddFloatingIPCompute, self).setUp()
self.app.client_manager.network_endpoint_enabled = False
# Get the command object to test
self.cmd = server.AddFloatingIP(self.app, None)
# Set add_floating_ip method to be tested.
self.methods = {
'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)
def test_server_add_floating_ip_default(self, fip_mock):
_floating_ip = compute_fakes.FakeFloatingIP.create_one_floating_ip()
arglist = [
servers[0].id,
'1.2.3.4',
] + extralist
'server1',
_floating_ip['ip'],
]
verifylist = [
('server', servers[0].id),
('ip_address', '1.2.3.4'),
('fixed_ip_address', fixed_ip_address),
('server', 'server1'),
('ip_address', _floating_ip['ip']),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.cmd.take_action(parsed_args)
servers[0].add_floating_ip.assert_called_once_with('1.2.3.4',
fixed_ip_address)
self.assertIsNone(result)
fip_mock.assert_called_once_with(
'server1',
_floating_ip['ip'],
fixed_address=None,
)
def test_server_add_floating_ip(self):
self._test_server_add_floating_ip([], None)
def test_server_add_floating_ip_fixed(self, fip_mock):
_floating_ip = compute_fakes.FakeFloatingIP.create_one_floating_ip()
arglist = [
'--fixed-ip-address', _floating_ip['fixed_ip'],
'server1',
_floating_ip['ip'],
]
verifylist = [
('fixed_ip_address', _floating_ip['fixed_ip']),
('server', 'server1'),
('ip_address', _floating_ip['ip']),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
def test_server_add_floating_ip_to_fixed_ip(self):
extralist = ['--fixed-ip-address', '5.6.7.8']
self._test_server_add_floating_ip(extralist, '5.6.7.8')
self.cmd.take_action(parsed_args)
fip_mock.assert_called_once_with(
'server1',
_floating_ip['ip'],
fixed_address=_floating_ip['fixed_ip'],
)
class TestServerAddFloatingIPNetwork(
TestServer,
network_fakes.TestNetworkV2,
):
def setUp(self):
super(TestServerAddFloatingIPNetwork, self).setUp()
self.app.client_manager.network = mock.Mock()
self.network = self.app.client_manager.network
self.network.update_ip = mock.Mock(return_value=None)
# Get the command object to test
self.cmd = server.AddFloatingIP(self.app, self.namespace)
def test_server_add_floating_ip_default(self):
_server = compute_fakes.FakeServer.create_one_server()
self.servers_mock.get.return_value = _server
_port = network_fakes.FakePort.create_one_port()
_floating_ip = network_fakes.FakeFloatingIP.create_one_floating_ip()
self.network.find_ip = mock.Mock(return_value=_floating_ip)
self.network.ports = mock.Mock(return_value=[_port])
arglist = [
_server.id,
_floating_ip['ip'],
]
verifylist = [
('server', _server.id),
('ip_address', _floating_ip['ip']),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.cmd.take_action(parsed_args)
attrs = {
'port_id': _port.id,
}
self.network.find_ip.assert_called_once_with(
_floating_ip['ip'],
ignore_missing=False,
)
self.network.ports.assert_called_once_with(
device_id=_server.id,
)
self.network.update_ip.assert_called_once_with(
_floating_ip,
**attrs
)
def test_server_add_floating_ip_fixed(self):
_server = compute_fakes.FakeServer.create_one_server()
self.servers_mock.get.return_value = _server
_port = network_fakes.FakePort.create_one_port()
_floating_ip = network_fakes.FakeFloatingIP.create_one_floating_ip()
self.network.find_ip = mock.Mock(return_value=_floating_ip)
self.network.ports = mock.Mock(return_value=[_port])
arglist = [
'--fixed-ip-address', _floating_ip['fixed_ip'],
_server.id,
_floating_ip['ip'],
]
verifylist = [
('fixed_ip_address', _floating_ip['fixed_ip']),
('server', _server.id),
('ip_address', _floating_ip['ip']),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.cmd.take_action(parsed_args)
attrs = {
'port_id': _port.id,
}
self.network.find_ip.assert_called_once_with(
_floating_ip['ip'],
ignore_missing=False,
)
self.network.ports.assert_called_once_with(
device_id=_server.id,
)
self.network.update_ip.assert_called_once_with(
_floating_ip,
**attrs
)
class TestServerAddPort(TestServer):
@ -2302,36 +2402,80 @@ class TestServerRescue(TestServer):
self.server.rescue.assert_called_with(image=None, password=password)
class TestServerRemoveFloatingIP(TestServer):
@mock.patch(
'openstackclient.api.compute_v2.APIv2.floating_ip_remove'
)
class TestServerRemoveFloatingIPCompute(compute_fakes.TestComputev2):
def setUp(self):
super(TestServerRemoveFloatingIP, self).setUp()
super(TestServerRemoveFloatingIPCompute, self).setUp()
self.app.client_manager.network_endpoint_enabled = False
# Get the command object to test
self.cmd = server.RemoveFloatingIP(self.app, None)
# Set unshelve method to be tested.
self.methods = {
'remove_floating_ip': None,
}
def test_server_remove_floating_ip(self):
servers = self.setup_servers_mock(count=1)
def test_server_remove_floating_ip(self, fip_mock):
_floating_ip = compute_fakes.FakeFloatingIP.create_one_floating_ip()
arglist = [
servers[0].id,
'1.2.3.4',
'server1',
_floating_ip['ip'],
]
verifylist = [
('server', servers[0].id),
('ip_address', '1.2.3.4'),
('server', 'server1'),
('ip_address', _floating_ip['ip']),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.cmd.take_action(parsed_args)
servers[0].remove_floating_ip.assert_called_once_with('1.2.3.4')
self.assertIsNone(result)
fip_mock.assert_called_once_with(
'server1',
_floating_ip['ip'],
)
class TestServerRemoveFloatingIPNetwork(network_fakes.TestNetworkV2):
def setUp(self):
super(TestServerRemoveFloatingIPNetwork, self).setUp()
self.app.client_manager.network = mock.Mock()
self.network = self.app.client_manager.network
self.network.update_ip = mock.Mock(return_value=None)
# Get the command object to test
self.cmd = server.RemoveFloatingIP(self.app, self.namespace)
def test_server_remove_floating_ip_default(self):
_server = compute_fakes.FakeServer.create_one_server()
_floating_ip = network_fakes.FakeFloatingIP.create_one_floating_ip()
self.network.find_ip = mock.Mock(return_value=_floating_ip)
arglist = [
_server.id,
_floating_ip['ip'],
]
verifylist = [
('server', _server.id),
('ip_address', _floating_ip['ip']),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
self.cmd.take_action(parsed_args)
attrs = {
'port_id': None,
}
self.network.find_ip.assert_called_once_with(
_floating_ip['ip'],
ignore_missing=False,
)
self.network.update_ip.assert_called_once_with(
_floating_ip,
**attrs
)
class TestServerRemovePort(TestServer):