openstack
/
neutron
Code Issues Proposed changes

Merge "Remove remaining do_delete from unit tests"

This commit is contained in:
Jenkins 2015-02-22 10:39:58 +00:00 committed by Gerrit Code Review
parent b0210d03ba 7322cb3cee
commit 194afde3a6
10 changed files with 43 additions and 67 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
neutron/tests/unit/bigswitch/test_restproxy_plugin.py
View File

@ -96,7 +96,6 @@ class TestBigSwitchProxyPortsV2(test_plugin.TestPortsV2,
with self.subnet(network=net) as sub:
with self.port(
subnet=sub,
do_delete=False,
device_owner=constants.DEVICE_OWNER_ROUTER_INTF
) as port:
# router ports should be immediately active

16
neutron/tests/unit/db/metering/test_db_metering.py
View File

@ -86,20 +86,17 @@ class MeteringPluginDbTestCaseMixin(object):
@contextlib.contextmanager
def metering_label(self, name='label', description='desc',
fmt=None, do_delete=True, **kwargs):
fmt=None, **kwargs):
if not fmt:
fmt = self.fmt
metering_label = self._make_metering_label(fmt, name,
description, **kwargs)
yield metering_label
if do_delete:
self._delete('metering-labels',
metering_label['metering_label']['id'])
@contextlib.contextmanager
def metering_label_rule(self, metering_label_id=None, direction='ingress',
remote_ip_prefix='10.0.0.0/24',
excluded='false', fmt=None, do_delete=True):
excluded='false', fmt=None):
if not fmt:
fmt = self.fmt
metering_label_rule = self._make_metering_label_rule(fmt,
@ -108,9 +105,6 @@ class MeteringPluginDbTestCaseMixin(object):
remote_ip_prefix,
excluded)
yield metering_label_rule
if do_delete:
self._delete('metering-label-rules',
metering_label_rule['metering_label_rule']['id'])
class MeteringPluginDbTestCase(test_db_plugin.NeutronDbPluginV2TestCase,
@ -163,8 +157,7 @@ class TestMetering(MeteringPluginDbTestCase):
name = 'my label'
description = 'my metering label'
with self.metering_label(name, description,
do_delete=False) as metering_label:
with self.metering_label(name, description) as metering_label:
metering_label_id = metering_label['metering_label']['id']
self._delete('metering-labels', metering_label_id, 204)
@ -214,8 +207,7 @@ class TestMetering(MeteringPluginDbTestCase):
with self.metering_label_rule(metering_label_id,
direction,
remote_ip_prefix,
excluded,
do_delete=False) as label_rule:
excluded) as label_rule:
rule_id = label_rule['metering_label_rule']['id']
self._delete('metering-label-rules', rule_id, 204)

8
neutron/tests/unit/ml2/test_ml2_plugin.py
View File

@ -307,7 +307,7 @@ class TestMl2PortsV2(test_plugin.TestPortsV2, Ml2PluginV2TestCase):
l3plugin = manager.NeutronManager.get_service_plugins().get(
service_constants.L3_ROUTER_NAT)
with contextlib.nested(
self.port(do_delete=False),
self.port(),
mock.patch.object(l3plugin, 'disassociate_floatingips'),
mock.patch.object(l3plugin, 'notify_routers_updated')
) as (port, disassociate_floatingips, notify):
@ -375,8 +375,7 @@ class TestMl2DvrPortsV2(TestMl2PortsV2):
mock.patch.object(manager.NeutronManager,
'get_service_plugins',
return_value=self.service_plugins),
self.port(do_delete=False,
device_owner=device_owner),
self.port(device_owner=device_owner),
mock.patch.object(self.l3plugin, 'notify_routers_updated'),
mock.patch.object(self.l3plugin, 'disassociate_floatingips',
return_value=fip_set),
@ -413,8 +412,7 @@ class TestMl2DvrPortsV2(TestMl2PortsV2):
mock.patch.object(manager.NeutronManager,
'get_service_plugins',
return_value=self.service_plugins),
self.port(do_delete=False,
device_owner='compute:None'),
self.port(device_owner='compute:None'),
mock.patch.object(self.l3plugin, 'dvr_deletens_if_no_port',
return_value=[ns_to_delete]),
mock.patch.object(self.l3plugin, 'remove_router_from_l3_agent',

17
neutron/tests/unit/nec/test_packet_filter.py
View File

@ -92,20 +92,17 @@ class TestNecPluginPacketFilterBase(test_nec_plugin.NecPluginV2TestCase):
return self.deserialize(fmt, res)
@contextlib.contextmanager
def packet_filter_on_network(self, network=None, fmt=None, do_delete=True,
**kwargs):
def packet_filter_on_network(self, network=None, fmt=None, **kwargs):
with test_plugin.optional_ctx(network, self.network) as network_to_use:
net_id = network_to_use['network']['id']
pf = self._make_packet_filter(fmt or self.fmt, net_id, **kwargs)
yield pf
if do_delete:
self._delete('packet_filters', pf['packet_filter']['id'])
if not network:
self._delete('networks', network_to_use['network']['id'])
@contextlib.contextmanager
def packet_filter_on_port(self, port=None, fmt=None, do_delete=True,
set_portinfo=True, **kwargs):
def packet_filter_on_port(self, port=None, fmt=None, set_portinfo=True,
**kwargs):
with test_plugin.optional_ctx(port, self.port) as port_to_use:
net_id = port_to_use['port']['network_id']
port_id = port_to_use['port']['id']
@ -122,8 +119,6 @@ class TestNecPluginPacketFilterBase(test_nec_plugin.NecPluginV2TestCase):
pf = self._make_packet_filter(fmt or self.fmt, net_id, **kwargs)
self.assertEqual(port_id, pf['packet_filter']['in_port'])
yield pf
if do_delete:
self._delete('packet_filters', pf['packet_filter']['id'])
class TestNecPluginPacketFilter(TestNecPluginPacketFilterBase):
@ -175,6 +170,7 @@ class TestNecPluginPacketFilter(TestNecPluginPacketFilterBase):
with self.packet_filter_on_port() as pf:
pf_id = pf['packet_filter']['id']
self.assertEqual(pf['packet_filter']['status'], 'ACTIVE')
self._delete('packet_filters', pf_id)
ctx = mock.ANY
pf_dict = mock.ANY
@ -554,8 +550,7 @@ class TestNecPluginPacketFilter(TestNecPluginPacketFilterBase):
self.assertEqual(self.ofc.delete_ofc_packet_filter.call_count, 2)
def test_auto_delete_pf_in_network_deletion(self):
with self.packet_filter_on_network(admin_state_up=False,
do_delete=False) as pf:
with self.packet_filter_on_network(admin_state_up=False) as pf:
pf_id = pf['packet_filter']['id']
self._show('packet_filters', pf_id,
@ -566,7 +561,7 @@ class TestNecPluginPacketFilter(TestNecPluginPacketFilterBase):
network = self._show('networks', port['port']['network_id'])
with self.packet_filter_on_network(network=network) as pfn:
with self.packet_filter_on_port(port=port, do_delete=False,
with self.packet_filter_on_port(port=port,
set_portinfo=False) as pf:
pf_id = pf['packet_filter']['id']
in_port_id = pf['packet_filter']['in_port']

3
neutron/tests/unit/nuage/test_netpartition.py
View File

@ -57,14 +57,11 @@ class NetPartitionTestCase(test_nuage_plugin.NuagePluginV2TestCase):
@contextlib.contextmanager
def netpartition(self, name='netpartition1',
do_delete=True,
fmt=None,
**kwargs):
netpart = self._make_netpartition(fmt or self.fmt, name)
yield netpart
if do_delete:
self._del_netpartition(netpart['net_partition']['id'])
def test_create_netpartition(self):
name = 'netpart1'

16
neutron/tests/unit/services/metering/test_metering_plugin.py
View File

@ -167,8 +167,10 @@ class TestMeteringPlugin(test_db_plugin.NeutronDbPluginV2TestCase,
with self.router(tenant_id=self.tenant_id, set_context=True):
with self.metering_label(tenant_id=self.tenant_id,
set_context=True):
set_context=True) as label:
self.mock_add.assert_called_with(self.ctx, expected)
self._delete('metering-labels',
label['metering_label']['id'])
self.mock_remove.assert_called_with(self.ctx, expected)
def test_remove_one_metering_label_rpc_call(self):
@ -199,8 +201,10 @@ class TestMeteringPlugin(test_db_plugin.NeutronDbPluginV2TestCase,
set_context=True):
self.mock_uuid.return_value = second_uuid
with self.metering_label(tenant_id=self.tenant_id,
set_context=True):
set_context=True) as label:
self.mock_add.assert_called_with(self.ctx, expected_add)
self._delete('metering-labels',
label['metering_label']['id'])
self.mock_remove.assert_called_with(self.ctx, expected_remove)
def test_update_metering_label_rules_rpc_call(self):
@ -246,16 +250,18 @@ class TestMeteringPlugin(test_db_plugin.NeutronDbPluginV2TestCase,
l = label['metering_label']
with self.metering_label_rule(l['id']):
self.mock_uuid.return_value = second_uuid
with self.metering_label_rule(l['id'], direction='egress'):
with self.metering_label_rule(l['id'],
direction='egress') as rule:
self.mock_update.assert_called_with(self.ctx,
expected_add)
self._delete('metering-label-rules',
rule['metering_label_rule']['id'])
self.mock_update.assert_called_with(self.ctx,
expected_del)
def test_delete_metering_label_does_not_clear_router_tenant_id(self):
tenant_id = '654f6b9d-0f36-4ae5-bd1b-01616794ca60'
with self.metering_label(tenant_id=tenant_id,
do_delete=False) as metering_label:
with self.metering_label(tenant_id=tenant_id) as metering_label:
with self.router(tenant_id=tenant_id, set_context=True) as r:
router = self._show('routers', r['router']['id'])
self.assertEqual(tenant_id, router['router']['tenant_id'])

12
neutron/tests/unit/test_extension_security_group.py
View File

@ -118,14 +118,11 @@ class SecurityGroupsTestCase(test_db_plugin.NeutronDbPluginV2TestCase):
@contextlib.contextmanager
def security_group(self, name='webservers', description='webservers',
fmt=None, do_delete=True):
fmt=None):
if not fmt:
fmt = self.fmt
security_group = self._make_security_group(fmt, name, description)
yield security_group
if do_delete:
self._delete('security-groups',
security_group['security_group']['id'])
@contextlib.contextmanager
def security_group_rule(self, security_group_id='4cd70774-cc67-4a87-9b39-7'
@ -133,7 +130,7 @@ class SecurityGroupsTestCase(test_db_plugin.NeutronDbPluginV2TestCase):
direction='ingress', protocol=const.PROTO_NAME_TCP,
port_range_min='22', port_range_max='22',
remote_ip_prefix=None, remote_group_id=None,
fmt=None, do_delete=True, ethertype=const.IPv4):
fmt=None, ethertype=const.IPv4):
if not fmt:
fmt = self.fmt
rule = self._build_security_group_rule(security_group_id,
@ -145,9 +142,6 @@ class SecurityGroupsTestCase(test_db_plugin.NeutronDbPluginV2TestCase):
ethertype=ethertype)
security_group_rule = self._make_security_group_rule(self.fmt, rule)
yield security_group_rule
if do_delete:
self._delete('security-group-rules',
security_group_rule['security_group_rule']['id'])
def _delete_default_security_group_egress_rules(self, security_group_id):
"""Deletes default egress rules given a security group ID."""
@ -563,7 +557,7 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
def test_delete_security_group(self):
name = 'webservers'
description = 'my webservers'
with self.security_group(name, description, do_delete=False) as sg:
with self.security_group(name, description) as sg:
remote_group_id = sg['security_group']['id']
self._delete('security-groups', remote_group_id,
webob.exc.HTTPNoContent.code)

23
neutron/tests/unit/test_security_groups_rpc.py
View File

@ -2850,19 +2850,18 @@ class SGNotificationTestMixin(object):
name = 'webservers'
description = 'my webservers'
with self.security_group(name, description) as sg:
with self.security_group(name, description) as sg2:
with self.security_group(name, description):
security_group_id = sg['security_group']['id']
direction = "ingress"
remote_group_id = sg2['security_group']['id']
protocol = const.PROTO_NAME_TCP
port_range_min = 88
port_range_max = 88
with self.security_group_rule(security_group_id, direction,
protocol, port_range_min,
port_range_max,
remote_group_id=remote_group_id
):
pass
rule = self._build_security_group_rule(
security_group_id,
direction='ingress',
proto=const.PROTO_NAME_TCP)
security_group_rule = self._make_security_group_rule(self.fmt,
rule)
self._delete('security-group-rules',
security_group_rule['security_group_rule']['id'])
self.notifier.assert_has_calls(
[mock.call.security_groups_rule_updated(mock.ANY,
[security_group_id]),

12
neutron/tests/unit/vmware/extensions/test_qosqueues.py
View File

@ -59,7 +59,7 @@ class TestQoSQueue(test_nsx_plugin.NsxPluginV2TestCase):
@contextlib.contextmanager
def qos_queue(self, name='foo', min='0', max='10',
qos_marking=None, dscp='0', default=None, do_delete=True):
qos_marking=None, dscp='0', default=None):
body = {'qos_queue': {'tenant_id': 'tenant',
'name': name,
@ -79,10 +79,6 @@ class TestQoSQueue(test_nsx_plugin.NsxPluginV2TestCase):
yield qos_queue
if do_delete:
self._delete('qos-queues',
qos_queue['qos_queue']['id'])
def test_create_qos_queue(self):
with self.qos_queue(name='fake_lqueue', min=34, max=44,
qos_marking='untrusted', default=False) as q:
@ -132,7 +128,7 @@ class TestQoSQueue(test_nsx_plugin.NsxPluginV2TestCase):
self.assertEqual(len(p['port'][ext_qos.QUEUE]), 36)
def test_create_shared_queue_networks(self):
with self.qos_queue(default=True, do_delete=False) as q1:
with self.qos_queue(default=True) as q1:
res = self._create_network('json', 'net1', True,
arg_list=(ext_qos.QUEUE,),
queue_id=q1['qos_queue']['id'])
@ -159,7 +155,7 @@ class TestQoSQueue(test_nsx_plugin.NsxPluginV2TestCase):
self._delete('ports', port2['port']['id'])
def test_remove_queue_in_use_fail(self):
with self.qos_queue(do_delete=False) as q1:
with self.qos_queue() as q1:
res = self._create_network('json', 'net1', True,
arg_list=(ext_qos.QUEUE,),
queue_id=q1['qos_queue']['id'])
@ -186,7 +182,7 @@ class TestQoSQueue(test_nsx_plugin.NsxPluginV2TestCase):
new_q['qos_queue']['id'])
def test_update_port_adding_device_id(self):
with self.qos_queue(do_delete=False) as q1:
with self.qos_queue() as q1:
res = self._create_network('json', 'net1', True,
arg_list=(ext_qos.QUEUE,),
queue_id=q1['qos_queue']['id'])

2
neutron/tests/unit/vmware/test_nsx_plugin.py
View File

@ -1016,7 +1016,7 @@ class TestL3NatTestCase(L3NatTest,
def test_router_add_interface_port_removes_security_group(self):
with self.router() as r:
with self.port(do_delete=False) as p:
with self.port() as p:
body = self._router_interface_action('add',
r['router']['id'],
None,