summaryrefslogtreecommitdiff
path: root/openstackclient/tests
diff options
context:
space:
mode:
Diffstat (limited to 'openstackclient/tests')
-rw-r--r--openstackclient/tests/functional/network/v2/test_network_qos_rule.py16
-rw-r--r--openstackclient/tests/unit/common/test_project_purge.py314
-rw-r--r--openstackclient/tests/unit/identity/v3/test_trust.py2
-rw-r--r--openstackclient/tests/unit/network/v2/fakes.py1
-rw-r--r--openstackclient/tests/unit/network/v2/test_network.py3
-rw-r--r--openstackclient/tests/unit/network/v2/test_network_qos_rule.py70
6 files changed, 381 insertions, 25 deletions
diff --git a/openstackclient/tests/functional/network/v2/test_network_qos_rule.py b/openstackclient/tests/functional/network/v2/test_network_qos_rule.py
index c6437d9c..056bc1f6 100644
--- a/openstackclient/tests/functional/network/v2/test_network_qos_rule.py
+++ b/openstackclient/tests/functional/network/v2/test_network_qos_rule.py
@@ -169,6 +169,8 @@ class NetworkQosRuleTestsBandwidthLimit(common.NetworkTests):
MAX_KBPS_MODIFIED = 15000
MAX_BURST_KBITS = 1400
MAX_BURST_KBITS_MODIFIED = 1800
+ RULE_DIRECTION = 'egress'
+ RULE_DIRECTION_MODIFIED = 'ingress'
HEADERS = ['ID']
FIELDS = ['id']
TYPE = 'bandwidth-limit'
@@ -187,6 +189,7 @@ class NetworkQosRuleTestsBandwidthLimit(common.NetworkTests):
'--type ' + cls.TYPE + ' ' +
'--max-kbps ' + str(cls.MAX_KBPS) + ' ' +
'--max-burst-kbits ' + str(cls.MAX_BURST_KBITS) + ' ' +
+ '--' + cls.RULE_DIRECTION + ' ' +
cls.QOS_POLICY_NAME +
opts
)
@@ -226,14 +229,13 @@ class NetworkQosRuleTestsBandwidthLimit(common.NetworkTests):
self.openstack('network qos rule set --max-kbps ' +
str(self.MAX_KBPS_MODIFIED) + ' --max-burst-kbits ' +
str(self.MAX_BURST_KBITS_MODIFIED) + ' ' +
+ '--' + self.RULE_DIRECTION_MODIFIED + ' ' +
self.QOS_POLICY_NAME + ' ' + self.RULE_ID)
- opts = self.get_opts(['max_kbps'])
+ opts = self.get_opts(['direction', 'max_burst_kbps', 'max_kbps'])
raw_output = self.openstack('network qos rule show ' +
self.QOS_POLICY_NAME + ' ' + self.RULE_ID +
opts)
- self.assertEqual(str(self.MAX_KBPS_MODIFIED) + "\n", raw_output)
- opts = self.get_opts(['max_burst_kbps'])
- raw_output = self.openstack('network qos rule show ' +
- self.QOS_POLICY_NAME + ' ' + self.RULE_ID +
- opts)
- self.assertEqual(str(self.MAX_BURST_KBITS_MODIFIED) + "\n", raw_output)
+ expected = (str(self.RULE_DIRECTION_MODIFIED) + "\n" +
+ str(self.MAX_BURST_KBITS_MODIFIED) + "\n" +
+ str(self.MAX_KBPS_MODIFIED) + "\n")
+ self.assertEqual(expected, raw_output)
diff --git a/openstackclient/tests/unit/common/test_project_purge.py b/openstackclient/tests/unit/common/test_project_purge.py
new file mode 100644
index 00000000..05a8aa3e
--- /dev/null
+++ b/openstackclient/tests/unit/common/test_project_purge.py
@@ -0,0 +1,314 @@
+# 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 mock
+
+from osc_lib import exceptions
+
+from openstackclient.common import project_purge
+from openstackclient.tests.unit.compute.v2 import fakes as compute_fakes
+from openstackclient.tests.unit import fakes
+from openstackclient.tests.unit.identity.v3 import fakes as identity_fakes
+from openstackclient.tests.unit.image.v2 import fakes as image_fakes
+from openstackclient.tests.unit import utils as tests_utils
+from openstackclient.tests.unit.volume.v2 import fakes as volume_fakes
+
+
+class TestProjectPurgeInit(tests_utils.TestCommand):
+
+ def setUp(self):
+ super(TestProjectPurgeInit, self).setUp()
+ compute_client = compute_fakes.FakeComputev2Client(
+ endpoint=fakes.AUTH_URL,
+ token=fakes.AUTH_TOKEN,
+ )
+ self.app.client_manager.compute = compute_client
+ self.servers_mock = compute_client.servers
+ self.servers_mock.reset_mock()
+
+ volume_client = volume_fakes.FakeVolumeClient(
+ endpoint=fakes.AUTH_URL,
+ token=fakes.AUTH_TOKEN,
+ )
+ self.app.client_manager.volume = volume_client
+ self.volumes_mock = volume_client.volumes
+ self.volumes_mock.reset_mock()
+ self.snapshots_mock = volume_client.volume_snapshots
+ self.snapshots_mock.reset_mock()
+ self.backups_mock = volume_client.backups
+ self.backups_mock.reset_mock()
+
+ identity_client = identity_fakes.FakeIdentityv3Client(
+ endpoint=fakes.AUTH_URL,
+ token=fakes.AUTH_TOKEN,
+ )
+ self.app.client_manager.identity = identity_client
+ self.domains_mock = identity_client.domains
+ self.domains_mock.reset_mock()
+ self.projects_mock = identity_client.projects
+ self.projects_mock.reset_mock()
+
+ image_client = image_fakes.FakeImagev2Client(
+ endpoint=fakes.AUTH_URL,
+ token=fakes.AUTH_TOKEN,
+ )
+ self.app.client_manager.image = image_client
+ self.images_mock = image_client.images
+ self.images_mock.reset_mock()
+
+
+class TestProjectPurge(TestProjectPurgeInit):
+
+ project = identity_fakes.FakeProject.create_one_project()
+ server = compute_fakes.FakeServer.create_one_server()
+ image = image_fakes.FakeImage.create_one_image()
+ volume = volume_fakes.FakeVolume.create_one_volume()
+ backup = volume_fakes.FakeBackup.create_one_backup()
+ snapshot = volume_fakes.FakeSnapshot.create_one_snapshot()
+
+ def setUp(self):
+ super(TestProjectPurge, self).setUp()
+ self.projects_mock.get.return_value = self.project
+ self.projects_mock.delete.return_value = None
+ self.images_mock.list.return_value = [self.image]
+ self.images_mock.delete.return_value = None
+ self.servers_mock.list.return_value = [self.server]
+ self.servers_mock.delete.return_value = None
+ self.volumes_mock.list.return_value = [self.volume]
+ self.volumes_mock.delete.return_value = None
+ self.volumes_mock.force_delete.return_value = None
+ self.snapshots_mock.list.return_value = [self.snapshot]
+ self.snapshots_mock.delete.return_value = None
+ self.backups_mock.list.return_value = [self.backup]
+ self.backups_mock.delete.return_value = None
+
+ self.cmd = project_purge.ProjectPurge(self.app, None)
+
+ def test_project_no_options(self):
+ arglist = []
+ verifylist = []
+
+ self.assertRaises(tests_utils.ParserException, self.check_parser,
+ self.cmd, arglist, verifylist)
+
+ def test_project_purge_with_project(self):
+ arglist = [
+ '--project', self.project.id,
+ ]
+ verifylist = [
+ ('dry_run', False),
+ ('keep_project', False),
+ ('auth_project', False),
+ ('project', self.project.id),
+ ('project_domain', None),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+ self.projects_mock.get.assert_called_once_with(self.project.id)
+ self.projects_mock.delete.assert_called_once_with(self.project.id)
+ self.servers_mock.list.assert_called_once_with(
+ search_opts={'tenant_id': self.project.id})
+ self.images_mock.list.assert_called_once_with(
+ owner=self.project.id)
+ volume_search_opts = {'project_id': self.project.id}
+ self.volumes_mock.list.assert_called_once_with(
+ search_opts=volume_search_opts)
+ self.snapshots_mock.list.assert_called_once_with(
+ search_opts=volume_search_opts)
+ self.backups_mock.list.assert_called_once_with(
+ search_opts=volume_search_opts)
+ self.servers_mock.delete.assert_called_once_with(self.server.id)
+ self.images_mock.delete.assert_called_once_with(self.image.id)
+ self.volumes_mock.force_delete.assert_called_once_with(self.volume.id)
+ self.snapshots_mock.delete.assert_called_once_with(self.snapshot.id)
+ self.backups_mock.delete.assert_called_once_with(self.backup.id)
+ self.assertIsNone(result)
+
+ def test_project_purge_with_dry_run(self):
+ arglist = [
+ '--dry-run',
+ '--project', self.project.id,
+ ]
+ verifylist = [
+ ('dry_run', True),
+ ('keep_project', False),
+ ('auth_project', False),
+ ('project', self.project.id),
+ ('project_domain', None),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+ self.projects_mock.get.assert_called_once_with(self.project.id)
+ self.projects_mock.delete.assert_not_called()
+ self.servers_mock.list.assert_called_once_with(
+ search_opts={'tenant_id': self.project.id})
+ self.images_mock.list.assert_called_once_with(
+ owner=self.project.id)
+ volume_search_opts = {'project_id': self.project.id}
+ self.volumes_mock.list.assert_called_once_with(
+ search_opts=volume_search_opts)
+ self.snapshots_mock.list.assert_called_once_with(
+ search_opts=volume_search_opts)
+ self.backups_mock.list.assert_called_once_with(
+ search_opts=volume_search_opts)
+ self.servers_mock.delete.assert_not_called()
+ self.images_mock.delete.assert_not_called()
+ self.volumes_mock.force_delete.assert_not_called()
+ self.snapshots_mock.delete.assert_not_called()
+ self.backups_mock.delete.assert_not_called()
+ self.assertIsNone(result)
+
+ def test_project_purge_with_keep_project(self):
+ arglist = [
+ '--keep-project',
+ '--project', self.project.id,
+ ]
+ verifylist = [
+ ('dry_run', False),
+ ('keep_project', True),
+ ('auth_project', False),
+ ('project', self.project.id),
+ ('project_domain', None),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+ self.projects_mock.get.assert_called_once_with(self.project.id)
+ self.projects_mock.delete.assert_not_called()
+ self.servers_mock.list.assert_called_once_with(
+ search_opts={'tenant_id': self.project.id})
+ self.images_mock.list.assert_called_once_with(
+ owner=self.project.id)
+ volume_search_opts = {'project_id': self.project.id}
+ self.volumes_mock.list.assert_called_once_with(
+ search_opts=volume_search_opts)
+ self.snapshots_mock.list.assert_called_once_with(
+ search_opts=volume_search_opts)
+ self.backups_mock.list.assert_called_once_with(
+ search_opts=volume_search_opts)
+ self.servers_mock.delete.assert_called_once_with(self.server.id)
+ self.images_mock.delete.assert_called_once_with(self.image.id)
+ self.volumes_mock.force_delete.assert_called_once_with(self.volume.id)
+ self.snapshots_mock.delete.assert_called_once_with(self.snapshot.id)
+ self.backups_mock.delete.assert_called_once_with(self.backup.id)
+ self.assertIsNone(result)
+
+ def test_project_purge_with_auth_project(self):
+ self.app.client_manager.auth_ref = mock.Mock()
+ self.app.client_manager.auth_ref.project_id = self.project.id
+ arglist = [
+ '--auth-project',
+ ]
+ verifylist = [
+ ('dry_run', False),
+ ('keep_project', False),
+ ('auth_project', True),
+ ('project', None),
+ ('project_domain', None),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+ self.projects_mock.get.assert_not_called()
+ self.projects_mock.delete.assert_called_once_with(self.project.id)
+ self.servers_mock.list.assert_called_once_with(
+ search_opts={'tenant_id': self.project.id})
+ self.images_mock.list.assert_called_once_with(
+ owner=self.project.id)
+ volume_search_opts = {'project_id': self.project.id}
+ self.volumes_mock.list.assert_called_once_with(
+ search_opts=volume_search_opts)
+ self.snapshots_mock.list.assert_called_once_with(
+ search_opts=volume_search_opts)
+ self.backups_mock.list.assert_called_once_with(
+ search_opts=volume_search_opts)
+ self.servers_mock.delete.assert_called_once_with(self.server.id)
+ self.images_mock.delete.assert_called_once_with(self.image.id)
+ self.volumes_mock.force_delete.assert_called_once_with(self.volume.id)
+ self.snapshots_mock.delete.assert_called_once_with(self.snapshot.id)
+ self.backups_mock.delete.assert_called_once_with(self.backup.id)
+ self.assertIsNone(result)
+
+ @mock.patch.object(project_purge.LOG, 'error')
+ def test_project_purge_with_exception(self, mock_error):
+ self.servers_mock.delete.side_effect = exceptions.CommandError()
+ arglist = [
+ '--project', self.project.id,
+ ]
+ verifylist = [
+ ('dry_run', False),
+ ('keep_project', False),
+ ('auth_project', False),
+ ('project', self.project.id),
+ ('project_domain', None),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+ self.projects_mock.get.assert_called_once_with(self.project.id)
+ self.projects_mock.delete.assert_called_once_with(self.project.id)
+ self.servers_mock.list.assert_called_once_with(
+ search_opts={'tenant_id': self.project.id})
+ self.images_mock.list.assert_called_once_with(
+ owner=self.project.id)
+ volume_search_opts = {'project_id': self.project.id}
+ self.volumes_mock.list.assert_called_once_with(
+ search_opts=volume_search_opts)
+ self.snapshots_mock.list.assert_called_once_with(
+ search_opts=volume_search_opts)
+ self.backups_mock.list.assert_called_once_with(
+ search_opts=volume_search_opts)
+ self.servers_mock.delete.assert_called_once_with(self.server.id)
+ self.images_mock.delete.assert_called_once_with(self.image.id)
+ self.volumes_mock.force_delete.assert_called_once_with(self.volume.id)
+ self.snapshots_mock.delete.assert_called_once_with(self.snapshot.id)
+ self.backups_mock.delete.assert_called_once_with(self.backup.id)
+ mock_error.assert_called_with("1 of 1 servers failed to delete.")
+ self.assertIsNone(result)
+
+ def test_project_purge_with_force_delete_backup(self):
+ self.backups_mock.delete.side_effect = [exceptions.CommandError, None]
+ arglist = [
+ '--project', self.project.id,
+ ]
+ verifylist = [
+ ('dry_run', False),
+ ('keep_project', False),
+ ('auth_project', False),
+ ('project', self.project.id),
+ ('project_domain', None),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+ self.projects_mock.get.assert_called_once_with(self.project.id)
+ self.projects_mock.delete.assert_called_once_with(self.project.id)
+ self.servers_mock.list.assert_called_once_with(
+ search_opts={'tenant_id': self.project.id})
+ self.images_mock.list.assert_called_once_with(
+ owner=self.project.id)
+ volume_search_opts = {'project_id': self.project.id}
+ self.volumes_mock.list.assert_called_once_with(
+ search_opts=volume_search_opts)
+ self.snapshots_mock.list.assert_called_once_with(
+ search_opts=volume_search_opts)
+ self.backups_mock.list.assert_called_once_with(
+ search_opts=volume_search_opts)
+ self.servers_mock.delete.assert_called_once_with(self.server.id)
+ self.images_mock.delete.assert_called_once_with(self.image.id)
+ self.volumes_mock.force_delete.assert_called_once_with(self.volume.id)
+ self.snapshots_mock.delete.assert_called_once_with(self.snapshot.id)
+ self.assertEqual(2, self.backups_mock.delete.call_count)
+ self.backups_mock.delete.assert_called_with(self.backup.id, force=True)
+ self.assertIsNone(result)
diff --git a/openstackclient/tests/unit/identity/v3/test_trust.py b/openstackclient/tests/unit/identity/v3/test_trust.py
index 93e8f63d..614aab54 100644
--- a/openstackclient/tests/unit/identity/v3/test_trust.py
+++ b/openstackclient/tests/unit/identity/v3/test_trust.py
@@ -94,7 +94,7 @@ class TestTrustCreate(TestTrust):
kwargs = {
'impersonation': False,
'project': identity_fakes.project_id,
- 'role_names': [identity_fakes.role_name],
+ 'role_ids': [identity_fakes.role_id],
'expires_at': None,
}
# TrustManager.create(trustee_id, trustor_id, impersonation=,
diff --git a/openstackclient/tests/unit/network/v2/fakes.py b/openstackclient/tests/unit/network/v2/fakes.py
index 69f28ee3..66b35cc5 100644
--- a/openstackclient/tests/unit/network/v2/fakes.py
+++ b/openstackclient/tests/unit/network/v2/fakes.py
@@ -923,6 +923,7 @@ class FakeNetworkQosRule(object):
if type == RULE_TYPE_BANDWIDTH_LIMIT:
qos_rule_attrs['max_kbps'] = randint(1, 10000)
qos_rule_attrs['max_burst_kbits'] = randint(1, 10000)
+ qos_rule_attrs['direction'] = 'egress'
elif type == RULE_TYPE_DSCP_MARKING:
qos_rule_attrs['dscp_mark'] = choice(VALID_DSCP_MARKS)
elif type == RULE_TYPE_MINIMUM_BANDWIDTH:
diff --git a/openstackclient/tests/unit/network/v2/test_network.py b/openstackclient/tests/unit/network/v2/test_network.py
index af2d08ba..bc8c4024 100644
--- a/openstackclient/tests/unit/network/v2/test_network.py
+++ b/openstackclient/tests/unit/network/v2/test_network.py
@@ -825,7 +825,6 @@ class TestSetNetwork(TestNetwork):
'--provider-network-type', 'vlan',
'--provider-physical-network', 'physnet1',
'--provider-segment', '400',
- '--no-transparent-vlan',
'--enable-port-security',
'--qos-policy', self.qos_policy.name,
]
@@ -840,7 +839,6 @@ class TestSetNetwork(TestNetwork):
('provider_network_type', 'vlan'),
('physical_network', 'physnet1'),
('segmentation_id', '400'),
- ('no_transparent_vlan', True),
('enable_port_security', True),
('qos_policy', self.qos_policy.name),
]
@@ -858,7 +856,6 @@ class TestSetNetwork(TestNetwork):
'provider:network_type': 'vlan',
'provider:physical_network': 'physnet1',
'provider:segmentation_id': '400',
- 'vlan_transparent': False,
'port_security_enabled': True,
'qos_policy_id': self.qos_policy.id,
}
diff --git a/openstackclient/tests/unit/network/v2/test_network_qos_rule.py b/openstackclient/tests/unit/network/v2/test_network_qos_rule.py
index 41ccae32..176bc86d 100644
--- a/openstackclient/tests/unit/network/v2/test_network_qos_rule.py
+++ b/openstackclient/tests/unit/network/v2/test_network_qos_rule.py
@@ -127,7 +127,7 @@ class TestCreateNetworkQosRuleMinimumBandwidth(TestNetworkQosRule):
self.cmd.take_action(parsed_args)
except exceptions.CommandError as e:
msg = ('"Create" rule command for type "minimum-bandwidth" '
- 'requires arguments min_kbps, direction')
+ 'requires arguments direction, min_kbps')
self.assertEqual(msg, str(e))
@@ -229,6 +229,7 @@ class TestCreateNetworkQosRuleBandwidtLimit(TestNetworkQosRule):
self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule(
attrs)
self.columns = (
+ 'direction',
'id',
'max_burst_kbits',
'max_kbps',
@@ -238,6 +239,7 @@ class TestCreateNetworkQosRuleBandwidtLimit(TestNetworkQosRule):
)
self.data = (
+ self.new_rule.direction,
self.new_rule.id,
self.new_rule.max_burst_kbits,
self.new_rule.max_kbps,
@@ -265,6 +267,7 @@ class TestCreateNetworkQosRuleBandwidtLimit(TestNetworkQosRule):
'--type', RULE_TYPE_BANDWIDTH_LIMIT,
'--max-kbps', str(self.new_rule.max_kbps),
'--max-burst-kbits', str(self.new_rule.max_burst_kbits),
+ '--egress',
self.new_rule.qos_policy_id,
]
@@ -272,6 +275,7 @@ class TestCreateNetworkQosRuleBandwidtLimit(TestNetworkQosRule):
('type', RULE_TYPE_BANDWIDTH_LIMIT),
('max_kbps', self.new_rule.max_kbps),
('max_burst_kbits', self.new_rule.max_burst_kbits),
+ ('egress', True),
('qos_policy', self.new_rule.qos_policy_id),
]
@@ -281,7 +285,8 @@ class TestCreateNetworkQosRuleBandwidtLimit(TestNetworkQosRule):
self.network.create_qos_bandwidth_limit_rule.assert_called_once_with(
self.qos_policy.id,
**{'max_kbps': self.new_rule.max_kbps,
- 'max_burst_kbps': self.new_rule.max_burst_kbits}
+ 'max_burst_kbps': self.new_rule.max_burst_kbits,
+ 'direction': self.new_rule.direction}
)
self.assertEqual(self.columns, columns)
self.assertEqual(self.data, data)
@@ -304,7 +309,7 @@ class TestCreateNetworkQosRuleBandwidtLimit(TestNetworkQosRule):
self.cmd.take_action(parsed_args)
except exceptions.CommandError as e:
msg = ('"Create" rule command for type "bandwidth-limit" '
- 'requires arguments max_kbps, max_burst_kbps')
+ 'requires arguments max_burst_kbps, max_kbps')
self.assertEqual(msg, str(e))
@@ -574,8 +579,8 @@ class TestSetNetworkQosRuleMinimumBandwidth(TestNetworkQosRule):
self.cmd.take_action(parsed_args)
except exceptions.CommandError as e:
msg = ('Failed to set Network QoS rule ID "%(rule)s": Rule type '
- '"minimum-bandwidth" only requires arguments min_kbps, '
- 'direction' % {'rule': self.new_rule.id})
+ '"minimum-bandwidth" only requires arguments direction, '
+ 'min_kbps' % {'rule': self.new_rule.id})
self.assertEqual(msg, str(e))
@@ -716,9 +721,13 @@ class TestSetNetworkQosRuleBandwidthLimit(TestNetworkQosRule):
def test_set_max_kbps_to_zero(self):
self._set_max_kbps(max_kbps=0)
+ def _reset_max_kbps(self, max_kbps):
+ self.new_rule.max_kbps = max_kbps
+
def _set_max_kbps(self, max_kbps=None):
if max_kbps:
- previous_max_kbps = self.new_rule.max_kbps
+ self.addCleanup(self._reset_max_kbps,
+ self.new_rule.max_kbps)
self.new_rule.max_kbps = max_kbps
arglist = [
@@ -742,18 +751,19 @@ class TestSetNetworkQosRuleBandwidthLimit(TestNetworkQosRule):
self.new_rule, self.qos_policy.id, **attrs)
self.assertIsNone(result)
- if max_kbps:
- self.new_rule.max_kbps = previous_max_kbps
-
def test_set_max_burst_kbits(self):
self._set_max_burst_kbits()
def test_set_max_burst_kbits_to_zero(self):
self._set_max_burst_kbits(max_burst_kbits=0)
+ def _reset_max_burst_kbits(self, max_burst_kbits):
+ self.new_rule.max_burst_kbits = max_burst_kbits
+
def _set_max_burst_kbits(self, max_burst_kbits=None):
if max_burst_kbits:
- previous_max_burst_kbits = self.new_rule.max_burst_kbits
+ self.addCleanup(self._reset_max_burst_kbits,
+ self.new_rule.max_burst_kbits)
self.new_rule.max_burst_kbits = max_burst_kbits
arglist = [
@@ -777,8 +787,38 @@ class TestSetNetworkQosRuleBandwidthLimit(TestNetworkQosRule):
self.new_rule, self.qos_policy.id, **attrs)
self.assertIsNone(result)
- if max_burst_kbits:
- self.new_rule.max_burst_kbits = previous_max_burst_kbits
+ def test_set_direction_egress(self):
+ self._set_direction('egress')
+
+ def test_set_direction_ingress(self):
+ self._set_direction('ingress')
+
+ def _reset_direction(self, direction):
+ self.new_rule.direction = direction
+
+ def _set_direction(self, direction):
+ self.addCleanup(self._reset_direction, self.new_rule.direction)
+
+ arglist = [
+ '--%s' % direction,
+ self.new_rule.qos_policy_id,
+ self.new_rule.id,
+ ]
+ verifylist = [
+ (direction, True),
+ ('qos_policy', self.new_rule.qos_policy_id),
+ ('id', self.new_rule.id),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+
+ attrs = {
+ 'direction': direction,
+ }
+ self.network.update_qos_bandwidth_limit_rule.assert_called_with(
+ self.new_rule, self.qos_policy.id, **attrs)
+ self.assertIsNone(result)
def test_set_wrong_options(self):
arglist = [
@@ -797,8 +837,8 @@ class TestSetNetworkQosRuleBandwidthLimit(TestNetworkQosRule):
self.cmd.take_action(parsed_args)
except exceptions.CommandError as e:
msg = ('Failed to set Network QoS rule ID "%(rule)s": Rule type '
- '"bandwidth-limit" only requires arguments max_kbps, '
- 'max_burst_kbps' % {'rule': self.new_rule.id})
+ '"bandwidth-limit" only requires arguments direction, '
+ 'max_burst_kbps, max_kbps' % {'rule': self.new_rule.id})
self.assertEqual(msg, str(e))
@@ -999,6 +1039,7 @@ class TestShowNetworkQosBandwidthLimit(TestNetworkQosRule):
attrs)
self.qos_policy.rules = [self.new_rule]
self.columns = (
+ 'direction',
'id',
'max_burst_kbits',
'max_kbps',
@@ -1007,6 +1048,7 @@ class TestShowNetworkQosBandwidthLimit(TestNetworkQosRule):
'type'
)
self.data = (
+ self.new_rule.direction,
self.new_rule.id,
self.new_rule.max_burst_kbits,
self.new_rule.max_kbps,