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