self.HAS_PORT_FILTER)
def _check_response_no_portbindings(self, port):
- self.assertTrue('status' in port)
+ self.assertIn('status', port)
self.assertNotIn(portbindings.VIF_TYPE, port)
self.assertNotIn(portbindings.CAPABILITIES, port)
r1_id = r1['router']['id']
body = self._router_interface_action('add', r_id, s_id,
None)
- self.assertTrue('port_id' in body)
+ self.assertIn('port_id', body)
r_port_id = body['port_id']
body = self._show('ports', r_port_id)
self.assertEqual(body['port']['device_id'], r_id)
s1_id = s1['subnet']['id']
body = self._router_interface_action('add', r1_id,
s1_id, None)
- self.assertTrue('port_id' in body)
+ self.assertIn('port_id', body)
r1_port_id = body['port_id']
body = self._show('ports', r1_port_id)
self.assertEqual(body['port']['device_id'], r1_id)
dummy_vif = ovs_lib.VifPort('port1', '1', 'id-1', 'mac-1', None)
self.state_rpc_api.assert_called_once_with('q-plugin')
- self.assertTrue('start_flag' in self.agent.agent_state)
+ self.assertIn('start_flag', self.agent.agent_state)
self._check_report_state([], 0, fail_mode, first=True)
self._check_report_state([dummy_vif] * 2, 2, fail_mode)
instance.create_network_gateway.assert_called_with(
mock.ANY, network_gateway=data)
self.assertEqual(res.status_int, exc.HTTPCreated.code)
- self.assertTrue(self._resource in res.json)
+ self.assertIn(self._resource, res.json)
nw_gw = res.json[self._resource]
self.assertEqual(nw_gw['id'], nw_gw_id)
instance.update_network_gateway.assert_called_with(
mock.ANY, nw_gw_id, network_gateway=data)
self.assertEqual(res.status_int, exc.HTTPOk.code)
- self.assertTrue(self._resource in res.json)
+ self.assertIn(self._resource, res.json)
nw_gw = res.json[self._resource]
self.assertEqual(nw_gw['id'], nw_gw_id)
self.assertEqual(nw_gw['name'], nw_gw_name)
net['network']['id'],
segmentation_type,
segmentation_id)
- self.assertTrue('connection_info' in body)
+ self.assertIn('connection_info', body)
connection_info = body['connection_info']
for attr in ('network_id', 'port_id',
'network_gateway_id'):
- self.assertTrue(attr in connection_info)
+ self.assertIn(attr, connection_info)
# fetch port and confirm device_id
gw_port_id = connection_info['port_id']
port_body = self._show('ports', gw_port_id)
a.available_local_vlans = set()
a.local_vlan_map[NET_UUID] = LVM
a.reclaim_local_vlan(NET_UUID)
- self.assertTrue(LVM.vlan in a.available_local_vlans)
+ self.assertIn(LVM.vlan, a.available_local_vlans)
self.mox.VerifyAll()
def test_reclaim_local_vlan_flat(self):
a.available_local_vlans = set()
a.local_vlan_map[NET_UUID] = LVM_FLAT
a.reclaim_local_vlan(NET_UUID)
- self.assertTrue(LVM_FLAT.vlan in a.available_local_vlans)
+ self.assertIn(LVM_FLAT.vlan, a.available_local_vlans)
self.mox.VerifyAll()
def test_reclaim_local_vlan_vlan(self):
a.available_local_vlans = set()
a.local_vlan_map[NET_UUID] = LVM_VLAN
a.reclaim_local_vlan(NET_UUID)
- self.assertTrue(LVM_VLAN.vlan in a.available_local_vlans)
+ self.assertIn(LVM_VLAN.vlan, a.available_local_vlans)
self.mox.VerifyAll()
def test_port_bound(self):
with self.port(name='myname') as port:
for k, v in keys:
self.assertEqual(port['port'][k], v)
- self.assertTrue('mac_address' in port['port'])
+ self.assertIn('mac_address', port['port'])
ips = port['port']['fixed_ips']
self.assertEqual(len(ips), 1)
self.assertEqual(ips[0]['ip_address'], '10.0.0.2')
port = self.deserialize(self.fmt, port_res)
for k, v in keys:
self.assertEqual(port['port'][k], v)
- self.assertTrue('mac_address' in port['port'])
+ self.assertIn('mac_address', port['port'])
self._delete('ports', port['port']['id'])
def test_create_port_public_network_with_ip(self):
port = self.deserialize(self.fmt, port_res)
for k, v in keys:
self.assertEqual(port['port'][k], v)
- self.assertTrue('mac_address' in port['port'])
+ self.assertIn('mac_address', port['port'])
self._delete('ports', port['port']['id'])
def test_create_ports_bulk_native(self):
# Shouldn't be reached
self.assertTrue(False)
except webtest.AppError as e:
- self.assertTrue('501' in e.message)
+ self.assertIn('501', e.message)
def test_resource_can_be_added_as_extension(self):
res_ext = extensions.ResourceExtension(
ext_mgr.add_extension(InvalidExtension())
ext_mgr.add_extension(ext_stubs.StubExtension("valid_extension"))
- self.assertTrue('valid_extension' in ext_mgr.extensions)
+ self.assertIn('valid_extension', ext_mgr.extensions)
self.assertNotIn('invalid_extension', ext_mgr.extensions)
ext_mgr.add_extension(ext_stubs.StubExtension("e2"))
ext_mgr.add_extension(ext_stubs.StubExtension("e3"))
- self.assertTrue("e1" in ext_mgr.extensions)
+ self.assertIn("e1", ext_mgr.extensions)
self.assertNotIn("e2", ext_mgr.extensions)
- self.assertTrue("e3" in ext_mgr.extensions)
+ self.assertIn("e3", ext_mgr.extensions)
def test_extensions_are_not_loaded_for_plugins_unaware_of_extensions(self):
class ExtensionUnawarePlugin(object):
ext_mgr.add_extension(
ext_stubs.ExtensionExpectingPluginInterface("supported_extension"))
- self.assertTrue("supported_extension" in ext_mgr.extensions)
+ self.assertIn("supported_extension", ext_mgr.extensions)
def test_extensions_expecting_neutron_plugin_interface_are_loaded(self):
class ExtensionForQuamtumPluginInterface(ext_stubs.StubExtension):
ext_mgr = extensions.PluginAwareExtensionManager('', plugin_info)
ext_mgr.add_extension(ExtensionForQuamtumPluginInterface("e1"))
- self.assertTrue("e1" in ext_mgr.extensions)
+ self.assertIn("e1", ext_mgr.extensions)
def test_extensions_without_need_for__plugin_interface_are_loaded(self):
class ExtensionWithNoNeedForPluginInterface(ext_stubs.StubExtension):
ext_mgr = extensions.PluginAwareExtensionManager('', plugin_info)
ext_mgr.add_extension(ExtensionWithNoNeedForPluginInterface("e1"))
- self.assertTrue("e1" in ext_mgr.extensions)
+ self.assertIn("e1", ext_mgr.extensions)
def test_extension_loaded_for_non_core_plugin(self):
class NonCorePluginExtenstion(ext_stubs.StubExtension):
ext_mgr = extensions.PluginAwareExtensionManager('', plugin_info)
ext_mgr.add_extension(NonCorePluginExtenstion("e1"))
- self.assertTrue("e1" in ext_mgr.extensions)
+ self.assertIn("e1", ext_mgr.extensions)
class ExtensionControllerTest(testlib_api.WebTestCase):
res = self._list_service_providers()
self.assertEqual(res.status_int, webexc.HTTPOk.code)
data = self.deserialize(res)
- self.assertTrue('service_providers' in data)
+ self.assertIn('service_providers', data)
self.assertEqual(len(data['service_providers']), 2)