diff --git a/vmware_nsxlib/tests/unit/v3/test_qos_switching_profile.py b/vmware_nsxlib/tests/unit/v3/test_qos_switching_profile.py index 8a9c840b..aaf3fb05 100644 --- a/vmware_nsxlib/tests/unit/v3/test_qos_switching_profile.py +++ b/vmware_nsxlib/tests/unit/v3/test_qos_switching_profile.py @@ -13,12 +13,14 @@ # See the License for the specific language governing permissions and # limitations under the License. # +import copy import mock from oslo_log import log from vmware_nsxlib.tests.unit.v3 import nsxlib_testcase from vmware_nsxlib.tests.unit.v3 import test_constants +from vmware_nsxlib.v3 import nsx_constants LOG = log.getLogger(__name__) @@ -46,14 +48,16 @@ class NsxLibQosTestCase(nsxlib_testcase.NsxClientTestCase): average_bandwidth=None, description=test_constants.FAKE_NAME, qos_marking=None, - dscp=0): - body = test_constants.FAKE_QOS_PROFILE + dscp=0, direction=nsx_constants.EGRESS): + body = copy.deepcopy(test_constants.FAKE_QOS_PROFILE) body["display_name"] = test_constants.FAKE_NAME body["description"] = description + resource_type = (nsx_constants.EGRESS_SHAPING + if direction == nsx_constants.EGRESS + else nsx_constants.INGRESS_SHAPING) for shaper in body["shaper_configuration"]: - # We currently support only shaping of Egress traffic - if shaper["resource_type"] == "EgressRateShaper": + if shaper["resource_type"] == resource_type: shaper["enabled"] = shaping_enabled if burst_size: shaper["burst_size_bytes"] = burst_size @@ -103,13 +107,14 @@ class NsxLibQosTestCase(nsxlib_testcase.NsxClientTestCase): % test_constants.FAKE_QOS_PROFILE['id'], self._body(description=new_description)) - def test_enable_qos_switching_profile_shaping(self): + def _enable_qos_switching_profile_shaping( + self, direction=nsx_constants.EGRESS): """Test updating a qos-switching profile returns the correct response """ - original_profile = self._body_with_shaping() + original_profile = self._body_with_shaping(direction=direction) burst_size = 100 peak_bandwidth = 200 average_bandwidth = 300 @@ -127,19 +132,32 @@ class NsxLibQosTestCase(nsxlib_testcase.NsxClientTestCase): peak_bandwidth=peak_bandwidth, average_bandwidth=average_bandwidth, qos_marking=qos_marking, - dscp=dscp) + dscp=dscp, direction=direction) - update.assert_called_with( - 'switching-profiles/%s' - % test_constants.FAKE_QOS_PROFILE['id'], - self._body_with_shaping( + actual_body = copy.deepcopy(update.call_args[0][1]) + actual_path = update.call_args[0][0] + expected_path = ('switching-profiles/%s' % + test_constants.FAKE_QOS_PROFILE['id']) + expected_body = self._body_with_shaping( shaping_enabled=True, burst_size=burst_size, peak_bandwidth=peak_bandwidth, average_bandwidth=average_bandwidth, - qos_marking="untrusted", dscp=10)) + qos_marking="untrusted", dscp=10, + direction=direction) + self.assertEqual(expected_path, actual_path) + self.assertEqual(expected_body, actual_body) - def test_disable_qos_switching_profile_shaping(self): + def test_enable_qos_switching_profile_egress_shaping(self): + self._enable_qos_switching_profile_shaping( + direction=nsx_constants.EGRESS) + + def test_enable_qos_switching_profile_ingress_shaping(self): + self._enable_qos_switching_profile_shaping( + direction=nsx_constants.INGRESS) + + def _disable_qos_switching_profile_shaping( + self, direction=nsx_constants.EGRESS): """Test updating a qos-switching profile returns the correct response @@ -153,7 +171,7 @@ class NsxLibQosTestCase(nsxlib_testcase.NsxClientTestCase): peak_bandwidth=peak_bandwidth, average_bandwidth=average_bandwidth, qos_marking="untrusted", - dscp=10) + dscp=10, direction=direction) with mock.patch.object(self.nsxlib.client, 'get', return_value=original_profile): @@ -161,12 +179,25 @@ class NsxLibQosTestCase(nsxlib_testcase.NsxClientTestCase): # update the bw shaping of the profile self.nsxlib.qos_switching_profile.update_shaping( test_constants.FAKE_QOS_PROFILE['id'], - shaping_enabled=False, qos_marking="trusted") + shaping_enabled=False, qos_marking="trusted", + direction=direction) - update.assert_called_with( - 'switching-profiles/%s' - % test_constants.FAKE_QOS_PROFILE['id'], - self._body_with_shaping(qos_marking="trusted")) + actual_body = copy.deepcopy(update.call_args[0][1]) + actual_path = update.call_args[0][0] + expected_path = ('switching-profiles/%s' % + test_constants.FAKE_QOS_PROFILE['id']) + expected_body = self._body_with_shaping(qos_marking="trusted", + direction=direction) + self.assertEqual(expected_path, actual_path) + self.assertEqual(expected_body, actual_body) + + def test_disable_qos_switching_profile_egress_shaping(self): + self._disable_qos_switching_profile_shaping( + direction=nsx_constants.EGRESS) + + def test_disable_qos_switching_profile_ingress_shaping(self): + self._disable_qos_switching_profile_shaping( + direction=nsx_constants.INGRESS) def test_delete_qos_switching_profile(self): """Test deleting qos-switching-profile""" diff --git a/vmware_nsxlib/v3/__init__.py b/vmware_nsxlib/v3/__init__.py index b513fd1a..8fd5d051 100644 --- a/vmware_nsxlib/v3/__init__.py +++ b/vmware_nsxlib/v3/__init__.py @@ -235,11 +235,17 @@ class NsxLibQosSwitchingProfile(utils.NsxLibApiBase): body["description"] = description return body + def _get_resource_type(self, direction): + if direction == nsx_constants.EGRESS: + return nsx_constants.EGRESS_SHAPING + return nsx_constants.INGRESS_SHAPING + def _enable_shaping_in_args(self, body, burst_size=None, - peak_bandwidth=None, average_bandwidth=None): + peak_bandwidth=None, average_bandwidth=None, + direction=None): + resource_type = self._get_resource_type(direction) for shaper in body["shaper_configuration"]: - # We currently supports only shaping of Egress traffic - if shaper["resource_type"] == "EgressRateShaper": + if shaper["resource_type"] == resource_type: shaper["enabled"] = True if burst_size: shaper["burst_size_bytes"] = burst_size @@ -251,10 +257,10 @@ class NsxLibQosSwitchingProfile(utils.NsxLibApiBase): return body - def _disable_shaping_in_args(self, body): + def _disable_shaping_in_args(self, body, direction=None): + resource_type = self._get_resource_type(direction) for shaper in body["shaper_configuration"]: - # We currently supports only shaping of Egress traffic - if shaper["resource_type"] == "EgressRateShaper": + if shaper["resource_type"] == resource_type: shaper["enabled"] = False shaper["burst_size_bytes"] = 0 shaper["peak_bandwidth_mbps"] = 0 @@ -289,7 +295,8 @@ class NsxLibQosSwitchingProfile(utils.NsxLibApiBase): burst_size=None, peak_bandwidth=None, average_bandwidth=None, - qos_marking=None, dscp=None): + qos_marking=None, dscp=None, + direction=nsx_constants.INGRESS): resource = 'switching-profiles/%s' % profile_id # get the current configuration body = self.get(profile_id) @@ -298,9 +305,10 @@ class NsxLibQosSwitchingProfile(utils.NsxLibApiBase): body = self._enable_shaping_in_args( body, burst_size=burst_size, peak_bandwidth=peak_bandwidth, - average_bandwidth=average_bandwidth) + average_bandwidth=average_bandwidth, + direction=direction) else: - body = self._disable_shaping_in_args(body) + body = self._disable_shaping_in_args(body, direction=direction) body = self._update_dscp_in_args(body, qos_marking, dscp) return self._update_resource_with_retry(resource, body) diff --git a/vmware_nsxlib/v3/nsx_constants.py b/vmware_nsxlib/v3/nsx_constants.py index 2a655c60..0b11199c 100644 --- a/vmware_nsxlib/v3/nsx_constants.py +++ b/vmware_nsxlib/v3/nsx_constants.py @@ -94,3 +94,9 @@ IPV4_IPV6 = 'IPV4_IPV6' LOCAL_IP_PREFIX = 'local_ip_prefix' LOGGING = 'logging' + +# QoS directions egress/ingress +EGRESS = 'egress' +INGRESS = 'ingress' +EGRESS_SHAPING = 'EgressRateShaper' +INGRESS_SHAPING = 'IngressRateShaper'