From dc2c46430158496878255dd2e4d6416804d7c622 Mon Sep 17 00:00:00 2001 From: Brad Hall Date: Mon, 19 Dec 2011 19:02:47 -0800 Subject: Add support for port security to QuantumManager If enabled, QuantumManager will now pass in the allowed_address_pairs that some quantum plugins understand in order to enforce port security on a given port. Any plugins that don't understand the extra fields will just ignore them. Change-Id: I640658036789b319ecefbb5e7dcdcf6b4f4ab34e --- nova/tests/test_quantum.py | 71 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 71 insertions(+) (limited to 'nova/tests') diff --git a/nova/tests/test_quantum.py b/nova/tests/test_quantum.py index 8e8a8511c..6c19f00e5 100644 --- a/nova/tests/test_quantum.py +++ b/nova/tests/test_quantum.py @@ -429,3 +429,74 @@ class QuantumNovaMACGenerationTestCase(QuantumNovaTestCase): project_id=project_id, requested_networks=requested_networks) self.assertEqual(nw_info[0][1]['mac'], fake_mac) + + +class QuantumNovaPortSecurityTestCase(QuantumNovaTestCase): + def test_port_securty(self): + self.flags(use_melange_mac_generation=True) + self.flags(quantum_use_port_security=True) + fake_mac = "ab:cd:ef:ab:cd:ef" + self.stubs.Set(melange_connection.MelangeConnection, "create_vif", + lambda w, x, y, z: fake_mac) + project_id = "fake_project1" + ctx = context.RequestContext('user1', project_id) + self._create_network(networks[0]) + + net_ids = self.net_man.q_conn.get_networks_for_tenant(project_id) + requested_networks = [(net_id, None) for net_id in net_ids['networks']] + + instance_ref = db.api.instance_create(ctx, + {"project_id": project_id}) + oldfunc = self.net_man.q_conn.create_and_attach_port + + # Make sure we get the appropriate mac set in allowed_address_pairs + # if port security is enabled. + def _instrumented_create_and_attach_port(tenant_id, net_id, + interface_id, **kwargs): + self.assertTrue('allowed_address_pairs' in kwargs.keys()) + pairs = kwargs['allowed_address_pairs'] + self.assertTrue(pairs[0]['mac_address'] == fake_mac) + self.net_man.q_conn.create_and_attach_port = oldfunc + return oldfunc(tenant_id, net_id, interface_id, **kwargs) + self.net_man.q_conn.create_and_attach_port = \ + _instrumented_create_and_attach_port + nw_info = self.net_man.allocate_for_instance(ctx, + instance_id=instance_ref['id'], host="", + instance_type_id=instance_ref['instance_type_id'], + project_id=project_id, + requested_networks=requested_networks) + self.assertEqual(nw_info[0][1]['mac'], fake_mac) + + def test_port_securty_negative(self): + self.flags(use_melange_mac_generation=True) + self.flags(quantum_use_port_security=False) + fake_mac = "ab:cd:ef:ab:cd:ef" + self.stubs.Set(melange_connection.MelangeConnection, "create_vif", + lambda w, x, y, z: fake_mac) + project_id = "fake_project1" + ctx = context.RequestContext('user1', project_id) + self._create_network(networks[0]) + + net_ids = self.net_man.q_conn.get_networks_for_tenant(project_id) + requested_networks = [(net_id, None) for net_id in net_ids['networks']] + + instance_ref = db.api.instance_create(ctx, + {"project_id": project_id}) + oldfunc = self.net_man.q_conn.create_and_attach_port + + # Make sure no pairs are passed in if port security is turned off + def _instrumented_create_and_attach_port(tenant_id, net_id, + interface_id, **kwargs): + self.assertTrue('allowed_address_pairs' in kwargs.keys()) + pairs = kwargs['allowed_address_pairs'] + self.assertTrue(len(pairs) == 0) + self.net_man.q_conn.create_and_attach_port = oldfunc + return oldfunc(tenant_id, net_id, interface_id, **kwargs) + self.net_man.q_conn.create_and_attach_port = \ + _instrumented_create_and_attach_port + nw_info = self.net_man.allocate_for_instance(ctx, + instance_id=instance_ref['id'], host="", + instance_type_id=instance_ref['instance_type_id'], + project_id=project_id, + requested_networks=requested_networks) + self.assertEqual(nw_info[0][1]['mac'], fake_mac) -- cgit