1# Copyright 2017 OVH SAS 2# Licensed under the Apache License, Version 2.0 (the "License"); you may 3# not use this file except in compliance with the License. You may obtain 4# a copy of the License at 5# 6# http://www.apache.org/licenses/LICENSE-2.0 7# 8# Unless required by applicable law or agreed to in writing, software 9# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 10# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 11# License for the specific language governing permissions and limitations 12# under the License. 13 14""" 15test_qos_bandwidth_limit_rule 16---------------------------------- 17 18Functional tests for `shade`QoS bandwidth limit methods. 19""" 20 21from openstack.cloud.exc import OpenStackCloudException 22from openstack.tests.functional import base 23 24 25class TestQosBandwidthLimitRule(base.BaseFunctionalTest): 26 def setUp(self): 27 super(TestQosBandwidthLimitRule, self).setUp() 28 if not self.operator_cloud.has_service('network'): 29 self.skipTest('Network service not supported by cloud') 30 if not self.operator_cloud._has_neutron_extension('qos'): 31 self.skipTest('QoS network extension not supported by cloud') 32 33 policy_name = self.getUniqueString('qos_policy') 34 self.policy = self.operator_cloud.create_qos_policy(name=policy_name) 35 36 self.addCleanup(self._cleanup_qos_policy) 37 38 def _cleanup_qos_policy(self): 39 try: 40 self.operator_cloud.delete_qos_policy(self.policy['id']) 41 except Exception as e: 42 raise OpenStackCloudException(e) 43 44 def test_qos_bandwidth_limit_rule_lifecycle(self): 45 max_kbps = 1500 46 max_burst_kbps = 500 47 updated_max_kbps = 2000 48 49 # Create bw limit rule 50 rule = self.operator_cloud.create_qos_bandwidth_limit_rule( 51 self.policy['id'], 52 max_kbps=max_kbps, 53 max_burst_kbps=max_burst_kbps) 54 self.assertIn('id', rule) 55 self.assertEqual(max_kbps, rule['max_kbps']) 56 self.assertEqual(max_burst_kbps, rule['max_burst_kbps']) 57 58 # Now try to update rule 59 updated_rule = self.operator_cloud.update_qos_bandwidth_limit_rule( 60 self.policy['id'], 61 rule['id'], 62 max_kbps=updated_max_kbps) 63 self.assertIn('id', updated_rule) 64 self.assertEqual(updated_max_kbps, updated_rule['max_kbps']) 65 self.assertEqual(max_burst_kbps, updated_rule['max_burst_kbps']) 66 67 # List rules from policy 68 policy_rules = self.operator_cloud.list_qos_bandwidth_limit_rules( 69 self.policy['id']) 70 self.assertEqual([updated_rule], policy_rules) 71 72 # Delete rule 73 self.operator_cloud.delete_qos_bandwidth_limit_rule( 74 self.policy['id'], updated_rule['id']) 75 76 # Check if there is no rules in policy 77 policy_rules = self.operator_cloud.list_qos_bandwidth_limit_rules( 78 self.policy['id']) 79 self.assertEqual([], policy_rules) 80 81 def test_create_qos_bandwidth_limit_rule_direction(self): 82 if not self.operator_cloud._has_neutron_extension( 83 'qos-bw-limit-direction'): 84 self.skipTest("'qos-bw-limit-direction' network extension " 85 "not supported by cloud") 86 max_kbps = 1500 87 direction = "ingress" 88 updated_direction = "egress" 89 90 # Create bw limit rule 91 rule = self.operator_cloud.create_qos_bandwidth_limit_rule( 92 self.policy['id'], 93 max_kbps=max_kbps, 94 direction=direction) 95 self.assertIn('id', rule) 96 self.assertEqual(max_kbps, rule['max_kbps']) 97 self.assertEqual(direction, rule['direction']) 98 99 # Now try to update direction in rule 100 updated_rule = self.operator_cloud.update_qos_bandwidth_limit_rule( 101 self.policy['id'], 102 rule['id'], 103 direction=updated_direction) 104 self.assertIn('id', updated_rule) 105 self.assertEqual(max_kbps, updated_rule['max_kbps']) 106 self.assertEqual(updated_direction, updated_rule['direction']) 107