test provider fw rules at the virt/ipteables layer.

lowercase protocol names in admin api to match what the firewall driver expects.
add provider fw rule chain in iptables6 as well.
fix a couple of small typos and copy-paste errors.

1 files changed, 64 insertions, 1 deletions

diff --git a/nova/tests/test_virt.py b/nova/tests/test_virt.py
index 958c8e3e2..34ecd09c6 100644
--- a/nova/tests/test_virt.py
+++ b/nova/tests/test_virt.py
@@ -566,7 +566,9 @@ class IptablesFirewallTestCase(test.TestCase):
         self.network = utils.import_object(FLAGS.network_manager)
 
         class FakeLibvirtConnection(object):
-            pass
+            def nwfilterDefineXML(*args, **kwargs):
+                """setup_basic_rules in nwfilter calls this."""
+                pass
         self.fake_libvirt_connection = FakeLibvirtConnection()
         self.fw = libvirt_conn.IptablesFirewallDriver(
                       get_connection=lambda: self.fake_libvirt_connection)
@@ -728,6 +730,67 @@ class IptablesFirewallTestCase(test.TestCase):
                         "TCP port 80/81 acceptance rule wasn't added")
         db.instance_destroy(admin_ctxt, instance_ref['id'])
 
+    def test_provider_firewall_rules(self):
+        # keep from changing state of actual firewall
+        #def fake_function(*args, **kwargs):
+        #    pass
+        #self.fw.iptables.apply = fake_function
+        #self.fw.nwfilter.setup_basic_filtering = fake_function
+
+        # setup basic instance data
+        instance_ref = db.instance_create(self.context,
+                                          {'user_id': 'fake',
+                                          'project_id': 'fake',
+                                          'mac_address': '56:12:12:12:12:12'})
+        ip = '10.11.12.13'
+        network_ref = db.project_get_network(self.context, 'fake')
+        admin_ctxt = context.get_admin_context()
+        fixed_ip = {'address': ip, 'network_id': network_ref['id']}
+        db.fixed_ip_create(admin_ctxt, fixed_ip)
+        db.fixed_ip_update(admin_ctxt, ip, {'allocated': True,
+                                            'instance_id': instance_ref['id']})
+        # FRAGILE: peeks at how the firewall names chains
+        chain_name = 'inst-%s' % instance_ref['id']
+
+        # create a firewall via setup_basic_filtering like libvirt_conn.spawn
+        # should have a chain with 0 rules
+        self.fw.setup_basic_filtering(instance_ref, network_info=None)
+        self.assertTrue('provider' in self.fw.iptables.ipv4['filter'].chains)
+        rules = [rule for rule in self.fw.iptables.ipv4['filter'].rules
+                      if rule.chain == 'provider']
+        self.assertEqual(0, len(rules))
+
+        # add a rule and send the update message, check for 1 rule
+        provider_fw0 = db.provider_fw_rule_create(admin_ctxt,
+                                                  {'protocol': 'tcp',
+                                                   'cidr': '10.99.99.99/32',
+                                                   'from_port': 1,
+                                                   'to_port': 65535})
+        self.fw.refresh_provider_fw_rules()
+        rules = [rule for rule in self.fw.iptables.ipv4['filter'].rules
+                      if rule.chain == 'provider']
+        self.assertEqual(1, len(rules))
+
+        # Add another, refresh, and make sure number of rules goes to two
+        provider_fw1 = db.provider_fw_rule_create(admin_ctxt,
+                                                  {'protocol': 'udp',
+                                                   'cidr': '10.99.99.99/32',
+                                                   'from_port': 1,
+                                                   'to_port': 65535})
+        self.fw.refresh_provider_fw_rules()
+        rules = [rule for rule in self.fw.iptables.ipv4['filter'].rules
+                      if rule.chain == 'provider']
+        self.assertEqual(2, len(rules))
+
+        # create the instance filter and make sure it has a jump rule
+        self.fw.prepare_instance_filter(instance_ref, network_info=None)
+        self.fw.apply_instance_filter(instance_ref)
+        inst_rules = [rule for rule in self.fw.iptables.ipv4['filter'].rules
+                           if rule.chain == chain_name]
+        jump_rules = [rule for rule in inst_rules if '-j' in rule.rule]
+        prov_rules = [rule for rule in jump_rules if 'provider' in rule.rule]
+        self.assertEqual(1, len(prov_rules))
+
 
 class NWFilterTestCase(test.TestCase):
     def setUp(self):