Add check that external gw port exist when metering-agent adds a rule

If router has no gateway port when metering-agent wants to add
a metering-label-rule method _process_metering_label_rules() fails
with error "cannot concatenate 'str' and 'NoneType' objects"
because there is no check that router has an external gateway port.
This patch adds this check and adds some unit test.

Closes-bug: #1527274
(cherry picked from commit f73dae0a65)

Change-Id: Ic9f626db41bfb6343187742e209402dd7d5232d1
This commit is contained in:
Sergey Belous 2016-04-18 14:39:45 +03:00 committed by Swaminathan Vasudevan
parent 6f3869d136
commit f346db1871
2 changed files with 35 additions and 1 deletions

View File

@ -135,7 +135,11 @@ class IptablesMeteringDriver(abstract_driver.MeteringAbstractDriver):
def _process_metering_label_rules(self, rm, rules, label_chain,
rules_chain):
im = rm.iptables_manager
ext_dev = self.get_external_device_name(rm.router['gw_port_id'])
ex_gw_port = rm.router.get('gw_port_id')
if not ex_gw_port:
return
ext_dev = self.get_external_device_name(ex_gw_port)
if not ext_dev:
return

View File

@ -166,6 +166,36 @@ class IptablesDriverTestCase(base.BaseTestCase):
self.v4filter_inst.assert_has_calls(calls)
def test_process_metering_label_rules_with_no_gateway_router(self):
routers = copy.deepcopy(TEST_ROUTERS)
for router in routers:
router['gw_port_id'] = None
self.metering.add_metering_label(None, routers)
calls = [mock.call.add_chain('neutron-meter-l-c5df2fe5-c60',
wrap=False),
mock.call.add_chain('neutron-meter-r-c5df2fe5-c60',
wrap=False),
mock.call.add_rule('neutron-meter-FORWARD', '-j '
'neutron-meter-r-c5df2fe5-c60',
wrap=False),
mock.call.add_rule('neutron-meter-l-c5df2fe5-c60',
'',
wrap=False),
mock.call.add_chain('neutron-meter-l-eeef45da-c60',
wrap=False),
mock.call.add_chain('neutron-meter-r-eeef45da-c60',
wrap=False),
mock.call.add_rule('neutron-meter-FORWARD', '-j '
'neutron-meter-r-eeef45da-c60',
wrap=False),
mock.call.add_rule('neutron-meter-l-eeef45da-c60',
'',
wrap=False)]
self.v4filter_inst.assert_has_calls(calls, any_order=False)
def test_add_metering_label_with_rules(self):
routers = copy.deepcopy(TEST_ROUTERS)
routers[1]['_metering_labels'][0]['rules'][0].update({