openstack
/
neutron
Code Issues Proposed changes

Replace assertItemsEqual with assertCountEqual 

assertItemsEqual was removed from Python's unittest.TestCase in
Python 3.3 [1][2]. We have been able to use them since then, because
testtools required unittest2, which still included it. With testtools
removing Python 2.7 support [3][4], we will lose support for
assertItemsEqual, so we should switch to use assertCountEqual.

NOTE(dmllr): added hacking check

[1] - https://bugs.python.org/issue17866
[2] - https://hg.python.org/cpython/rev/d9921cb6e3cd
[3] - testing-cabal/testtools#286
[4] - testing-cabal/testtools#277

Change-Id: I7c20fec08e5dc9f67b34100c925ea6724bbd25f0
This commit is contained in:
gugug 2020-07-12 11:19:55 +08:00 committed by Dirk Mueller
parent d7ab8f2a78
commit 62729d4e74
36 changed files with 164 additions and 154 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
neutron/hacking/checks.py
View File

@ -87,6 +87,16 @@ def check_asserttruefalse(logical_line, filename):
yield (0, msg)
@core.flake8ext
def check_assertitemsequal(logical_line, filename):
"""N328 - Don't use assertItemsEqual."""
if 'neutron/tests/' in filename:
if re.search(r"assertItemsEqual\(", logical_line):
msg = ("N329: Use assertCountEqual() instead of "
"assertItemsEqual()")
yield (0, msg)
@core.flake8ext
def check_assertempty(logical_line, filename):
"""N330 - Enforce using assertEqual parameter ordering in case of empty

8
neutron/tests/functional/agent/linux/test_ip_lib.py
View File

@ -386,7 +386,7 @@ class IpLibTestCase(IpLibTestFramework):
'scope': 'link'}]
routes = ip_lib.get_routing_table(4, namespace=attr.namespace)
self.assertItemsEqual(expected_routes, routes)
self.assertCountEqual(expected_routes, routes)
self.assertIsInstance(routes, list)
expected_routes6 = [{'nexthop': "fd00::2",
@ -399,7 +399,7 @@ class IpLibTestCase(IpLibTestFramework):
netaddr.IPNetwork(attr.ip_cidrs[1]).cidr),
'scope': 'universe'}]
routes6 = ip_lib.get_routing_table(6, namespace=attr.namespace)
self.assertItemsEqual(expected_routes6, routes6)
self.assertCountEqual(expected_routes6, routes6)
self.assertIsInstance(routes6, list)
def test_get_routing_table_no_namespace(self):
@ -419,7 +419,7 @@ class IpLibTestCase(IpLibTestFramework):
'device': attr.name}]
neighs = device.neigh.dump(4)
self.assertItemsEqual(expected_neighs, neighs)
self.assertCountEqual(expected_neighs, neighs)
self.assertIsInstance(neighs, list)
device.neigh.delete(TEST_IP_NEIGH, mac_address)
@ -573,7 +573,7 @@ class IpLibTestCase(IpLibTestFramework):
ip_info['scope'],
ip_info['broadcast']) for
ip_info in device.addr.list()]
self.assertItemsEqual(ip_addresses, device_ips_info)
self.assertCountEqual(ip_addresses, device_ips_info)
def _flush_ips(self, device, ip_version):
device.addr.flush(ip_version)

4
neutron/tests/functional/agent/test_ovs_lib.py
View File

@ -486,7 +486,7 @@ class OVSBridgeTestCase(OVSBridgeTestBase):
expected = self.br.initial_protocols.union(protocols)
self.br.ovsdb.db_add("Bridge", self.br.br_name, "protocols",
*protocols).execute(check_error=True)
self.assertItemsEqual(expected,
self.assertCountEqual(expected,
self.br.db_get_val('Bridge',
self.br.br_name, "protocols"))
@ -669,4 +669,4 @@ class OVSLibTestCase(base.BaseOVSLinuxTestCase):
self.assertTrue(tags_present)
tags_42 = [t for t in tags_present if t['tag'] == 42]
self.assertEqual(tags_42, single_value.result)
self.assertItemsEqual(len_0_list.result, tags_present)
self.assertCountEqual(len_0_list.result, tags_present)

2
neutron/tests/functional/plugins/ml2/drivers/ovn/mech_driver/ovsdb/test_impl_idl.py
View File

@ -72,7 +72,7 @@ class TestSbApi(n_base.BaseLoggingTestCase,
dp, iface, phys = self.api.get_chassis_data_for_ml2_bind_port(host)
self.assertEqual('', dp)
self.assertEqual('', iface)
self.assertItemsEqual(phys, ['private', 'public'])
self.assertCountEqual(phys, ['private', 'public'])
def test_chassis_exists(self):
self.assertTrue(self.api.chassis_exists(

8
neutron/tests/functional/plugins/ml2/drivers/ovn/mech_driver/ovsdb/test_ovn_db_resources.py
View File

@ -56,7 +56,7 @@ class TestNBDbResources(base.TestOVNFunctionalBase):
'cidr': row.cidr, 'external_ids': ext_ids,
'options': row.options})
self.assertItemsEqual(expected_dhcp_options_rows,
self.assertCountEqual(expected_dhcp_options_rows,
observed_dhcp_options_rows)
def _verify_dhcp_option_row_for_port(self, port_id,
@ -822,7 +822,7 @@ class TestPortSecurity(base.TestOVNFunctionalBase):
def _verify_port_acls(self, port_id, expected_acls):
port_acls = self._get_port_related_acls(port_id)
self.assertItemsEqual(expected_acls, port_acls)
self.assertCountEqual(expected_acls, port_acls)
def test_port_security_port_group(self):
n1 = self._make_network(self.fmt, 'n1', True)
@ -913,7 +913,7 @@ class TestDNSRecords(base.TestOVNFunctionalBase):
observed_dns_records.append(
{'external_ids': dns_row.external_ids,
'records': dns_row.records})
self.assertItemsEqual(expected_dns_records, observed_dns_records)
self.assertCountEqual(expected_dns_records, observed_dns_records)
def _validate_ls_dns_records(self, lswitch_name, expected_dns_records):
ls = idlutils.row_by_value(self.nb_api.idl,
@ -923,7 +923,7 @@ class TestDNSRecords(base.TestOVNFunctionalBase):
observed_dns_records.append(
{'external_ids': dns_row.external_ids,
'records': dns_row.records})
self.assertItemsEqual(expected_dns_records, observed_dns_records)
self.assertCountEqual(expected_dns_records, observed_dns_records)
def setUp(self):
ovn_config.cfg.CONF.set_override('dns_domain', 'ovn.test')

116
neutron/tests/functional/plugins/ml2/drivers/ovn/mech_driver/ovsdb/test_ovn_db_sync.py
View File

@ -848,25 +848,25 @@ class TestOvnNbSync(base.TestOVNFunctionalBase):
if row.name.startswith(ovn_const.OVN_PROVNET_PORT_NAME_PREFIX)]
if should_match:
self.assertItemsEqual(db_net_ids, plugin_lswitch_ids)
self.assertItemsEqual(db_net_ids, monitor_lswitch_ids)
self.assertItemsEqual(db_provnet_ports, plugin_provnet_ports)
self.assertItemsEqual(db_provnet_ports, monitor_provnet_ports)
self.assertCountEqual(db_net_ids, plugin_lswitch_ids)
self.assertCountEqual(db_net_ids, monitor_lswitch_ids)
self.assertCountEqual(db_provnet_ports, plugin_provnet_ports)
self.assertCountEqual(db_provnet_ports, monitor_provnet_ports)
else:
self.assertRaises(
AssertionError, self.assertItemsEqual, db_net_ids,
AssertionError, self.assertCountEqual, db_net_ids,
plugin_lswitch_ids)
self.assertRaises(
AssertionError, self.assertItemsEqual, db_net_ids,
AssertionError, self.assertCountEqual, db_net_ids,
monitor_lswitch_ids)
self.assertRaises(
AssertionError, self.assertItemsEqual, db_provnet_ports,
AssertionError, self.assertCountEqual, db_provnet_ports,
plugin_provnet_ports)
self.assertRaises(
AssertionError, self.assertItemsEqual, db_provnet_ports,
AssertionError, self.assertCountEqual, db_provnet_ports,
monitor_provnet_ports)
def _validate_metadata_ports(self, should_match=True):
@ -894,9 +894,9 @@ class TestOvnNbSync(base.TestOVNFunctionalBase):
if should_match:
# Check that metadata ports exist in both Neutron and OVN dbs.
self.assertItemsEqual(db_metadata_ports_ids, plugin_metadata_ports)
self.assertCountEqual(db_metadata_ports_ids, plugin_metadata_ports)
# Check that all networks have one and only one metadata port.
self.assertItemsEqual(db_metadata_ports_nets, db_net_ids)
self.assertCountEqual(db_metadata_ports_nets, db_net_ids)
else:
metadata_sync = (sorted(db_metadata_ports_ids) ==
sorted(plugin_metadata_ports))
@ -942,23 +942,23 @@ class TestOvnNbSync(base.TestOVNFunctionalBase):
if row.dhcpv6_options]
if should_match:
self.assertItemsEqual(db_port_ids, plugin_lport_ids)
self.assertItemsEqual(db_port_ids, monitor_lport_ids)
self.assertCountEqual(db_port_ids, plugin_lport_ids)
self.assertCountEqual(db_port_ids, monitor_lport_ids)
expected_dhcpv4_options_ports_ids = (
db_port_ids_dhcp_valid.difference(
set(self.lport_dhcpv4_disabled.keys())))
self.assertItemsEqual(expected_dhcpv4_options_ports_ids,
self.assertCountEqual(expected_dhcpv4_options_ports_ids,
plugin_lport_ids_dhcpv4_enabled)
self.assertItemsEqual(expected_dhcpv4_options_ports_ids,
self.assertCountEqual(expected_dhcpv4_options_ports_ids,
monitor_lport_ids_dhcpv4_enabled)
expected_dhcpv6_options_ports_ids = (
db_port_ids_dhcp_valid.difference(
set(self.lport_dhcpv6_disabled.keys())))
self.assertItemsEqual(expected_dhcpv6_options_ports_ids,
self.assertCountEqual(expected_dhcpv6_options_ports_ids,
plugin_lport_ids_dhcpv6_enabled)
self.assertItemsEqual(expected_dhcpv6_options_ports_ids,
self.assertCountEqual(expected_dhcpv6_options_ports_ids,
monitor_lport_ids_dhcpv6_enabled)
# Check if unknow address is set for the expected lports.
@ -969,19 +969,19 @@ class TestOvnNbSync(base.TestOVNFunctionalBase):
else:
self.assertRaises(
AssertionError, self.assertItemsEqual, db_port_ids,
AssertionError, self.assertCountEqual, db_port_ids,
plugin_lport_ids)
self.assertRaises(
AssertionError, self.assertItemsEqual, db_port_ids,
AssertionError, self.assertCountEqual, db_port_ids,
monitor_lport_ids)
self.assertRaises(
AssertionError, self.assertItemsEqual, db_port_ids,
AssertionError, self.assertCountEqual, db_port_ids,
plugin_lport_ids_dhcpv4_enabled)
self.assertRaises(
AssertionError, self.assertItemsEqual, db_port_ids,
AssertionError, self.assertCountEqual, db_port_ids,
monitor_lport_ids_dhcpv4_enabled)
@staticmethod
@ -1026,18 +1026,18 @@ class TestOvnNbSync(base.TestOVNFunctionalBase):
'options': opts})
if should_match:
self.assertItemsEqual(self.expected_dhcp_options_rows,
self.assertCountEqual(self.expected_dhcp_options_rows,
observed_plugin_dhcp_options_rows)
self.assertItemsEqual(self.expected_dhcp_options_rows,
self.assertCountEqual(self.expected_dhcp_options_rows,
observed_monitor_dhcp_options_rows)
else:
self.assertRaises(
AssertionError, self.assertItemsEqual,
AssertionError, self.assertCountEqual,
self.expected_dhcp_options_rows,
observed_plugin_dhcp_options_rows)
self.assertRaises(
AssertionError, self.assertItemsEqual,
AssertionError, self.assertCountEqual,
self.expected_dhcp_options_rows,
observed_monitor_dhcp_options_rows)
@ -1088,14 +1088,14 @@ class TestOvnNbSync(base.TestOVNFunctionalBase):
monitor_acls.append(self._build_acl_to_compare(acl))
if should_match:
self.assertItemsEqual(db_acls, plugin_acls)
self.assertItemsEqual(db_acls, monitor_acls)
self.assertCountEqual(db_acls, plugin_acls)
self.assertCountEqual(db_acls, monitor_acls)
else:
self.assertRaises(
AssertionError, self.assertItemsEqual,
AssertionError, self.assertCountEqual,
db_acls, plugin_acls)
self.assertRaises(
AssertionError, self.assertItemsEqual,
AssertionError, self.assertCountEqual,
db_acls, monitor_acls)
def _validate_routers_and_router_ports(self, should_match=True):
@ -1153,15 +1153,15 @@ class TestOvnNbSync(base.TestOVNFunctionalBase):
self.nb_api.tables['Logical_Router'].rows.values())]
if should_match:
self.assertItemsEqual(db_router_ids, plugin_lrouter_ids)
self.assertItemsEqual(db_router_ids, monitor_lrouter_ids)
self.assertCountEqual(db_router_ids, plugin_lrouter_ids)
self.assertCountEqual(db_router_ids, monitor_lrouter_ids)
else:
self.assertRaises(
AssertionError, self.assertItemsEqual, db_router_ids,
AssertionError, self.assertCountEqual, db_router_ids,
plugin_lrouter_ids)
self.assertRaises(
AssertionError, self.assertItemsEqual, db_router_ids,
AssertionError, self.assertCountEqual, db_router_ids,
monitor_lrouter_ids)
def _get_networks_for_router_port(port_fixed_ips):
@ -1247,64 +1247,64 @@ class TestOvnNbSync(base.TestOVNFunctionalBase):
monitor_nats = []
if should_match:
self.assertItemsEqual(r_port_ids, plugin_lrouter_port_ids)
self.assertItemsEqual(r_port_ids, monitor_lrouter_port_ids)
self.assertCountEqual(r_port_ids, plugin_lrouter_port_ids)
self.assertCountEqual(r_port_ids, monitor_lrouter_port_ids)
for p in plugin_lport_networks:
self.assertItemsEqual(r_port_networks[p],
self.assertCountEqual(r_port_networks[p],
plugin_lport_networks[p])
self.assertItemsEqual(r_port_ipv6_ra_configs[p],
self.assertCountEqual(r_port_ipv6_ra_configs[p],
plugin_lport_ra_configs[p])
for p in monitor_lport_networks:
self.assertItemsEqual(r_port_networks[p],
self.assertCountEqual(r_port_networks[p],
monitor_lport_networks[p])
self.assertItemsEqual(r_port_ipv6_ra_configs[p],
self.assertCountEqual(r_port_ipv6_ra_configs[p],
monitor_lport_ra_configs[p])
self.assertItemsEqual(r_routes, plugin_routes)
self.assertItemsEqual(r_routes, monitor_routes)
self.assertItemsEqual(r_nats, plugin_nats)
self.assertItemsEqual(r_nats, monitor_nats)
self.assertCountEqual(r_routes, plugin_routes)
self.assertCountEqual(r_routes, monitor_routes)
self.assertCountEqual(r_nats, plugin_nats)
self.assertCountEqual(r_nats, monitor_nats)
else:
self.assertRaises(
AssertionError, self.assertItemsEqual, r_port_ids,
AssertionError, self.assertCountEqual, r_port_ids,
plugin_lrouter_port_ids)
self.assertRaises(
AssertionError, self.assertItemsEqual, r_port_ids,
AssertionError, self.assertCountEqual, r_port_ids,
monitor_lrouter_port_ids)
for _p in self.update_lrouter_ports:
p = _p[0].replace('lrp-', '')
if p in plugin_lport_networks:
self.assertRaises(
AssertionError, self.assertItemsEqual,
AssertionError, self.assertCountEqual,
r_port_networks[p], plugin_lport_networks[p])
self.assertRaises(
AssertionError, self.assertItemsEqual,
AssertionError, self.assertCountEqual,
r_port_ipv6_ra_configs[p],
plugin_lport_ra_configs[p])
if p in monitor_lport_networks:
self.assertRaises(
AssertionError, self.assertItemsEqual,
AssertionError, self.assertCountEqual,
r_port_networks[p], monitor_lport_networks[p])
self.assertRaises(
AssertionError, self.assertItemsEqual,
AssertionError, self.assertCountEqual,
r_port_ipv6_ra_configs[p],
monitor_lport_ra_configs[p])
self.assertRaises(
AssertionError, self.assertItemsEqual, r_routes,
AssertionError, self.assertCountEqual, r_routes,
plugin_routes)
self.assertRaises(
AssertionError, self.assertItemsEqual, r_routes,
AssertionError, self.assertCountEqual, r_routes,
monitor_routes)
self.assertRaises(
AssertionError, self.assertItemsEqual, r_nats,
AssertionError, self.assertCountEqual, r_nats,
plugin_nats)
self.assertRaises(
AssertionError, self.assertItemsEqual, r_nats,
AssertionError, self.assertCountEqual, r_nats,
monitor_nats)
def _validate_port_groups(self, should_match=True):
@ -1322,12 +1322,12 @@ class TestOvnNbSync(base.TestOVNFunctionalBase):
mn_pgs.append(getattr(row, 'name', ''))
if should_match:
self.assertItemsEqual(nb_pgs, db_pgs)
self.assertItemsEqual(mn_pgs, db_pgs)
self.assertCountEqual(nb_pgs, db_pgs)
self.assertCountEqual(mn_pgs, db_pgs)
else:
self.assertRaises(AssertionError, self.assertItemsEqual,
self.assertRaises(AssertionError, self.assertCountEqual,
nb_pgs, db_pgs)
self.assertRaises(AssertionError, self.assertItemsEqual,
self.assertRaises(AssertionError, self.assertCountEqual,
mn_pgs, db_pgs)
def _delete_metadata_ports(self):
@ -1368,10 +1368,10 @@ class TestOvnNbSync(base.TestOVNFunctionalBase):
{'external_ids': dns_row.external_ids,
'records': dns_row.records})
if should_match:
self.assertItemsEqual(self.expected_dns_records,
self.assertCountEqual(self.expected_dns_records,
observed_dns_records)
else:
self.assertRaises(AssertionError, self.assertItemsEqual,
self.assertRaises(AssertionError, self.assertCountEqual,
self.expected_dns_records, observed_dns_records)
def _validate_resources(self, should_match=True):

2
neutron/tests/functional/plugins/ml2/drivers/ovn/mech_driver/test_mech_driver.py
View File

@ -642,7 +642,7 @@ class TestCreateDefaultDropPortGroup(base.BaseLoggingTestCase,
expected_ports = expected_ports or []
mech_driver.create_default_drop_port_group(self.api)
port_group = self.api.get_port_group(self.PG_NAME)
self.assertItemsEqual(
self.assertCountEqual(
expected_ports, [port.name for port in port_group.ports])
def test_with_ports_available(self):

4
neutron/tests/functional/scheduler/test_dhcp_agent_scheduler.py
View File

@ -171,7 +171,7 @@ class TestWeightScheduleNetwork(test_dhcp_sch.TestDhcpSchedulerBaseTestCase,
sorted_unscheduled_active_agents = sorted(
unscheduled_active_agents,
key=attrgetter('load'))[0:self.expected_scheduled_agent_count]
self.assertItemsEqual(
self.assertCountEqual(
(agent['id'] for agent in actual_scheduled_agents),
(agent['id'] for agent in sorted_unscheduled_active_agents))
self.assertEqual(self.expected_scheduled_agent_count,
@ -409,7 +409,7 @@ class TestAutoSchedule(test_dhcp_sch.TestDhcpSchedulerBaseTestCase,
network.Network.get_objects(self.ctx, id=hosted_net_ids)]
expected_hosted_networks = self.expected_hosted_networks[
'agent-%s' % host_index]
self.assertItemsEqual(
self.assertCountEqual(
hosted_net_names, expected_hosted_networks, msg)

8
neutron/tests/functional/services/ovn_l3/test_plugin.py
View File

@ -85,7 +85,7 @@ class TestRouter(base.TestOVNFunctionalBase):
'Logical_Router_Port'].rows.values():
if self._l3_ha_supported():
chassis = [gwc.chassis_name for gwc in row.gateway_chassis]
self.assertItemsEqual(expected, chassis)
self.assertCountEqual(expected, chassis)
else:
rc = row.options.get(ovn_const.OVN_GATEWAY_CHASSIS_KEY)
self.assertIn(rc, expected)
@ -100,7 +100,7 @@ class TestRouter(base.TestOVNFunctionalBase):
# candidates.
def fake_select(*args, **kwargs):
self.assertItemsEqual(candidates, kwargs['candidates'])
self.assertCountEqual(candidates, kwargs['candidates'])
# We are not interested in further processing, let us return
# INVALID_CHASSIS to avoid erros
return [ovn_const.OVN_GATEWAY_INVALID_CHASSIS]
@ -289,7 +289,7 @@ class TestRouter(base.TestOVNFunctionalBase):
self.assertIsNotNone(gw_port)
expected_networks = ['%s/24' % subnet4_ip, '%s/64' % subnet6_ip]
self.assertItemsEqual(
self.assertCountEqual(
expected_networks, gw_port.networks,
'networks in ovn port must match fixed_ips in neutron')
@ -341,7 +341,7 @@ class TestRouter(base.TestOVNFunctionalBase):
self.candidates = []
def fake_select(*args, **kwargs):
self.assertItemsEqual(self.candidates, kwargs['candidates'])
self.assertCountEqual(self.candidates, kwargs['candidates'])
# We are not interested in further processing, let us return
# INVALID_CHASSIS to avoid erros
return [ovn_const.OVN_GATEWAY_INVALID_CHASSIS]

2
neutron/tests/functional/services/trunk/drivers/ovn/test_trunk_driver.py
View File

@ -77,7 +77,7 @@ class TestOVNTrunkDriver(base.TestOVNFunctionalBase):
self.context, port_id=subport['port_id'], host='')
self.assertEqual(n_consts.PORT_STATUS_ACTIVE, binding['status'])
self.assertItemsEqual(ovn_subports_info, neutron_subports_info)
self.assertCountEqual(ovn_subports_info, neutron_subports_info)
self.assertEqual(has_items, len(neutron_subports_info) != 0)
if trunk.get('status'):

2
neutron/tests/unit/agent/linux/openvswitch_firewall/test_iptables.py
View File

@ -35,7 +35,7 @@ class TestHelper(base.BaseTestCase):
self.helper.int_br.get_port_name_list.return_value = present_ports
expected_hybrid_ports = ['qvo-1234', 'qvo-fghfhfh']
observed = self.helper.get_hybrid_ports()
self.assertItemsEqual(expected_hybrid_ports, observed)
self.assertCountEqual(expected_hybrid_ports, observed)
def test_has_not_been_cleaned_no_value(self):
other_config = {'foo': 'bar'}

2
neutron/tests/unit/agent/linux/openvswitch_firewall/test_rules.py
View File

@ -451,7 +451,7 @@ class TestMergeRules(base.BaseTestCase):
({'direction': 'ingress', 'ethertype': 'IPv4',
'protocol': 1}, 24)]
result = rules.merge_common_rules(rule_conj_list)
self.assertItemsEqual(
self.assertCountEqual(
[({'direction': 'ingress', 'ethertype': 'IPv4',
'protocol': 1}, [8, 24]),
({'direction': 'ingress', 'ethertype': 'IPv4',

2
neutron/tests/unit/agent/linux/test_dhcp.py
View File

@ -2745,7 +2745,7 @@ class TestDnsmasq(TestBase):
result = dhcp.Dnsmasq.existing_dhcp_networks(self.conf)
mock_listdir.assert_called_once_with(path)
self.assertItemsEqual(['aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa',
self.assertCountEqual(['aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa',
'bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb'],
result)

8
neutron/tests/unit/agent/test_resource_cache.py
View File

@ -86,7 +86,7 @@ class RemoteResourceCacheTestCase(base.BaseTestCase):
self._pullmock.bulk_pull.assert_called_once_with(
mock.ANY, 'goose', filter_kwargs={'id': (67, )})
self.assertItemsEqual(
self.assertCountEqual(
resources, [rec['updated'] for rec in received_kw])
def test_bulk_pull_doesnt_wipe_out_newer_data(self):
@ -106,9 +106,9 @@ class RemoteResourceCacheTestCase(base.BaseTestCase):
self.rcache.record_resource_update(self.ctx, 'goose', goose)
is_large = {'size': ('large', )}
is_small = {'size': ('small', )}
self.assertItemsEqual([geese[0], geese[2]],
self.assertCountEqual([geese[0], geese[2]],
self.rcache.get_resources('goose', is_large))
self.assertItemsEqual([geese[3]],
self.assertCountEqual([geese[3]],
self.rcache.get_resources('goose', is_small))
def test_match_resources_with_func(self):
@ -117,7 +117,7 @@ class RemoteResourceCacheTestCase(base.BaseTestCase):
for goose in geese:
self.rcache.record_resource_update(self.ctx, 'goose', goose)
has_large = lambda o: 'large' in o.size
self.assertItemsEqual([geese[0], geese[2]],
self.assertCountEqual([geese[0], geese[2]],
self.rcache.match_resources_with_func('goose',
has_large))

2
neutron/tests/unit/api/rpc/handlers/test_resources_rpc.py
View File

@ -231,7 +231,7 @@ class ResourcesPullRpcCallbackTestCase(ResourcesRpcBaseTestCase):
objs = self.callbacks.bulk_pull(
self.context, resource_type=FakeResource.obj_name(),
version=TEST_VERSION)
self.assertItemsEqual([r1.obj_to_primitive(),
self.assertCountEqual([r1.obj_to_primitive(),
r2.obj_to_primitive()],
objs)
objs = self.callbacks.bulk_pull(

2
neutron/tests/unit/common/test_utils.py
View File

@ -332,7 +332,7 @@ class TestPortRuleMasking(base.BaseTestCase):
def compare_port_ranges_results(self, port_min, port_max):
observed = utils.port_rule_masking(port_min, port_max)
expected = _port_rule_masking(port_min, port_max)
self.assertItemsEqual(expected, observed)
self.assertCountEqual(expected, observed)
def test_port_rule_masking_random_ranges(self):
# calling randint a bunch of times is really slow

6
neutron/tests/unit/db/test_db_base_plugin_v2.py
View File

@ -663,7 +663,7 @@ class NeutronDbPluginV2TestCase(testlib_api.WebTestCase):
expected_code=expected_code)
if expected_code == webob.exc.HTTPOk.code:
resource = resource.replace('-', '_')
self.assertItemsEqual([i['id'] for i in res['%ss' % resource]],
self.assertCountEqual([i['id'] for i in res['%ss' % resource]],
[i[resource]['id'] for i in items])
@contextlib.contextmanager
@ -6030,7 +6030,7 @@ class TestSubnetPoolsV2(NeutronDbPluginV2TestCase):
initial_subnetpool['subnetpool']['id'])
api = self._api_for_resource('subnetpools')
res = self.deserialize(self.fmt, req.get_response(api))
self.assertItemsEqual(res['subnetpool']['prefixes'],
self.assertCountEqual(res['subnetpool']['prefixes'],
['10.10.8.0/21', '3.3.3.0/24', '2.2.2.0/24'])
def test_update_subnetpool_prefix_list_compaction(self):
@ -6045,7 +6045,7 @@ class TestSubnetPoolsV2(NeutronDbPluginV2TestCase):
initial_subnetpool['subnetpool']['id'])
api = self._api_for_resource('subnetpools')
res = self.deserialize(self.fmt, req.get_response(api))
self.assertItemsEqual(res['subnetpool']['prefixes'],
self.assertCountEqual(res['subnetpool']['prefixes'],
['10.10.10.0/23'])
def test_illegal_subnetpool_prefix_list_update(self):

2
neutron/tests/unit/db/test_dvr_mac_db.py
View File

@ -210,5 +210,5 @@ class DvrDbMixinTestCase(test_plugin.Ml2PluginV2TestCase):
dvr_ports = self.mixin.get_ports_on_host_by_subnet(
self.ctx, HOST, subnet['subnet']['id'])
self.assertEqual(len(expected_ids), len(dvr_ports))
self.assertItemsEqual(expected_ids,
self.assertCountEqual(expected_ids,
[port['id'] for port in dvr_ports])

6
neutron/tests/unit/db/test_extraroute_db.py
View File

@ -45,7 +45,7 @@ class TestExtraRouteDb(testlib_api.SqlTestCase):
}
}
router = self._plugin.create_router(ctx, create_request)
self.assertItemsEqual(router['routes'], [])
self.assertCountEqual(router['routes'], [])
router_id = router['id']
routes = [
{'destination': '10.0.0.0/24', 'nexthop': '1.1.1.4'},
@ -71,9 +71,9 @@ class TestExtraRouteDb(testlib_api.SqlTestCase):
update_request)
mock_cb.assert_called_with('router', events.PRECOMMIT_UPDATE,
self._plugin, payload=mock.ANY)
self.assertItemsEqual(updated_router['routes'], routes)
self.assertCountEqual(updated_router['routes'], routes)
got_router = self._plugin.get_router(ctx, router_id)
self.assertItemsEqual(got_router['routes'], routes)
self.assertCountEqual(got_router['routes'], routes)
def assertEqualRoutes(self, a, b):
"""Compare a list of routes without caring for the list order."""

6
neutron/tests/unit/db/test_ipam_backend_mixin.py
View File

@ -94,9 +94,9 @@ class TestIpamBackendMixin(base.BaseTestCase):
new_ips,
owner)
self.assertItemsEqual(expected.add, change.add)
self.assertItemsEqual(expected.original, change.original)
self.assertItemsEqual(expected.remove, change.remove)
self.assertCountEqual(expected.add, change.add)
self.assertCountEqual(expected.original, change.original)
self.assertCountEqual(expected.remove, change.remove)
def test__get_changed_ips_for_port(self):
new_ips = self._prepare_ips(self.default_new_ips)

2
neutron/tests/unit/db/test_l3_dvr_db.py
View File

@ -1346,7 +1346,7 @@ class L3DvrTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
dvr_subnet_ports = self.mixin.get_ports_under_dvr_connected_subnet(
self.ctx, subnet['subnet']['id'])
dvr_subnet_ports_ids = [p['id'] for p in dvr_subnet_ports]
self.assertItemsEqual(fake_bound_ports_ids, dvr_subnet_ports_ids)
self.assertCountEqual(fake_bound_ports_ids, dvr_subnet_ports_ids)
@mock.patch.object(plugin_utils, 'can_port_be_bound_to_virtual_bridge',
return_value=True)

18
neutron/tests/unit/extensions/test_availability_zone.py
View File

@ -72,12 +72,12 @@ class TestAZAgentCase(AZTestCommon):
{'name': 'nova3', 'resource': 'router', 'state': 'unavailable'}]
res = self._list('availability_zones')
azs = res['availability_zones']
self.assertItemsEqual(expected, azs)
self.assertCountEqual(expected, azs)
# not admin case
ctx = context.Context('', 'noadmin')
res = self._list('availability_zones', neutron_context=ctx)
azs = res['availability_zones']
self.assertItemsEqual(expected, azs)
self.assertCountEqual(expected, azs)
def test_list_availability_zones_with_filter(self):
self._register_azs()
@ -90,27 +90,27 @@ class TestAZAgentCase(AZTestCommon):
{'name': 'nova3', 'resource': 'router', 'state': 'unavailable'}]
res = self._list('availability_zones')
azs = res['availability_zones']
self.assertItemsEqual(expected, azs)
self.assertCountEqual(expected, azs)
# list with filter of 'name'
res = self._list('availability_zones',
query_params="name=nova1")
azs = res['availability_zones']
self.assertItemsEqual(expected[:1], azs)
self.assertCountEqual(expected[:1], azs)
# list with filter of 'resource'
res = self._list('availability_zones',
query_params="resource=router")
azs = res['availability_zones']
self.assertItemsEqual(expected[-2:], azs)
self.assertCountEqual(expected[-2:], azs)
# list with filter of 'state' as 'available'
res = self._list('availability_zones',
query_params="state=available")
azs = res['availability_zones']
self.assertItemsEqual(expected[:3], azs)
self.assertCountEqual(expected[:3], azs)
# list with filter of 'state' as 'unavailable'
res = self._list('availability_zones',
query_params="state=unavailable")
azs = res['availability_zones']
self.assertItemsEqual(expected[-1:], azs)
self.assertCountEqual(expected[-1:], azs)
def test_list_agent_with_az(self):
helpers.register_dhcp_agent(host='host1', az='nova1')
@ -145,7 +145,7 @@ class TestAZNetworkCase(AZTestCommon):
az_hints = ['nova1']
with self.network(availability_zone_hints=az_hints) as net:
res = self._show('networks', net['network']['id'])
self.assertItemsEqual(az_hints,
self.assertCountEqual(az_hints,
res['network']['availability_zone_hints'])
def test_create_network_with_azs(self):
@ -153,7 +153,7 @@ class TestAZNetworkCase(AZTestCommon):
az_hints = ['nova1', 'nova2']
with self.network(availability_zone_hints=az_hints) as net:
res = self._show('networks', net['network']['id'])
self.assertItemsEqual(az_hints,
self.assertCountEqual(az_hints,
res['network']['availability_zone_hints'])
def test_create_network_without_az(self):

2
neutron/tests/unit/extensions/test_dns.py
View File

@ -115,7 +115,7 @@ class DnsExtensionTestCase(test_plugin.Ml2PluginV2TestCase):
neutron_context=neutron_context,
query_params=query_params)
resource = resource.replace('-', '_')
self.assertItemsEqual([i['id'] for i in res['%ss' % resource]],
self.assertCountEqual([i['id'] for i in res['%ss' % resource]],
[i[resource]['id'] for i in items])
return res

4
neutron/tests/unit/extensions/test_router_availability_zone.py
View File

@ -63,7 +63,7 @@ class TestAZRouterCase(test_az.AZTestCommon, test_l3.L3NatTestCaseMixin):
az_hints = ['nova2']
with self.router(availability_zone_hints=az_hints) as router:
res = self._show('routers', router['router']['id'])
self.assertItemsEqual(az_hints,
self.assertCountEqual(az_hints,
res['router']['availability_zone_hints'])
def test_create_router_with_azs(self):
@ -71,7 +71,7 @@ class TestAZRouterCase(test_az.AZTestCommon, test_l3.L3NatTestCaseMixin):
az_hints = ['nova2', 'nova3']
with self.router(availability_zone_hints=az_hints) as router:
res = self._show('routers', router['router']['id'])
self.assertItemsEqual(az_hints,
self.assertCountEqual(az_hints,
res['router']['availability_zone_hints'])
def test_create_router_without_az(self):

2
neutron/tests/unit/objects/qos/test_policy.py
View File

@ -113,7 +113,7 @@ class QosPolicyObjectTestCase(test_base.BaseObjectIfaceTestCase):
objs = self._test_class.get_objects(self.context)
self.get_objects_mock.assert_any_call(
self._test_class, self.context, _pager=None)
self.assertItemsEqual(
self.assertCountEqual(
[test_base.get_obj_persistent_fields(obj) for obj in self.objs],
[test_base.get_obj_persistent_fields(obj) for obj in objs])

12
neutron/tests/unit/objects/test_base.py
View File

@ -839,7 +839,7 @@ class BaseObjectIfaceTestCase(_BaseObjectTestCase, test_base.BaseTestCase):
obj_db_api, 'get_objects',
side_effect=self.fake_get_objects) as get_objects_mock:
objs = self._test_class.get_objects(self.context)
self.assertItemsEqual(
self.assertCountEqual(
[get_obj_persistent_fields(obj) for obj in self.objs],
[get_obj_persistent_fields(obj) for obj in objs])
get_objects_mock.assert_any_call(
@ -903,7 +903,7 @@ class BaseObjectIfaceTestCase(_BaseObjectTestCase, test_base.BaseTestCase):
objs = self._test_class.get_objects(self.context,
validate_filters=False,
unknown_filter='value')
self.assertItemsEqual(
self.assertCountEqual(
[get_obj_persistent_fields(obj) for obj in self.objs],
[get_obj_persistent_fields(obj) for obj in objs])
@ -914,7 +914,7 @@ class BaseObjectIfaceTestCase(_BaseObjectTestCase, test_base.BaseTestCase):
obj_db_api, 'get_values',
side_effect=self.fake_get_values) as get_values_mock:
values = self._test_class.get_values(self.context, field)
self.assertItemsEqual(
self.assertCountEqual(
[getattr(obj, field) for obj in self.objs], values)
get_values_mock.assert_any_call(
self._test_class, self.context, db_field
@ -935,7 +935,7 @@ class BaseObjectIfaceTestCase(_BaseObjectTestCase, test_base.BaseTestCase):
values = self._test_class.get_values(self.context, field,
validate_filters=False,
unknown_filter='value')
self.assertItemsEqual(
self.assertCountEqual(
[getattr(obj, field) for obj in self.objs], values)
def test_get_values_mixed_field(self):
@ -1079,7 +1079,7 @@ class BaseObjectIfaceTestCase(_BaseObjectTestCase, test_base.BaseTestCase):
fake_field='xxx')
def _check_equal(self, expected, observed):
self.assertItemsEqual(get_obj_persistent_fields(expected),
self.assertCountEqual(get_obj_persistent_fields(expected),
get_obj_persistent_fields(observed))
def test_count_validate_filters_false(self):
@ -1840,7 +1840,7 @@ class BaseDbObjectTestCase(_BaseObjectTestCase,
else:
filters = {field: obj[field]}
new = self._test_class.get_objects(self.context, **filters)
self.assertItemsEqual(
self.assertCountEqual(
[obj._get_composite_keys()],
[obj_._get_composite_keys() for obj_ in new],
'Filtering by %s failed.' % field)

2
neutron/tests/unit/objects/test_network_segment_range.py
View File

@ -136,7 +136,7 @@ class NetworkSegmentRangeDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
physical_network='foo')
obj = self._create_network_segment_range(range_minimum, range_maximum)
available_alloc = self._test_class._get_available_allocation(obj)
self.assertItemsEqual(not_to_alloc, available_alloc)
self.assertCountEqual(not_to_alloc, available_alloc)
def test__get_used_allocation_mapping(self):
alloc_mapping = {}

4
neutron/tests/unit/objects/test_subnetpool.py
View File

@ -63,14 +63,14 @@ class SubnetPoolDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
pool.update()
new_pool = self._test_class.get_object(self.context, id=pool.id)
self.assertItemsEqual(prefixes, new_pool.prefixes)
self.assertCountEqual(prefixes, new_pool.prefixes)
prefixes.pop()
pool.prefixes = prefixes
pool.update()
new_pool = self._test_class.get_object(self.context, id=pool.id)
self.assertItemsEqual(prefixes, new_pool.prefixes)
self.assertCountEqual(prefixes, new_pool.prefixes)
def test_get_objects_queries_constant(self):
# TODO(korzen) SubnetPool is using SubnetPoolPrefix object to reload

4
neutron/tests/unit/plugins/ml2/drivers/openvswitch/agent/test_vlanmanager.py
View File

@ -60,7 +60,7 @@ class TestLocalVlanManager(base.BaseTestCase):
self.vlan_manager.add(1, None, None, None, None)
new_vlan_manager = vlanmanager.LocalVlanManager()
self.assertIs(new_vlan_manager, self.vlan_manager)
self.assertItemsEqual(new_vlan_manager.mapping,
self.assertCountEqual(new_vlan_manager.mapping,
self.vlan_manager.mapping)
def test_in_operator_on_key(self):
@ -74,7 +74,7 @@ class TestLocalVlanManager(base.BaseTestCase):
self.vlan_manager.add(val, val, val, val, val)
created_vlans.append(self.vlan_manager.get(val))
self.assertItemsEqual(created_vlans, list(self.vlan_manager))
self.assertCountEqual(created_vlans, list(self.vlan_manager))
def test_get_net_uuid_existing(self):
port_id = 'port-id'

44
neutron/tests/unit/plugins/ml2/drivers/ovn/mech_driver/ovsdb/test_impl_idl_ovn.py
View File

@ -422,7 +422,7 @@ class TestNBImplIdlOvn(TestDBImplIdlOvn):
def test_get_all_logical_switches_with_ports(self):
# Test empty
mapping = self.nb_ovn_idl.get_all_logical_switches_with_ports()
self.assertItemsEqual(mapping, {})
self.assertCountEqual(mapping, {})
# Test loaded values
self._load_nb_db()
mapping = self.nb_ovn_idl.get_all_logical_switches_with_ports()
@ -440,12 +440,12 @@ class TestNBImplIdlOvn(TestDBImplIdlOvn):
'ports': ['lsp-id-51', 'lsp-id-52', 'lsp-rp-id-5',
'lsp-vpn-id-5'],
'provnet_ports': []}]
self.assertItemsEqual(mapping, expected)
self.assertCountEqual(mapping, expected)
def test_get_all_logical_routers_with_rports(self):
# Test empty
mapping = self.nb_ovn_idl.get_all_logical_switches_with_ports()
self.assertItemsEqual(mapping, {})
self.assertCountEqual(mapping, {})
# Test loaded values
self._load_nb_db()
mapping = self.nb_ovn_idl.get_all_logical_routers_with_rports()
@ -481,7 +481,7 @@ class TestNBImplIdlOvn(TestDBImplIdlOvn):
'snats': [], 'dnat_and_snats': []},
{'name': 'lr-id-e', 'ports': {}, 'static_routes': [],
'snats': [], 'dnat_and_snats': []}]
self.assertItemsEqual(mapping, expected)
self.assertCountEqual(mapping, expected)
def test_get_acls_for_lswitches(self):
self._load_nb_db()
@ -538,7 +538,7 @@ class TestNBImplIdlOvn(TestDBImplIdlOvn):
'direction': 'to-lport',
'match': 'outport == "lsp-id-52" && ip4.src == $as_ip4_id_5'}
]}
self.assertItemsEqual(acl_values, excepted_acl_values)
self.assertCountEqual(acl_values, excepted_acl_values)
self.assertEqual(len(acl_objs), 8)
self.assertEqual(len(lswitch_ovsdb_dict), len(lswitches))
@ -546,7 +546,7 @@ class TestNBImplIdlOvn(TestDBImplIdlOvn):
lswitches = ['ls-id-4']
acl_values, acl_objs, lswitch_ovsdb_dict = \
self.nb_ovn_idl.get_acls_for_lswitches(lswitches)
self.assertItemsEqual(acl_values, {})
self.assertCountEqual(acl_values, {})
self.assertEqual(len(acl_objs), 0)
self.assertEqual(len(lswitch_ovsdb_dict), 0)
@ -558,15 +558,15 @@ class TestNBImplIdlOvn(TestDBImplIdlOvn):
'host-2': [utils.ovn_lrouter_port_name('orp-id-b2')],
ovn_const.OVN_GATEWAY_INVALID_CHASSIS: [
utils.ovn_name('orp-id-a3')]}
self.assertItemsEqual(bindings, expected)
self.assertCountEqual(bindings, expected)
bindings = self.nb_ovn_idl.get_all_chassis_gateway_bindings([])
self.assertItemsEqual(bindings, expected)
self.assertCountEqual(bindings, expected)
bindings = self.nb_ovn_idl.get_all_chassis_gateway_bindings(['host-1'])
expected = {'host-1': [utils.ovn_lrouter_port_name('orp-id-a1'),
utils.ovn_lrouter_port_name('orp-id-a2')]}
self.assertItemsEqual(bindings, expected)
self.assertCountEqual(bindings, expected)
def test_get_gateway_chassis_binding(self):
self._load_nb_db()
@ -600,13 +600,13 @@ class TestNBImplIdlOvn(TestDBImplIdlOvn):
port_physnet_dict, {'host-1': 'physnet1', 'host-2': 'physnet3'},
['host-1', 'host-2'])
expected = ['lrp-orp-id-a3']
self.assertItemsEqual(unhosted_gateways, expected)
self.assertCountEqual(unhosted_gateways, expected)
# Test both host-1, host-2 in valid list
unhosted_gateways = self.nb_ovn_idl.get_unhosted_gateways(
port_physnet_dict, {'host-1': 'physnet1', 'host-2': 'physnet2'},
['host-1', 'host-2'])
expected = ['lrp-orp-id-a3', 'lrp-orp-id-b6']
self.assertItemsEqual(unhosted_gateways, expected)
self.assertCountEqual(unhosted_gateways, expected)
def test_get_unhosted_gateways_deleted_physnet(self):
self._load_nb_db()
@ -622,12 +622,12 @@ class TestNBImplIdlOvn(TestDBImplIdlOvn):
['host-1', 'host-2'])
# Make sure that lrp is rescheduled, because host-1 has physet1
expected = ['lrp-orp-id-a1']
self.assertItemsEqual(unhosted_gateways, expected)
self.assertCountEqual(unhosted_gateways, expected)
# Spoof that there is no valid host with required physnet.
unhosted_gateways = self.nb_ovn_idl.get_unhosted_gateways(
port_physnet_dict, {'host-1': 'physnet4', 'host-2': 'physnet3'},
['host-1', 'host-2'])
self.assertItemsEqual(unhosted_gateways, [])
self.assertCountEqual(unhosted_gateways, [])
def _test_get_unhosted_gateway_max_chassis(self, r):
gw_chassis_table = fakes.FakeOvsdbTable.create_one_ovsdb_table()
@ -650,7 +650,7 @@ class TestNBImplIdlOvn(TestDBImplIdlOvn):
['host-%s' % x for x in range(1, 7)])
# We don't have required number of chassis
expected = []
self.assertItemsEqual(unhosted_gateways, expected)
self.assertCountEqual(unhosted_gateways, expected)
def test_get_unhosted_gateway_max_chassis(self):
# We have required number of chassis, and lrp
@ -663,7 +663,7 @@ class TestNBImplIdlOvn(TestDBImplIdlOvn):
'host-5': 'physnet1', 'host-6': 'physnet1'},
['host-%s' % x for x in range(1, 7)])
expected = []
self.assertItemsEqual(unhosted_gateways, expected)
self.assertCountEqual(unhosted_gateways, expected)
def test_get_unhosed_gateway_schedule_to_max(self):
# The LRP is not yet scheduled on all chassis
@ -676,7 +676,7 @@ class TestNBImplIdlOvn(TestDBImplIdlOvn):
'host-5': 'physnet1', 'host-6': 'physnet1'},
['host-%s' % x for x in range(1, 7)])
expected = ['lrp-orp-id-a1']
self.assertItemsEqual(unhosted_gateways, expected)
self.assertCountEqual(unhosted_gateways, expected)
def test_get_subnet_dhcp_options(self):
self._load_nb_db()
@ -701,7 +701,7 @@ class TestNBImplIdlOvn(TestDBImplIdlOvn):
# Test empty
subnet_options = self.nb_ovn_idl.get_subnet_dhcp_options(
'subnet-id-10-0-1-0', with_ports=True)
self.assertItemsEqual({'subnet': None, 'ports': []}, subnet_options)
self.assertCountEqual({'subnet': None, 'ports': []}, subnet_options)
# Test loaded values
self._load_nb_db()
# Test getting both subnet and port dhcp options
@ -714,7 +714,7 @@ class TestNBImplIdlOvn(TestDBImplIdlOvn):
'external_ids': dhcp_row.external_ids,
'options': dhcp_row.options,
'uuid': dhcp_row.uuid} for dhcp_row in dhcp_rows]
self.assertItemsEqual(expected_rows, [
self.assertCountEqual(expected_rows, [
subnet_options['subnet']] + subnet_options['ports'])
# Test getting only subnet dhcp options
subnet_options = self.nb_ovn_idl.get_subnet_dhcp_options(
@ -725,12 +725,12 @@ class TestNBImplIdlOvn(TestDBImplIdlOvn):
'external_ids': dhcp_row.external_ids,
'options': dhcp_row.options,
'uuid': dhcp_row.uuid} for dhcp_row in dhcp_rows]
self.assertItemsEqual(expected_rows, [
self.assertCountEqual(expected_rows, [
subnet_options['subnet']] + subnet_options['ports'])
# Test getting no dhcp options
subnet_options = self.nb_ovn_idl.get_subnet_dhcp_options(
'subnet-id-11-0-2-0', with_ports=True)
self.assertItemsEqual({'subnet': None, 'ports': []}, subnet_options)
self.assertCountEqual({'subnet': None, 'ports': []}, subnet_options)
def test_get_subnets_dhcp_options(self):
self._load_nb_db()
@ -745,13 +745,13 @@ class TestNBImplIdlOvn(TestDBImplIdlOvn):
get_row_dict(
self._find_ovsdb_fake_row(self.dhcp_table, 'cidr', cidr))
for cidr in ('10.0.1.0/24', '10.0.2.0/24')]
self.assertItemsEqual(expected_rows, subnets_options)
self.assertCountEqual(expected_rows, subnets_options)
subnets_options = self.nb_ovn_idl.get_subnets_dhcp_options(
['subnet-id-11-0-2-0', 'subnet-id-20-0-1-0'])
expected_row = get_row_dict(
self._find_ovsdb_fake_row(self.dhcp_table, 'cidr', '20.0.1.0/24'))
self.assertItemsEqual([expected_row], subnets_options)
self.assertCountEqual([expected_row], subnets_options)
subnets_options = self.nb_ovn_idl.get_subnets_dhcp_options(
['port-id-30-0-1-0', 'fake-not-exist'])

8
neutron/tests/unit/plugins/ml2/drivers/ovn/mech_driver/ovsdb/test_ovsdb_monitor.py
View File

@ -107,20 +107,20 @@ class TestOvnDbNotifyHandler(base.BaseTestCase):
ovn_event = mock.Mock()
unknown_event = mock.Mock()
self.assertItemsEqual(set(), self.watched_events)
self.assertCountEqual(set(), self.watched_events)
expected_events.add(networking_event)
self.handler.watch_event(networking_event)
self.assertItemsEqual(expected_events, self.watched_events)
self.assertCountEqual(expected_events, self.watched_events)
expected_events.add(ovn_event)
self.handler.watch_events([ovn_event])
self.assertItemsEqual(expected_events, self.watched_events)
self.assertCountEqual(expected_events, self.watched_events)
self.handler.unwatch_events([networking_event, ovn_event])
self.handler.unwatch_event(unknown_event)
self.handler.unwatch_events([unknown_event])
self.assertItemsEqual(set(), self.watched_events)
self.assertCountEqual(set(), self.watched_events)
def test_shutdown(self):
self.handler.shutdown()

2
neutron/tests/unit/scheduler/test_l3_agent_scheduler.py
View File

@ -2053,7 +2053,7 @@ class TestGetL3AgentsWithAgentModeFilter(TestGetL3AgentsWithFilter):
self.assertEqual(len(self.expected_agent_modes), len(l3_agents))
returned_agent_modes = [self._get_agent_mode(agent)
for agent in l3_agents]
self.assertItemsEqual(self.expected_agent_modes, returned_agent_modes)
self.assertCountEqual(self.expected_agent_modes, returned_agent_modes)
class TestGetL3AgentsWithHostFilter(TestGetL3AgentsWithFilter):

6
neutron/tests/unit/scheduler/test_l3_ovn_scheduler.py
View File

@ -119,7 +119,7 @@ class OVNGatewayChanceScheduler(TestOVNGatewayScheduler):
mapping = self.fake_chassis_gateway_mappings['Multiple1']
gateway_name = self.new_gateway_name
chassis = self.select(mapping, gateway_name)
self.assertItemsEqual(chassis, mapping.get('Chassis'))
self.assertCountEqual(chassis, mapping.get('Chassis'))
def test_filter_existing_chassis(self):
# filter_existing_chassis is scheduler independent, but calling
@ -182,7 +182,7 @@ class OVNGatewayLeastLoadedScheduler(TestOVNGatewayScheduler):
mapping = self.fake_chassis_gateway_mappings['Multiple1']
gateway_name = self.new_gateway_name
chassis = self.select(mapping, gateway_name)
self.assertItemsEqual(chassis, mapping.get('Chassis'))
self.assertCountEqual(chassis, mapping.get('Chassis'))
# least loaded will be the first one in the list,
# networking-ovn will assign highest priority to this first element
self.assertEqual(['hv5', 'hv4', 'hv3', 'hv2', 'hv1'], chassis)
@ -222,7 +222,7 @@ class OVNGatewayLeastLoadedScheduler(TestOVNGatewayScheduler):
chassis_info.append(('lrp', 2))
actual = self.l3_scheduler._get_chassis_load_by_prios(chassis_info)
expected = {1: 5, 2: 5}
self.assertItemsEqual(expected.items(), actual)
self.assertCountEqual(expected.items(), actual)
def test__get_chassis_load_by_prios_no_ports(self):
self.assertFalse(self.l3_scheduler._get_chassis_load_by_prios([]))

2
neutron/tests/unit/services/ovn_l3/test_plugin.py
View File

@ -349,7 +349,7 @@ class TestOVNL3RouterPlugin(test_mech_driver.Ml2PluginV2TestCase):
self.l3_inst._ovn.update_lrouter_port.call_args_list[0][1])
self.assertEqual(1, self.l3_inst._ovn.update_lrouter_port.call_count)
self.assertItemsEqual(fake_rtr_intf_networks,
self.assertCountEqual(fake_rtr_intf_networks,
called_args_dict.get('networks', []))
self.l3_inst._ovn.set_lrouter_port_in_lswitch_port.\
assert_called_once_with(

2
neutron/tests/unit/services/trunk/rpc/test_server.py
View File

@ -67,7 +67,7 @@ class TrunkSkeletonTest(test_plugin.Ml2PluginV2TestCase):
self.mock_registry_provide.assert_called_with(
server.trunk_by_port_provider,
resources.TRUNK)
self.assertItemsEqual(('trunk', [test_obj],),
self.assertCountEqual(('trunk', [test_obj],),
mock_conn.mock_calls[1][1])
def test_update_subport_bindings(self):

2
neutron/tests/unit/tests/test_base.py
View File

@ -70,7 +70,7 @@ class SystemExitTestCase(base.DietTestCase):
self.fail('SystemExit escaped!')
self.assertEqual([], result.errors)
self.assertItemsEqual(set(id(t) for t in expectedFails),
self.assertCountEqual(set(id(t) for t in expectedFails),
set(id(t) for (t, traceback) in result.failures))