openstack
/
nova
Code Issues Proposed changes

Fix order of arguments in assertEqual 

Some tests used incorrect order assertEqual(observed, expected).

The correct order expected by testtools is
assertEqual(expected, observed).

Replaced assertEqual with assertTrue/assertFalse wherever required.

Partial-Bug: #1259292
Change-Id: If0a1f3850d92d011aae32ae34e8c8664f2ee4313
This commit is contained in:
Rajesh Tailor 2015-09-04 02:37:27 -07:00
parent 806dc025db
commit 830010b443
10 changed files with 336 additions and 333 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
nova/tests/unit/network/security_group/test_neutron_driver.py
View File

@ -320,7 +320,7 @@ class TestNeutronDriver(test.NoDBTestCase):
sg_api = neutron_driver.SecurityGroupAPI()
result = sg_api.get_instances_security_groups_bindings(
self.context, servers)
self.assertEqual(result, sg_bindings)
self.assertEqual(sg_bindings, result)
def _test_instances_security_group_bindings_scale(self, num_servers):
max_query = 150
@ -356,7 +356,7 @@ class TestNeutronDriver(test.NoDBTestCase):
sg_api = neutron_driver.SecurityGroupAPI()
result = sg_api.get_instances_security_groups_bindings(
self.context, servers)
self.assertEqual(result, sg_bindings)
self.assertEqual(sg_bindings, result)
def test_instances_security_group_bindings_less_than_max(self):
self._test_instances_security_group_bindings_scale(100)
@ -388,7 +388,7 @@ class TestNeutronDriver(test.NoDBTestCase):
sg_api = neutron_driver.SecurityGroupAPI()
result = sg_api.get_instances_security_groups_bindings(
self.context, servers)
self.assertEqual(result, sg_bindings)
self.assertEqual(sg_bindings, result)
def test_instance_empty_security_groups(self):

46
nova/tests/unit/network/test_linux_net.py
View File

@ -403,8 +403,8 @@ class LinuxNetworkTestCase(test.NoDBTestCase):
def _test_add_snat_rule(self, expected, is_external):
def verify_add_rule(chain, rule):
self.assertEqual(chain, 'snat')
self.assertEqual(rule, expected)
self.assertEqual('snat', chain)
self.assertEqual(expected, rule)
self.called = True
self.stubs.Set(linux_net.iptables_manager.ipv4['nat'],
@ -505,7 +505,7 @@ class LinuxNetworkTestCase(test.NoDBTestCase):
actual_hosts = self.driver.get_dhcp_hosts(self.context, networks[0],
fixedips)
self.assertEqual(actual_hosts, expected)
self.assertEqual(expected, actual_hosts)
def test_get_dhcp_hosts_for_nw01(self):
self.flags(use_single_default_gateway=True)
@ -519,7 +519,7 @@ class LinuxNetworkTestCase(test.NoDBTestCase):
fixedips = self._get_fixedips(networks[1], host='fake_instance01')
actual_hosts = self.driver.get_dhcp_hosts(self.context, networks[1],
fixedips)
self.assertEqual(actual_hosts, expected)
self.assertEqual(expected, actual_hosts)
def test_get_dns_hosts_for_nw00(self):
expected = (
@ -528,7 +528,7 @@ class LinuxNetworkTestCase(test.NoDBTestCase):
"192.168.0.102\tfake_instance00.novalocal"
)
actual_hosts = self.driver.get_dns_hosts(self.context, networks[0])
self.assertEqual(actual_hosts, expected)
self.assertEqual(expected, actual_hosts)
def test_get_dns_hosts_for_nw01(self):
expected = (
@ -537,7 +537,7 @@ class LinuxNetworkTestCase(test.NoDBTestCase):
"192.168.1.102\tfake_instance01.novalocal"
)
actual_hosts = self.driver.get_dns_hosts(self.context, networks[1])
self.assertEqual(actual_hosts, expected)
self.assertEqual(expected, actual_hosts)
def test_get_dhcp_opts_for_nw00(self):
self.flags(use_single_default_gateway=True)
@ -546,7 +546,7 @@ class LinuxNetworkTestCase(test.NoDBTestCase):
actual_opts = self.driver.get_dhcp_opts(self.context, networks[0],
fixedips)
self.assertEqual(actual_opts, expected_opts)
self.assertEqual(expected_opts, actual_opts)
def test_get_dhcp_opts_for_nw00_no_single_default_gateway(self):
self.flags(use_single_default_gateway=False)
@ -555,7 +555,7 @@ class LinuxNetworkTestCase(test.NoDBTestCase):
actual_opts = self.driver.get_dhcp_opts(self.context, networks[0],
fixedips)
self.assertEqual(actual_opts, expected_opts)
self.assertEqual(expected_opts, actual_opts)
def test_get_dhcp_opts_for_nw01(self):
self.flags(use_single_default_gateway=True)
@ -564,7 +564,7 @@ class LinuxNetworkTestCase(test.NoDBTestCase):
actual_opts = self.driver.get_dhcp_opts(self.context, networks[1],
fixedips)
self.assertEqual(actual_opts, expected_opts)
self.assertEqual(expected_opts, actual_opts)
def test_get_dhcp_leases_for_nw00(self):
timestamp = timeutils.utcnow()
@ -605,7 +605,7 @@ class LinuxNetworkTestCase(test.NoDBTestCase):
fixedip = objects.FixedIPList.get_by_network(self.context,
{'id': 0})[0]
actual = self.driver._host_dhcp_opts(fixedip.virtual_interface_id)
self.assertEqual(actual, expected)
self.assertEqual(expected, actual)
def test_host_dhcp_without_default_gateway_network(self):
expected = ','.join(['DE:AD:BE:EF:00:00',
@ -614,7 +614,7 @@ class LinuxNetworkTestCase(test.NoDBTestCase):
fixedip = objects.FixedIPList.get_by_network(self.context,
{'id': 0})[0]
actual = self.driver._host_dhcp(fixedip)
self.assertEqual(actual, expected)
self.assertEqual(expected, actual)
def test_host_dhcp_truncated_hostname(self):
expected = ','.join(['DE:AD:BE:EF:00:07',
@ -631,7 +631,7 @@ class LinuxNetworkTestCase(test.NoDBTestCase):
fixedip = objects.FixedIPList.get_by_network(self.context,
{'id': 0})[0]
actual = self.driver._host_dns(fixedip)
self.assertEqual(actual, expected)
self.assertEqual(expected, actual)
def test_linux_bridge_driver_plug(self):
"""Makes sure plug doesn't drop FORWARD by default.
@ -644,7 +644,7 @@ class LinuxNetworkTestCase(test.NoDBTestCase):
self.stubs.Set(utils, 'execute', fake_execute)
def verify_add_rule(chain, rule):
self.assertEqual(chain, 'FORWARD')
self.assertEqual('FORWARD', chain)
self.assertIn('ACCEPT', rule)
self.stubs.Set(linux_net.iptables_manager.ipv4['filter'],
'add_rule', verify_add_rule)
@ -693,10 +693,10 @@ class LinuxNetworkTestCase(test.NoDBTestCase):
}
self.flags(vlan_interface="")
driver.plug(network, "fakemac")
self.assertEqual(info['passed_interface'], "base_interface")
self.assertEqual("base_interface", info['passed_interface'])
self.flags(vlan_interface="override_interface")
driver.plug(network, "fakemac")
self.assertEqual(info['passed_interface'], "override_interface")
self.assertEqual("override_interface", info['passed_interface'])
driver.plug(network, "fakemac")
def test_flat_override(self):
@ -722,10 +722,10 @@ class LinuxNetworkTestCase(test.NoDBTestCase):
"share_address": False,
}
driver.plug(network, "fakemac")
self.assertEqual(info['passed_interface'], "base_interface")
self.assertEqual("base_interface", info['passed_interface'])
self.flags(flat_interface="override_interface")
driver.plug(network, "fakemac")
self.assertEqual(info['passed_interface'], "override_interface")
self.assertEqual("override_interface", info['passed_interface'])
def _test_dnsmasq_execute(self, extra_expected=None):
network_ref = {'id': 'fake',
@ -866,7 +866,7 @@ class LinuxNetworkTestCase(test.NoDBTestCase):
('ip6tables-save', '-c'),
('ip6tables-restore', '-c'),
]
self.assertEqual(executes, expected)
self.assertEqual(expected, executes)
executes = []
@ -890,7 +890,7 @@ class LinuxNetworkTestCase(test.NoDBTestCase):
iface, '--ip-protocol', 'udp', '--ip-destination-port', '67:68',
'-j', 'DROP'),
]
self.assertEqual(executes, expected)
self.assertEqual(expected, executes)
def _test_initialize_gateway(self, existing, expected, routes=''):
self.flags(fake_network=False)
@ -910,7 +910,7 @@ class LinuxNetworkTestCase(test.NoDBTestCase):
'broadcast': '192.168.1.255',
'cidr_v6': '2001:db8::/64'}
self.driver.initialize_gateway_device('eth0', network)
self.assertEqual(executes, expected)
self.assertEqual(expected, executes)
def test_initialize_gateway_moves_wrong_ip(self):
existing = ("2: eth0: <BROADCAST,MULTICAST,UP,LOWER_UP> "
@ -1059,7 +1059,7 @@ class LinuxNetworkTestCase(test.NoDBTestCase):
def _test_add_metadata_accept_rule(self, expected):
def verify_add_rule(chain, rule):
self.assertEqual(chain, 'INPUT')
self.assertEqual('INPUT', chain)
self.assertEqual(expected, rule)
self.stubs.Set(linux_net.iptables_manager.ipv4['filter'],
@ -1068,7 +1068,7 @@ class LinuxNetworkTestCase(test.NoDBTestCase):
def _test_add_metadata_accept_ipv6_rule(self, expected):
def verify_add_rule(chain, rule):
self.assertEqual(chain, 'INPUT')
self.assertEqual('INPUT', chain)
self.assertEqual(expected, rule)
self.stubs.Set(linux_net.iptables_manager.ipv6['filter'],
@ -1105,7 +1105,7 @@ class LinuxNetworkTestCase(test.NoDBTestCase):
def _test_add_metadata_forward_rule(self, expected):
def verify_add_rule(chain, rule):
self.assertEqual(chain, 'PREROUTING')
self.assertEqual('PREROUTING', chain)
self.assertEqual(expected, rule)
self.stubs.Set(linux_net.iptables_manager.ipv4['nat'],

164
nova/tests/unit/network/test_manager.py
View File

@ -460,7 +460,7 @@ class FlatNetworkTestCase(test.TestCase):
nw_info = self.network.get_instance_nw_info(self.context, None,
None, None)
for i, vif in enumerate(nw_info):
self.assertEqual(vif['network']['bridge'], objs[i].network.bridge)
self.assertEqual(objs[i].network.bridge, vif['network']['bridge'])
@mock.patch.object(objects.Network, 'get_by_id')
def test_add_fixed_ip_instance_using_id_without_vpn(self, get_by_id):
@ -526,16 +526,16 @@ class FlatNetworkTestCase(test.TestCase):
driver.modify_address("hostfour", "10.0.0.1", zone1)
driver.modify_address("hostthree", "10.0.0.1", zone1)
names = driver.get_entries_by_address("10.0.0.1", zone1)
self.assertEqual(len(names), 2)
self.assertEqual(2, len(names))
self.assertIn('hostthree', names)
self.assertIn('hostfour', names)
names = driver.get_entries_by_address("10.0.0.5", zone2)
self.assertEqual(len(names), 1)
self.assertEqual(1, len(names))
self.assertIn('hostfive', names)
addresses = driver.get_entries_by_name("hosttwo", zone1)
self.assertEqual(len(addresses), 1)
self.assertEqual(1, len(addresses))
self.assertIn('10.0.0.2', addresses)
self.assertRaises(exception.InvalidInput,
@ -550,11 +550,11 @@ class FlatNetworkTestCase(test.TestCase):
driver = self.network.instance_dns_manager
driver.create_entry("HostTen", "10.0.0.10", "A", zone1)
addresses = driver.get_entries_by_address("10.0.0.10", zone1)
self.assertEqual(len(addresses), 1)
self.assertEqual(1, len(addresses))
for n in addresses:
driver.delete_entry(n, zone1)
addresses = driver.get_entries_by_address("10.0.0.10", zone1)
self.assertEqual(len(addresses), 0)
self.assertEqual(0, len(addresses))
def test_allocate_fixed_ip_instance_dns(self):
# Test DNS entries are created when allocating a fixed IP.
@ -629,8 +629,8 @@ class FlatNetworkTestCase(test.TestCase):
res = self.network._get_networks_by_uuids(self.context,
requested_networks)
self.assertEqual(res[0]['id'], 1)
self.assertEqual(res[1]['id'], 0)
self.assertEqual(1, res[0]['id'])
self.assertEqual(0, res[1]['id'])
@mock.patch('nova.objects.instance.Instance.get_by_uuid')
@mock.patch('nova.objects.quotas.Quotas.reserve')
@ -1049,7 +1049,7 @@ class VlanNetworkTestCase(test.TestCase):
vlan_start=100, cidr='192.168.3.1/24',
network_size=100)
self.assertEqual(networks[0]["vlan"], 102)
self.assertEqual(102, networks[0]["vlan"])
def test_vlan_start_multiple(self):
# VLAN 100 and 101 are used, so these networks shoud be created in 102
@ -1059,8 +1059,8 @@ class VlanNetworkTestCase(test.TestCase):
vlan_start=100, cidr='192.168.3.1/24',
network_size=100)
self.assertEqual(networks[0]["vlan"], 102)
self.assertEqual(networks[1]["vlan"], 103)
self.assertEqual(102, networks[0]["vlan"])
self.assertEqual(103, networks[1]["vlan"])
def test_vlan_start_used(self):
# VLAN 100 and 101 are used, but vlan_start=99.
@ -1069,7 +1069,7 @@ class VlanNetworkTestCase(test.TestCase):
vlan_start=99, cidr='192.168.3.1/24',
network_size=100)
self.assertEqual(networks[0]["vlan"], 102)
self.assertEqual(102, networks[0]["vlan"])
def test_vlan_parameter(self):
# vlan parameter could not be greater than 4094
@ -1106,8 +1106,8 @@ class VlanNetworkTestCase(test.TestCase):
vlan_start=100, cidr='192.168.3.1/24',
network_size=100)
self.assertEqual(networks[0]["dhcp_server"], "192.168.3.1")
self.assertEqual(networks[1]["dhcp_server"], "192.168.3.129")
self.assertEqual("192.168.3.1", networks[0]["dhcp_server"])
self.assertEqual("192.168.3.129", networks[1]["dhcp_server"])
def test_vlan_multiple_with_dhcp_server(self):
networks = self.network.create_networks(
@ -1115,8 +1115,8 @@ class VlanNetworkTestCase(test.TestCase):
vlan_start=100, cidr='192.168.3.1/24',
network_size=100, dhcp_server='192.168.3.1')
self.assertEqual(networks[0]["dhcp_server"], "192.168.3.1")
self.assertEqual(networks[1]["dhcp_server"], "192.168.3.1")
self.assertEqual("192.168.3.1", networks[0]["dhcp_server"])
self.assertEqual("192.168.3.1", networks[1]["dhcp_server"])
def test_validate_networks(self):
self.mox.StubOutWithMock(db, "fixed_ip_get_by_address")
@ -1884,8 +1884,8 @@ class VlanNetworkTestCase(test.TestCase):
res = self.network._get_networks_by_uuids(self.context,
requested_networks)
self.assertEqual(res[0]['id'], 1)
self.assertEqual(res[1]['id'], 0)
self.assertEqual(1, res[0]['id'])
self.assertEqual(0, res[1]['id'])
@mock.patch('nova.objects.fixed_ip.FixedIP.get_by_id')
@mock.patch('nova.objects.floating_ip.FloatingIPList.get_by_host')
@ -2080,7 +2080,7 @@ class CommonNetworkTestCase(test.TestCase):
HOST,
'10.0.0.1')
self.assertEqual(manager.deallocate_called, '10.0.0.1')
self.assertEqual('10.0.0.1', manager.deallocate_called)
disassociate.assert_called_once_with(self.context, '10.0.0.1')
@mock.patch('nova.db.fixed_ip_get_by_instance')
@ -2294,7 +2294,7 @@ class CommonNetworkTestCase(test.TestCase):
# Greedy get eveything
res = manager.get_instance_uuids_by_ip_filter(fake_context,
{'ip': '.*'})
self.assertEqual(len(res), len(_vifs))
self.assertEqual(len(_vifs), len(res))
# Doesn't exist
res = manager.get_instance_uuids_by_ip_filter(fake_context,
@ -2305,31 +2305,31 @@ class CommonNetworkTestCase(test.TestCase):
res = manager.get_instance_uuids_by_ip_filter(fake_context,
{'ip': '172.16.0.2'})
self.assertTrue(res)
self.assertEqual(len(res), 1)
self.assertEqual(res[0]['instance_uuid'], _vifs[1]['instance_uuid'])
self.assertEqual(1, len(res))
self.assertEqual(_vifs[1]['instance_uuid'], res[0]['instance_uuid'])
# Get instance 2
res = manager.get_instance_uuids_by_ip_filter(fake_context,
{'ip': '173.16.0.2'})
self.assertTrue(res)
self.assertEqual(len(res), 1)
self.assertEqual(res[0]['instance_uuid'], _vifs[2]['instance_uuid'])
self.assertEqual(1, len(res))
self.assertEqual(_vifs[2]['instance_uuid'], res[0]['instance_uuid'])
# Get instance 0 and 1
res = manager.get_instance_uuids_by_ip_filter(fake_context,
{'ip': '172.16.0.*'})
self.assertTrue(res)
self.assertEqual(len(res), 2)
self.assertEqual(res[0]['instance_uuid'], _vifs[0]['instance_uuid'])
self.assertEqual(res[1]['instance_uuid'], _vifs[1]['instance_uuid'])
self.assertEqual(2, len(res))
self.assertEqual(_vifs[0]['instance_uuid'], res[0]['instance_uuid'])
self.assertEqual(_vifs[1]['instance_uuid'], res[1]['instance_uuid'])
# Get instance 1 and 2
res = manager.get_instance_uuids_by_ip_filter(fake_context,
{'ip': '17..16.0.2'})
self.assertTrue(res)
self.assertEqual(len(res), 2)
self.assertEqual(res[0]['instance_uuid'], _vifs[1]['instance_uuid'])
self.assertEqual(res[1]['instance_uuid'], _vifs[2]['instance_uuid'])
self.assertEqual(2, len(res))
self.assertEqual(_vifs[1]['instance_uuid'], res[0]['instance_uuid'])
self.assertEqual(_vifs[2]['instance_uuid'], res[1]['instance_uuid'])
@mock.patch('nova.db.network_get')
def test_get_instance_uuids_by_ipv6_regex(self, network_get):
@ -2345,7 +2345,7 @@ class CommonNetworkTestCase(test.TestCase):
# Greedy get eveything
res = manager.get_instance_uuids_by_ip_filter(fake_context,
{'ip6': '.*'})
self.assertEqual(len(res), len(_vifs))
self.assertEqual(len(_vifs), len(res))
# Doesn't exist
res = manager.get_instance_uuids_by_ip_filter(fake_context,
@ -2356,33 +2356,33 @@ class CommonNetworkTestCase(test.TestCase):
res = manager.get_instance_uuids_by_ip_filter(fake_context,
{'ip6': '2001:.*2'})
self.assertTrue(res)
self.assertEqual(len(res), 1)
self.assertEqual(res[0]['instance_uuid'], _vifs[1]['instance_uuid'])
self.assertEqual(1, len(res))
self.assertEqual(_vifs[1]['instance_uuid'], res[0]['instance_uuid'])
# Get instance 2
ip6 = '2001:db8:69:1f:dead:beff:feff:ef03'
res = manager.get_instance_uuids_by_ip_filter(fake_context,
{'ip6': ip6})
self.assertTrue(res)
self.assertEqual(len(res), 1)
self.assertEqual(res[0]['instance_uuid'], _vifs[2]['instance_uuid'])
self.assertEqual(1, len(res))
self.assertEqual(_vifs[2]['instance_uuid'], res[0]['instance_uuid'])
# Get instance 0 and 1
res = manager.get_instance_uuids_by_ip_filter(fake_context,
{'ip6': '.*ef0[1,2]'})
self.assertTrue(res)
self.assertEqual(len(res), 2)
self.assertEqual(res[0]['instance_uuid'], _vifs[0]['instance_uuid'])
self.assertEqual(res[1]['instance_uuid'], _vifs[1]['instance_uuid'])
self.assertEqual(2, len(res))
self.assertEqual(_vifs[0]['instance_uuid'], res[0]['instance_uuid'])
self.assertEqual(_vifs[1]['instance_uuid'], res[1]['instance_uuid'])
# Get instance 1 and 2
ip6 = '2001:db8:69:1.:dead:beff:feff:ef0.'
res = manager.get_instance_uuids_by_ip_filter(fake_context,
{'ip6': ip6})
self.assertTrue(res)
self.assertEqual(len(res), 2)
self.assertEqual(res[0]['instance_uuid'], _vifs[1]['instance_uuid'])
self.assertEqual(res[1]['instance_uuid'], _vifs[2]['instance_uuid'])
self.assertEqual(2, len(res))
self.assertEqual(_vifs[1]['instance_uuid'], res[0]['instance_uuid'])
self.assertEqual(_vifs[2]['instance_uuid'], res[1]['instance_uuid'])
@mock.patch('nova.db.network_get')
@mock.patch('nova.db.fixed_ips_by_virtual_interface')
@ -2410,16 +2410,16 @@ class CommonNetworkTestCase(test.TestCase):
res = manager.get_instance_uuids_by_ip_filter(fake_context,
{'fixed_ip': ip})
self.assertTrue(res)
self.assertEqual(len(res), 1)
self.assertEqual(res[0]['instance_uuid'], _vifs[1]['instance_uuid'])
self.assertEqual(1, len(res))
self.assertEqual(_vifs[1]['instance_uuid'], res[0]['instance_uuid'])
# Get instance 2
ip = '173.16.0.2'
res = manager.get_instance_uuids_by_ip_filter(fake_context,
{'fixed_ip': ip})
self.assertTrue(res)
self.assertEqual(len(res), 1)
self.assertEqual(res[0]['instance_uuid'], _vifs[2]['instance_uuid'])
self.assertEqual(1, len(res))
self.assertEqual(_vifs[2]['instance_uuid'], res[0]['instance_uuid'])
@mock.patch('nova.db.network_get_by_uuid')
def test_get_network(self, get):
@ -2428,7 +2428,7 @@ class CommonNetworkTestCase(test.TestCase):
get.return_value = dict(test_network.fake_network, **networks[0])
uuid = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'
network = manager.get_network(fake_context, uuid)
self.assertEqual(network['uuid'], uuid)
self.assertEqual(uuid, network['uuid'])
@mock.patch('nova.db.network_get_by_uuid')
def test_get_network_not_found(self, get):
@ -2446,11 +2446,11 @@ class CommonNetworkTestCase(test.TestCase):
get_all.return_value = [dict(test_network.fake_network, **net)
for net in networks]
output = manager.get_all_networks(fake_context)
self.assertEqual(len(networks), 2)
self.assertEqual(output[0]['uuid'],
'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa')
self.assertEqual(output[1]['uuid'],
'bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb')
self.assertEqual(2, len(networks))
self.assertEqual('aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa',
output[0]['uuid'])
self.assertEqual('bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb',
output[1]['uuid'])
@mock.patch('nova.db.network_get_by_uuid')
@mock.patch('nova.db.network_disassociate')
@ -2672,7 +2672,7 @@ class RPCAllocateTestCase(test.NoDBTestCase):
rval = self.rpc_fixed._rpc_allocate_fixed_ip(self.context,
'fake_instance',
'fake_network')
self.assertEqual(rval, address)
self.assertEqual(address, rval)
class TestFloatingIPManager(floating_ips.FloatingIP,
@ -3027,7 +3027,7 @@ class FloatingIPTestCase(test.TestCase):
source='fake_source',
dest='fake_dest')
self.assertEqual(called['count'], 2)
self.assertEqual(2, called['count'])
@mock.patch('nova.db.fixed_ip_get')
@mock.patch('nova.db.floating_ip_update')
@ -3069,7 +3069,7 @@ class FloatingIPTestCase(test.TestCase):
project_id=self.project_id,
source='fake_source')
self.assertEqual(called['count'], 2)
self.assertEqual(2, called['count'])
def test_floating_dns_create_conflict(self):
zone = "example.org"
@ -3095,14 +3095,14 @@ class FloatingIPTestCase(test.TestCase):
self.network.add_dns_entry(self.context, address1, name2, "A", zone)
entries = self.network.get_dns_entries_by_address(self.context,
address1, zone)
self.assertEqual(len(entries), 2)
self.assertEqual(entries[0], name1)
self.assertEqual(entries[1], name2)
self.assertEqual(2, len(entries))
self.assertEqual(name1, entries[0])
self.assertEqual(name2, entries[1])
entries = self.network.get_dns_entries_by_name(self.context,
name1, zone)
self.assertEqual(len(entries), 1)
self.assertEqual(entries[0], address1)
self.assertEqual(1, len(entries))
self.assertEqual(address1, entries[0])
def test_floating_dns_delete(self):
zone = "example.org"
@ -3116,8 +3116,8 @@ class FloatingIPTestCase(test.TestCase):
entries = self.network.get_dns_entries_by_address(self.context,
address1, zone)
self.assertEqual(len(entries), 1)
self.assertEqual(entries[0], name2)
self.assertEqual(1, len(entries))
self.assertEqual(name2, entries[0])
self.assertRaises(exception.NotFound,
self.network.delete_dns_entry, self.context,
@ -3135,18 +3135,18 @@ class FloatingIPTestCase(test.TestCase):
'fakeproject')
domains = self.network.get_dns_domains(self.context)
self.assertEqual(len(domains), 2)
self.assertEqual(domains[0]['domain'], domain1)
self.assertEqual(domains[1]['domain'], domain2)
self.assertEqual(domains[0]['project'], 'testproject')
self.assertEqual(domains[1]['project'], 'fakeproject')
self.assertEqual(2, len(domains))
self.assertEqual(domain1, domains[0]['domain'])
self.assertEqual(domain2, domains[1]['domain'])
self.assertEqual('testproject', domains[0]['project'])
self.assertEqual('fakeproject', domains[1]['project'])
self.network.add_dns_entry(self.context, address1, entryname,
'A', domain1)
entries = self.network.get_dns_entries_by_name(self.context,
entryname, domain1)
self.assertEqual(len(entries), 1)
self.assertEqual(entries[0], address1)
self.assertEqual(1, len(entries))
self.assertEqual(address1, entries[0])
self.network.delete_dns_domain(self.context, domain1)
self.network.delete_dns_domain(self.context, domain2)
@ -3187,7 +3187,7 @@ class FloatingIPTestCase(test.TestCase):
entries = self.network.get_dns_entries_by_address(self.context,
address,
domain['domain'])
self.assertEqual(len(entries), 2)
self.assertEqual(2, len(entries))
self.network._delete_all_entries_for_ip(self.context, address)
@ -3234,7 +3234,7 @@ class FloatingIPTestCase(test.TestCase):
# Attempt to add another and make sure that both MACs are consumed
# by the retry loop
self.network._add_virtual_interface(ctxt, 'fake_uuid', 123)
self.assertEqual(macs, [])
self.assertEqual([], macs)
def test_deallocate_client_exceptions(self):
# Ensure that FloatingIpNotFoundForAddress is wrapped.
@ -3363,9 +3363,9 @@ class InstanceDNSTestCase(test.TestCase):
self.network.create_private_dns_domain(self.context, domain1, zone1)
domains = self.network.get_dns_domains(self.context)
self.assertEqual(len(domains), 1)
self.assertEqual(domains[0]['domain'], domain1)
self.assertEqual(domains[0]['availability_zone'], zone1)
self.assertEqual(1, len(domains))
self.assertEqual(domain1, domains[0]['domain'])
self.assertEqual(zone1, domains[0]['availability_zone'])
self.network.delete_dns_domain(self.context, domain1)
@ -3400,7 +3400,7 @@ class LdapDNSTestCase(test.NoDBTestCase):
def test_ldap_dns_domains(self):
domains = self.driver.get_domains()
self.assertEqual(len(domains), 2)
self.assertEqual(2, len(domains))
self.assertIn(domain1, domains)
self.assertIn(domain2, domains)
@ -3424,13 +3424,13 @@ class LdapDNSTestCase(test.NoDBTestCase):
self.driver.create_entry(name1, address1, "A", domain1)
self.driver.create_entry(name2, address1, "A", domain1)
entries = self.driver.get_entries_by_address(address1, domain1)
self.assertEqual(len(entries), 2)
self.assertEqual(entries[0], name1)
self.assertEqual(entries[1], name2)
self.assertEqual(2, len(entries))
self.assertEqual(name1, entries[0])
self.assertEqual(name2, entries[1])
entries = self.driver.get_entries_by_name(name1, domain1)
self.assertEqual(len(entries), 1)
self.assertEqual(entries[0], address1)
self.assertEqual(1, len(entries))
self.assertEqual(address1, entries[0])
def test_ldap_dns_delete(self):
address1 = "10.10.10.11"
@ -3440,13 +3440,13 @@ class LdapDNSTestCase(test.NoDBTestCase):
self.driver.create_entry(name1, address1, "A", domain1)
self.driver.create_entry(name2, address1, "A", domain1)
entries = self.driver.get_entries_by_address(address1, domain1)
self.assertEqual(len(entries), 2)
self.assertEqual(2, len(entries))
self.driver.delete_entry(name1, domain1)
entries = self.driver.get_entries_by_address(address1, domain1)
LOG.debug("entries: %s" % entries)
self.assertEqual(len(entries), 1)
self.assertEqual(entries[0], name2)
self.assertEqual(1, len(entries))
self.assertEqual(name2, entries[0])
self.assertRaises(exception.NotFound,
self.driver.delete_entry,

175
nova/tests/unit/network/test_network_info.py
View File

@ -27,9 +27,9 @@ class RouteTests(test.NoDBTestCase):
def test_create_route_with_attrs(self):
route = fake_network_cache_model.new_route()
fake_network_cache_model.new_ip(dict(address='192.168.1.1'))
self.assertEqual(route['cidr'], '0.0.0.0/24')
self.assertEqual(route['gateway']['address'], '192.168.1.1')
self.assertEqual(route['interface'], 'eth0')
self.assertEqual('0.0.0.0/24', route['cidr'])
self.assertEqual('192.168.1.1', route['gateway']['address'])
self.assertEqual('eth0', route['interface'])
def test_routes_equal(self):
route1 = model.Route()
@ -54,7 +54,7 @@ class RouteTests(test.NoDBTestCase):
{'gateway': fake_network_cache_model.new_ip(
dict(address='192.168.1.1'))})
self.assertIsNone(route['cidr'])
self.assertEqual(route['gateway']['address'], '192.168.1.1')
self.assertEqual('192.168.1.1', route['gateway']['address'])
self.assertIsNone(route['interface'])
@ -81,17 +81,17 @@ class IPTests(test.NoDBTestCase):
class FixedIPTests(test.NoDBTestCase):
def test_createnew_fixed_ip_with_attrs(self):
fixed_ip = model.FixedIP(address='192.168.1.100')
self.assertEqual(fixed_ip['address'], '192.168.1.100')
self.assertEqual(fixed_ip['floating_ips'], [])
self.assertEqual(fixed_ip['type'], 'fixed')
self.assertEqual(fixed_ip['version'], 4)
self.assertEqual('192.168.1.100', fixed_ip['address'])
self.assertEqual([], fixed_ip['floating_ips'])
self.assertEqual('fixed', fixed_ip['type'])
self.assertEqual(4, fixed_ip['version'])
def test_create_fixed_ipv6(self):
fixed_ip = model.FixedIP(address='::1')
self.assertEqual(fixed_ip['address'], '::1')
self.assertEqual(fixed_ip['floating_ips'], [])
self.assertEqual(fixed_ip['type'], 'fixed')
self.assertEqual(fixed_ip['version'], 6)
self.assertEqual('::1', fixed_ip['address'])
self.assertEqual([], fixed_ip['floating_ips'])
self.assertEqual('fixed', fixed_ip['type'])
self.assertEqual(6, fixed_ip['version'])
def test_create_fixed_bad_ip_fails(self):
self.assertRaises(exception.InvalidIpAddressError,
@ -122,21 +122,21 @@ class FixedIPTests(test.NoDBTestCase):
def test_hydrate(self):
fixed_ip = model.FixedIP.hydrate({})
self.assertEqual(fixed_ip['floating_ips'], [])
self.assertEqual([], fixed_ip['floating_ips'])
self.assertIsNone(fixed_ip['address'])
self.assertEqual(fixed_ip['type'], 'fixed')
self.assertEqual('fixed', fixed_ip['type'])
self.assertIsNone(fixed_ip['version'])
def test_add_floating_ip(self):
fixed_ip = model.FixedIP(address='192.168.1.100')
fixed_ip.add_floating_ip('192.168.1.101')
self.assertEqual(fixed_ip['floating_ips'], ['192.168.1.101'])
self.assertEqual(['192.168.1.101'], fixed_ip['floating_ips'])
def test_add_floating_ip_repeatedly_only_one_instance(self):
fixed_ip = model.FixedIP(address='192.168.1.100')
for i in range(10):
fixed_ip.add_floating_ip('192.168.1.101')
self.assertEqual(fixed_ip['floating_ips'], ['192.168.1.101'])
self.assertEqual(['192.168.1.101'], fixed_ip['floating_ips'])
class SubnetTests(test.NoDBTestCase):
@ -145,18 +145,19 @@ class SubnetTests(test.NoDBTestCase):
route1 = fake_network_cache_model.new_route()
self.assertEqual(subnet['cidr'], '10.10.0.0/24')
self.assertEqual(subnet['dns'],
[fake_network_cache_model.new_ip(dict(address='1.2.3.4')),
fake_network_cache_model.new_ip(dict(address='2.3.4.5'))])
self.assertEqual(subnet['gateway']['address'], '10.10.0.1')
self.assertEqual(subnet['ips'],
self.assertEqual('10.10.0.0/24', subnet['cidr'])
self.assertEqual(
[fake_network_cache_model.new_ip(dict(address='1.2.3.4')),
fake_network_cache_model.new_ip(dict(address='2.3.4.5'))],
subnet['dns'])
self.assertEqual('10.10.0.1', subnet['gateway']['address'])
self.assertEqual(
[fake_network_cache_model.new_fixed_ip(
dict(address='10.10.0.2')),
fake_network_cache_model.new_fixed_ip(
dict(address='10.10.0.3'))])
self.assertEqual(subnet['routes'], [route1])
self.assertEqual(subnet['version'], 4)
dict(address='10.10.0.3'))], subnet['ips'])
self.assertEqual([route1], subnet['routes'])
self.assertEqual(4, subnet['version'])
def test_subnet_equal(self):
subnet1 = fake_network_cache_model.new_subnet()
@ -193,7 +194,7 @@ class SubnetTests(test.NoDBTestCase):
route1 = fake_network_cache_model.new_route()
route2 = fake_network_cache_model.new_route({'cidr': '1.1.1.1/24'})
subnet.add_route(route2)
self.assertEqual(subnet['routes'], [route1, route2])
self.assertEqual([route1, route2], subnet['routes'])
def test_add_route_a_lot(self):
subnet = fake_network_cache_model.new_subnet()
@ -201,51 +202,53 @@ class SubnetTests(test.NoDBTestCase):
route2 = fake_network_cache_model.new_route({'cidr': '1.1.1.1/24'})
for i in range(10):
subnet.add_route(route2)
self.assertEqual(subnet['routes'], [route1, route2])
self.assertEqual([route1, route2], subnet['routes'])
def test_add_dns(self):
subnet = fake_network_cache_model.new_subnet()
dns = fake_network_cache_model.new_ip(dict(address='9.9.9.9'))
subnet.add_dns(dns)
self.assertEqual(subnet['dns'],
self.assertEqual(
[fake_network_cache_model.new_ip(dict(address='1.2.3.4')),
fake_network_cache_model.new_ip(dict(address='2.3.4.5')),
fake_network_cache_model.new_ip(dict(address='9.9.9.9'))])
fake_network_cache_model.new_ip(dict(address='9.9.9.9'))],
subnet['dns'])
def test_add_dns_a_lot(self):
subnet = fake_network_cache_model.new_subnet()
for i in range(10):
subnet.add_dns(fake_network_cache_model.new_ip(
dict(address='9.9.9.9')))
self.assertEqual(subnet['dns'],
self.assertEqual(
[fake_network_cache_model.new_ip(dict(address='1.2.3.4')),
fake_network_cache_model.new_ip(dict(address='2.3.4.5')),
fake_network_cache_model.new_ip(dict(address='9.9.9.9'))])
fake_network_cache_model.new_ip(dict(address='9.9.9.9'))],
subnet['dns'])
def test_add_ip(self):
subnet = fake_network_cache_model.new_subnet()
subnet.add_ip(fake_network_cache_model.new_ip(
dict(address='192.168.1.102')))
self.assertEqual(subnet['ips'],
self.assertEqual(
[fake_network_cache_model.new_fixed_ip(
dict(address='10.10.0.2')),
fake_network_cache_model.new_fixed_ip(
dict(address='10.10.0.3')),
fake_network_cache_model.new_ip(
dict(address='192.168.1.102'))])
dict(address='192.168.1.102'))], subnet['ips'])
def test_add_ip_a_lot(self):
subnet = fake_network_cache_model.new_subnet()
for i in range(10):
subnet.add_ip(fake_network_cache_model.new_fixed_ip(
dict(address='192.168.1.102')))
self.assertEqual(subnet['ips'],
self.assertEqual(
[fake_network_cache_model.new_fixed_ip(
dict(address='10.10.0.2')),
fake_network_cache_model.new_fixed_ip(
dict(address='10.10.0.3')),
fake_network_cache_model.new_fixed_ip(
dict(address='192.168.1.102'))])
dict(address='192.168.1.102'))], subnet['ips'])
def test_hydrate(self):
subnet_dict = {
@ -259,48 +262,50 @@ class SubnetTests(test.NoDBTestCase):
dict(address='3.3.3.3'))}
subnet = model.Subnet.hydrate(subnet_dict)
self.assertEqual(subnet['cidr'], '255.255.255.0')
self.assertEqual(subnet['dns'], [fake_network_cache_model.new_ip(
dict(address='1.1.1.1'))])
self.assertEqual(subnet['gateway']['address'], '3.3.3.3')
self.assertEqual(subnet['ips'], [fake_network_cache_model.new_fixed_ip(
dict(address='2.2.2.2'))])
self.assertEqual(subnet['routes'], [
fake_network_cache_model.new_route()])
self.assertEqual(subnet['version'], 4)
self.assertEqual('255.255.255.0', subnet['cidr'])
self.assertEqual([fake_network_cache_model.new_ip(
dict(address='1.1.1.1'))], subnet['dns'])
self.assertEqual('3.3.3.3', subnet['gateway']['address'])
self.assertEqual([fake_network_cache_model.new_fixed_ip(
dict(address='2.2.2.2'))], subnet['ips'])
self.assertEqual([fake_network_cache_model.new_route()],
subnet['routes'])
self.assertEqual(4, subnet['version'])
class NetworkTests(test.NoDBTestCase):
def test_create_network(self):
network = fake_network_cache_model.new_network()
self.assertEqual(network['id'], 1)
self.assertEqual(network['bridge'], 'br0')
self.assertEqual(network['label'], 'public')
self.assertEqual(network['subnets'],
self.assertEqual(1, network['id'])
self.assertEqual('br0', network['bridge'])
self.assertEqual('public', network['label'])
self.assertEqual(
[fake_network_cache_model.new_subnet(),
fake_network_cache_model.new_subnet(
dict(cidr='255.255.255.255'))])
dict(cidr='255.255.255.255'))], network['subnets'])
def test_add_subnet(self):
network = fake_network_cache_model.new_network()
network.add_subnet(fake_network_cache_model.new_subnet(
dict(cidr='0.0.0.0')))
self.assertEqual(network['subnets'],
self.assertEqual(
[fake_network_cache_model.new_subnet(),
fake_network_cache_model.new_subnet(
dict(cidr='255.255.255.255')),
fake_network_cache_model.new_subnet(dict(cidr='0.0.0.0'))])
fake_network_cache_model.new_subnet(dict(cidr='0.0.0.0'))],
network['subnets'])
def test_add_subnet_a_lot(self):
network = fake_network_cache_model.new_network()
for i in range(10):
network.add_subnet(fake_network_cache_model.new_subnet(
dict(cidr='0.0.0.0')))
self.assertEqual(network['subnets'],
self.assertEqual(
[fake_network_cache_model.new_subnet(),
fake_network_cache_model.new_subnet(
dict(cidr='255.255.255.255')),
fake_network_cache_model.new_subnet(dict(cidr='0.0.0.0'))])
fake_network_cache_model.new_subnet(dict(cidr='0.0.0.0'))],
network['subnets'])
def test_network_equal(self):
network1 = model.Network()
@ -329,22 +334,22 @@ class NetworkTests(test.NoDBTestCase):
fake_network_cache_model.new_subnet(dict(cidr='255.255.255.255'))
network = model.Network.hydrate(fake_network_cache_model.new_network())
self.assertEqual(network['id'], 1)
self.assertEqual(network['bridge'], 'br0')
self.assertEqual(network['label'], 'public')
self.assertEqual(network['subnets'],
self.assertEqual(1, network['id'])
self.assertEqual('br0', network['bridge'])
self.assertEqual('public', network['label'])
self.assertEqual(
[fake_network_cache_model.new_subnet(),
fake_network_cache_model.new_subnet(
dict(cidr='255.255.255.255'))])
dict(cidr='255.255.255.255'))], network['subnets'])
class VIFTests(test.NoDBTestCase):
def test_create_vif(self):
vif = fake_network_cache_model.new_vif()
self.assertEqual(vif['id'], 1)
self.assertEqual(vif['address'], 'aa:aa:aa:aa:aa:aa')
self.assertEqual(vif['network'],
fake_network_cache_model.new_network())
self.assertEqual(1, vif['id'])
self.assertEqual('aa:aa:aa:aa:aa:aa', vif['address'])
self.assertEqual(fake_network_cache_model.new_network(),
vif['network'])
def test_vif_equal(self):
vif1 = model.VIF()
@ -403,11 +408,11 @@ class VIFTests(test.NoDBTestCase):
network=fake_network_cache_model.new_network(),
type='bridge')
vif = fake_network_cache_model.new_vif(vif_dict)
self.assertEqual(vif['id'], 1)
self.assertEqual(vif['address'], 'aa:aa:aa:aa:aa:aa')
self.assertEqual(vif['type'], 'bridge')
self.assertEqual(vif['network'],
fake_network_cache_model.new_network())
self.assertEqual(1, vif['id'])
self.assertEqual('aa:aa:aa:aa:aa:aa', vif['address'])
self.assertEqual('bridge', vif['type'])
self.assertEqual(fake_network_cache_model.new_network(),
vif['network'])
def test_vif_get_fixed_ips(self):
vif = fake_network_cache_model.new_vif()
@ -422,7 +427,7 @@ class VIFTests(test.NoDBTestCase):
vif = fake_network_cache_model.new_vif()
vif['network']['subnets'][0]['ips'][0].add_floating_ip('192.168.1.1')
floating_ips = vif.floating_ips()
self.assertEqual(floating_ips, ['192.168.1.1'])
self.assertEqual(['192.168.1.1'], floating_ips)
def test_vif_get_labeled_ips(self):
vif = fake_network_cache_model.new_vif()
@ -434,15 +439,15 @@ class VIFTests(test.NoDBTestCase):
fake_network_cache_model.new_ip(
{'address': '10.10.0.3', 'type': 'fixed'})] * 2,
'network_label': 'public'}
self.assertEqual(labeled_ips, ip_dict)
self.assertEqual(ip_dict, labeled_ips)
def test_hydrate(self):
fake_network_cache_model.new_network()
vif = model.VIF.hydrate(fake_network_cache_model.new_vif())
self.assertEqual(vif['id'], 1)
self.assertEqual(vif['address'], 'aa:aa:aa:aa:aa:aa')
self.assertEqual(vif['network'],
fake_network_cache_model.new_network())
self.assertEqual(1, vif['id'])
self.assertEqual('aa:aa:aa:aa:aa:aa', vif['address'])
self.assertEqual(fake_network_cache_model.new_network(),
vif['network'])
def test_hydrate_vif_with_type(self):
vif_dict = dict(
@ -451,11 +456,11 @@ class VIFTests(test.NoDBTestCase):
network=fake_network_cache_model.new_network(),
type='bridge')
vif = model.VIF.hydrate(fake_network_cache_model.new_vif(vif_dict))
self.assertEqual(vif['id'], 1)
self.assertEqual(vif['address'], 'aa:aa:aa:aa:aa:aa')
self.assertEqual(vif['type'], 'bridge')
self.assertEqual(vif['network'],
fake_network_cache_model.new_network())
self.assertEqual(1, vif['id'])
self.assertEqual('aa:aa:aa:aa:aa:aa', vif['address'])
self.assertEqual('bridge', vif['type'])
self.assertEqual(fake_network_cache_model.new_network(),
vif['network'])
class NetworkInfoTests(test.NoDBTestCase):
@ -463,11 +468,11 @@ class NetworkInfoTests(test.NoDBTestCase):
ninfo = model.NetworkInfo([fake_network_cache_model.new_vif(),
fake_network_cache_model.new_vif(
{'address': 'bb:bb:bb:bb:bb:bb'})])
self.assertEqual(ninfo.fixed_ips(),
self.assertEqual(
[fake_network_cache_model.new_fixed_ip(
{'address': '10.10.0.2'}),
fake_network_cache_model.new_fixed_ip(
{'address': '10.10.0.3'})] * 4)
{'address': '10.10.0.3'})] * 4, ninfo.fixed_ips())
def test_create_async_model(self):
def async_wrapper():
@ -477,11 +482,11 @@ class NetworkInfoTests(test.NoDBTestCase):
{'address': 'bb:bb:bb:bb:bb:bb'})])
ninfo = model.NetworkInfoAsyncWrapper(async_wrapper)
self.assertEqual(ninfo.fixed_ips(),
self.assertEqual(
[fake_network_cache_model.new_fixed_ip(
{'address': '10.10.0.2'}),
fake_network_cache_model.new_fixed_ip(
{'address': '10.10.0.3'})] * 4)
{'address': '10.10.0.3'})] * 4, ninfo.fixed_ips())
def test_create_async_model_exceptions(self):
def async_wrapper():
@ -504,18 +509,18 @@ class NetworkInfoTests(test.NoDBTestCase):
ninfo = model.NetworkInfo([vif,
fake_network_cache_model.new_vif(
{'address': 'bb:bb:bb:bb:bb:bb'})])
self.assertEqual(ninfo.floating_ips(), ['192.168.1.1'])
self.assertEqual(['192.168.1.1'], ninfo.floating_ips())
def test_hydrate(self):
ninfo = model.NetworkInfo([fake_network_cache_model.new_vif(),
fake_network_cache_model.new_vif(
{'address': 'bb:bb:bb:bb:bb:bb'})])
model.NetworkInfo.hydrate(ninfo)
self.assertEqual(ninfo.fixed_ips(),
self.assertEqual(
[fake_network_cache_model.new_fixed_ip(
{'address': '10.10.0.2'}),
fake_network_cache_model.new_fixed_ip(
{'address': '10.10.0.3'})] * 4)
{'address': '10.10.0.3'})] * 4, ninfo.fixed_ips())
def _setup_injected_network_scenario(self, should_inject=True,
use_ipv4=True, use_ipv6=False,

149
nova/tests/unit/network/test_neutronv2.py
View File

@ -816,8 +816,8 @@ class TestNeutronv2(TestNeutronv2Base):
self.assertEqual(index, len(nw_infs))
# ensure that nic ordering is preserved
for iface_index in range(index):
self.assertEqual(nw_infs[iface_index]['id'],
port_ids[iface_index])
self.assertEqual(port_ids[iface_index],
nw_infs[iface_index]['id'])
def test_get_instance_nw_info_without_subnet(self):
# Test get instance_nw_info for a port without subnet.
@ -895,7 +895,7 @@ class TestNeutronv2(TestNeutronv2Base):
port_req_body = {'port': {}}
api._populate_neutron_extension_values(self.context, instance,
None, port_req_body)
self.assertEqual(port_req_body['port']['rxtx_factor'], 1)
self.assertEqual(1, port_req_body['port']['rxtx_factor'])
def test_allocate_for_instance_1(self):
# Allocate one port in one network env.
@ -1058,7 +1058,7 @@ class TestNeutronv2(TestNeutronv2Base):
{'networks': model.NetworkInfo([])})
self.mox.ReplayAll()
nwinfo = api.allocate_for_instance(self.context, self.instance)
self.assertEqual(len(nwinfo), 0)
self.assertEqual(0, len(nwinfo))
@mock.patch('nova.network.neutronv2.api.API._get_preexisting_port_ids')
@mock.patch('nova.network.neutronv2.api.API._unbind_ports')
@ -1196,7 +1196,7 @@ class TestNeutronv2(TestNeutronv2Base):
new_port = {'id': 'fake'}
self._returned_nw_info = self.port_data1 + [new_port]
nw_info = self._allocate_for_instance()
self.assertEqual(nw_info, [new_port])
self.assertEqual([new_port], nw_info)
def test_allocate_for_instance_port_in_use(self):
# If a port is already in use, an exception should be raised.
@ -1432,9 +1432,9 @@ class TestNeutronv2(TestNeutronv2Base):
instance = self._fake_instance_object_with_info_cache(self.instance)
nwinfo = api.deallocate_port_for_instance(self.context, instance,
port_data[0]['id'])
self.assertEqual(len(nwinfo), len(port_data[1:]))
self.assertEqual(len(port_data[1:]), len(nwinfo))
if len(port_data) > 1:
self.assertEqual(nwinfo[0]['network']['id'], 'my_netid2')
self.assertEqual('my_netid2', nwinfo[0]['network']['id'])
def test_deallocate_port_for_instance_1(self):
# Test to deallocate the first and only port
@ -1734,7 +1734,7 @@ class TestNeutronv2(TestNeutronv2Base):
api = neutronapi.API()
max_count = api.validate_networks(self.context,
requested_networks, 1)
self.assertEqual(max_count, 0)
self.assertEqual(0, max_count)
def test_validate_networks_with_ports_and_networks(self):
# Test validation for a request for one instance needing
@ -1759,7 +1759,7 @@ class TestNeutronv2(TestNeutronv2Base):
api = neutronapi.API()
max_count = api.validate_networks(self.context,
requested_networks, 1)
self.assertEqual(max_count, 1)
self.assertEqual(1, max_count)
def test_validate_networks_one_port_and_no_networks(self):
# Test that show quota is not called if no networks are
@ -1774,7 +1774,7 @@ class TestNeutronv2(TestNeutronv2Base):
api = neutronapi.API()
max_count = api.validate_networks(self.context,
requested_networks, 1)
self.assertEqual(max_count, 1)
self.assertEqual(1, max_count)
def test_validate_networks_some_quota(self):
# Test validation for a request for two instance needing
@ -1796,7 +1796,7 @@ class TestNeutronv2(TestNeutronv2Base):
api = neutronapi.API()
max_count = api.validate_networks(self.context,
requested_networks, 2)
self.assertEqual(max_count, 1)
self.assertEqual(1, max_count)
def test_validate_networks_unlimited_quota(self):
# Test validation for a request for two instance needing
@ -1818,7 +1818,7 @@ class TestNeutronv2(TestNeutronv2Base):
api = neutronapi.API()
max_count = api.validate_networks(self.context,
requested_networks, 2)
self.assertEqual(max_count, 2)
self.assertEqual(2, max_count)
def test_validate_networks_no_quota_but_ports_supplied(self):
port_a = self.port_data3[0]
@ -1841,7 +1841,7 @@ class TestNeutronv2(TestNeutronv2Base):
api = neutronapi.API()
max_count = api.validate_networks(self.context,
requested_networks, 1)
self.assertEqual(max_count, 1)
self.assertEqual(1, max_count)
def _mock_list_ports(self, port_data=None):
if port_data is None:
@ -1895,7 +1895,7 @@ class TestNeutronv2(TestNeutronv2Base):
context if context else self.context,
self.instance['project_id'],
req_ids)
self.assertEqual(rets, nets)
self.assertEqual(nets, rets)
def test_get_available_networks_all_private(self):
self._get_available_networks(prv_nets=self.nets2, pub_nets=[])
@ -2085,7 +2085,7 @@ class TestNeutronv2(TestNeutronv2Base):
AndReturn({'floatingip': self.fip_unassociated})
self.mox.ReplayAll()
fip = api.allocate_floating_ip(self.context, 'ext_net')
self.assertEqual(fip, self.fip_unassociated['floating_ip_address'])
self.assertEqual(self.fip_unassociated['floating_ip_address'], fip)
def test_allocate_floating_ip_addr_gen_fail(self):
api = neutronapi.API()
@ -2132,7 +2132,7 @@ class TestNeutronv2(TestNeutronv2Base):
AndReturn({'floatingip': self.fip_unassociated})
self.mox.ReplayAll()
fip = api.allocate_floating_ip(self.context, pool_id)
self.assertEqual(fip, self.fip_unassociated['floating_ip_address'])
self.assertEqual(self.fip_unassociated['floating_ip_address'], fip)
def test_allocate_floating_ip_with_default_pool(self):
api = neutronapi.API()
@ -2148,7 +2148,7 @@ class TestNeutronv2(TestNeutronv2Base):
AndReturn({'floatingip': self.fip_unassociated})
self.mox.ReplayAll()
fip = api.allocate_floating_ip(self.context)
self.assertEqual(fip, self.fip_unassociated['floating_ip_address'])
self.assertEqual(self.fip_unassociated['floating_ip_address'], fip)
def test_release_floating_ip(self):
api = neutronapi.API()
@ -2343,7 +2343,7 @@ class TestNeutronv2(TestNeutronv2Base):
neutronapi.get_client('fake')
floatingips = api._get_floating_ips_by_fixed_and_port(
self.moxed_client, '1.1.1.1', 1)
self.assertEqual(floatingips, [])
self.assertEqual([], floatingips)
def test_nw_info_get_ips(self):
fake_port = {
@ -2359,9 +2359,9 @@ class TestNeutronv2(TestNeutronv2Base):
self.mox.ReplayAll()
neutronapi.get_client('fake')
result = api._nw_info_get_ips(self.moxed_client, fake_port)
self.assertEqual(len(result), 1)
self.assertEqual(result[0]['address'], '1.1.1.1')
self.assertEqual(result[0]['floating_ips'][0]['address'], '10.0.0.1')
self.assertEqual(1, len(result))
self.assertEqual('1.1.1.1', result[0]['address'])
self.assertEqual('10.0.0.1', result[0]['floating_ips'][0]['address'])
def test_nw_info_get_subnets(self):
fake_port = {
@ -2379,9 +2379,9 @@ class TestNeutronv2(TestNeutronv2Base):
self.mox.ReplayAll()
neutronapi.get_client('fake')
subnets = api._nw_info_get_subnets(self.context, fake_port, fake_ips)
self.assertEqual(len(subnets), 1)
self.assertEqual(len(subnets[0]['ips']), 1)
self.assertEqual(subnets[0]['ips'][0]['address'], '1.1.1.1')
self.assertEqual(1, len(subnets))
self.assertEqual(1, len(subnets[0]['ips']))
self.assertEqual('1.1.1.1', subnets[0]['ips'][0]['address'])
def _test_nw_info_build_network(self, vif_type):
fake_port = {
@ -2397,18 +2397,18 @@ class TestNeutronv2(TestNeutronv2Base):
neutronapi.get_client('fake')
net, iid = api._nw_info_build_network(fake_port, fake_nets,
fake_subnets)
self.assertEqual(net['subnets'], fake_subnets)
self.assertEqual(net['id'], 'net-id')
self.assertEqual(net['label'], 'foo')
self.assertEqual(net.get_meta('tenant_id'), 'tenant')
self.assertEqual(net.get_meta('injected'), CONF.flat_injected)
self.assertEqual(fake_subnets, net['subnets'])
self.assertEqual('net-id', net['id'])
self.assertEqual('foo', net['label'])
self.assertEqual('tenant', net.get_meta('tenant_id'))
self.assertEqual(CONF.flat_injected, net.get_meta('injected'))
return net, iid
def test_nw_info_build_network_ovs(self):
net, iid = self._test_nw_info_build_network(model.VIF_TYPE_OVS)
self.assertEqual(net['bridge'], CONF.neutron.ovs_bridge)
self.assertEqual(CONF.neutron.ovs_bridge, net['bridge'])
self.assertNotIn('should_create_bridge', net)
self.assertEqual(iid, 'port-id')
self.assertEqual('port-id', iid)
def test_nw_info_build_network_dvs(self):
net, iid = self._test_nw_info_build_network(model.VIF_TYPE_DVS)
@ -2419,7 +2419,7 @@ class TestNeutronv2(TestNeutronv2Base):
def test_nw_info_build_network_bridge(self):
net, iid = self._test_nw_info_build_network(model.VIF_TYPE_BRIDGE)
self.assertEqual(net['bridge'], 'brqnet-id')
self.assertEqual('brqnet-id', net['bridge'])
self.assertTrue(net['should_create_bridge'])
self.assertIsNone(iid)
@ -2472,14 +2472,14 @@ class TestNeutronv2(TestNeutronv2Base):
neutronapi.get_client('fake')
net, iid = api._nw_info_build_network(fake_port, fake_nets,
fake_subnets)
self.assertEqual(net['subnets'], fake_subnets)
self.assertEqual(net['id'], 'net-id')
self.assertEqual(net['label'], 'foo')
self.assertEqual(net.get_meta('tenant_id'), 'tenant')
self.assertEqual(net.get_meta('injected'), CONF.flat_injected)
self.assertEqual(net['bridge'], CONF.neutron.ovs_bridge)
self.assertEqual(fake_subnets, net['subnets'])
self.assertEqual('net-id', net['id'])
self.assertEqual('foo', net['label'])
self.assertEqual('tenant', net.get_meta('tenant_id'))
self.assertEqual(CONF.flat_injected, net.get_meta('injected'))
self.assertEqual(CONF.neutron.ovs_bridge, net['bridge'])
self.assertNotIn('should_create_bridge', net)
self.assertEqual(iid, 'port-id')
self.assertEqual('port-id', iid)
def test_build_network_info_model(self):
api = neutronapi.API()
@ -2615,39 +2615,38 @@ class TestNeutronv2(TestNeutronv2Base):
fake_ports[5]['id']],
preexisting_port_ids=['port3'])
self.assertEqual(len(nw_infos), 6)
self.assertEqual(6, len(nw_infos))
index = 0
for nw_info in nw_infos:
self.assertEqual(nw_info['address'],
requested_ports[index]['mac_address'])
self.assertEqual(nw_info['devname'], 'tapport' + str(index))
self.assertEqual(requested_ports[index]['mac_address'],
nw_info['address'])
self.assertEqual('tapport' + str(index), nw_info['devname'])
self.assertIsNone(nw_info['ovs_interfaceid'])
self.assertEqual(nw_info['type'],
requested_ports[index]['binding:vif_type'])
self.assertEqual(requested_ports[index]['binding:vif_type'],
nw_info['type'])
if nw_info['type'] == model.VIF_TYPE_BRIDGE:
self.assertEqual(nw_info['network']['bridge'], 'brqnet-id')
self.assertEqual(nw_info['vnic_type'],
requested_ports[index].get('binding:vnic_type',
model.VNIC_TYPE_NORMAL))
self.assertEqual(nw_info.get('details'),
requested_ports[index].get('binding:vif_details'))
self.assertEqual(nw_info.get('profile'),
requested_ports[index].get('binding:profile'))
self.assertEqual('brqnet-id', nw_info['network']['bridge'])
self.assertEqual(requested_ports[index].get('binding:vnic_type',
model.VNIC_TYPE_NORMAL), nw_info['vnic_type'])
self.assertEqual(requested_ports[index].get('binding:vif_details'),
nw_info.get('details'))
self.assertEqual(requested_ports[index].get('binding:profile'),
nw_info.get('profile'))
index += 1
self.assertEqual(nw_infos[0]['active'], False)
self.assertEqual(nw_infos[1]['active'], True)
self.assertEqual(nw_infos[2]['active'], True)
self.assertEqual(nw_infos[3]['active'], True)
self.assertEqual(nw_infos[4]['active'], True)
self.assertEqual(nw_infos[5]['active'], True)
self.assertFalse(nw_infos[0]['active'])
self.assertTrue(nw_infos[1]['active'])
self.assertTrue(nw_infos[2]['active'])
self.assertTrue(nw_infos[3]['active'])
self.assertTrue(nw_infos[4]['active'])
self.assertTrue(nw_infos[5]['active'])
self.assertEqual(nw_infos[0]['id'], 'port0')
self.assertEqual(nw_infos[1]['id'], 'port1')
self.assertEqual(nw_infos[2]['id'], 'port2')
self.assertEqual(nw_infos[3]['id'], 'port3')
self.assertEqual(nw_infos[4]['id'], 'port4')
self.assertEqual(nw_infos[5]['id'], 'port5')
self.assertEqual('port0', nw_infos[0]['id'])
self.assertEqual('port1', nw_infos[1]['id'])
self.assertEqual('port2', nw_infos[2]['id'])
self.assertEqual('port3', nw_infos[3]['id'])
self.assertEqual('port4', nw_infos[4]['id'])
self.assertEqual('port5', nw_infos[5]['id'])
self.assertFalse(nw_infos[0]['preserve_on_delete'])
self.assertFalse(nw_infos[1]['preserve_on_delete'])
@ -2706,7 +2705,7 @@ class TestNeutronv2(TestNeutronv2Base):
nw_infos = api._build_network_info_model(
self.context, fake_inst)
self.assertEqual(len(nw_infos), 1)
self.assertEqual(1, len(nw_infos))
def test_get_subnets_from_port(self):
api = neutronapi.API()
@ -2727,12 +2726,12 @@ class TestNeutronv2(TestNeutronv2Base):
subnets = api._get_subnets_from_port(self.context, port_data)
self.assertEqual(len(subnets), 1)
self.assertEqual(len(subnets[0]['routes']), 1)
self.assertEqual(subnets[0]['routes'][0]['cidr'],
subnet_data1[0]['host_routes'][0]['destination'])
self.assertEqual(subnets[0]['routes'][0]['gateway']['address'],
subnet_data1[0]['host_routes'][0]['nexthop'])
self.assertEqual(1, len(subnets))
self.assertEqual(1, len(subnets[0]['routes']))
self.assertEqual(subnet_data1[0]['host_routes'][0]['destination'],
subnets[0]['routes'][0]['cidr'])
self.assertEqual(subnet_data1[0]['host_routes'][0]['nexthop'],
subnets[0]['routes'][0]['gateway']['address'])
def test_get_all_empty_list_networks(self):
api = neutronapi.API()
@ -2766,7 +2765,7 @@ class TestNeutronv2(TestNeutronv2Base):
test_port['port']['network_id'],
fields='provider:physical_network')
self.assertEqual(model.VNIC_TYPE_DIRECT, vnic_type)
self.assertEqual(phynet_name, 'phynet1')
self.assertEqual('phynet1', phynet_name)
def _test_get_port_vnic_info(self, mock_get_client,
binding_vnic_type=None):
@ -3625,7 +3624,7 @@ class TestNeutronv2Portbinding(TestNeutronv2Base):
port_req_body = {'port': {}}
api._populate_neutron_extension_values(self.context, instance,
None, port_req_body)
self.assertEqual(port_req_body['port']['binding:host_id'], host_id)
self.assertEqual(host_id, port_req_body['port']['binding:host_id'])
self.assertFalse(port_req_body['port'].get('binding:profile'))
@mock.patch.object(pci_whitelist, 'get_pci_device_devspec')
@ -3657,7 +3656,7 @@ class TestNeutronv2Portbinding(TestNeutronv2Base):
api._populate_neutron_binding_profile(instance,
pci_req_id, port_req_body)
self.assertEqual(port_req_body['port']['binding:profile'], profile)
self.assertEqual(profile, port_req_body['port']['binding:profile'])
def _test_update_port_binding_false(self, func_name, *args):
api = neutronapi.API()

4
nova/tests/unit/network/test_rpcapi.py
View File

@ -46,7 +46,7 @@ class NetworkRpcAPITestCase(test.NoDBTestCase):
rpcapi = network_rpcapi.NetworkAPI()
self.assertIsNotNone(rpcapi.client)
self.assertEqual(rpcapi.client.target.topic, CONF.network_topic)
self.assertEqual(CONF.network_topic, rpcapi.client.target.topic)
expected_retval = 'foo' if rpc_method == 'call' else None
expected_version = kwargs.pop('version', None)
@ -108,7 +108,7 @@ class NetworkRpcAPITestCase(test.NoDBTestCase):
self.mox.ReplayAll()
retval = getattr(rpcapi, method)(ctxt, **kwargs)
self.assertEqual(retval, expected_retval)
self.assertEqual(expected_retval, retval)
def test_create_networks(self):
self._test_network_api('create_networks', rpc_method='call',

2
nova/tests/unit/objects/test_bandwidth_usage.py
View File

@ -88,7 +88,7 @@ class _TestBandwidthUsage(test.TestCase):
bw_usages = bandwidth_usage.BandwidthUsageList.get_by_uuids(
self.context, ['fake_uuid'],
start_period=self.expected_bw_usage['start_period'])
self.assertEqual(len(bw_usages), 1)
self.assertEqual(1, len(bw_usages))
self._compare(self, self.expected_bw_usage, bw_usages[0])
@mock.patch.object(db, 'bw_usage_update')

12
nova/tests/unit/objects/test_block_device.py
View File

@ -83,8 +83,8 @@ class _TestBlockDeviceMappingObject(object):
self.assertTrue(len(cells_update_mock.call_args[0]) > 1)
self.assertIsInstance(cells_update_mock.call_args[0][1],
block_device_obj.BlockDeviceMapping)
self.assertEqual(cells_update_mock.call_args[1], {'create':
create})
self.assertEqual({'create': create},
cells_update_mock.call_args[1])
def test_save_nocells(self):
self._test_save()
@ -191,12 +191,12 @@ class _TestBlockDeviceMappingObject(object):
if cell_type == 'compute' and 'device_name' in values:
self.assertEqual(1, cells_update_mock.call_count)
self.assertTrue(len(cells_update_mock.call_args[0]) > 1)
self.assertEqual(cells_update_mock.call_args[0][0],
self.context)
self.assertEqual(self.context,
cells_update_mock.call_args[0][0])
self.assertIsInstance(cells_update_mock.call_args[0][1],
block_device_obj.BlockDeviceMapping)
self.assertEqual(cells_update_mock.call_args[1],
{'create': update_or_create or None})
self.assertEqual({'create': update_or_create or None},
cells_update_mock.call_args[1])
else:
self.assertFalse(cells_update_mock.called)

107
nova/tests/unit/objects/test_instance.py
View File

@ -88,7 +88,7 @@ class _TestInstanceObject(object):
self.assertJsonEqual(primitive, expected)
inst2 = objects.Instance.obj_from_primitive(primitive)
self.assertIsInstance(inst2.launched_at, datetime.datetime)
self.assertEqual(inst2.launched_at, red_letter_date)
self.assertEqual(red_letter_date, inst2.launched_at)
def test_ip_deserialization(self):
inst = objects.Instance(uuid='fake-uuid', access_ip_v4='1.2.3.4',
@ -107,8 +107,8 @@ class _TestInstanceObject(object):
inst2 = objects.Instance.obj_from_primitive(primitive)
self.assertIsInstance(inst2.access_ip_v4, netaddr.IPAddress)
self.assertIsInstance(inst2.access_ip_v6, netaddr.IPAddress)
self.assertEqual(inst2.access_ip_v4, netaddr.IPAddress('1.2.3.4'))
self.assertEqual(inst2.access_ip_v6, netaddr.IPAddress('::1'))
self.assertEqual(netaddr.IPAddress('1.2.3.4'), inst2.access_ip_v4)
self.assertEqual(netaddr.IPAddress('::1'), inst2.access_ip_v6)
def test_get_without_expected(self):
self.mox.StubOutWithMock(db, 'instance_get_by_uuid')
@ -184,7 +184,7 @@ class _TestInstanceObject(object):
).AndReturn(self.fake_instance)
self.mox.ReplayAll()
inst = objects.Instance.get_by_id(self.context, 'instid')
self.assertEqual(inst.uuid, self.fake_instance['uuid'])
self.assertEqual(self.fake_instance['uuid'], inst.uuid)
def test_load(self):
self.mox.StubOutWithMock(db, 'instance_get_by_uuid')
@ -204,11 +204,11 @@ class _TestInstanceObject(object):
inst = objects.Instance.get_by_uuid(self.context, fake_uuid)
self.assertFalse(hasattr(inst, '_obj_metadata'))
meta = inst.metadata
self.assertEqual(meta, {'foo': 'bar'})
self.assertEqual({'foo': 'bar'}, meta)
self.assertTrue(hasattr(inst, '_obj_metadata'))
# Make sure we don't run load again
meta2 = inst.metadata
self.assertEqual(meta2, {'foo': 'bar'})
self.assertEqual({'foo': 'bar'}, meta2)
def test_load_invalid(self):
inst = objects.Instance(context=self.context, uuid='fake-uuid')
@ -226,13 +226,13 @@ class _TestInstanceObject(object):
).AndReturn(fake_instance)
self.mox.ReplayAll()
inst = objects.Instance.get_by_uuid(self.context, 'fake-uuid')
self.assertEqual(inst.id, fake_instance['id'])
self.assertEqual(inst.launched_at.replace(tzinfo=None),
fake_instance['launched_at'])
self.assertEqual(str(inst.access_ip_v4),
fake_instance['access_ip_v4'])
self.assertEqual(str(inst.access_ip_v6),
fake_instance['access_ip_v6'])
self.assertEqual(fake_instance['id'], inst.id)
self.assertEqual(fake_instance['launched_at'],
inst.launched_at.replace(tzinfo=None))
self.assertEqual(fake_instance['access_ip_v4'],
str(inst.access_ip_v4))
self.assertEqual(fake_instance['access_ip_v6'],
str(inst.access_ip_v6))
def test_refresh(self):
self.mox.StubOutWithMock(db, 'instance_get_by_uuid')
@ -254,9 +254,9 @@ class _TestInstanceObject(object):
instance_info_cache.InstanceInfoCache.refresh()
self.mox.ReplayAll()
inst = objects.Instance.get_by_uuid(self.context, fake_uuid)
self.assertEqual(inst.host, 'orig-host')
self.assertEqual('orig-host', inst.host)
inst.refresh()
self.assertEqual(inst.host, 'new-host')
self.assertEqual('new-host', inst.host)
self.assertEqual(set([]), inst.obj_what_changed())
def test_refresh_does_not_recurse(self):
@ -484,11 +484,11 @@ class _TestInstanceObject(object):
inst.vcpu_model = test_vcpu_model.fake_vcpumodel
inst.save()
self.assertTrue(mock_update.called)
self.assertEqual(mock_update.call_count, 1)
self.assertEqual(1, mock_update.call_count)
actual_args = mock_update.call_args
self.assertEqual(self.context, actual_args[0][0])
self.assertEqual(inst.uuid, actual_args[0][1])
self.assertEqual(list(actual_args[0][2].keys()), ['vcpu_model'])
self.assertEqual(['vcpu_model'], list(actual_args[0][2].keys()))
self.assertJsonEqual(jsonutils.dumps(
test_vcpu_model.fake_vcpumodel.obj_to_primitive()),
actual_args[0][2]['vcpu_model'])
@ -505,11 +505,11 @@ class _TestInstanceObject(object):
self.context)
inst.save()
self.assertTrue(mock_update.called)
self.assertEqual(mock_update.call_count, 1)
self.assertEqual(1, mock_update.call_count)
actual_args = mock_update.call_args
self.assertEqual(self.context, actual_args[0][0])
self.assertEqual(inst.uuid, actual_args[0][1])
self.assertEqual(list(actual_args[0][2].keys()), ['migration_context'])
self.assertEqual(['migration_context'], list(actual_args[0][2].keys()))
self.assertIsInstance(
objects.MigrationContext.obj_from_db_obj(
actual_args[0][2]['migration_context']),
@ -609,7 +609,7 @@ class _TestInstanceObject(object):
self.mox.ReplayAll()
inst = objects.Instance.get_by_uuid(self.context, fake_uuid)
# NOTE(danms): Make sure it's actually a bool
self.assertEqual(inst.deleted, True)
self.assertTrue(inst.deleted)
def test_get_not_cleaned(self):
fake_inst = dict(self.fake_instance, id=123, cleaned=None)
@ -623,7 +623,7 @@ class _TestInstanceObject(object):
self.mox.ReplayAll()
inst = objects.Instance.get_by_uuid(self.context, fake_uuid)
# NOTE(mikal): Make sure it's actually a bool
self.assertEqual(inst.cleaned, False)
self.assertFalse(inst.cleaned)
def test_get_cleaned(self):
fake_inst = dict(self.fake_instance, id=123, cleaned=1)
@ -637,7 +637,7 @@ class _TestInstanceObject(object):
self.mox.ReplayAll()
inst = objects.Instance.get_by_uuid(self.context, fake_uuid)
# NOTE(mikal): Make sure it's actually a bool
self.assertEqual(inst.cleaned, True)
self.assertTrue(inst.cleaned)
def test_with_info_cache(self):
fake_inst = dict(self.fake_instance)
@ -663,8 +663,8 @@ class _TestInstanceObject(object):
{'network_info': nwinfo2_json}).AndReturn(fake_info_cache)
self.mox.ReplayAll()
inst = objects.Instance.get_by_uuid(self.context, fake_uuid)
self.assertEqual(inst.info_cache.network_info, nwinfo1)
self.assertEqual(inst.info_cache.instance_uuid, fake_uuid)
self.assertEqual(nwinfo1, inst.info_cache.network_info)
self.assertEqual(fake_uuid, inst.info_cache.instance_uuid)
inst.info_cache.network_info = nwinfo2
inst.save()
@ -706,17 +706,17 @@ class _TestInstanceObject(object):
).AndReturn(fake_inst['security_groups'][0])
self.mox.ReplayAll()
inst = objects.Instance.get_by_uuid(self.context, fake_uuid)
self.assertEqual(len(inst.security_groups), 2)
self.assertEqual(2, len(inst.security_groups))
for index, group in enumerate(fake_inst['security_groups']):
for key in group:
self.assertEqual(group[key],
inst.security_groups[index][key])
self.assertIsInstance(inst.security_groups[index],
security_group.SecurityGroup)
self.assertEqual(inst.security_groups.obj_what_changed(), set())
self.assertEqual(set(), inst.security_groups.obj_what_changed())
inst.security_groups[0].description = 'changed'
inst.save()
self.assertEqual(inst.security_groups.obj_what_changed(), set())
self.assertEqual(set(), inst.security_groups.obj_what_changed())
def test_with_empty_security_groups(self):
fake_inst = dict(self.fake_instance, security_groups=[])
@ -742,7 +742,7 @@ class _TestInstanceObject(object):
self.mox.ReplayAll()
inst = objects.Instance.get_by_uuid(self.context, fake_uuid,
['pci_devices'])
self.assertEqual(len(inst.pci_devices), 0)
self.assertEqual(0, len(inst.pci_devices))
def test_with_pci_devices(self):
fake_inst = dict(self.fake_instance)
@ -792,9 +792,9 @@ class _TestInstanceObject(object):
self.mox.ReplayAll()
inst = objects.Instance.get_by_uuid(self.context, fake_uuid,
['pci_devices'])
self.assertEqual(len(inst.pci_devices), 2)
self.assertEqual(inst.pci_devices[0].instance_uuid, fake_uuid)
self.assertEqual(inst.pci_devices[1].instance_uuid, fake_uuid)
self.assertEqual(2, len(inst.pci_devices))
self.assertEqual(fake_uuid, inst.pci_devices[0].instance_uuid)
self.assertEqual(fake_uuid, inst.pci_devices[1].instance_uuid)
def test_with_fault(self):
fake_inst = dict(self.fake_instance)
@ -832,10 +832,9 @@ class _TestInstanceObject(object):
def test_iteritems_with_extra_attrs(self):
self.stubs.Set(objects.Instance, 'name', 'foo')
inst = objects.Instance(uuid='fake-uuid')
self.assertEqual(sorted(inst.items()),
sorted({'uuid': 'fake-uuid',
self.assertEqual(sorted({'uuid': 'fake-uuid',
'name': 'foo',
}.items()))
}.items()), sorted(inst.items()))
def _test_metadata_change_tracking(self, which):
inst = objects.Instance(uuid='fake-uuid')
@ -860,7 +859,7 @@ class _TestInstanceObject(object):
scheduled_at=None,
system_metadata={'foo': 'bar'})
inst.create()
self.assertEqual(inst.host, 'foo-host')
self.assertEqual('foo-host', inst.host)
def test_metadata_change_tracking(self):
self._test_metadata_change_tracking('metadata')
@ -897,9 +896,9 @@ class _TestInstanceObject(object):
project_id=self.context.project_id,
host='foo-host')
inst1.create()
self.assertEqual(inst1.host, 'foo-host')
self.assertEqual('foo-host', inst1.host)
inst2 = objects.Instance.get_by_uuid(self.context, inst1.uuid)
self.assertEqual(inst2.host, 'foo-host')
self.assertEqual('foo-host', inst2.host)
def test_create_with_extras(self):
inst = objects.Instance(context=self.context,
@ -970,8 +969,8 @@ class _TestInstanceObject(object):
inst = objects.Instance(context=self.context, id=1, uuid='fake-uuid',
host='foo')
inst.destroy()
self.assertEqual(timeutils.normalize_time(inst.deleted_at),
timeutils.normalize_time(deleted_at))
self.assertEqual(timeutils.normalize_time(deleted_at),
timeutils.normalize_time(inst.deleted_at))
self.assertTrue(inst.deleted)
def test_destroy(self):
@ -1415,7 +1414,7 @@ class _TestInstanceListObject(object):
for i in range(0, len(fakes)):
self.assertIsInstance(inst_list.objects[i], objects.Instance)
self.assertEqual(inst_list.objects[i].uuid, fakes[i]['uuid'])
self.assertEqual(fakes[i]['uuid'], inst_list.objects[i].uuid)
def test_get_all_by_filters_sorted(self):
fakes = [self.fake_instance(1), self.fake_instance(2)]
@ -1433,7 +1432,7 @@ class _TestInstanceListObject(object):
for i in range(0, len(fakes)):
self.assertIsInstance(inst_list.objects[i], objects.Instance)
self.assertEqual(inst_list.objects[i].uuid, fakes[i]['uuid'])
self.assertEqual(fakes[i]['uuid'], inst_list.objects[i].uuid)
@mock.patch.object(db, 'instance_get_all_by_filters_sort')
@mock.patch.object(db, 'instance_get_all_by_filters')
@ -1486,7 +1485,7 @@ class _TestInstanceListObject(object):
self.assertEqual(1, len(inst_list))
self.assertIsInstance(inst_list.objects[0], objects.Instance)
self.assertEqual(inst_list.objects[0].uuid, fakes[1]['uuid'])
self.assertEqual(fakes[1]['uuid'], inst_list.objects[0].uuid)
def test_get_by_host(self):
fakes = [self.fake_instance(1),
@ -1499,9 +1498,9 @@ class _TestInstanceListObject(object):
inst_list = objects.InstanceList.get_by_host(self.context, 'foo')
for i in range(0, len(fakes)):
self.assertIsInstance(inst_list.objects[i], objects.Instance)
self.assertEqual(inst_list.objects[i].uuid, fakes[i]['uuid'])
self.assertEqual(inst_list.objects[i]._context, self.context)
self.assertEqual(inst_list.obj_what_changed(), set())
self.assertEqual(fakes[i]['uuid'], inst_list.objects[i].uuid)
self.assertEqual(self.context, inst_list.objects[i]._context)
self.assertEqual(set(), inst_list.obj_what_changed())
def test_get_by_host_and_node(self):
fakes = [self.fake_instance(1),
@ -1515,7 +1514,7 @@ class _TestInstanceListObject(object):
'foo', 'bar')
for i in range(0, len(fakes)):
self.assertIsInstance(inst_list.objects[i], objects.Instance)
self.assertEqual(inst_list.objects[i].uuid, fakes[i]['uuid'])
self.assertEqual(fakes[i]['uuid'], inst_list.objects[i].uuid)
def test_get_by_host_and_not_type(self):
fakes = [self.fake_instance(1),
@ -1529,7 +1528,7 @@ class _TestInstanceListObject(object):
self.context, 'foo', 'bar')
for i in range(0, len(fakes)):
self.assertIsInstance(inst_list.objects[i], objects.Instance)
self.assertEqual(inst_list.objects[i].uuid, fakes[i]['uuid'])
self.assertEqual(fakes[i]['uuid'], inst_list.objects[i].uuid)
@mock.patch('nova.objects.instance._expected_cols')
@mock.patch('nova.db.instance_get_all')
@ -1543,7 +1542,7 @@ class _TestInstanceListObject(object):
self.context, columns_to_join=mock.sentinel.exp_att)
for i in range(0, len(fakes)):
self.assertIsInstance(inst_list.objects[i], objects.Instance)
self.assertEqual(inst_list.objects[i].uuid, fakes[i]['uuid'])
self.assertEqual(fakes[i]['uuid'], inst_list.objects[i].uuid)
def test_get_hung_in_rebooting(self):
fakes = [self.fake_instance(1),
@ -1557,7 +1556,7 @@ class _TestInstanceListObject(object):
dt)
for i in range(0, len(fakes)):
self.assertIsInstance(inst_list.objects[i], objects.Instance)
self.assertEqual(inst_list.objects[i].uuid, fakes[i]['uuid'])
self.assertEqual(fakes[i]['uuid'], inst_list.objects[i].uuid)
def test_get_active_by_window_joined(self):
fakes = [self.fake_instance(1), self.fake_instance(2)]
@ -1582,7 +1581,7 @@ class _TestInstanceListObject(object):
for fake, obj in zip(fakes, inst_list.objects):
self.assertIsInstance(obj, objects.Instance)
self.assertEqual(obj.uuid, fake['uuid'])
self.assertEqual(fake['uuid'], obj.uuid)
def test_with_fault(self):
fake_insts = [
@ -1638,12 +1637,12 @@ class _TestInstanceListObject(object):
inst_list._context = self.context
inst_list.objects = insts
faulty = inst_list.fill_faults()
self.assertEqual(list(faulty), ['uuid1'])
self.assertEqual(inst_list[0].fault.message,
db_faults['uuid1'][0]['message'])
self.assertEqual(['uuid1'], list(faulty))
self.assertEqual(db_faults['uuid1'][0]['message'],
inst_list[0].fault.message)
self.assertIsNone(inst_list[1].fault)
for inst in inst_list:
self.assertEqual(inst.obj_what_changed(), set())
self.assertEqual(set(), inst.obj_what_changed())
def test_get_by_security_group(self):
fake_secgroup = dict(test_security_group.fake_secgroup)

4
nova/tests/unit/objects/test_instance_info_cache.py
View File

@ -45,8 +45,8 @@ class _TestInstanceInfoCacheObject(object):
self.mox.ReplayAll()
obj = instance_info_cache.InstanceInfoCache.get_by_instance_uuid(
self.context, 'fake-uuid')
self.assertEqual(obj.instance_uuid, 'fake-uuid')
self.assertEqual(obj.network_info, nwinfo)
self.assertEqual('fake-uuid', obj.instance_uuid)
self.assertEqual(nwinfo, obj.network_info)
def test_get_by_instance_uuid_no_entries(self):
self.mox.StubOutWithMock(db, 'instance_info_cache_get')