Update port functional tests for qos agent

Change-Id: I4a1f4ec1ed9a9104fe7e5bbce66147d8ea6c0f27
Partially-Implements: quantum-qos-api

diff --git a/neutron/tests/common/agents/l2_extensions.py b/neutron/tests/common/agents/l2_extensions.py
index 39ae0bdd741bbd4365a5946c558f35fc7dcfd629..0d46d3676d438bcf068a21349c7bf0dce36124c2 100644 (file)
--- a/neutron/tests/common/agents/l2_extensions.py
+++ b/neutron/tests/common/agents/l2_extensions.py
@@ -18,9 +18,10 @@ from neutron.agent.linux import utils as agent_utils
 
 def wait_until_bandwidth_limit_rule_applied(bridge, port_vif, rule):
     def _bandwidth_limit_rule_applied():
-        max_rate, burst = (
-            bridge.get_qos_bw_limit_for_port(port_vif))
-        return (max_rate == rule.max_kbps and
-                burst == rule.max_burst_kbps)
+        bw_rule = bridge.get_qos_bw_limit_for_port(port_vif)
+        expected = None, None
+        if rule:
+            expected = rule.max_kbps, rule.max_burst_kbps
+        return bw_rule == expected
 
     agent_utils.wait_until_true(_bandwidth_limit_rule_applied)

diff --git a/neutron/tests/functional/agent/l2/extensions/test_ovs_agent_qos_extension.py b/neutron/tests/functional/agent/l2/extensions/test_ovs_agent_qos_extension.py
index 32c13be61e4bf8646b1a8df87d2e9a4704f4646b..8fd8ee18b40ee1e104ff0123e7c1ee1a3b6e805a 100644 (file)
--- a/neutron/tests/functional/agent/l2/extensions/test_ovs_agent_qos_extension.py
+++ b/neutron/tests/functional/agent/l2/extensions/test_ovs_agent_qos_extension.py
@@ -151,3 +151,36 @@ class TestOVSAgentQosExtension(OVSAgentQoSExtensionTestFramework):
                                                      policy_copy.rules[0])
         self._assert_bandwidth_limit_rule_is_set(self.ports[0],
                                                  policy_copy.rules[0])
+
+    def test_port_qos_disassociation(self):
+        """Test that qos_policy_id set to None will remove all qos rules from
+           given port.
+        """
+        port_dict = self._create_test_port_dict()
+        port_dict['qos_policy_id'] = TEST_POLICY_ID1
+        self.setup_agent_and_ports([port_dict])
+        self.wait_until_ports_state(self.ports, up=True)
+        self.wait_until_bandwidth_limit_rule_applied(port_dict,
+                                                     TEST_BW_LIMIT_RULE_1)
+
+        port_dict['qos_policy_id'] = None
+        self.agent.port_update(None, port=port_dict)
+
+        self.wait_until_bandwidth_limit_rule_applied(port_dict, None)
+
+    def test_port_qos_update_policy_id(self):
+        """Test that change of qos policy id on given port refreshes all its
+           rules.
+        """
+        port_dict = self._create_test_port_dict()
+        port_dict['qos_policy_id'] = TEST_POLICY_ID1
+        self.setup_agent_and_ports([port_dict])
+        self.wait_until_ports_state(self.ports, up=True)
+        self.wait_until_bandwidth_limit_rule_applied(port_dict,
+                                                     TEST_BW_LIMIT_RULE_1)
+
+        port_dict['qos_policy_id'] = TEST_POLICY_ID2
+        self.agent.port_update(None, port=port_dict)
+
+        self.wait_until_bandwidth_limit_rule_applied(port_dict,
+                                                     TEST_BW_LIMIT_RULE_2)