"""Test cloudbridge.security modules.""" import cloudbridge.cloud.base.helpers as cb_helpers from cloudbridge.cloud.interfaces.exceptions import DuplicateResourceException from cloudbridge.cloud.interfaces.resources import KeyPair from cloudbridge.cloud.interfaces.resources import TrafficDirection from cloudbridge.cloud.interfaces.resources import VMFirewall from cloudbridge.cloud.interfaces.resources import VMFirewallRule from test import helpers from test.helpers import ProviderTestBase from test.helpers import standard_interface_tests as sit class CloudSecurityServiceTestCase(ProviderTestBase): _multiprocess_can_split_ = True @helpers.skipIfNoService(['security.vm_firewalls']) def test_storage_services_event_pattern(self): expected_event = ".".join((self.provider.PROVIDER_ID, "security.key_pairs")) self.assertEqual( self.provider.security.key_pairs._service_event_pattern, expected_event, "Event pattern for {} service should be '{}', " "but found '{}'.".format("key_pairs", expected_event, self.provider.security. key_pairs. _service_event_pattern)) expected_event = ".".join((self.provider.PROVIDER_ID, "security.vm_firewalls")) self.assertEqual( self.provider.security.vm_firewalls._service_event_pattern, expected_event, "Event pattern for {} service should be '{}', " "but found '{}'.".format("vm_firewalls", expected_event, self.provider.security.vm_firewalls. _service_event_pattern)) @helpers.skipIfNoService(['security.key_pairs']) def test_crud_key_pair_service(self): def create_kp(name): return self.provider.security.key_pairs.create(name=name) def cleanup_kp(kp): if kp: self.provider.security.key_pairs.delete(kp.id) def extra_tests(kp): # Recreating existing keypair should raise an exception with self.assertRaises(DuplicateResourceException): self.provider.security.key_pairs.create(name=kp.name) sit.check_crud(self, self.provider.security.key_pairs, KeyPair, "cb-crudkp", create_kp, cleanup_kp, extra_test_func=extra_tests) @helpers.skipIfNoService(['security.key_pairs']) def test_key_pair_properties(self): name = 'cb-kpprops-{0}'.format(helpers.get_uuid()) kp = self.provider.security.key_pairs.create(name=name) with cb_helpers.cleanup_action(lambda: kp.delete()): self.assertIsNotNone( kp.material, "KeyPair material is empty but it should not be.") # get the keypair again - keypair material should now be empty kp = self.provider.security.key_pairs.get(kp.id) self.assertIsNone(kp.material, "Keypair material should now be empty") @helpers.skipIfNoService(['security.key_pairs']) def test_import_key_pair(self): name = 'cb-kpimport-{0}'.format(helpers.get_uuid()) public_key, _ = cb_helpers.generate_key_pair() kp = self.provider.security.key_pairs.create( name=name, public_key_material=public_key) with cb_helpers.cleanup_action(lambda: kp.delete()): self.assertIsNone(kp.material, "Private KeyPair material should" " be None when key is imported.") @helpers.skipIfNoService(['security.vm_firewalls']) def test_crud_vm_firewall(self): subnet = helpers.get_or_create_default_subnet(self.provider) net = subnet.network def create_fw(label): return self.provider.security.vm_firewalls.create( label=label, description=label, network=net.id) def cleanup_fw(fw): if fw: fw.delete() def network_id_test(fw): # Checking that the network ID is returned correctly self.assertEqual(fw.network_id, net.id) sit.check_crud(self, self.provider.security.vm_firewalls, VMFirewall, "cb-crudfw", create_fw, cleanup_fw, extra_test_func=network_id_test) @helpers.skipIfNoService(['security.vm_firewalls']) def test_vm_firewall_properties(self): label = 'cb-propfw-{0}'.format(helpers.get_uuid()) # Declare these variables and late binding will allow # the cleanup method access to the most current values fw = None with cb_helpers.cleanup_action(lambda: helpers.cleanup_test_resources( vm_firewall=fw)): subnet = helpers.get_or_create_default_subnet(self.provider) net = subnet.network fw = self.provider.security.vm_firewalls.create( label=label, description=label, network=net.id) self.assertEqual(label, fw.description) @helpers.skipIfNoService(['security.vm_firewalls']) def test_crud_vm_firewall_rules(self): label = 'cb-crudfw-rules-{0}'.format(helpers.get_uuid()) subnet = helpers.get_or_create_default_subnet(self.provider) net = subnet.network fw = None with cb_helpers.cleanup_action(lambda: fw.delete()): fw = self.provider.security.vm_firewalls.create( label=label, description=label, network=net.id) def create_fw_rule(label): return fw.rules.create( direction=TrafficDirection.INBOUND, protocol='tcp', from_port=1111, to_port=1111, cidr='0.0.0.0/0') def cleanup_fw_rule(rule): if rule: rule.delete() sit.check_crud(self, fw.rules, VMFirewallRule, "cb-crudfwrule", create_fw_rule, cleanup_fw_rule, skip_name_check=True) @helpers.skipIfNoService(['security.vm_firewalls']) def test_vm_firewall_rule_properties(self): label = 'cb-propfwrule-{0}'.format(helpers.get_uuid()) # Declare these variables and late binding will allow # the cleanup method access to the most current values fw = None with cb_helpers.cleanup_action(lambda: helpers.cleanup_test_resources( vm_firewall=fw)): subnet = helpers.get_or_create_default_subnet(self.provider) net = subnet.network fw = self.provider.security.vm_firewalls.create( label=label, description=label, network=net.id) rule = fw.rules.create( direction=TrafficDirection.INBOUND, protocol='tcp', from_port=1111, to_port=1111, cidr='0.0.0.0/0') self.assertEqual(rule.direction, TrafficDirection.INBOUND) self.assertEqual(rule.protocol, 'tcp') self.assertEqual(rule.from_port, 1111) self.assertEqual(rule.to_port, 1111) self.assertEqual(rule.cidr, '0.0.0.0/0') @helpers.skipIfNoService(['security.vm_firewalls']) def test_vm_firewall_rule_add_twice(self): label = 'cb-fwruletwice-{0}'.format(helpers.get_uuid()) # Declare these variables and late binding will allow # the cleanup method access to the most current values fw = None with cb_helpers.cleanup_action(lambda: helpers.cleanup_test_resources( vm_firewall=fw)): subnet = helpers.get_or_create_default_subnet(self.provider) net = subnet.network fw = self.provider.security.vm_firewalls.create( label=label, description=label, network=net.id) rule = fw.rules.create( direction=TrafficDirection.INBOUND, protocol='tcp', from_port=1111, to_port=1111, cidr='0.0.0.0/0') # attempting to add the same rule twice should succeed same_rule = fw.rules.create( direction=TrafficDirection.INBOUND, protocol='tcp', from_port=1111, to_port=1111, cidr='0.0.0.0/0') self.assertEqual(rule, same_rule) @helpers.skipIfNoService(['security.vm_firewalls']) def test_vm_firewall_group_rule(self): label = 'cb-fwrule-{0}'.format(helpers.get_uuid()) # Declare these variables and late binding will allow # the cleanup method access to the most current values fw = None with cb_helpers.cleanup_action(lambda: helpers.cleanup_test_resources( vm_firewall=fw)): subnet = helpers.get_or_create_default_subnet(self.provider) net = subnet.network fw = self.provider.security.vm_firewalls.create( label=label, description=label, network=net.id) rule = fw.rules.create( direction=TrafficDirection.INBOUND, src_dest_fw=fw, protocol='tcp', from_port=1, to_port=65535) self.assertTrue( rule.src_dest_fw.label == fw.label, "Expected VM firewall rule label {0}. Got {1}." .format(fw.label, rule.src_dest_fw.label)) for r in fw.rules: r.delete() fw = self.provider.security.vm_firewalls.get(fw.id) # update self.assertTrue( len(list(fw.rules)) == 0, "Deleting VMFirewallRule should delete it: {0}".format( fw.rules)) fwl = self.provider.security.vm_firewalls.list() found_fw = [f for f in fwl if f.label == label] self.assertTrue( len(found_fw) == 0, "VM firewall {0} should have been deleted but still exists." .format(label))