openstack
/
neutron
Code Issues Proposed changes

Replaces assertEquals to assertEqual 

assertEquals() has been deprecated in 2.7
Replaced assertEquals to assertEqual in all test scripts

Change-Id: I04b4ac873e151837ade9127f8c1eba32f4707dd4
This commit is contained in:
Zhongyue Luo 2012-12-14 09:32:24 +08:00
parent bc1a0d4912
commit e36ff0f569
14 changed files with 356 additions and 356 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
quantum/tests/unit/bigswitch/test_restproxy_plugin.py
View File

@ -109,4 +109,4 @@ class TestBigSwitchProxySync(BigSwitchProxyPluginV2TestCase):
def test_send_data(self):
plugin_obj = QuantumManager.get_plugin()
result = plugin_obj._send_all_data()
self.assertEquals(result[0], 200)
self.assertEqual(result[0], 200)

2
quantum/tests/unit/test_agent_linux_utils.py
View File

@ -69,4 +69,4 @@ class AgentUtilsGetInterfaceMAC(unittest.TestCase):
'\x01\x02\x03\x04\x05\x06',
'\x00' * 232])
actual_val = utils.get_interface_mac('eth0')
self.assertEquals(actual_val, expect_val)
self.assertEqual(actual_val, expect_val)

2
quantum/tests/unit/test_agent_netns_cleanup.py
View File

@ -141,7 +141,7 @@ class TestNetnsCleanup(unittest.TestCase):
mock_get_bridge_for_iface.assert_called_once_with(
conf.root_helper, 'tap1')
self.assertEquals(ovs_br_cls.mock_calls, [])
self.assertEqual(ovs_br_cls.mock_calls, [])
self.assertTrue(debug.called)
def _test_destroy_namespace_helper(self, force, num_devices):

42
quantum/tests/unit/test_attributes.py
View File

@ -41,14 +41,14 @@ class TestAttributes(unittest2.TestCase):
self.assertIsNone(msg)
msg = attributes._validate_values(7, [4, 6])
self.assertEquals(msg, "'7' is not in [4, 6]")
self.assertEqual(msg, "'7' is not in [4, 6]")
msg = attributes._validate_values(7, (4, 6))
self.assertEquals(msg, "'7' is not in (4, 6)")
self.assertEqual(msg, "'7' is not in (4, 6)")
def test_validate_string(self):
msg = attributes._validate_string(None, None)
self.assertEquals(msg, "'None' is not a valid string")
self.assertEqual(msg, "'None' is not a valid string")
# 0 == len(data) == max_len
msg = attributes._validate_string("", 0)
@ -68,7 +68,7 @@ class TestAttributes(unittest2.TestCase):
# 0 < max_len < len(data)
msg = attributes._validate_string("1234567890", 9)
self.assertEquals(msg, "'1234567890' exceeds maximum length of 9")
self.assertEqual(msg, "'1234567890' exceeds maximum length of 9")
msg = attributes._validate_string("123456789", None)
self.assertIsNone(msg)
@ -93,10 +93,10 @@ class TestAttributes(unittest2.TestCase):
self.assertIsNone(msg)
msg = attributes._validate_range(0, [1, 9])
self.assertEquals(msg, "'0' is not in range 1 through 9")
self.assertEqual(msg, "'0' is not in range 1 through 9")
msg = attributes._validate_range(10, (1, 9))
self.assertEquals(msg, "'10' is not in range 1 through 9")
self.assertEqual(msg, "'10' is not in range 1 through 9")
def test_validate_mac_address(self):
mac_addr = "ff:16:3e:4f:00:00"
@ -105,7 +105,7 @@ class TestAttributes(unittest2.TestCase):
mac_addr = "ffa:16:3e:4f:00:00"
msg = attributes._validate_mac_address(mac_addr)
self.assertEquals(msg, "'%s' is not a valid MAC address" % mac_addr)
self.assertEqual(msg, "'%s' is not a valid MAC address" % mac_addr)
def test_validate_ip_address(self):
ip_addr = '1.1.1.1'
@ -114,7 +114,7 @@ class TestAttributes(unittest2.TestCase):
ip_addr = '1111.1.1.1'
msg = attributes._validate_ip_address(ip_addr)
self.assertEquals(msg, "'%s' is not a valid IP address" % ip_addr)
self.assertEqual(msg, "'%s' is not a valid IP address" % ip_addr)
def test_validate_ip_pools(self):
pools = [[{'end': '10.0.0.254'}],
@ -223,7 +223,7 @@ class TestAttributes(unittest2.TestCase):
ip_addr = '1111.1.1.1'
msg = attributes._validate_ip_address_or_none(ip_addr)
self.assertEquals(msg, "'%s' is not a valid IP address" % ip_addr)
self.assertEqual(msg, "'%s' is not a valid IP address" % ip_addr)
def test_hostname_pattern(self):
data = '@openstack'
@ -322,39 +322,39 @@ class TestAttributes(unittest2.TestCase):
msg = attributes._validate_subnet(cidr,
None)
error = "'%s' is not a valid IP subnet" % cidr
self.assertEquals(msg, error)
self.assertEqual(msg, error)
# Invalid - IPv6 without final octets, missing mask
cidr = "fe80::"
msg = attributes._validate_subnet(cidr,
None)
error = "'%s' is not a valid IP subnet" % cidr
self.assertEquals(msg, error)
self.assertEqual(msg, error)
# Invalid - IPv6 with final octets, missing mask
cidr = "fe80::0"
msg = attributes._validate_subnet(cidr,
None)
error = "'%s' is not a valid IP subnet" % cidr
self.assertEquals(msg, error)
self.assertEqual(msg, error)
# Invalid - Address format error
cidr = 'invalid'
msg = attributes._validate_subnet(cidr,
None)
error = "'%s' is not a valid IP subnet" % cidr
self.assertEquals(msg, error)
self.assertEqual(msg, error)
def test_validate_regex(self):
pattern = '[hc]at'
data = None
msg = attributes._validate_regex(data, pattern)
self.assertEquals(msg, "'%s' is not a valid input" % data)
self.assertEqual(msg, "'%s' is not a valid input" % data)
data = 'bat'
msg = attributes._validate_regex(data, pattern)
self.assertEquals(msg, "'%s' is not a valid input" % data)
self.assertEqual(msg, "'%s' is not a valid input" % data)
data = 'hat'
msg = attributes._validate_regex(data, pattern)
@ -366,7 +366,7 @@ class TestAttributes(unittest2.TestCase):
def test_validate_uuid(self):
msg = attributes._validate_uuid('garbage')
self.assertEquals(msg, "'garbage' is not a valid UUID")
self.assertEqual(msg, "'garbage' is not a valid UUID")
msg = attributes._validate_uuid('00000000-ffff-ffff-ffff-000000000000')
self.assertIsNone(msg)
@ -400,13 +400,13 @@ class TestConvertToBoolean(unittest2.TestCase):
class TestConvertToInt(unittest2.TestCase):
def test_convert_to_int_int(self):
self.assertEquals(attributes.convert_to_int(-1), -1)
self.assertEquals(attributes.convert_to_int(0), 0)
self.assertEquals(attributes.convert_to_int(1), 1)
self.assertEqual(attributes.convert_to_int(-1), -1)
self.assertEqual(attributes.convert_to_int(0), 0)
self.assertEqual(attributes.convert_to_int(1), 1)
def test_convert_to_int_str(self):
self.assertEquals(attributes.convert_to_int('4'), 4)
self.assertEquals(attributes.convert_to_int('6'), 6)
self.assertEqual(attributes.convert_to_int('4'), 4)
self.assertEqual(attributes.convert_to_int('6'), 6)
self.assertRaises(q_exc.InvalidInput,
attributes.convert_to_int,
'garbage')

394
quantum/tests/unit/test_db_plugin.py
View File

File diff suppressed because it is too large Load Diff

2
quantum/tests/unit/test_db_rpc_base.py
View File

@ -59,7 +59,7 @@ class TestDhcpRpcCallackMixin(unittest.TestCase):
self.plugin.get_ports.return_value = port_retval
retval = self.callbacks.get_network_info(mock.Mock(), network_id='a')
self.assertEquals(retval, network_retval)
self.assertEqual(retval, network_retval)
self.assertEqual(retval['subnets'], subnet_retval)
self.assertEqual(retval['ports'], port_retval)

90
quantum/tests/unit/test_extension_security_group.py
View File

@ -228,7 +228,7 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
keys = [('name', name,), ('description', description)]
with self.security_group(name, description) as security_group:
for k, v, in keys:
self.assertEquals(security_group['security_group'][k], v)
self.assertEqual(security_group['security_group'][k], v)
def test_create_security_group_external_id(self):
cfg.CONF.SECURITYGROUP.proxy_mode = True
@ -239,13 +239,13 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
('external_id', external_id)]
with self.security_group(name, description, external_id) as sg:
for k, v, in keys:
self.assertEquals(sg['security_group'][k], v)
self.assertEqual(sg['security_group'][k], v)
def test_default_security_group(self):
with self.network():
res = self.new_list_request('security-groups')
groups = self.deserialize('json', res.get_response(self.ext_api))
self.assertEquals(len(groups['security_groups']), 1)
self.assertEqual(len(groups['security_groups']), 1)
def test_create_security_group_proxy_mode_not_admin(self):
cfg.CONF.SECURITYGROUP.proxy_mode = True
@ -254,27 +254,27 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
tenant_id='bad_tenant',
set_context=True)
self.deserialize('json', res)
self.assertEquals(res.status_int, 500)
self.assertEqual(res.status_int, 500)
def test_create_security_group_no_external_id_proxy_mode(self):
cfg.CONF.SECURITYGROUP.proxy_mode = True
res = self._create_security_group('json', 'webservers',
'webservers')
self.deserialize('json', res)
self.assertEquals(res.status_int, 400)
self.assertEqual(res.status_int, 400)
def test_create_security_group_no_external_id_not_proxy_mode(self):
res = self._create_security_group('json', 'webservers',
'webservers', '1')
self.deserialize('json', res)
self.assertEquals(res.status_int, 409)
self.assertEqual(res.status_int, 409)
def test_create_default_security_group_fail(self):
name = 'default'
description = 'my webservers'
res = self._create_security_group('json', name, description)
self.deserialize('json', res)
self.assertEquals(res.status_int, 409)
self.assertEqual(res.status_int, 409)
def test_create_security_group_duplicate_external_id(self):
cfg.CONF.SECURITYGROUP.proxy_mode = True
@ -285,7 +285,7 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
res = self._create_security_group('json', name, description,
external_id)
self.deserialize('json', res)
self.assertEquals(res.status_int, 409)
self.assertEqual(res.status_int, 409)
def test_list_security_groups(self):
name = 'webservers'
@ -293,7 +293,7 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
with self.security_group(name, description):
res = self.new_list_request('security-groups')
groups = self.deserialize('json', res.get_response(self.ext_api))
self.assertEquals(len(groups['security_groups']), 2)
self.assertEqual(len(groups['security_groups']), 2)
def test_get_security_group(self):
name = 'webservers'
@ -302,7 +302,7 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
source_group_id = sg['security_group']['id']
res = self.new_show_request('security-groups', source_group_id)
group = self.deserialize('json', res.get_response(self.ext_api))
self.assertEquals(group['security_group']['id'], source_group_id)
self.assertEqual(group['security_group']['id'], source_group_id)
def test_delete_security_group(self):
name = 'webservers'
@ -322,16 +322,16 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
with self.network():
res = self.new_list_request('security-groups')
groups = self.deserialize('json', res.get_response(self.ext_api))
self.assertEquals(len(groups['security_groups']), 1)
self.assertEqual(len(groups['security_groups']), 1)
res = self.new_list_request('security-group-rules')
rules = self.deserialize('json', res.get_response(self.ext_api))
self.assertEquals(len(rules['security_group_rules']), 2)
self.assertEqual(len(rules['security_group_rules']), 2)
# just generic rules to allow default egress and
# intergroup communicartion
for rule in rules['security_group_rules']:
self.assertEquals(rule['port_range_max'], None)
self.assertEquals(rule['port_range_min'], None)
self.assertEquals(rule['protocol'], None)
self.assertEqual(rule['port_range_max'], None)
self.assertEqual(rule['port_range_min'], None)
self.assertEqual(rule['protocol'], None)
def test_create_security_group_rule_source_ip_prefix(self):
name = 'webservers'
@ -354,7 +354,7 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
port_range_max,
source_ip_prefix) as rule:
for k, v, in keys:
self.assertEquals(rule['security_group_rule'][k], v)
self.assertEqual(rule['security_group_rule'][k], v)
def test_create_security_group_rule_group_id(self):
name = 'webservers'
@ -379,7 +379,7 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
source_group_id=source_group_id
) as rule:
for k, v, in keys:
self.assertEquals(rule['security_group_rule'][k], v)
self.assertEqual(rule['security_group_rule'][k], v)
def test_create_security_group_source_group_ip_and_ip_prefix(self):
security_group_id = "4cd70774-cc67-4a87-9b39-7d1db38eb087"
@ -396,7 +396,7 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
source_group_id)
res = self._create_security_group_rule('json', rule)
self.deserialize('json', res)
self.assertEquals(res.status_int, 400)
self.assertEqual(res.status_int, 400)
def test_create_security_group_rule_bad_security_group_id(self):
security_group_id = "4cd70774-cc67-4a87-9b39-7d1db38eb087"
@ -411,7 +411,7 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
source_ip_prefix)
res = self._create_security_group_rule('json', rule)
self.deserialize('json', res)
self.assertEquals(res.status_int, 404)
self.assertEqual(res.status_int, 404)
def test_create_security_group_rule_bad_tenant(self):
with self.security_group() as sg:
@ -425,7 +425,7 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
res = self._create_security_group_rule('json', rule)
self.deserialize('json', res)
self.assertEquals(res.status_int, 404)
self.assertEqual(res.status_int, 404)
def test_create_security_group_rule_exteral_id_proxy_mode(self):
cfg.CONF.SECURITYGROUP.proxy_mode = True
@ -442,7 +442,7 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
res = self._create_security_group_rule('json', rule)
self.deserialize('json', res)
self.assertEquals(res.status_int, 201)
self.assertEqual(res.status_int, 201)
def test_create_security_group_rule_exteral_id_not_proxy_mode(self):
with self.security_group() as sg:
@ -458,7 +458,7 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
res = self._create_security_group_rule('json', rule)
self.deserialize('json', res)
self.assertEquals(res.status_int, 409)
self.assertEqual(res.status_int, 409)
def test_create_security_group_rule_not_admin(self):
cfg.CONF.SECURITYGROUP.proxy_mode = True
@ -477,7 +477,7 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
tenant_id='bad_tenant',
set_context=True)
self.deserialize('json', res)
self.assertEquals(res.status_int, 500)
self.assertEqual(res.status_int, 500)
def test_create_security_group_rule_bad_tenant_source_group_id(self):
with self.security_group() as sg:
@ -498,7 +498,7 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
tenant_id='bad_tenant',
set_context=True)
self.deserialize('json', res)
self.assertEquals(res.status_int, 404)
self.assertEqual(res.status_int, 404)
def test_create_security_group_rule_bad_tenant_security_group_rule(self):
with self.security_group() as sg:
@ -518,7 +518,7 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
tenant_id='bad_tenant',
set_context=True)
self.deserialize('json', res)
self.assertEquals(res.status_int, 404)
self.assertEqual(res.status_int, 404)
def test_create_security_group_rule_bad_source_group_id(self):
name = 'webservers'
@ -536,7 +536,7 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
source_group_id=source_group_id)
res = self._create_security_group_rule('json', rule)
self.deserialize('json', res)
self.assertEquals(res.status_int, 404)
self.assertEqual(res.status_int, 404)
def test_create_security_group_rule_duplicate_rules(self):
name = 'webservers'
@ -549,7 +549,7 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
self._create_security_group_rule('json', rule)
res = self._create_security_group_rule('json', rule)
self.deserialize('json', res)
self.assertEquals(res.status_int, 409)
self.assertEqual(res.status_int, 409)
def test_create_security_group_rule_min_port_greater_max(self):
name = 'webservers'
@ -562,7 +562,7 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
self._create_security_group_rule('json', rule)
res = self._create_security_group_rule('json', rule)
self.deserialize('json', res)
self.assertEquals(res.status_int, 400)
self.assertEqual(res.status_int, 400)
def test_create_security_group_rule_ports_but_no_protocol(self):
name = 'webservers'
@ -575,7 +575,7 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
self._create_security_group_rule('json', rule)
res = self._create_security_group_rule('json', rule)
self.deserialize('json', res)
self.assertEquals(res.status_int, 400)
self.assertEqual(res.status_int, 400)
def test_update_port_with_security_group(self):
with self.network() as n:
@ -592,8 +592,8 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
req = self.new_update_request('ports', data,
port['port']['id'])
res = self.deserialize('json', req.get_response(self.api))
self.assertEquals(res['port'][ext_sg.SECURITYGROUP][0],
sg['security_group']['id'])
self.assertEqual(res['port'][ext_sg.SECURITYGROUP][0],
sg['security_group']['id'])
self._delete('ports', port['port']['id'])
def test_update_port_with_multiple_security_groups(self):
@ -606,7 +606,7 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
security_groups=[sg1['security_group']['id'],
sg2['security_group']['id']])
port = self.deserialize('json', res)
self.assertEquals(len(
self.assertEqual(len(
port['port'][ext_sg.SECURITYGROUP]), 2)
self._delete('ports', port['port']['id'])
@ -625,7 +625,7 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
req = self.new_update_request('ports', data,
port['port']['id'])
res = self.deserialize('json', req.get_response(self.api))
self.assertEquals(res['port'][ext_sg.SECURITYGROUP], [])
self.assertEqual(res['port'][ext_sg.SECURITYGROUP], [])
self._delete('ports', port['port']['id'])
def test_create_port_with_bad_security_group(self):
@ -635,7 +635,7 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
security_groups=['bad_id'])
self.deserialize('json', res)
self.assertEquals(res.status_int, 404)
self.assertEqual(res.status_int, 404)
def test_create_delete_security_group_port_in_use(self):
with self.network() as n:
@ -645,8 +645,8 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
security_groups=(
[sg['security_group']['id']]))
port = self.deserialize('json', res)
self.assertEquals(port['port'][ext_sg.SECURITYGROUP][0],
sg['security_group']['id'])
self.assertEqual(port['port'][ext_sg.SECURITYGROUP][0],
sg['security_group']['id'])
# try to delete security group that's in use
res = self._delete('security-groups',
sg['security_group']['id'], 409)
@ -668,7 +668,7 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
rule2['security_group_rule']]}
res = self._create_security_group_rule('json', rules)
self.deserialize('json', res)
self.assertEquals(res.status_int, 201)
self.assertEqual(res.status_int, 201)
def test_create_security_group_rule_bulk_emulated(self):
real_has_attr = hasattr
@ -693,7 +693,7 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
}
res = self._create_security_group_rule('json', rules)
self.deserialize('json', res)
self.assertEquals(res.status_int, 201)
self.assertEqual(res.status_int, 201)
def test_create_security_group_rule_duplicate_rule_in_post(self):
if self._skip_native_bulk:
@ -707,7 +707,7 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
rule['security_group_rule']]}
res = self._create_security_group_rule('json', rules)
rule = self.deserialize('json', res)
self.assertEquals(res.status_int, 409)
self.assertEqual(res.status_int, 409)
def test_create_security_group_rule_duplicate_rule_in_post_emulated(self):
real_has_attr = hasattr
@ -729,7 +729,7 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
rule['security_group_rule']]}
res = self._create_security_group_rule('json', rules)
rule = self.deserialize('json', res)
self.assertEquals(res.status_int, 409)
self.assertEqual(res.status_int, 409)
def test_create_security_group_rule_duplicate_rule_db(self):
if self._skip_native_bulk:
@ -743,7 +743,7 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
self._create_security_group_rule('json', rules)
res = self._create_security_group_rule('json', rules)
rule = self.deserialize('json', res)
self.assertEquals(res.status_int, 409)
self.assertEqual(res.status_int, 409)
def test_create_security_group_rule_duplicate_rule_db_emulated(self):
real_has_attr = hasattr
@ -764,7 +764,7 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
self._create_security_group_rule('json', rules)
res = self._create_security_group_rule('json', rule)
self.deserialize('json', res)
self.assertEquals(res.status_int, 409)
self.assertEqual(res.status_int, 409)
def test_create_security_group_rule_differnt_security_group_ids(self):
if self._skip_native_bulk:
@ -784,7 +784,7 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
}
res = self._create_security_group_rule('json', rules)
self.deserialize('json', res)
self.assertEquals(res.status_int, 400)
self.assertEqual(res.status_int, 400)
def test_create_security_group_rule_with_invalid_ethertype(self):
security_group_id = "4cd70774-cc67-4a87-9b39-7d1db38eb087"
@ -802,7 +802,7 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
ethertype='IPv5')
res = self._create_security_group_rule('json', rule)
self.deserialize('json', res)
self.assertEquals(res.status_int, 400)
self.assertEqual(res.status_int, 400)
def test_create_security_group_rule_with_invalid_protocol(self):
security_group_id = "4cd70774-cc67-4a87-9b39-7d1db38eb087"
@ -819,4 +819,4 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
source_group_id)
res = self._create_security_group_rule('json', rule)
self.deserialize('json', res)
self.assertEquals(res.status_int, 400)
self.assertEqual(res.status_int, 400)

14
quantum/tests/unit/test_l3_agent.py
View File

@ -107,12 +107,12 @@ class TestBasicRouterOperations(unittest2.TestCase):
self.device_exists.return_value = False
agent.internal_network_added(ri, ex_gw_port, network_id,
port_id, cidr, mac)
self.assertEquals(self.mock_driver.plug.call_count, 1)
self.assertEquals(self.mock_driver.init_l3.call_count, 1)
self.assertEqual(self.mock_driver.plug.call_count, 1)
self.assertEqual(self.mock_driver.init_l3.call_count, 1)
elif action == 'remove':
self.device_exists.return_value = True
agent.internal_network_removed(ri, ex_gw_port, port_id, cidr)
self.assertEquals(self.mock_driver.unplug.call_count, 1)
self.assertEqual(self.mock_driver.unplug.call_count, 1)
else:
raise Exception("Invalid action %s" % action)
@ -140,8 +140,8 @@ class TestBasicRouterOperations(unittest2.TestCase):
if action == 'add':
self.device_exists.return_value = False
agent.external_gateway_added(ri, ex_gw_port, internal_cidrs)
self.assertEquals(self.mock_driver.plug.call_count, 1)
self.assertEquals(self.mock_driver.init_l3.call_count, 1)
self.assertEqual(self.mock_driver.plug.call_count, 1)
self.assertEqual(self.mock_driver.init_l3.call_count, 1)
arping_cmd = ['arping', '-A', '-U',
'-I', interface_name,
'-c', self.conf.send_arp_for_ha,
@ -156,7 +156,7 @@ class TestBasicRouterOperations(unittest2.TestCase):
elif action == 'remove':
self.device_exists.return_value = True
agent.external_gateway_removed(ri, ex_gw_port, internal_cidrs)
self.assertEquals(self.mock_driver.unplug.call_count, 1)
self.assertEqual(self.mock_driver.unplug.call_count, 1)
else:
raise Exception("Invalid action %s" % action)
@ -277,7 +277,7 @@ class TestBasicRouterOperations(unittest2.TestCase):
agent.router_deleted(None, routers[0]['id'])
# verify that remove is called
self.assertEquals(self.mock_ip.get_devices.call_count, 1)
self.assertEqual(self.mock_ip.get_devices.call_count, 1)
self.device_exists.assert_has_calls(
[mock.call(self.conf.external_network_bridge)])

86
quantum/tests/unit/test_l3_plugin.py
View File

@ -363,7 +363,7 @@ class L3NatDBTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
router_req = self.new_create_request('routers', data, 'json')
res = router_req.get_response(self.ext_api)
router = self.deserialize('json', res)
self.assertEquals(
self.assertEqual(
s['subnet']['network_id'],
router['router']['external_gateway_info']['network_id'])
self._delete('routers', router['router']['id'])
@ -394,13 +394,13 @@ class L3NatDBTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
rname2 = "nachorouter"
with self.router(name=rname1) as r:
body = self._show('routers', r['router']['id'])
self.assertEquals(body['router']['name'], rname1)
self.assertEqual(body['router']['name'], rname1)
body = self._update('routers', r['router']['id'],
{'router': {'name': rname2}})
body = self._show('routers', r['router']['id'])
self.assertEquals(body['router']['name'], rname2)
self.assertEqual(body['router']['name'], rname2)
def test_router_update_gateway(self):
with self.router() as r:
@ -413,7 +413,7 @@ class L3NatDBTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
body = self._show('routers', r['router']['id'])
net_id = (body['router']
['external_gateway_info']['network_id'])
self.assertEquals(net_id, s1['subnet']['network_id'])
self.assertEqual(net_id, s1['subnet']['network_id'])
self._set_net_external(s2['subnet']['network_id'])
self._add_external_gateway_to_router(
r['router']['id'],
@ -421,7 +421,7 @@ class L3NatDBTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
body = self._show('routers', r['router']['id'])
net_id = (body['router']
['external_gateway_info']['network_id'])
self.assertEquals(net_id, s2['subnet']['network_id'])
self.assertEqual(net_id, s2['subnet']['network_id'])
self._remove_external_gateway_from_router(
r['router']['id'],
s2['subnet']['network_id'])
@ -453,7 +453,7 @@ class L3NatDBTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
# fetch port and confirm device_id
r_port_id = body['port_id']
body = self._show('ports', r_port_id)
self.assertEquals(body['port']['device_id'], r['router']['id'])
self.assertEqual(body['port']['device_id'], r['router']['id'])
body = self._router_interface_action('remove',
r['router']['id'],
@ -545,11 +545,11 @@ class L3NatDBTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
None,
p['port']['id'])
self.assertTrue('port_id' in body)
self.assertEquals(body['port_id'], p['port']['id'])
self.assertEqual(body['port_id'], p['port']['id'])
# fetch port and confirm device_id
body = self._show('ports', p['port']['id'])
self.assertEquals(body['port']['device_id'], r['router']['id'])
self.assertEqual(body['port']['device_id'], r['router']['id'])
# clean-up
self._router_interface_action('remove',
@ -712,13 +712,13 @@ class L3NatDBTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
s['subnet']['network_id'])
body = self._show('routers', r['router']['id'])
net_id = body['router']['external_gateway_info']['network_id']
self.assertEquals(net_id, s['subnet']['network_id'])
self.assertEqual(net_id, s['subnet']['network_id'])
self._remove_external_gateway_from_router(
r['router']['id'],
s['subnet']['network_id'])
body = self._show('routers', r['router']['id'])
gw_info = body['router']['external_gateway_info']
self.assertEquals(gw_info, None)
self.assertEqual(gw_info, None)
def test_router_add_gateway_invalid_network_returns_404(self):
with self.router() as r:
@ -919,13 +919,13 @@ class L3NatDBTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
def _validate_floating_ip(self, fip):
body = self._list('floatingips')
self.assertEquals(len(body['floatingips']), 1)
self.assertEquals(body['floatingips'][0]['id'],
fip['floatingip']['id'])
self.assertEqual(len(body['floatingips']), 1)
self.assertEqual(body['floatingips'][0]['id'],
fip['floatingip']['id'])
body = self._show('floatingips', fip['floatingip']['id'])
self.assertEquals(body['floatingip']['id'],
fip['floatingip']['id'])
self.assertEqual(body['floatingip']['id'],
fip['floatingip']['id'])
@contextlib.contextmanager
def floatingip_with_assoc(self, port_id=None, fmt='json'):
@ -995,7 +995,7 @@ class L3NatDBTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
# post-delete, check that it is really gone
body = self._list('floatingips')
self.assertEquals(len(body['floatingips']), 0)
self.assertEqual(len(body['floatingips']), 0)
self._show('floatingips', fip['floatingip']['id'],
expected_code=exc.HTTPNotFound.code)
@ -1044,25 +1044,25 @@ class L3NatDBTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
p['port']['fixed_ips'][0]['subnet_id']}}
with self.floatingip_no_assoc(private_sub) as fip:
body = self._show('floatingips', fip['floatingip']['id'])
self.assertEquals(body['floatingip']['port_id'], None)
self.assertEquals(body['floatingip']['fixed_ip_address'], None)
self.assertEqual(body['floatingip']['port_id'], None)
self.assertEqual(body['floatingip']['fixed_ip_address'], None)
port_id = p['port']['id']
ip_address = p['port']['fixed_ips'][0]['ip_address']
fixed_ip = p['port']['fixed_ips'][0]['ip_address']
body = self._update('floatingips', fip['floatingip']['id'],
{'floatingip': {'port_id': port_id}})
self.assertEquals(body['floatingip']['port_id'], port_id)
self.assertEquals(body['floatingip']['fixed_ip_address'],
ip_address)
self.assertEqual(body['floatingip']['port_id'], port_id)
self.assertEqual(body['floatingip']['fixed_ip_address'],
ip_address)
def test_floatingip_with_assoc(self):
with self.floatingip_with_assoc() as fip:
body = self._show('floatingips', fip['floatingip']['id'])
self.assertEquals(body['floatingip']['id'],
fip['floatingip']['id'])
self.assertEquals(body['floatingip']['port_id'],
fip['floatingip']['port_id'])
self.assertEqual(body['floatingip']['id'],
fip['floatingip']['id'])
self.assertEqual(body['floatingip']['port_id'],
fip['floatingip']['port_id'])
self.assertTrue(body['floatingip']['fixed_ip_address'] is not None)
self.assertTrue(body['floatingip']['router_id'] is not None)
@ -1077,11 +1077,11 @@ class L3NatDBTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
# deleted, which is what we want to test. We want to confirm
# that the fields are set back to None
body = self._show('floatingips', fip['floatingip']['id'])
self.assertEquals(body['floatingip']['id'],
fip['floatingip']['id'])
self.assertEquals(body['floatingip']['port_id'], None)
self.assertEquals(body['floatingip']['fixed_ip_address'], None)
self.assertEquals(body['floatingip']['router_id'], None)
self.assertEqual(body['floatingip']['id'],
fip['floatingip']['id'])
self.assertEqual(body['floatingip']['port_id'], None)
self.assertEqual(body['floatingip']['fixed_ip_address'], None)
self.assertEqual(body['floatingip']['router_id'], None)
def test_two_fips_one_port_invalid_return_409(self):
with self.floatingip_with_assoc() as fip1:
@ -1195,15 +1195,15 @@ class L3NatDBTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
self._set_net_external(n1['network']['id'])
with self.network() as n2:
body = self._list('networks')
self.assertEquals(len(body['networks']), 2)
self.assertEqual(len(body['networks']), 2)
body = self._list('networks',
query_params="%s=True" % l3.EXTERNAL)
self.assertEquals(len(body['networks']), 1)
self.assertEqual(len(body['networks']), 1)
body = self._list('networks',
query_params="%s=False" % l3.EXTERNAL)
self.assertEquals(len(body['networks']), 1)
self.assertEqual(len(body['networks']), 1)
def test_get_network_succeeds_without_filter(self):
plugin = manager.QuantumManager.get_plugin()
@ -1237,7 +1237,7 @@ class L3NatDBTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
set_context='True',
tenant_id='noadmin'):
pass
self.assertEquals(ctx_manager.exception.code, 403)
self.assertEqual(ctx_manager.exception.code, 403)
def test_create_port_external_network_admin_suceeds(self):
with self.network(router__external=True) as ext_net:
@ -1252,7 +1252,7 @@ class L3NatDBTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
set_context='True',
tenant_id='noadmin'):
pass
self.assertEquals(ctx_manager.exception.code, 403)
self.assertEqual(ctx_manager.exception.code, 403)
def test_create_external_network_admin_suceeds(self):
with self.network(router__external=True) as ext_net:
@ -1285,7 +1285,7 @@ class L3NatDBTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
self._remove_external_gateway_from_router(
r['router']['id'],
s['subnet']['network_id'])
self.assertEquals(
self.assertEqual(
2, notifyApi.routers_updated.call_count)
def test_router_gateway_op_agent(self):
@ -1302,7 +1302,7 @@ class L3NatDBTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
r['router']['id'],
None,
p['port']['id'])
self.assertEquals(2, notifyApi.routers_updated.call_count)
self.assertEqual(2, notifyApi.routers_updated.call_count)
def test_interfaces_op_agent(self):
with self.router() as r:
@ -1314,7 +1314,7 @@ class L3NatDBTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
pass
# add gateway, add interface, associate, deletion of floatingip,
# delete gateway, delete interface
self.assertEquals(6, notifyApi.routers_updated.call_count)
self.assertEqual(6, notifyApi.routers_updated.call_count)
def test_floatingips_op_agent(self):
self._test_notify_op_agent(self._test_floatingips_op_agent)
@ -1382,7 +1382,7 @@ class L3NatDBTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
[r['router']['id']])
self.assertEqual(1, len(routers))
gw_port = routers[0]['gw_port']
self.assertEquals(s['subnet']['id'], gw_port['subnet']['id'])
self.assertEqual(s['subnet']['id'], gw_port['subnet']['id'])
self._remove_external_gateway_from_router(
r['router']['id'],
s['subnet']['network_id'])
@ -1395,9 +1395,9 @@ class L3NatDBTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
self.assertEqual(1, len(routers))
floatingips = routers[0][l3_constants.FLOATINGIP_KEY]
self.assertEqual(1, len(floatingips))
self.assertEquals(floatingips[0]['id'],
fip['floatingip']['id'])
self.assertEquals(floatingips[0]['port_id'],
fip['floatingip']['port_id'])
self.assertEqual(floatingips[0]['id'],
fip['floatingip']['id'])
self.assertEqual(floatingips[0]['port_id'],
fip['floatingip']['port_id'])
self.assertTrue(floatingips[0]['fixed_ip_address'] is not None)
self.assertTrue(floatingips[0]['router_id'] is not None)

2
quantum/tests/unit/test_linux_dhcp.py
View File

@ -176,7 +176,7 @@ class TestDhcpBase(unittest.TestCase):
c = SubClass()
c.restart()
self.assertEquals(c.called, ['disable True', 'enable'])
self.assertEqual(c.called, ['disable True', 'enable'])
class LocalChild(dhcp.DhcpLocalProcess):

2
quantum/tests/unit/test_linux_interface.py
View File

@ -315,7 +315,7 @@ class TestBridgeInterfaceDriver(TestBase):
'tap0',
'aa:bb:cc:dd:ee:ff')
self.ip_dev.assert_has_calls([])
self.assertEquals(log.call_count, 1)
self.assertEqual(log.call_count, 1)
def test_plug_mtu(self):
self.device_exists.return_value = False

40
quantum/tests/unit/test_linux_ip_lib.py
View File

@ -144,11 +144,11 @@ class TestIpWrapper(unittest.TestCase):
def test_get_devices(self):
self.execute.return_value = '\n'.join(LINK_SAMPLE)
retval = ip_lib.IPWrapper('sudo').get_devices()
self.assertEquals(retval,
[ip_lib.IPDevice('lo'),
ip_lib.IPDevice('eth0'),
ip_lib.IPDevice('br-int'),
ip_lib.IPDevice('gw-ddc717df-49')])
self.assertEqual(retval,
[ip_lib.IPDevice('lo'),
ip_lib.IPDevice('eth0'),
ip_lib.IPDevice('br-int'),
ip_lib.IPDevice('gw-ddc717df-49')])
self.execute.assert_called_once_with('o', 'link', ('list',),
'sudo', None)
@ -156,11 +156,11 @@ class TestIpWrapper(unittest.TestCase):
def test_get_devices_malformed_line(self):
self.execute.return_value = '\n'.join(LINK_SAMPLE + ['gibberish'])
retval = ip_lib.IPWrapper('sudo').get_devices()
self.assertEquals(retval,
[ip_lib.IPDevice('lo'),
ip_lib.IPDevice('eth0'),
ip_lib.IPDevice('br-int'),
ip_lib.IPDevice('gw-ddc717df-49')])
self.assertEqual(retval,
[ip_lib.IPDevice('lo'),
ip_lib.IPDevice('eth0'),
ip_lib.IPDevice('br-int'),
ip_lib.IPDevice('gw-ddc717df-49')])
self.execute.assert_called_once_with('o', 'link', ('list',),
'sudo', None)
@ -168,10 +168,10 @@ class TestIpWrapper(unittest.TestCase):
def test_get_namespaces(self):
self.execute.return_value = '\n'.join(NETNS_SAMPLE)
retval = ip_lib.IPWrapper.get_namespaces('sudo')
self.assertEquals(retval,
['12345678-1234-5678-abcd-1234567890ab',
'bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb',
'cccccccc-cccc-cccc-cccc-cccccccccccc'])
self.assertEqual(retval,
['12345678-1234-5678-abcd-1234567890ab',
'bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb',
'cccccccc-cccc-cccc-cccc-cccccccccccc'])
self.execute.assert_called_once_with('', 'netns', ('list',),
root_helper='sudo')
@ -440,7 +440,7 @@ class TestIpLinkCommand(TestIPCmdBase):
'brd': 'ff:ff:ff:ff:ff:ff',
'link/ether': 'cc:dd:ee:ff:ab:cd'}
self.parent._execute = mock.Mock(return_value=LINK_SAMPLE[1])
self.assertEquals(self.link_cmd.attributes, expected)
self.assertEqual(self.link_cmd.attributes, expected)
self._assert_call('o', ('show', 'eth0'))
@ -495,7 +495,7 @@ class TestIpAddrCommand(TestIPCmdBase):
broadcast='::')]
self.parent._run = mock.Mock(return_value=ADDR_SAMPLE)
self.assertEquals(self.addr_cmd.list(), expected)
self.assertEqual(self.addr_cmd.list(), expected)
self._assert_call([], ('show', 'tap0'))
def test_list_filtered(self):
@ -506,8 +506,8 @@ class TestIpAddrCommand(TestIPCmdBase):
output = '\n'.join(ADDR_SAMPLE.split('\n')[0:4])
self.parent._run.return_value = output
self.assertEquals(self.addr_cmd.list('global', filters=['permanent']),
expected)
self.assertEqual(self.addr_cmd.list('global', filters=['permanent']),
expected)
self._assert_call([], ('show', 'tap0', 'permanent', 'scope', 'global'))
@ -547,8 +547,8 @@ class TestIpRouteCommand(TestIPCmdBase):
'expected': {'gateway': '10.35.19.254'}}]
for test_case in test_cases:
self.parent._run = mock.Mock(return_value=test_case['sample'])
self.assertEquals(self.route_cmd.get_gateway(),
test_case['expected'])
self.assertEqual(self.route_cmd.get_gateway(),
test_case['expected'])
def test_pullup_route(self):
# interface is not the first in the list - requires

8
quantum/tests/unit/test_quantum_context.py
View File

@ -34,14 +34,14 @@ class TestQuantumContext(unittest.TestCase):
def testQuantumContextCreate(self):
cxt = context.Context('user_id', 'tenant_id')
self.assertEquals('user_id', cxt.user_id)
self.assertEquals('tenant_id', cxt.project_id)
self.assertEqual('user_id', cxt.user_id)
self.assertEqual('tenant_id', cxt.project_id)
def testQuantumContextToDict(self):
cxt = context.Context('user_id', 'tenant_id')
cxt_dict = cxt.to_dict()
self.assertEquals('user_id', cxt_dict['user_id'])
self.assertEquals('tenant_id', cxt_dict['project_id'])
self.assertEqual('user_id', cxt_dict['user_id'])
self.assertEqual('tenant_id', cxt_dict['project_id'])
def testQuantumContextAdminToDict(self):
self.db_api_session.return_value = 'fakesession'

26
quantum/tests/unit/test_quota_per_tenant_ext.py
View File

@ -112,17 +112,17 @@ class QuotaExtensionTestCase(unittest.TestCase):
def test_quotas_loaded_right(self):
res = self.api.get(_get_path('quotas'))
self.assertEquals(200, res.status_int)
self.assertEqual(200, res.status_int)
def test_quotas_defaul_values(self):
tenant_id = 'tenant_id1'
env = {'quantum.context': context.Context('', tenant_id)}
res = self.api.get(_get_path('quotas', id=tenant_id),
extra_environ=env)
self.assertEquals(10, res.json['quota']['network'])
self.assertEquals(10, res.json['quota']['subnet'])
self.assertEquals(50, res.json['quota']['port'])
self.assertEquals(-1, res.json['quota']['extra1'])
self.assertEqual(10, res.json['quota']['network'])
self.assertEqual(10, res.json['quota']['subnet'])
self.assertEqual(50, res.json['quota']['port'])
self.assertEqual(-1, res.json['quota']['extra1'])
def test_show_quotas_with_admin(self):
tenant_id = 'tenant_id1'
@ -130,7 +130,7 @@ class QuotaExtensionTestCase(unittest.TestCase):
is_admin=True)}
res = self.api.get(_get_path('quotas', id=tenant_id),
extra_environ=env)
self.assertEquals(200, res.status_int)
self.assertEqual(200, res.status_int)
def test_show_quotas_without_admin_forbidden(self):
tenant_id = 'tenant_id1'
@ -138,7 +138,7 @@ class QuotaExtensionTestCase(unittest.TestCase):
is_admin=False)}
res = self.api.get(_get_path('quotas', id=tenant_id),
extra_environ=env, expect_errors=True)
self.assertEquals(403, res.status_int)
self.assertEqual(403, res.status_int)
def test_update_quotas_without_admin_forbidden(self):
tenant_id = 'tenant_id1'
@ -149,7 +149,7 @@ class QuotaExtensionTestCase(unittest.TestCase):
fmt='json'),
quotas, extra_environ=env,
expect_errors=True)
self.assertEquals(403, res.status_int)
self.assertEqual(403, res.status_int)
def test_update_quotas_with_admin(self):
tenant_id = 'tenant_id1'
@ -158,11 +158,11 @@ class QuotaExtensionTestCase(unittest.TestCase):
quotas = {'quota': {'network': 100}}
res = self.api.put_json(_get_path('quotas', id=tenant_id, fmt='json'),
quotas, extra_environ=env)
self.assertEquals(200, res.status_int)
self.assertEqual(200, res.status_int)
env2 = {'quantum.context': context.Context('', tenant_id)}
res = self.api.get(_get_path('quotas', id=tenant_id),
extra_environ=env2).json
self.assertEquals(100, res['quota']['network'])
self.assertEqual(100, res['quota']['network'])
def test_delete_quotas_with_admin(self):
tenant_id = 'tenant_id1'
@ -170,7 +170,7 @@ class QuotaExtensionTestCase(unittest.TestCase):
is_admin=True)}
res = self.api.delete(_get_path('quotas', id=tenant_id, fmt='json'),
extra_environ=env)
self.assertEquals(204, res.status_int)
self.assertEqual(204, res.status_int)
def test_delete_quotas_without_admin_forbidden(self):
tenant_id = 'tenant_id1'
@ -178,13 +178,13 @@ class QuotaExtensionTestCase(unittest.TestCase):
is_admin=False)}
res = self.api.delete(_get_path('quotas', id=tenant_id, fmt='json'),
extra_environ=env, expect_errors=True)
self.assertEquals(403, res.status_int)
self.assertEqual(403, res.status_int)
def test_quotas_loaded_bad(self):
self.testflag = 2
try:
res = self.api.get(_get_path('quotas'), expect_errors=True)
self.assertEquals(404, res.status_int)
self.assertEqual(404, res.status_int)
except Exception:
pass
self.testflag = 1