openstack
/
neutron
Code Issues Proposed changes

Merge "Use more specific asserts in tests"

This commit is contained in:
Jenkins 2016-10-11 11:41:55 +00:00 committed by Gerrit Code Review
parent 316e7a1fd1 7deba33bf3
commit 65a3b89116
15 changed files with 52 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
neutron/tests/functional/agent/ovsdb/test_impl_idl.py
View File

@ -56,7 +56,7 @@ class ImplIdlTestCase(base.BaseSudoTestCase):
self._add_br()
ofport = self.ovs.db_get_val("Interface", self.brname, "ofport")
self.assertTrue(int(ofport))
self.assertTrue(ofport > -1)
self.assertGreater(ofport, -1)
def test_post_commit_vswitchd_completed_no_failures(self):
self._add_br_and_test()

13
neutron/tests/functional/agent/test_ovs_flows.py
View File

@ -381,9 +381,9 @@ class OVSFlowTestCase(OVSAgentTestBase):
"nw_proto=1,nw_tos=0,nw_ttl=128,"
"icmp_type=8,icmp_code=0,"
"dl_vlan=%(vlan_tag)d" % kwargs)
self.assertTrue("vlan_tci=0x0000" in trace["Final flow"])
self.assertTrue(("dl_src=%(gateway_mac)s" % kwargs) in
trace["Final flow"])
self.assertIn("vlan_tci=0x0000", trace["Final flow"])
self.assertIn(("dl_src=%(gateway_mac)s" % kwargs),
trace["Final flow"])
def test_install_flood_to_tun(self):
attrs = {
@ -401,12 +401,11 @@ class OVSFlowTestCase(OVSAgentTestBase):
"nw_tos=0,nw_ttl=128,icmp_type=8,icmp_code=0,"
"dl_vlan=%(vlan)d,dl_vlan_pcp=0" % kwargs)
trace = self._run_trace(self.tun_br.br_name, test_packet)
self.assertTrue(("tun_id=0x%(tun_id)x" % kwargs) in
trace["Final flow"])
self.assertTrue("vlan_tci=0x0000," in trace["Final flow"])
self.assertIn(("tun_id=0x%(tun_id)x" % kwargs), trace["Final flow"])
self.assertIn("vlan_tci=0x0000,", trace["Final flow"])
self.br_tun.delete_flood_to_tun(kwargs['vlan'])
trace = self._run_trace(self.tun_br.br_name, test_packet)
self.assertEqual(" unchanged", trace["Final flow"])
self.assertTrue("drop" in trace["Datapath actions"])
self.assertIn("drop", trace["Datapath actions"])

4
neutron/tests/functional/db/test_migrations.py
View File

@ -334,8 +334,8 @@ class TestModelsMigrationsMysql(testlib_api.MySQLTestCaseMixin,
insp = sqlalchemy.engine.reflection.Inspector.from_engine(engine)
# Test that table creation on MySQL only builds InnoDB tables
tables = insp.get_table_names()
self.assertTrue(len(tables) > 0,
"No tables found. Wrong schema?")
self.assertGreater(len(tables), 0,
"No tables found. Wrong schema?")
res = [table for table in tables if
insp.get_table_options(table)['mysql_engine'] != 'InnoDB'
and table != 'alembic_version']

4
neutron/tests/tempest/api/base.py
View File

@ -594,7 +594,7 @@ class BaseSearchCriteriaTest(BaseNetworkTest):
}
body = self.list_method(**pagination_args)
resources = self._extract_resources(body)
self.assertTrue(len(resources) >= len(self.resource_names))
self.assertGreaterEqual(len(resources), len(self.resource_names))
def _test_list_pagination_iteratively(self, lister):
# first, collect all resources for later comparison
@ -709,7 +709,7 @@ class BaseSearchCriteriaTest(BaseNetworkTest):
self.plural_name, uri
)
resources_ = self._extract_resources(body)
self.assertTrue(page_size >= len(resources_))
self.assertGreaterEqual(page_size, len(resources_))
resources.extend(reversed(resources_))
self.assertSameOrder(expected_resources, reversed(resources))

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

@ -3009,7 +3009,7 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
fip = self._make_floatingip(self.fmt, n['network']['id'])
fip_set = netaddr.IPSet(netaddr.IPNetwork("192.168.1.0/24"))
fip_ip = fip['floatingip']['floating_ip_address']
self.assertTrue(netaddr.IPAddress(fip_ip) in fip_set)
self.assertIn(netaddr.IPAddress(fip_ip), fip_set)
def test_create_floatingip_with_assoc_to_ipv6_subnet(self):
with self.subnet() as public_sub:

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

@ -291,7 +291,7 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
def side_effect(context, security_group, default_sg):
# can't always raise, or create_security_group will hang
self.assertTrue(default_sg)
self.assertTrue(num_called[0] < 2)
self.assertLess(num_called[0], 2)
num_called[0] += 1
ret = original_func(context, security_group, default_sg)
if num_called[0] == 1:

10
neutron/tests/unit/extensions/test_segment.py
View File

@ -751,7 +751,7 @@ class TestMl2HostSegmentMappingAgentServerSynch(HostSegmentMappingTestCase):
physical_network = 'phys_net1'
self._register_agent(host, mappings={physical_network: 'br-eth-1'},
plugin=self.plugin, start_flag=False)
self.assertTrue(host in db.reported_hosts)
self.assertIn(host, db.reported_hosts)
self.assertEqual(1, mock_function.call_count)
expected_call = mock.call(mock.ANY, host, set())
mock_function.assert_has_calls([expected_call])
@ -762,10 +762,10 @@ class TestMl2HostSegmentMappingAgentServerSynch(HostSegmentMappingTestCase):
physical_network = 'phys_net1'
self._register_agent(host, mappings={physical_network: 'br-eth-1'},
plugin=self.plugin, start_flag=False)
self.assertTrue(host in db.reported_hosts)
self.assertIn(host, db.reported_hosts)
self._register_agent(host, mappings={physical_network: 'br-eth-1'},
plugin=self.plugin, start_flag=True)
self.assertTrue(host in db.reported_hosts)
self.assertIn(host, db.reported_hosts)
self.assertEqual(2, mock_function.call_count)
expected_call = mock.call(mock.ANY, host, set())
mock_function.assert_has_calls([expected_call, expected_call])
@ -776,11 +776,11 @@ class TestMl2HostSegmentMappingAgentServerSynch(HostSegmentMappingTestCase):
physical_network = 'phys_net1'
self._register_agent(host, mappings={physical_network: 'br-eth-1'},
plugin=self.plugin, start_flag=False)
self.assertTrue(host in db.reported_hosts)
self.assertIn(host, db.reported_hosts)
mock_function.reset_mock()
self._register_agent(host, mappings={physical_network: 'br-eth-1'},
plugin=self.plugin, start_flag=False)
self.assertTrue(host in db.reported_hosts)
self.assertIn(host, db.reported_hosts)
mock_function.assert_not_called()

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

@ -191,8 +191,8 @@ class QosPolicyDbObjectTestCase(test_base.BaseDbObjectTestCase,
networks = obj.get_bound_networks()
self.assertEqual(2, len(networks))
self.assertTrue(net1_id in networks)
self.assertTrue(net2_id in networks)
self.assertIn(net1_id, networks)
self.assertIn(net2_id, networks)
def test_attach_port_nonexistent_port(self):
@ -248,8 +248,8 @@ class QosPolicyDbObjectTestCase(test_base.BaseDbObjectTestCase,
ports = obj.get_bound_ports()
self.assertEqual(2, len(ports))
self.assertTrue(port1_id in ports)
self.assertTrue(port2_id in ports)
self.assertIn(port1_id, ports)
self.assertIn(port2_id, ports)
def test_attach_port_get_policy_port(self):

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

@ -789,7 +789,7 @@ class BaseObjectIfaceTestCase(_BaseObjectTestCase, test_base.BaseTestCase):
for field in self._test_class.obj_extra_fields:
# field is accessible and cannot be set by any value
getattr(obj, field)
self.assertTrue(field in obj.to_dict().keys())
self.assertIn(field, obj.to_dict().keys())
self.assertRaises(AttributeError, setattr, obj, field, "1")
def test_to_dict_makes_primitive_field_value(self):

40
neutron/tests/unit/plugins/ml2/drivers/agent/test__common_agent.py
View File

@ -121,9 +121,7 @@ class TestCommonAgentLoop(base.BaseTestCase):
self.assertTrue(fn_udd.called)
self.assertTrue(fn_rdf.called)
self.assertTrue(ext_mgr_delete_port.called)
self.assertTrue(
PORT_DATA not in agent.network_ports[NETWORK_ID]
)
self.assertNotIn(PORT_DATA, agent.network_ports[NETWORK_ID])
def test_treat_devices_removed_with_not_existed_device(self):
agent = self.agent
@ -144,9 +142,7 @@ class TestCommonAgentLoop(base.BaseTestCase):
self.assertTrue(fn_udd.called)
self.assertTrue(fn_rdf.called)
self.assertTrue(ext_mgr_delete_port.called)
self.assertTrue(
PORT_DATA not in agent.network_ports[NETWORK_ID]
)
self.assertNotIn(PORT_DATA, agent.network_ports[NETWORK_ID])
def test_treat_devices_removed_failed(self):
agent = self.agent
@ -164,9 +160,7 @@ class TestCommonAgentLoop(base.BaseTestCase):
self.assertTrue(fn_udd.called)
self.assertTrue(fn_rdf.called)
self.assertTrue(ext_mgr_delete_port.called)
self.assertTrue(
PORT_DATA not in agent.network_ports[NETWORK_ID]
)
self.assertNotIn(PORT_DATA, agent.network_ports[NETWORK_ID])
def test_treat_devices_removed_with_prevent_arp_spoofing_true(self):
agent = self.agent
@ -458,9 +452,9 @@ class TestCommonAgentLoop(base.BaseTestCase):
constants.DEVICE_OWNER_NETWORK_PREFIX)
self.assertTrue(agent.plugin_rpc.update_device_up.called)
self.assertTrue(agent.ext_manager.handle_port.called)
self.assertTrue(mock_port_data in agent.network_ports[
self.assertIn(mock_port_data, agent.network_ports[
mock_details['network_id']]
)
)
def test_treat_devices_added_updated_prevent_arp_spoofing_true(self):
agent = self.agent
@ -550,12 +544,8 @@ class TestCommonAgentLoop(base.BaseTestCase):
self.agent._update_network_ports(
NETWORK_2_ID, port_2_data['port_id'], port_2_data['device']
)
self.assertTrue(
port_2_data not in self.agent.network_ports[NETWORK_ID]
)
self.assertTrue(
port_2_data in self.agent.network_ports[NETWORK_2_ID]
)
self.assertNotIn(port_2_data, self.agent.network_ports[NETWORK_ID])
self.assertIn(port_2_data, self.agent.network_ports[NETWORK_2_ID])
def test_clean_network_ports(self):
port_1_data = PORT_DATA
@ -571,21 +561,13 @@ class TestCommonAgentLoop(base.BaseTestCase):
)
#check removing port from network when other ports are still there:
cleaned_port_id = self.agent._clean_network_ports(DEVICE_1)
self.assertTrue(
NETWORK_ID in self.agent.network_ports.keys()
)
self.assertTrue(
port_1_data not in self.agent.network_ports[NETWORK_ID]
)
self.assertTrue(
port_2_data in self.agent.network_ports[NETWORK_ID]
)
self.assertIn(NETWORK_ID, self.agent.network_ports.keys())
self.assertNotIn(port_1_data, self.agent.network_ports[NETWORK_ID])
self.assertIn(port_2_data, self.agent.network_ports[NETWORK_ID])
self.assertEqual(PORT_1, cleaned_port_id)
#and now remove last port from network:
cleaned_port_id = self.agent._clean_network_ports(
port_2_data['device']
)
self.assertTrue(
NETWORK_ID not in self.agent.network_ports.keys()
)
self.assertNotIn(NETWORK_ID, self.agent.network_ports.keys())
self.assertEqual(port_2_data['port_id'], cleaned_port_id)

18
neutron/tests/unit/plugins/ml2/extensions/test_dns_integration.py
View File

@ -183,13 +183,13 @@ class DNSIntegrationTestCase(test_plugin.Ml2PluginV2TestCase):
V6UUID))
mock_client.recordsets.create.assert_has_calls(expected,
any_order=True)
self.assertTrue(
len(mock_client.recordsets.create.call_args_list) ==
self.assertEqual(
len(mock_client.recordsets.create.call_args_list),
len(expected))
mock_client.recordsets.delete.assert_has_calls(expected_delete,
any_order=True)
self.assertTrue(
len(mock_client.recordsets.delete.call_args_list) ==
self.assertEqual(
len(mock_client.recordsets.delete.call_args_list),
len(expected_delete))
expected = []
expected_delete = []
@ -217,18 +217,18 @@ class DNSIntegrationTestCase(test_plugin.Ml2PluginV2TestCase):
in_addr_name))
mock_admin_client.recordsets.create.assert_has_calls(
expected, any_order=True)
self.assertTrue(
len(mock_admin_client.recordsets.create.call_args_list) ==
self.assertEqual(
len(mock_admin_client.recordsets.create.call_args_list),
len(expected))
mock_admin_client.recordsets.delete.assert_has_calls(
expected_delete, any_order=True)
self.assertTrue(
len(mock_admin_client.recordsets.delete.call_args_list) ==
self.assertEqual(
len(mock_admin_client.recordsets.delete.call_args_list),
len(expected_delete))
else:
if not dns_name:
self.assertEqual('', port[dns.DNSNAME])
self.assertTrue(dns_data_db is None)
self.assertIsNone(dns_data_db)
self.assertFalse(mock_client.recordsets.create.call_args_list)
self.assertFalse(
mock_admin_client.recordsets.create.call_args_list)

8
neutron/tests/unit/plugins/ml2/test_plugin.py
View File

@ -1740,7 +1740,7 @@ class TestMultiSegmentNetworks(Ml2PluginV2TestCase):
self.assertEqual('vlan', dynamic_segment[driver_api.NETWORK_TYPE])
self.assertEqual('physnet1',
dynamic_segment[driver_api.PHYSICAL_NETWORK])
self.assertTrue(dynamic_segment[driver_api.SEGMENTATION_ID] > 0)
self.assertGreater(dynamic_segment[driver_api.SEGMENTATION_ID], 0)
segment2 = {driver_api.NETWORK_TYPE: 'vlan',
driver_api.SEGMENTATION_ID: 1234,
driver_api.PHYSICAL_NETWORK: 'physnet3'}
@ -1770,7 +1770,7 @@ class TestMultiSegmentNetworks(Ml2PluginV2TestCase):
self.assertEqual('physnet1',
dynamic_segment[driver_api.PHYSICAL_NETWORK])
dynamic_segmentation_id = dynamic_segment[driver_api.SEGMENTATION_ID]
self.assertTrue(dynamic_segmentation_id > 0)
self.assertGreater(dynamic_segmentation_id, 0)
dynamic_segment1 = segments_db.get_dynamic_segment(
self.context.session, network_id, 'physnet1')
dynamic_segment1_id = dynamic_segment1[driver_api.SEGMENTATION_ID]
@ -1801,7 +1801,7 @@ class TestMultiSegmentNetworks(Ml2PluginV2TestCase):
self.assertEqual('physnet1',
dynamic_segment[driver_api.PHYSICAL_NETWORK])
dynamic_segmentation_id = dynamic_segment[driver_api.SEGMENTATION_ID]
self.assertTrue(dynamic_segmentation_id > 0)
self.assertGreater(dynamic_segmentation_id, 0)
self.driver.type_manager.release_dynamic_segment(
self.context.session, dynamic_segment[driver_api.ID])
self.assertIsNone(segments_db.get_dynamic_segment(
@ -1937,7 +1937,7 @@ class TestMultiSegmentNetworks(Ml2PluginV2TestCase):
self.assertEqual('vlan', dynamic_segment[driver_api.NETWORK_TYPE])
self.assertEqual('physnet2',
dynamic_segment[driver_api.PHYSICAL_NETWORK])
self.assertTrue(dynamic_segment[driver_api.SEGMENTATION_ID] > 0)
self.assertGreater(dynamic_segment[driver_api.SEGMENTATION_ID], 0)
with mock.patch.object(type_vlan.VlanTypeDriver,
'release_segment') as rs:

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

@ -579,7 +579,7 @@ class L3SchedulerTestBaseMixin(object):
rid = router['router']['id']
scheduler.bind_router(ctx, rid, agent)
results = (session.query(db).filter_by(router_id=rid).all())
self.assertTrue(len(results) > 0)
self.assertGreater(len(results), 0)
self.assertIn(agent.id, [bind.l3_agent_id for bind in results])
def test_bind_new_router(self):

6
neutron/tests/unit/services/metering/agents/test_metering_agent.py
View File

@ -160,11 +160,11 @@ class TestMeteringOperations(base.BaseTestCase):
payload = n['payload']
self.assertEqual(TENANT_ID, payload['tenant_id'])
self.assertEqual(LABEL_ID, payload['label_id'])
self.assertTrue((payload['time'] - report_interval)
< measure_interval, payload)
self.assertLess((payload['time'] - report_interval),
measure_interval, payload)
interval = (payload['last_update'] - payload['first_update']) \
- report_interval
self.assertTrue(interval < measure_interval, payload)
self.assertLess(interval, measure_interval, payload)
def test_router_deleted(self):
label_id = _uuid()

2
neutron/tests/unit/services/test_provider_configuration.py
View File

@ -70,7 +70,7 @@ class ParseServiceProviderConfigurationTestCase(base.BaseTestCase):
res = provconf.parse_service_provider_opt()
# This parsing crosses repos if additional projects are installed,
# so check that at least what we expect is there; there may be more.
self.assertTrue(len(res) >= 3)
self.assertGreaterEqual(len(res), 3)
def test_parse_service_provider_invalid_format(self):
self._set_override([constants.LOADBALANCER +