| # Copyright 2014 NEC Corporation. All rights reserved. |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); you may |
| # not use this file except in compliance with the License. You may obtain |
| # a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT |
| # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the |
| # License for the specific language governing permissions and limitations |
| # under the License. |
| |
| import six |
| from tempest_lib.common.utils import data_utils |
| from tempest_lib import exceptions as lib_exc |
| |
| from tempest.api.network import base |
| from tempest import config |
| from tempest import exceptions |
| from tempest import test |
| |
| CONF = config.CONF |
| |
| |
| class FWaaSExtensionTestJSON(base.BaseNetworkTest): |
| """ |
| Tests the following operations in the Neutron API using the REST client for |
| Neutron: |
| |
| List firewall rules |
| Create firewall rule |
| Update firewall rule |
| Delete firewall rule |
| Show firewall rule |
| List firewall policies |
| Create firewall policy |
| Update firewall policy |
| Insert firewall rule to policy |
| Remove firewall rule from policy |
| Insert firewall rule after/before rule in policy |
| Update firewall policy audited attribute |
| Delete firewall policy |
| Show firewall policy |
| List firewall |
| Create firewall |
| Update firewall |
| Delete firewall |
| Show firewall |
| """ |
| |
| @classmethod |
| def skip_checks(cls): |
| super(FWaaSExtensionTestJSON, cls).skip_checks() |
| if not test.is_extension_enabled('fwaas', 'network'): |
| msg = "FWaaS Extension not enabled." |
| raise cls.skipException(msg) |
| |
| @classmethod |
| def resource_setup(cls): |
| super(FWaaSExtensionTestJSON, cls).resource_setup() |
| cls.fw_rule = cls.create_firewall_rule("allow", "tcp") |
| cls.fw_policy = cls.create_firewall_policy() |
| |
| def _try_delete_policy(self, policy_id): |
| # delete policy, if it exists |
| try: |
| self.client.delete_firewall_policy(policy_id) |
| # if policy is not found, this means it was deleted in the test |
| except lib_exc.NotFound: |
| pass |
| |
| def _try_delete_rule(self, rule_id): |
| # delete rule, if it exists |
| try: |
| self.client.delete_firewall_rule(rule_id) |
| # if rule is not found, this means it was deleted in the test |
| except lib_exc.NotFound: |
| pass |
| |
| def _try_delete_firewall(self, fw_id): |
| # delete firewall, if it exists |
| try: |
| self.client.delete_firewall(fw_id) |
| # if firewall is not found, this means it was deleted in the test |
| except lib_exc.NotFound: |
| pass |
| |
| self.client.wait_for_resource_deletion('firewall', fw_id) |
| |
| def _wait_until_ready(self, fw_id): |
| target_states = ('ACTIVE', 'CREATED') |
| |
| def _wait(): |
| firewall = self.client.show_firewall(fw_id) |
| firewall = firewall['firewall'] |
| return firewall['status'] in target_states |
| |
| if not test.call_until_true(_wait, CONF.network.build_timeout, |
| CONF.network.build_interval): |
| status = self.client.show_firewall(fw_id)['firewall']['status'] |
| m = ("Timed out waiting for firewall %s to reach %s state(s) " |
| "after %ss, currently in %s state." % |
| (fw_id, |
| target_states, |
| CONF.network.build_interval, |
| status)) |
| raise exceptions.TimeoutException(m) |
| |
| @test.idempotent_id('1b84cf01-9c09-4ce7-bc72-b15e39076468') |
| def test_list_firewall_rules(self): |
| # List firewall rules |
| fw_rules = self.client.list_firewall_rules() |
| fw_rules = fw_rules['firewall_rules'] |
| self.assertIn((self.fw_rule['id'], |
| self.fw_rule['name'], |
| self.fw_rule['action'], |
| self.fw_rule['protocol'], |
| self.fw_rule['ip_version'], |
| self.fw_rule['enabled']), |
| [(m['id'], |
| m['name'], |
| m['action'], |
| m['protocol'], |
| m['ip_version'], |
| m['enabled']) for m in fw_rules]) |
| |
| @test.idempotent_id('563564f7-7077-4f5e-8cdc-51f37ae5a2b9') |
| def test_create_update_delete_firewall_rule(self): |
| # Create firewall rule |
| body = self.client.create_firewall_rule( |
| name=data_utils.rand_name("fw-rule"), |
| action="allow", |
| protocol="tcp") |
| fw_rule_id = body['firewall_rule']['id'] |
| |
| # Update firewall rule |
| body = self.client.update_firewall_rule(fw_rule_id, |
| shared=True) |
| self.assertTrue(body["firewall_rule"]['shared']) |
| |
| # Delete firewall rule |
| self.client.delete_firewall_rule(fw_rule_id) |
| # Confirm deletion |
| fw_rules = self.client.list_firewall_rules() |
| self.assertNotIn(fw_rule_id, |
| [m['id'] for m in fw_rules['firewall_rules']]) |
| |
| @test.idempotent_id('3ff8c08e-26ff-4034-ae48-810ed213a998') |
| def test_show_firewall_rule(self): |
| # show a created firewall rule |
| fw_rule = self.client.show_firewall_rule(self.fw_rule['id']) |
| for key, value in six.iteritems(fw_rule['firewall_rule']): |
| self.assertEqual(self.fw_rule[key], value) |
| |
| @test.idempotent_id('1086dd93-a4c0-4bbb-a1bd-6d4bc62c199f') |
| def test_list_firewall_policies(self): |
| fw_policies = self.client.list_firewall_policies() |
| fw_policies = fw_policies['firewall_policies'] |
| self.assertIn((self.fw_policy['id'], |
| self.fw_policy['name'], |
| self.fw_policy['firewall_rules']), |
| [(m['id'], |
| m['name'], |
| m['firewall_rules']) for m in fw_policies]) |
| |
| @test.idempotent_id('bbf37b6c-498c-421e-9c95-45897d3ed775') |
| def test_create_update_delete_firewall_policy(self): |
| # Create firewall policy |
| body = self.client.create_firewall_policy( |
| name=data_utils.rand_name("fw-policy")) |
| fw_policy_id = body['firewall_policy']['id'] |
| self.addCleanup(self._try_delete_policy, fw_policy_id) |
| |
| # Update firewall policy |
| body = self.client.update_firewall_policy(fw_policy_id, |
| shared=True, |
| name="updated_policy") |
| updated_fw_policy = body["firewall_policy"] |
| self.assertTrue(updated_fw_policy['shared']) |
| self.assertEqual("updated_policy", updated_fw_policy['name']) |
| |
| # Delete firewall policy |
| self.client.delete_firewall_policy(fw_policy_id) |
| # Confirm deletion |
| fw_policies = self.client.list_firewall_policies() |
| fw_policies = fw_policies['firewall_policies'] |
| self.assertNotIn(fw_policy_id, [m['id'] for m in fw_policies]) |
| |
| @test.idempotent_id('1df59b3a-517e-41d4-96f6-fc31cf4ecff2') |
| def test_show_firewall_policy(self): |
| # show a created firewall policy |
| fw_policy = self.client.show_firewall_policy(self.fw_policy['id']) |
| fw_policy = fw_policy['firewall_policy'] |
| for key, value in six.iteritems(fw_policy): |
| self.assertEqual(self.fw_policy[key], value) |
| |
| @test.idempotent_id('02082a03-3cdd-4789-986a-1327dd80bfb7') |
| def test_create_show_delete_firewall(self): |
| # Create tenant network resources required for an ACTIVE firewall |
| network = self.create_network() |
| subnet = self.create_subnet(network) |
| router = self.create_router( |
| data_utils.rand_name('router-'), |
| admin_state_up=True) |
| self.client.add_router_interface_with_subnet_id( |
| router['id'], subnet['id']) |
| |
| # Create firewall |
| body = self.client.create_firewall( |
| name=data_utils.rand_name("firewall"), |
| firewall_policy_id=self.fw_policy['id']) |
| created_firewall = body['firewall'] |
| firewall_id = created_firewall['id'] |
| self.addCleanup(self._try_delete_firewall, firewall_id) |
| |
| # Wait for the firewall resource to become ready |
| self._wait_until_ready(firewall_id) |
| |
| # show a created firewall |
| firewall = self.client.show_firewall(firewall_id) |
| firewall = firewall['firewall'] |
| |
| for key, value in six.iteritems(firewall): |
| if key == 'status': |
| continue |
| self.assertEqual(created_firewall[key], value) |
| |
| # list firewall |
| firewalls = self.client.list_firewalls() |
| firewalls = firewalls['firewalls'] |
| self.assertIn((created_firewall['id'], |
| created_firewall['name'], |
| created_firewall['firewall_policy_id']), |
| [(m['id'], |
| m['name'], |
| m['firewall_policy_id']) for m in firewalls]) |
| |
| # Delete firewall |
| self.client.delete_firewall(firewall_id) |
| |
| @test.idempotent_id('53305b4b-9897-4e01-87c0-2ae386083180') |
| def test_firewall_rule_insertion_position_removal_rule_from_policy(self): |
| # Create firewall rule |
| body = self.client.create_firewall_rule( |
| name=data_utils.rand_name("fw-rule"), |
| action="allow", |
| protocol="tcp") |
| fw_rule_id1 = body['firewall_rule']['id'] |
| self.addCleanup(self._try_delete_rule, fw_rule_id1) |
| # Create firewall policy |
| body = self.client.create_firewall_policy( |
| name=data_utils.rand_name("fw-policy")) |
| fw_policy_id = body['firewall_policy']['id'] |
| self.addCleanup(self._try_delete_policy, fw_policy_id) |
| |
| # Insert rule to firewall policy |
| self.client.insert_firewall_rule_in_policy( |
| fw_policy_id, fw_rule_id1, '', '') |
| |
| # Verify insertion of rule in policy |
| self.assertIn(fw_rule_id1, self._get_list_fw_rule_ids(fw_policy_id)) |
| # Create another firewall rule |
| body = self.client.create_firewall_rule( |
| name=data_utils.rand_name("fw-rule"), |
| action="allow", |
| protocol="icmp") |
| fw_rule_id2 = body['firewall_rule']['id'] |
| self.addCleanup(self._try_delete_rule, fw_rule_id2) |
| |
| # Insert rule to firewall policy after the first rule |
| self.client.insert_firewall_rule_in_policy( |
| fw_policy_id, fw_rule_id2, fw_rule_id1, '') |
| |
| # Verify the posiition of rule after insertion |
| fw_rule = self.client.show_firewall_rule( |
| fw_rule_id2) |
| |
| self.assertEqual(int(fw_rule['firewall_rule']['position']), 2) |
| # Remove rule from the firewall policy |
| self.client.remove_firewall_rule_from_policy( |
| fw_policy_id, fw_rule_id2) |
| # Insert rule to firewall policy before the first rule |
| self.client.insert_firewall_rule_in_policy( |
| fw_policy_id, fw_rule_id2, '', fw_rule_id1) |
| # Verify the posiition of rule after insertion |
| fw_rule = self.client.show_firewall_rule( |
| fw_rule_id2) |
| self.assertEqual(int(fw_rule['firewall_rule']['position']), 1) |
| # Remove rule from the firewall policy |
| self.client.remove_firewall_rule_from_policy( |
| fw_policy_id, fw_rule_id2) |
| # Verify removal of rule from firewall policy |
| self.assertNotIn(fw_rule_id2, self._get_list_fw_rule_ids(fw_policy_id)) |
| |
| # Remove rule from the firewall policy |
| self.client.remove_firewall_rule_from_policy( |
| fw_policy_id, fw_rule_id1) |
| |
| # Verify removal of rule from firewall policy |
| self.assertNotIn(fw_rule_id1, self._get_list_fw_rule_ids(fw_policy_id)) |
| |
| def _get_list_fw_rule_ids(self, fw_policy_id): |
| fw_policy = self.client.show_firewall_policy( |
| fw_policy_id) |
| return [ruleid for ruleid in fw_policy['firewall_policy'] |
| ['firewall_rules']] |
| |
| @test.idempotent_id('8515ca8a-0d2f-4298-b5ff-6f924e4587ca') |
| def test_update_firewall_policy_audited_attribute(self): |
| # Create firewall rule |
| body = self.client.create_firewall_rule( |
| name=data_utils.rand_name("fw-rule"), |
| action="allow", |
| protocol="icmp") |
| fw_rule_id = body['firewall_rule']['id'] |
| self.addCleanup(self._try_delete_rule, fw_rule_id) |
| # Create firewall policy |
| body = self.client.create_firewall_policy( |
| name=data_utils.rand_name('fw-policy')) |
| fw_policy_id = body['firewall_policy']['id'] |
| self.addCleanup(self._try_delete_policy, fw_policy_id) |
| self.assertFalse(body['firewall_policy']['audited']) |
| # Update firewall policy audited attribute to ture |
| self.client.update_firewall_policy(fw_policy_id, |
| audited=True) |
| # Insert Firewall rule to firewall policy |
| self.client.insert_firewall_rule_in_policy( |
| fw_policy_id, fw_rule_id, '', '') |
| body = self.client.show_firewall_policy( |
| fw_policy_id) |
| self.assertFalse(body['firewall_policy']['audited']) |