openstack
/
neutron
Code Issues Proposed changes

remove E251 exemption from pep8 check 

This change removes the exemption for E251 by addressing unexpected spaces
around keyword/parameter equals

Change-Id: Iff17477e37bef2a97fc58a538d08bcfc35c67751
Partial-Bug: 1329017
This commit is contained in:
Mark McClain 2014-06-11 20:44:43 -04:00
parent 941849bf66
commit b4f025be06
17 changed files with 226 additions and 175 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
neutron/agent/metadata/agent.py
View File

@ -209,9 +209,10 @@ class MetadataProxyHandler(object):
req.query_string,
''))
h = httplib2.Http(ca_certs=self.conf.auth_ca_cert,
disable_ssl_certificate_validation=
self.conf.nova_metadata_insecure)
h = httplib2.Http(
ca_certs=self.conf.auth_ca_cert,
disable_ssl_certificate_validation=self.conf.nova_metadata_insecure
)
if self.conf.nova_client_cert and self.conf.nova_client_priv_key:
h.add_certificate(self.conf.nova_client_priv_key,
self.conf.nova_client_cert,

52
neutron/db/firewall/firewall_db.py
View File

@ -187,8 +187,8 @@ class Firewall_db_mixin(firewall.FirewallPluginBase, base_db.CommonDbMixin):
# If we find an invalid rule in the list we
# do not perform the update since this breaks
# the integrity of this list.
raise firewall.FirewallRuleNotFound(firewall_rule_id=
fwrule_id)
raise firewall.FirewallRuleNotFound(
firewall_rule_id=fwrule_id)
elif rules_dict[fwrule_id]['firewall_policy_id']:
if (rules_dict[fwrule_id]['firewall_policy_id'] !=
fwp_db['id']):
@ -273,14 +273,14 @@ class Firewall_db_mixin(firewall.FirewallPluginBase, base_db.CommonDbMixin):
status = (const.CREATED
if cfg.CONF.router_distributed else const.PENDING_CREATE)
with context.session.begin(subtransactions=True):
firewall_db = Firewall(id=uuidutils.generate_uuid(),
tenant_id=tenant_id,
name=fw['name'],
description=fw['description'],
firewall_policy_id=
fw['firewall_policy_id'],
admin_state_up=fw['admin_state_up'],
status=status)
firewall_db = Firewall(
id=uuidutils.generate_uuid(),
tenant_id=tenant_id,
name=fw['name'],
description=fw['description'],
firewall_policy_id=fw['firewall_policy_id'],
admin_state_up=fw['admin_state_up'],
status=status)
context.session.add(firewall_db)
return self._make_firewall_dict(firewall_db)
@ -396,22 +396,22 @@ class Firewall_db_mixin(firewall.FirewallPluginBase, base_db.CommonDbMixin):
dst_port_min, dst_port_max = self._get_min_max_ports_from_range(
fwr['destination_port'])
with context.session.begin(subtransactions=True):
fwr_db = FirewallRule(id=uuidutils.generate_uuid(),
tenant_id=tenant_id,
name=fwr['name'],
description=fwr['description'],
shared=fwr['shared'],
protocol=fwr['protocol'],
ip_version=fwr['ip_version'],
source_ip_address=fwr['source_ip_address'],
destination_ip_address=
fwr['destination_ip_address'],
source_port_range_min=src_port_min,
source_port_range_max=src_port_max,
destination_port_range_min=dst_port_min,
destination_port_range_max=dst_port_max,
action=fwr['action'],
enabled=fwr['enabled'])
fwr_db = FirewallRule(
id=uuidutils.generate_uuid(),
tenant_id=tenant_id,
name=fwr['name'],
description=fwr['description'],
shared=fwr['shared'],
protocol=fwr['protocol'],
ip_version=fwr['ip_version'],
source_ip_address=fwr['source_ip_address'],
destination_ip_address=fwr['destination_ip_address'],
source_port_range_min=src_port_min,
source_port_range_max=src_port_max,
destination_port_range_min=dst_port_min,
destination_port_range_max=dst_port_max,
action=fwr['action'],
enabled=fwr['enabled'])
context.session.add(fwr_db)
return self._make_firewall_rule_dict(fwr_db)

4
neutron/db/metering/metering_db.py
View File

@ -158,8 +158,8 @@ class MeteringDbMixin(metering.MeteringPluginBase,
cidrs = [r['remote_ip_prefix'] for r in r_ips]
new_cidr_ipset = netaddr.IPSet([remote_ip_prefix])
if (netaddr.IPSet(cidrs) & new_cidr_ipset):
raise metering.MeteringLabelRuleOverlaps(remote_ip_prefix=
remote_ip_prefix)
raise metering.MeteringLabelRuleOverlaps(
remote_ip_prefix=remote_ip_prefix)
def create_metering_label_rule(self, context, metering_label_rule):
m = metering_label_rule['metering_label_rule']

12
neutron/extensions/firewall.py
View File

@ -152,15 +152,15 @@ def convert_protocol(value):
if 0 <= val <= 255:
return val
else:
raise FirewallRuleInvalidProtocol(protocol=value,
values=
fw_valid_protocol_values)
raise FirewallRuleInvalidProtocol(
protocol=value,
values=fw_valid_protocol_values)
elif value.lower() in fw_valid_protocol_values:
return value.lower()
else:
raise FirewallRuleInvalidProtocol(protocol=value,
values=
fw_valid_protocol_values)
raise FirewallRuleInvalidProtocol(
protocol=value,
values=fw_valid_protocol_values)
def convert_action_to_case_insensitive(value):

2
neutron/plugins/cisco/db/l3/device_handling_db.py
View File

@ -432,7 +432,7 @@ class DeviceHandlingMixin(object):
with context.session.begin(subtransactions=True):
hd_db = l3_models.HostingDevice(
id=hd.get('id') or uuidutils.generate_uuid(),
complementary_id = hd.get('complementary_id'),
complementary_id=hd.get('complementary_id'),
tenant_id=tenant_id,
device_id=hd.get('device_id'),
admin_state_up=hd.get('admin_state_up', True),

6
neutron/plugins/cisco/db/n1kv_db_v2.py
View File

@ -943,9 +943,9 @@ def update_profile_binding(db_session, profile_id, tenants, profile_type):
profile_id=profile_id, profile_type=profile_type).delete()
new_tenants_set = set(tenants)
for tenant_id in new_tenants_set:
tenant = n1kv_models_v2.ProfileBinding(profile_type = profile_type,
tenant_id = tenant_id,
profile_id = profile_id)
tenant = n1kv_models_v2.ProfileBinding(profile_type=profile_type,
tenant_id=tenant_id,
profile_id=profile_id)
db_session.add(tenant)

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

@ -155,9 +155,10 @@ class TestBigSwitchProxyPortsV2(test_plugin.TestPortsV2,
# stop normal patch
self.httpPatch.stop()
with patch(HTTPCON, new=fake_server.HTTPConnectionMock500):
self._delete('ports', port['port']['id'],
expected_code=
webob.exc.HTTPInternalServerError.code)
self._delete(
'ports',
port['port']['id'],
expected_code=webob.exc.HTTPInternalServerError.code)
self.httpPatch.start()
port = self._get_ports(n['network']['id'])[0]
self.assertEqual('BUILD', port['status'])

2
neutron/tests/unit/cisco/l3/test_l3_router_appliance_plugin.py
View File

@ -326,7 +326,7 @@ class RouterSchedulingTestCase(L3RouterApplianceTestCaseBase,
self.assertIsNotNone(hosting_device_1)
cfg_dh_rpc.report_non_responding_hosting_devices(
self.adminContext,
host = None,
host=None,
hosting_device_ids=[hosting_device_1['id']])
self.assertEqual(1, mock_notify.call_count)
mock_notify.assert_called_with(

6
neutron/tests/unit/cisco/n1kv/test_n1kv_plugin.py
View File

@ -587,7 +587,7 @@ class TestN1kvNetworkProfiles(N1kvPluginTestCase):
net_p_req = self.new_create_request('network_profiles', data)
net_p_req.environ['neutron.context'] = context.Context('',
self.tenant_id,
is_admin = True)
is_admin=True)
res = net_p_req.get_response(self.ext_api)
self.assertEqual(201, res.status_int)
net_p = self.deserialize(self.fmt, res)
@ -612,7 +612,7 @@ class TestN1kvNetworkProfiles(N1kvPluginTestCase):
net_p['network_profile']['id'])
update_req.environ['neutron.context'] = context.Context('',
self.tenant_id,
is_admin = True)
is_admin=True)
update_res = update_req.get_response(self.ext_api)
self.assertEqual(200, update_res.status_int)
db_session = db.get_session()
@ -637,7 +637,7 @@ class TestN1kvNetworkProfiles(N1kvPluginTestCase):
net_p['network_profile']['id'])
update_req.environ['neutron.context'] = context.Context('',
self.tenant_id,
is_admin = True)
is_admin=True)
update_res = update_req.get_response(self.ext_api)
self.assertEqual(200, update_res.status_int)
# current tenant_id should always present

29
neutron/tests/unit/db/firewall/test_db_firewall.py
View File

@ -624,9 +624,9 @@ class TestFirewallDBPlugin(FirewallPluginDbTestCase):
with self.firewall_policy() as fwp:
fwp_id = fwp['firewall_policy']['id']
attrs['firewall_policy_id'] = fwp_id
with self.firewall(firewall_policy_id=fwp_id,
admin_state_up=
ADMIN_STATE_UP):
with self.firewall(
firewall_policy_id=fwp_id,
admin_state_up=ADMIN_STATE_UP):
req = self.new_delete_request('firewall_policies', fwp_id)
res = req.get_response(self.ext_api)
self.assertEqual(res.status_int, 409)
@ -896,10 +896,11 @@ class TestFirewallDBPlugin(FirewallPluginDbTestCase):
with self.firewall_policy() as fwp:
fwp_id = fwp['firewall_policy']['id']
attrs['firewall_policy_id'] = fwp_id
with self.firewall(name=attrs['name'],
firewall_policy_id=fwp_id,
admin_state_up=
ADMIN_STATE_UP) as firewall:
with self.firewall(
name=attrs['name'],
firewall_policy_id=fwp_id,
admin_state_up=ADMIN_STATE_UP
) as firewall:
for k, v in attrs.iteritems():
self.assertEqual(firewall['firewall'][k], v)
@ -919,10 +920,10 @@ class TestFirewallDBPlugin(FirewallPluginDbTestCase):
with self.firewall_policy() as fwp:
fwp_id = fwp['firewall_policy']['id']
attrs['firewall_policy_id'] = fwp_id
with self.firewall(name=name,
firewall_policy_id=fwp_id,
admin_state_up=
ADMIN_STATE_UP) as firewall:
with self.firewall(
name=name,
firewall_policy_id=fwp_id,
admin_state_up=ADMIN_STATE_UP) as firewall:
req = self.new_show_request('firewalls',
firewall['firewall']['id'],
fmt=self.fmt)
@ -953,9 +954,9 @@ class TestFirewallDBPlugin(FirewallPluginDbTestCase):
with self.firewall_policy() as fwp:
fwp_id = fwp['firewall_policy']['id']
attrs['firewall_policy_id'] = fwp_id
with self.firewall(firewall_policy_id=fwp_id,
admin_state_up=
ADMIN_STATE_UP) as firewall:
with self.firewall(
firewall_policy_id=fwp_id,
admin_state_up=ADMIN_STATE_UP) as firewall:
data = {'firewall': {'name': name}}
req = self.new_update_request('firewalls', data,
firewall['firewall']['id'])

8
neutron/tests/unit/ml2/drivers/arista/test_arista_l3_driver.py
View File

@ -416,7 +416,7 @@ class AristaL3DriverTestCasesMlag_one_switch_failed(base.BaseTestCase):
tenant = '123'
# Make one of the switches throw an exception - i.e. fail
self.drv._servers[0].runCmds = mock.Mock(side_effect = Exception())
self.drv._servers[0].runCmds = mock.Mock(side_effect=Exception)
self.drv.create_router(None, tenant, router)
def test_delete_router_when_one_switch_fails(self):
@ -426,7 +426,7 @@ class AristaL3DriverTestCasesMlag_one_switch_failed(base.BaseTestCase):
router_id = '345'
# Make one of the switches throw an exception - i.e. fail
self.drv._servers[1].runCmds = mock.Mock(side_effect = Exception())
self.drv._servers[1].runCmds = mock.Mock(side_effect=Exception)
self.drv.delete_router(None, tenant, router_id, router)
def test_add_router_interface_when_one_switch_fails(self):
@ -439,7 +439,7 @@ class AristaL3DriverTestCasesMlag_one_switch_failed(base.BaseTestCase):
router['gip'] = '10.10.10.1'
# Make one of the switches throw an exception - i.e. fail
self.drv._servers[1].runCmds = mock.Mock(side_effect = Exception())
self.drv._servers[1].runCmds = mock.Mock(side_effect=Exception)
self.drv.add_router_interface(None, router)
def test_remove_router_interface_when_one_switch_fails(self):
@ -452,5 +452,5 @@ class AristaL3DriverTestCasesMlag_one_switch_failed(base.BaseTestCase):
router['gip'] = '10.10.10.1'
# Make one of the switches throw an exception - i.e. fail
self.drv._servers[0].runCmds = mock.Mock(side_effect = Exception())
self.drv._servers[0].runCmds = mock.Mock(side_effect=Exception)
self.drv.remove_router_interface(None, router)

79
neutron/tests/unit/ofagent/test_ofa_flows.py
View File

@ -115,15 +115,27 @@ class TestOFAgentFlows(ofa_test_base.OFATestBase):
(dp, ofp, ofpp) = br._get_dp()
call = mock.call
expected_calls = [
call(ofpp.OFPFlowMod(dp, instructions=[
ofpp.OFPInstructionActions(ofp.OFPIT_APPLY_ACTIONS,
[ofpp.OFPActionSetField(tunnel_id=112),
ofpp.OFPActionOutput(port=113),
ofpp.OFPActionOutput(port=114)]),
ofpp.OFPInstructionGotoTable(table_id=111)],
match=ofpp.OFPMatch(metadata=
meta.mk_metadata(111, meta.LOCAL)),
priority=1, table_id=110))
call(
ofpp.OFPFlowMod(
dp,
instructions=[
ofpp.OFPInstructionActions(
ofp.OFPIT_APPLY_ACTIONS,
[
ofpp.OFPActionSetField(tunnel_id=112),
ofpp.OFPActionOutput(port=113),
ofpp.OFPActionOutput(port=114)
]
),
ofpp.OFPInstructionGotoTable(table_id=111)
],
match=ofpp.OFPMatch(
metadata=meta.mk_metadata(111, meta.LOCAL)
),
priority=1,
table_id=110
)
)
]
sendmsg.assert_has_calls(expected_calls)
@ -134,11 +146,17 @@ class TestOFAgentFlows(ofa_test_base.OFATestBase):
(dp, ofp, ofpp) = br._get_dp()
call = mock.call
expected_calls = [
call(ofpp.OFPFlowMod(dp, command=ofp.OFPFC_DELETE,
match=ofpp.OFPMatch(metadata=
meta.mk_metadata(111, meta.LOCAL)),
out_group=ofp.OFPG_ANY,
out_port=ofp.OFPP_ANY, priority=0, table_id=110))
call(
ofpp.OFPFlowMod(
dp,
command=ofp.OFPFC_DELETE,
match=ofpp.OFPMatch(
metadata=meta.mk_metadata(111, meta.LOCAL)
),
out_group=ofp.OFPG_ANY,
out_port=ofp.OFPP_ANY, priority=0, table_id=110
)
)
]
sendmsg.assert_has_calls(expected_calls)
@ -189,15 +207,30 @@ class TestOFAgentFlows(ofa_test_base.OFATestBase):
match=ofpp.OFPMatch(in_port=99,
vlan_vid=151 | ofp.OFPVID_PRESENT),
priority=1, table_id=0)),
call(ofpp.OFPFlowMod(dp, instructions=[
ofpp.OFPInstructionActions(ofp.OFPIT_APPLY_ACTIONS, [
ofpp.OFPActionPushVlan(),
ofpp.OFPActionSetField(vlan_vid=151 | ofp.OFPVID_PRESENT),
ofpp.OFPActionOutput(port=99), ofpp.OFPActionPopVlan()]),
ofpp.OFPInstructionGotoTable(table_id=13)],
match=ofpp.OFPMatch(metadata=
meta.mk_metadata(150, meta.LOCAL)),
priority=1, table_id=12))
call(
ofpp.OFPFlowMod(
dp,
instructions=[
ofpp.OFPInstructionActions(
ofp.OFPIT_APPLY_ACTIONS,
[
ofpp.OFPActionPushVlan(),
ofpp.OFPActionSetField(
vlan_vid=151 | ofp.OFPVID_PRESENT
),
ofpp.OFPActionOutput(port=99),
ofpp.OFPActionPopVlan()
]
),
ofpp.OFPInstructionGotoTable(table_id=13)
],
match=ofpp.OFPMatch(
metadata=meta.mk_metadata(150, meta.LOCAL)
),
priority=1,
table_id=12
)
)
]
sendmsg.assert_has_calls(expected_calls)

73
neutron/tests/unit/services/firewall/test_fwaas_plugin.py
View File

@ -43,9 +43,10 @@ class TestFirewallCallbacks(test_db_firewall.FirewallPluginDbTestCase):
ctx = context.get_admin_context()
with self.firewall_policy() as fwp:
fwp_id = fwp['firewall_policy']['id']
with self.firewall(firewall_policy_id=fwp_id,
admin_state_up=
test_db_firewall.ADMIN_STATE_UP) as fw:
with self.firewall(
firewall_policy_id=fwp_id,
admin_state_up=test_db_firewall.ADMIN_STATE_UP
) as fw:
fw_id = fw['firewall']['id']
res = self.callbacks.set_firewall_status(ctx, fw_id,
const.ACTIVE,
@ -63,9 +64,10 @@ class TestFirewallCallbacks(test_db_firewall.FirewallPluginDbTestCase):
ctx = context.get_admin_context()
with self.firewall_policy() as fwp:
fwp_id = fwp['firewall_policy']['id']
with self.firewall(firewall_policy_id=fwp_id,
admin_state_up=
test_db_firewall.ADMIN_STATE_UP) as fw:
with self.firewall(
firewall_policy_id=fwp_id,
admin_state_up=test_db_firewall.ADMIN_STATE_UP
) as fw:
fw_id = fw['firewall']['id']
fw_db = self.plugin._get_firewall(ctx, fw_id)
fw_db['status'] = const.PENDING_DELETE
@ -131,10 +133,10 @@ class TestFirewallCallbacks(test_db_firewall.FirewallPluginDbTestCase):
res = req.get_response(self.ext_api)
attrs = self._get_test_firewall_attrs()
attrs['firewall_policy_id'] = fwp_id
with self.firewall(firewall_policy_id=fwp_id,
tenant_id=tenant_id,
admin_state_up=
test_db_firewall.ADMIN_STATE_UP) as fw:
with self.firewall(
firewall_policy_id=fwp_id,
tenant_id=tenant_id,
admin_state_up=test_db_firewall.ADMIN_STATE_UP) as fw:
fw_id = fw['firewall']['id']
res = self.callbacks.get_firewalls_for_tenant(ctx,
host='dummy')
@ -229,9 +231,10 @@ class TestFirewallPluginBase(test_db_firewall.TestFirewallDBPlugin):
with self.firewall_policy() as fwp:
fwp_id = fwp['firewall_policy']['id']
attrs['firewall_policy_id'] = fwp_id
with self.firewall(firewall_policy_id=fwp_id,
admin_state_up=
test_db_firewall.ADMIN_STATE_UP) as firewall:
with self.firewall(
firewall_policy_id=fwp_id,
admin_state_up=test_db_firewall.ADMIN_STATE_UP
) as firewall:
fw_id = firewall['firewall']['id']
res = self.callbacks.set_firewall_status(ctx, fw_id,
const.ACTIVE)
@ -252,9 +255,10 @@ class TestFirewallPluginBase(test_db_firewall.TestFirewallDBPlugin):
with self.firewall_policy() as fwp:
fwp_id = fwp['firewall_policy']['id']
attrs['firewall_policy_id'] = fwp_id
with self.firewall(firewall_policy_id=fwp_id,
admin_state_up=
test_db_firewall.ADMIN_STATE_UP) as firewall:
with self.firewall(
firewall_policy_id=fwp_id,
admin_state_up=test_db_firewall.ADMIN_STATE_UP
) as firewall:
fw_id = firewall['firewall']['id']
data = {'firewall': {'name': name}}
req = self.new_update_request('firewalls', data, fw_id)
@ -265,10 +269,11 @@ class TestFirewallPluginBase(test_db_firewall.TestFirewallDBPlugin):
ctx = context.get_admin_context()
with self.firewall_policy() as fwp:
fwp_id = fwp['firewall_policy']['id']
with self.firewall(firewall_policy_id=fwp_id,
admin_state_up=
test_db_firewall.ADMIN_STATE_UP,
tenant_id='noadmin') as firewall:
with self.firewall(
firewall_policy_id=fwp_id,
admin_state_up=test_db_firewall.ADMIN_STATE_UP,
tenant_id='noadmin'
) as firewall:
fw_id = firewall['firewall']['id']
self.callbacks.set_firewall_status(ctx, fw_id,
const.ACTIVE)
@ -286,9 +291,10 @@ class TestFirewallPluginBase(test_db_firewall.TestFirewallDBPlugin):
with self.firewall_policy() as fwp:
fwp_id = fwp['firewall_policy']['id']
attrs['firewall_policy_id'] = fwp_id
with self.firewall(firewall_policy_id=fwp_id,
admin_state_up=
test_db_firewall.ADMIN_STATE_UP):
with self.firewall(
firewall_policy_id=fwp_id,
admin_state_up=test_db_firewall.ADMIN_STATE_UP
):
data = {'firewall_policy': {'name': name}}
req = self.new_update_request('firewall_policies',
data, fwp_id)
@ -306,9 +312,10 @@ class TestFirewallPluginBase(test_db_firewall.TestFirewallDBPlugin):
req = self.new_update_request('firewall_policies', data,
fwp_id)
req.get_response(self.ext_api)
with self.firewall(firewall_policy_id=fwp_id,
admin_state_up=
test_db_firewall.ADMIN_STATE_UP):
with self.firewall(
firewall_policy_id=fwp_id,
admin_state_up=test_db_firewall.ADMIN_STATE_UP
):
data = {'firewall_rule': {'protocol': 'udp'}}
req = self.new_update_request('firewall_rules',
data, fr_id)
@ -323,9 +330,10 @@ class TestFirewallPluginBase(test_db_firewall.TestFirewallDBPlugin):
with self.firewall_policy() as fwp:
fwp_id = fwp['firewall_policy']['id']
attrs['firewall_policy_id'] = fwp_id
with self.firewall(firewall_policy_id=fwp_id,
admin_state_up=
test_db_firewall.ADMIN_STATE_UP) as firewall:
with self.firewall(
firewall_policy_id=fwp_id,
admin_state_up=test_db_firewall.ADMIN_STATE_UP
) as firewall:
fw_id = firewall['firewall']['id']
attrs = self._replace_firewall_status(attrs,
const.PENDING_CREATE,
@ -367,9 +375,10 @@ class TestFirewallPluginBase(test_db_firewall.TestFirewallDBPlugin):
req.get_response(self.ext_api)
attrs = self._get_test_firewall_attrs()
attrs['firewall_policy_id'] = fwp_id
with self.firewall(firewall_policy_id=fwp_id,
admin_state_up=
test_db_firewall.ADMIN_STATE_UP) as fw:
with self.firewall(
firewall_policy_id=fwp_id,
admin_state_up=test_db_firewall.ADMIN_STATE_UP
) as fw:
fw_id = fw['firewall']['id']
fw_rules = (
self.plugin._make_firewall_dict_with_rules(ctx,

27
neutron/tests/unit/services/loadbalancer/drivers/haproxy/test_namespace_driver.py
View File

@ -272,10 +272,11 @@ class TestHaproxyNSDriver(base.BaseTestCase):
'test_interface',
'mac_addr',
namespace='test_ns')
self.vif_driver.init_l3.assert_called_once_with('test_interface',
['10.0.0.2/24'],
namespace=
'test_ns')
self.vif_driver.init_l3.assert_called_once_with(
'test_interface',
['10.0.0.2/24'],
namespace='test_ns'
)
cmd = ['route', 'add', 'default', 'gw', '10.0.0.1']
cmd_arping = ['arping', '-U', '-I',
'test_interface', '-c',
@ -339,10 +340,11 @@ class TestHaproxyNSDriver(base.BaseTestCase):
'test_interface',
'mac_addr',
namespace='test_ns')
self.vif_driver.init_l3.assert_called_once_with('test_interface',
['10.0.0.2/24'],
namespace=
'test_ns')
self.vif_driver.init_l3.assert_called_once_with(
'test_interface',
['10.0.0.2/24'],
namespace='test_ns'
)
self.assertFalse(ip_wrap.called)
dev_exists.return_value = True
self.assertRaises(exceptions.PreexistingDeviceFailure,
@ -375,10 +377,11 @@ class TestHaproxyNSDriver(base.BaseTestCase):
'test_interface',
'mac_addr',
namespace='test_ns')
self.vif_driver.init_l3.assert_called_once_with('test_interface',
['10.0.0.2/24'],
namespace=
'test_ns')
self.vif_driver.init_l3.assert_called_once_with(
'test_interface',
['10.0.0.2/24'],
namespace='test_ns'
)
cmd = ['route', 'add', 'default', 'gw', '10.0.0.1']
ip_wrap.assert_has_calls([
mock.call('sudo_test', namespace='test_ns'),

29
neutron/tests/unit/test_extension_firewall.py
View File

@ -148,10 +148,9 @@ class FirewallExtensionTestCase(test_api_v2_extension.ExtensionTestCase):
res = self.api.post(_get_path('fw/firewall_rules', fmt=self.fmt),
self.serialize(data),
content_type='application/%s' % self.fmt)
instance.create_firewall_rule.assert_called_with(mock.ANY,
firewall_rule=
{'firewall_rule':
expected_call_args})
instance.create_firewall_rule.assert_called_with(
mock.ANY,
firewall_rule={'firewall_rule': expected_call_args})
self.assertEqual(res.status_int, exc.HTTPCreated.code)
res = self.deserialize(res)
self.assertIn('firewall_rule', res)
@ -213,10 +212,10 @@ class FirewallExtensionTestCase(test_api_v2_extension.ExtensionTestCase):
fmt=self.fmt),
self.serialize(update_data))
instance.update_firewall_rule.assert_called_with(mock.ANY,
rule_id,
firewall_rule=
update_data)
instance.update_firewall_rule.assert_called_with(
mock.ANY,
rule_id,
firewall_rule=update_data)
self.assertEqual(res.status_int, exc.HTTPOk.code)
res = self.deserialize(res)
self.assertIn('firewall_rule', res)
@ -242,9 +241,9 @@ class FirewallExtensionTestCase(test_api_v2_extension.ExtensionTestCase):
fmt=self.fmt),
self.serialize(data),
content_type='application/%s' % self.fmt)
instance.create_firewall_policy.assert_called_with(mock.ANY,
firewall_policy=
data)
instance.create_firewall_policy.assert_called_with(
mock.ANY,
firewall_policy=data)
self.assertEqual(res.status_int, exc.HTTPCreated.code)
res = self.deserialize(res)
self.assertIn('firewall_policy', res)
@ -299,10 +298,10 @@ class FirewallExtensionTestCase(test_api_v2_extension.ExtensionTestCase):
fmt=self.fmt),
self.serialize(update_data))
instance.update_firewall_policy.assert_called_with(mock.ANY,
policy_id,
firewall_policy=
update_data)
instance.update_firewall_policy.assert_called_with(
mock.ANY,
policy_id,
firewall_policy=update_data)
self.assertEqual(res.status_int, exc.HTTPOk.code)
res = self.deserialize(res)
self.assertIn('firewall_policy', res)

55
neutron/tests/unit/vmware/vshield/test_fwaas_plugin.py
View File

@ -130,12 +130,13 @@ class FirewallPluginTestCase(test_db_firewall.FirewallPluginDbTestCase,
fwp_id = fwp['firewall_policy']['id']
attrs['firewall_policy_id'] = fwp_id
attrs['router_id'] = self._create_and_get_router()
with self.firewall(name=name,
firewall_policy_id=fwp_id,
router_id=attrs['router_id'],
admin_state_up=
test_db_firewall.ADMIN_STATE_UP,
expected_res_status=201) as fw:
with self.firewall(
name=name,
firewall_policy_id=fwp_id,
router_id=attrs['router_id'],
admin_state_up=test_db_firewall.ADMIN_STATE_UP,
expected_res_status=201
) as fw:
attrs = self._replace_firewall_status(
attrs, const.PENDING_CREATE, const.ACTIVE)
for k, v in attrs.iteritems():
@ -146,11 +147,12 @@ class FirewallPluginTestCase(test_db_firewall.FirewallPluginDbTestCase,
attrs = self._get_test_firewall_attrs(name)
attrs['router_id'] = self._create_and_get_router()
with self.firewall(name=name,
router_id=attrs['router_id'],
admin_state_up=
test_db_firewall.ADMIN_STATE_UP,
expected_res_status=201) as fw:
with self.firewall(
name=name,
router_id=attrs['router_id'],
admin_state_up=test_db_firewall.ADMIN_STATE_UP,
expected_res_status=201
) as fw:
attrs = self._replace_firewall_status(
attrs, const.PENDING_CREATE, const.ACTIVE)
for k, v in attrs.iteritems():
@ -201,12 +203,13 @@ class FirewallPluginTestCase(test_db_firewall.FirewallPluginDbTestCase,
router_id = self._create_and_get_router()
with self.firewall_policy() as fwp:
fwp_id = fwp['firewall_policy']['id']
with self.firewall(name='fw',
firewall_policy_id=fwp_id,
router_id=router_id,
admin_state_up=
test_db_firewall.ADMIN_STATE_UP,
expected_res_status=201):
with self.firewall(
name='fw',
firewall_policy_id=fwp_id,
router_id=router_id,
admin_state_up=test_db_firewall.ADMIN_STATE_UP,
expected_res_status=201
):
self._delete('routers', router_id,
expected_code=webob.exc.HTTPConflict.code)
@ -311,10 +314,11 @@ class FirewallPluginTestCase(test_db_firewall.FirewallPluginDbTestCase,
firewall_rules=None,
audited=test_db_firewall.AUDITED) as fwp:
fwp_id = fwp['firewall_policy']['id']
with self.firewall(firewall_policy_id=fwp_id,
router_id=self._create_and_get_router(),
admin_state_up=
test_db_firewall.ADMIN_STATE_UP):
with self.firewall(
firewall_policy_id=fwp_id,
router_id=self._create_and_get_router(),
admin_state_up=test_db_firewall.ADMIN_STATE_UP
):
data = {'firewall_policy': {'name': name}}
req = self.new_update_request(
'firewall_policies', data, fwp['firewall_policy']['id'])
@ -372,10 +376,11 @@ class FirewallPluginTestCase(test_db_firewall.FirewallPluginDbTestCase,
with self.firewall_policy() as fwp:
fwp_id = fwp['firewall_policy']['id']
attrs['firewall_policy_id'] = fwp_id
with self.firewall(firewall_policy_id=fwp_id,
router_id=self._create_and_get_router(),
admin_state_up=
test_db_firewall.ADMIN_STATE_UP):
with self.firewall(
firewall_policy_id=fwp_id,
router_id=self._create_and_get_router(),
admin_state_up=test_db_firewall.ADMIN_STATE_UP
):
fwr_id = fwr['firewall_rule']['id']
data = {'firewall_policy': {'firewall_rules': [fwr_id]}}
req = self.new_update_request(

3
tox.ini
View File

@ -64,7 +64,6 @@ commands = python setup.py build_sphinx
# E126 continuation line over-indented for hanging indent
# E128 continuation line under-indented for visual indent
# E129 visually indented line with same indent as next logical line
# E251 unexpected spaces around keyword / parameter equals
# E265 block comment should start with ‘# ‘
# E713 test for membership should be ‘not in’
# F402 import module shadowed by loop variable
@ -79,7 +78,7 @@ commands = python setup.py build_sphinx
# H405 multi line docstring summary not separated with an empty line
# H904 Wrap long lines in parentheses instead of a backslash
# TODO(marun) H404 multi line docstring should start with a summary
ignore = E125,E126,E128,E129,E251,E265,E713,F402,F811,F812,H104,H237,H305,H307,H401,H402,H404,H405,H904
ignore = E125,E126,E128,E129,E265,E713,F402,F811,F812,H104,H237,H305,H307,H401,H402,H404,H405,H904
show-source = true
builtins = _
exclude = .venv,.git,.tox,dist,doc,*openstack/common*,*lib/python*,*egg,build,tools,.ropeproject,rally-scenarios