x/vmware-nsx
Code Issues Proposed changes

Ensure to count firewalls in target tenant

Browse Source 
Previously admin tenant cannot create a firewall if other tenant
already created a firewall. We need to count firewalls only in
a target tenant.

Change-Id: I3e6d151d00d4a487bdd858e94929fab8960511a2
Closes-Bug: #1258438
This commit is contained in:
Akihiro Motoki 2014-03-15 07:57:34 +09:00
parent 061f27b489
commit e3b7ceeb3b
3 changed files with 15 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
neutron/services/firewall/fwaas_plugin.py
View File

@ -225,7 +225,8 @@ class FirewallPlugin(firewall_db.Firewall_db_mixin):
LOG.debug(_("create_firewall() called")) LOG.debug(_("create_firewall() called"))
tenant_id = self._get_tenant_id_for_create(context, tenant_id = self._get_tenant_id_for_create(context,
firewall['firewall']) firewall['firewall'])
fw_count = self.get_firewalls_count(context) fw_count = self.get_firewalls_count(context,
filters={'tenant_id': [tenant_id]})
if fw_count: if fw_count:
raise FirewallCountExceeded(tenant_id=tenant_id) raise FirewallCountExceeded(tenant_id=tenant_id)
firewall['firewall']['status'] = const.PENDING_CREATE firewall['firewall']['status'] = const.PENDING_CREATE

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

@ -153,9 +153,10 @@ class FirewallPluginDbTestCase(test_db_plugin.NeutronDbPluginV2TestCase):
def _create_firewall_policy(self, fmt, name, description, shared, def _create_firewall_policy(self, fmt, name, description, shared,
firewall_rules, audited, firewall_rules, audited,
expected_res_status=None, **kwargs): expected_res_status=None, **kwargs):
tenant_id = kwargs.get('tenant_id', self._tenant_id)
data = {'firewall_policy': {'name': name, data = {'firewall_policy': {'name': name,
'description': description, 'description': description,
'tenant_id': self._tenant_id, 'tenant_id': tenant_id,
'shared': shared, 'shared': shared,
'firewall_rules': firewall_rules, 'firewall_rules': firewall_rules,
'audited': audited}} 'audited': audited}}
@ -199,8 +200,9 @@ class FirewallPluginDbTestCase(test_db_plugin.NeutronDbPluginV2TestCase):
destination_ip_address, source_port, destination_ip_address, source_port,
destination_port, action, enabled, destination_port, action, enabled,
expected_res_status=None, **kwargs): expected_res_status=None, **kwargs):
tenant_id = kwargs.get('tenant_id', self._tenant_id)
data = {'firewall_rule': {'name': name, data = {'firewall_rule': {'name': name,
'tenant_id': self._tenant_id, 'tenant_id': tenant_id,
'shared': shared, 'shared': shared,
'protocol': protocol, 'protocol': protocol,
'ip_version': ip_version, 'ip_version': ip_version,
@ -248,11 +250,12 @@ class FirewallPluginDbTestCase(test_db_plugin.NeutronDbPluginV2TestCase):
def _create_firewall(self, fmt, name, description, firewall_policy_id, def _create_firewall(self, fmt, name, description, firewall_policy_id,
admin_state_up=True, expected_res_status=None, admin_state_up=True, expected_res_status=None,
**kwargs): **kwargs):
tenant_id = kwargs.get('tenant_id', self._tenant_id)
data = {'firewall': {'name': name, data = {'firewall': {'name': name,
'description': description, 'description': description,
'firewall_policy_id': firewall_policy_id, 'firewall_policy_id': firewall_policy_id,
'admin_state_up': admin_state_up, 'admin_state_up': admin_state_up,
'tenant_id': self._tenant_id}} 'tenant_id': tenant_id}}
firewall_req = self.new_create_request('firewalls', data, fmt) firewall_req = self.new_create_request('firewalls', data, fmt)
firewall_res = firewall_req.get_response(self.ext_api) firewall_res = firewall_req.get_response(self.ext_api)

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

@ -200,6 +200,13 @@ class TestFirewallPluginBase(test_db_firewall.TestFirewallDBPlugin):
firewall_policy_id=None, admin_state_up=True) firewall_policy_id=None, admin_state_up=True)
self.assertEqual(res.status_int, 500) self.assertEqual(res.status_int, 500)
def test_create_firewall_admin_not_affected_by_other_tenant(self):
# Create fw with admin after creating fw with other tenant
with self.firewall(tenant_id='other-tenant') as fw1:
with self.firewall() as fw2:
self.assertEqual('other-tenant', fw1['firewall']['tenant_id'])
self.assertEqual(self._tenant_id, fw2['firewall']['tenant_id'])
def test_update_firewall(self): def test_update_firewall(self):
ctx = context.get_admin_context() ctx = context.get_admin_context()
name = "new_firewall1" name = "new_firewall1"