summaryrefslogtreecommitdiff
path: root/tests
diff options
context:
space:
mode:
Diffstat (limited to 'tests')
-rw-r--r--tests/unit/__init__.py16
-rw-r--r--tests/unit/lb/__init__.py16
-rw-r--r--tests/unit/lb/test_cli20_healthmonitor.py219
-rw-r--r--tests/unit/lb/test_cli20_member.py134
-rw-r--r--tests/unit/lb/test_cli20_pool.py171
-rw-r--r--tests/unit/lb/test_cli20_vip.py214
-rw-r--r--tests/unit/test_auth.py320
-rw-r--r--tests/unit/test_casual_args.py99
-rw-r--r--tests/unit/test_cli20.py504
-rw-r--r--tests/unit/test_cli20_extensions.py49
-rw-r--r--tests/unit/test_cli20_floatingips.py139
-rw-r--r--tests/unit/test_cli20_network.py533
-rw-r--r--tests/unit/test_cli20_nvp_queue.py88
-rw-r--r--tests/unit/test_cli20_nvpnetworkgateway.py114
-rw-r--r--tests/unit/test_cli20_port.py304
-rw-r--r--tests/unit/test_cli20_router.py197
-rw-r--r--tests/unit/test_cli20_securitygroup.py333
-rw-r--r--tests/unit/test_cli20_subnet.py402
-rw-r--r--tests/unit/test_name_or_id.py132
-rw-r--r--tests/unit/test_quota.py45
-rw-r--r--tests/unit/test_shell.py173
-rw-r--r--tests/unit/test_utils.py190
22 files changed, 0 insertions, 4392 deletions
diff --git a/tests/unit/__init__.py b/tests/unit/__init__.py
deleted file mode 100644
index 1668497..0000000
--- a/tests/unit/__init__.py
+++ /dev/null
@@ -1,16 +0,0 @@
-# Copyright 2013 OpenStack LLC.
-# All Rights Reserved
-#
-# 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.
-#
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
diff --git a/tests/unit/lb/__init__.py b/tests/unit/lb/__init__.py
deleted file mode 100644
index 1668497..0000000
--- a/tests/unit/lb/__init__.py
+++ /dev/null
@@ -1,16 +0,0 @@
-# Copyright 2013 OpenStack LLC.
-# All Rights Reserved
-#
-# 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.
-#
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
diff --git a/tests/unit/lb/test_cli20_healthmonitor.py b/tests/unit/lb/test_cli20_healthmonitor.py
deleted file mode 100644
index 0b87ff9..0000000
--- a/tests/unit/lb/test_cli20_healthmonitor.py
+++ /dev/null
@@ -1,219 +0,0 @@
-# Copyright 2013 Mirantis Inc.
-# All Rights Reserved
-#
-# 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.
-#
-# @author: Ilya Shakhat, Mirantis Inc.
-#
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-import sys
-
-import mox
-
-from quantumclient.quantum.v2_0.lb import healthmonitor
-from tests.unit import test_cli20
-
-
-class CLITestV20LbHealthmonitorJSON(test_cli20.CLITestV20Base):
- def test_create_healthmonitor_with_mandatory_params(self):
- """lb-healthmonitor-create with mandatory params only."""
- resource = 'health_monitor'
- cmd = healthmonitor.CreateHealthMonitor(test_cli20.MyApp(sys.stdout),
- None)
- admin_state_up = False
- delay = '60'
- max_retries = '2'
- timeout = '10'
- type = 'tcp'
- tenant_id = 'my-tenant'
- my_id = 'my-id'
- args = ['--admin-state-down',
- '--delay', delay,
- '--max-retries', max_retries,
- '--timeout', timeout,
- '--type', type,
- '--tenant-id', tenant_id]
- position_names = ['admin_state_up', 'delay', 'max_retries', 'timeout',
- 'type', 'tenant_id']
- position_values = [admin_state_up, delay, max_retries, timeout, type,
- tenant_id]
- self._test_create_resource(resource, cmd, '', my_id, args,
- position_names, position_values)
-
- def test_create_healthmonitor_with_all_params(self):
- """lb-healthmonitor-create with all params set."""
- resource = 'health_monitor'
- cmd = healthmonitor.CreateHealthMonitor(test_cli20.MyApp(sys.stdout),
- None)
- admin_state_up = False
- delay = '60'
- expected_codes = '200-202,204'
- http_method = 'HEAD'
- max_retries = '2'
- timeout = '10'
- type = 'tcp'
- tenant_id = 'my-tenant'
- url_path = '/health'
- my_id = 'my-id'
- args = ['--admin-state-down',
- '--delay', delay,
- '--expected-codes', expected_codes,
- '--http-method', http_method,
- '--max-retries', max_retries,
- '--timeout', timeout,
- '--type', type,
- '--tenant-id', tenant_id,
- '--url-path', url_path]
- position_names = ['admin_state_up', 'delay',
- 'expected_codes', 'http_method',
- 'max_retries', 'timeout',
- 'type', 'tenant_id', 'url_path']
- position_values = [admin_state_up, delay,
- expected_codes, http_method,
- max_retries, timeout,
- type, tenant_id, url_path]
- self._test_create_resource(resource, cmd, '', my_id, args,
- position_names, position_values)
-
- def test_list_healthmonitors(self):
- """lb-healthmonitor-list."""
- resources = "health_monitors"
- cmd = healthmonitor.ListHealthMonitor(test_cli20.MyApp(sys.stdout),
- None)
- self._test_list_resources(resources, cmd, True)
-
- def test_list_healthmonitors_pagination(self):
- """lb-healthmonitor-list."""
- resources = "health_monitors"
- cmd = healthmonitor.ListHealthMonitor(test_cli20.MyApp(sys.stdout),
- None)
- self._test_list_resources_with_pagination(resources, cmd)
-
- def test_list_healthmonitors_sort(self):
- """lb-healthmonitor-list --sort-key name --sort-key id --sort-key asc
- --sort-key desc
- """
- resources = "health_monitors"
- cmd = healthmonitor.ListHealthMonitor(test_cli20.MyApp(sys.stdout),
- None)
- self._test_list_resources(resources, cmd,
- sort_key=["name", "id"],
- sort_dir=["asc", "desc"])
-
- def test_list_healthmonitors_limit(self):
- """lb-healthmonitor-list -P."""
- resources = "health_monitors"
- cmd = healthmonitor.ListHealthMonitor(test_cli20.MyApp(sys.stdout),
- None)
- self._test_list_resources(resources, cmd, page_size=1000)
-
- def test_show_healthmonitor_id(self):
- """lb-healthmonitor-show test_id."""
- resource = 'health_monitor'
- cmd = healthmonitor.ShowHealthMonitor(test_cli20.MyApp(sys.stdout),
- None)
- args = ['--fields', 'id', self.test_id]
- self._test_show_resource(resource, cmd, self.test_id, args, ['id'])
-
- def test_show_healthmonitor_id_name(self):
- """lb-healthmonitor-show."""
- resource = 'health_monitor'
- cmd = healthmonitor.ShowHealthMonitor(test_cli20.MyApp(sys.stdout),
- None)
- args = ['--fields', 'id', '--fields', 'name', self.test_id]
- self._test_show_resource(resource, cmd, self.test_id,
- args, ['id', 'name'])
-
- def test_update_health_monitor(self):
- """lb-healthmonitor-update myid --name myname --tags a b."""
- resource = 'health_monitor'
- cmd = healthmonitor.UpdateHealthMonitor(test_cli20.MyApp(sys.stdout),
- None)
- self._test_update_resource(resource, cmd, 'myid',
- ['myid', '--timeout', '5'],
- {'timeout': '5', })
-
- def test_delete_healthmonitor(self):
- """lb-healthmonitor-delete my-id."""
- resource = 'health_monitor'
- cmd = healthmonitor.DeleteHealthMonitor(test_cli20.MyApp(sys.stdout),
- None)
- my_id = 'my-id'
- args = [my_id]
- self._test_delete_resource(resource, cmd, my_id, args)
-
- def test_associate_healthmonitor(self):
- cmd = healthmonitor.AssociateHealthMonitor(
- test_cli20.MyApp(sys.stdout),
- None)
- resource = 'health_monitor'
- health_monitor_id = 'hm-id'
- pool_id = 'p_id'
- args = [health_monitor_id, pool_id]
-
- self.mox.StubOutWithMock(cmd, "get_client")
- self.mox.StubOutWithMock(self.client.httpclient, "request")
- cmd.get_client().MultipleTimes().AndReturn(self.client)
-
- body = {resource: {'id': health_monitor_id}}
- result = {resource: {'id': health_monitor_id}, }
- result_str = self.client.serialize(result)
-
- path = getattr(self.client,
- "associate_pool_health_monitors_path") % pool_id
- return_tup = (test_cli20.MyResp(200), result_str)
- self.client.httpclient.request(
- test_cli20.end_url(path), 'POST',
- body=test_cli20.MyComparator(body, self.client),
- headers=mox.ContainsKeyValue(
- 'X-Auth-Token', test_cli20.TOKEN)).AndReturn(return_tup)
- self.mox.ReplayAll()
- cmd_parser = cmd.get_parser('test_' + resource)
- parsed_args = cmd_parser.parse_args(args)
- cmd.run(parsed_args)
- self.mox.VerifyAll()
- self.mox.UnsetStubs()
-
- def test_disassociate_healthmonitor(self):
- cmd = healthmonitor.DisassociateHealthMonitor(
- test_cli20.MyApp(sys.stdout),
- None)
- resource = 'health_monitor'
- health_monitor_id = 'hm-id'
- pool_id = 'p_id'
- args = [health_monitor_id, pool_id]
-
- self.mox.StubOutWithMock(cmd, "get_client")
- self.mox.StubOutWithMock(self.client.httpclient, "request")
- cmd.get_client().MultipleTimes().AndReturn(self.client)
-
- path = (getattr(self.client,
- "disassociate_pool_health_monitors_path") %
- {'pool': pool_id, 'health_monitor': health_monitor_id})
- return_tup = (test_cli20.MyResp(204), None)
- self.client.httpclient.request(
- test_cli20.end_url(path), 'DELETE',
- body=None,
- headers=mox.ContainsKeyValue(
- 'X-Auth-Token', test_cli20.TOKEN)).AndReturn(return_tup)
- self.mox.ReplayAll()
- cmd_parser = cmd.get_parser('test_' + resource)
- parsed_args = cmd_parser.parse_args(args)
- cmd.run(parsed_args)
- self.mox.VerifyAll()
- self.mox.UnsetStubs()
-
-
-class CLITestV20LbHealthmonitorXML(CLITestV20LbHealthmonitorJSON):
- format = 'xml'
diff --git a/tests/unit/lb/test_cli20_member.py b/tests/unit/lb/test_cli20_member.py
deleted file mode 100644
index 9b75367..0000000
--- a/tests/unit/lb/test_cli20_member.py
+++ /dev/null
@@ -1,134 +0,0 @@
-# Copyright 2013 Mirantis Inc.
-# All Rights Reserved
-#
-# 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.
-#
-# @author: Ilya Shakhat, Mirantis Inc.
-#
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-import sys
-
-from quantumclient.quantum.v2_0.lb import member
-from tests.unit import test_cli20
-
-
-class CLITestV20LbMemberJSON(test_cli20.CLITestV20Base):
- def setUp(self):
- super(CLITestV20LbMemberJSON, self).setUp(plurals={'tags': 'tag'})
-
- def test_create_member(self):
- """lb-member-create with mandatory params only."""
- resource = 'member'
- cmd = member.CreateMember(test_cli20.MyApp(sys.stdout), None)
- address = '10.0.0.1'
- port = '8080'
- tenant_id = 'my-tenant'
- my_id = 'my-id'
- pool_id = 'pool-id'
- args = ['--address', address, '--protocol-port', port,
- '--tenant-id', tenant_id, pool_id]
- position_names = ['address', 'protocol_port', 'tenant_id', 'pool_id',
- 'admin_state_up']
- position_values = [address, port, tenant_id, pool_id, True]
- self._test_create_resource(resource, cmd, None, my_id, args,
- position_names, position_values,
- admin_state_up=None)
-
- def test_create_member_all_params(self):
- """lb-member-create with all available params."""
- resource = 'member'
- cmd = member.CreateMember(test_cli20.MyApp(sys.stdout), None)
- address = '10.0.0.1'
- admin_state_up = False
- port = '8080'
- weight = '1'
- tenant_id = 'my-tenant'
- my_id = 'my-id'
- pool_id = 'pool-id'
- args = ['--address', address, '--admin-state-down',
- '--protocol-port', port, '--weight', weight,
- '--tenant-id', tenant_id, pool_id]
- position_names = [
- 'address', 'admin_state_up', 'protocol_port', 'weight',
- 'tenant_id', 'pool_id'
- ]
- position_values = [address, admin_state_up, port, weight,
- tenant_id, pool_id]
- self._test_create_resource(resource, cmd, None, my_id, args,
- position_names, position_values,
- admin_state_up=None)
-
- def test_list_members(self):
- """lb-member-list."""
- resources = "members"
- cmd = member.ListMember(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd, True)
-
- def test_list_members_pagination(self):
- """lb-member-list."""
- resources = "members"
- cmd = member.ListMember(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources_with_pagination(resources, cmd)
-
- def test_list_members_sort(self):
- """lb-member-list --sort-key name --sort-key id --sort-key asc
- --sort-key desc
- """
- resources = "members"
- cmd = member.ListMember(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd,
- sort_key=["name", "id"],
- sort_dir=["asc", "desc"])
-
- def test_list_members_limit(self):
- """lb-member-list -P."""
- resources = "members"
- cmd = member.ListMember(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd, page_size=1000)
-
- def test_show_member_id(self):
- """lb-member-show test_id."""
- resource = 'member'
- cmd = member.ShowMember(test_cli20.MyApp(sys.stdout), None)
- args = ['--fields', 'id', self.test_id]
- self._test_show_resource(resource, cmd, self.test_id, args, ['id'])
-
- def test_show_member_id_name(self):
- """lb-member-show."""
- resource = 'member'
- cmd = member.ShowMember(test_cli20.MyApp(sys.stdout), None)
- args = ['--fields', 'id', '--fields', 'name', self.test_id]
- self._test_show_resource(resource, cmd, self.test_id,
- args, ['id', 'name'])
-
- def test_update_member(self):
- """lb-member-update myid --name myname --tags a b."""
- resource = 'member'
- cmd = member.UpdateMember(test_cli20.MyApp(sys.stdout), None)
- self._test_update_resource(resource, cmd, 'myid',
- ['myid', '--name', 'myname',
- '--tags', 'a', 'b'],
- {'name': 'myname', 'tags': ['a', 'b'], })
-
- def test_delete_member(self):
- """lb-member-delete my-id."""
- resource = 'member'
- cmd = member.DeleteMember(test_cli20.MyApp(sys.stdout), None)
- my_id = 'my-id'
- args = [my_id]
- self._test_delete_resource(resource, cmd, my_id, args)
-
-
-class CLITestV20LbMemberXML(CLITestV20LbMemberJSON):
- format = 'xml'
diff --git a/tests/unit/lb/test_cli20_pool.py b/tests/unit/lb/test_cli20_pool.py
deleted file mode 100644
index 709c3d1..0000000
--- a/tests/unit/lb/test_cli20_pool.py
+++ /dev/null
@@ -1,171 +0,0 @@
-# Copyright 2013 Mirantis Inc.
-# All Rights Reserved
-#
-# 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.
-#
-# @author: Ilya Shakhat, Mirantis Inc.
-#
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-import sys
-
-import mox
-
-from quantumclient.quantum.v2_0.lb import pool
-from tests.unit import test_cli20
-
-
-class CLITestV20LbPoolJSON(test_cli20.CLITestV20Base):
-
- def test_create_pool_with_mandatory_params(self):
- """lb-pool-create with mandatory params only."""
- resource = 'pool'
- cmd = pool.CreatePool(test_cli20.MyApp(sys.stdout), None)
- name = 'my-name'
- lb_method = 'round-robin'
- protocol = 'http'
- subnet_id = 'subnet-id'
- tenant_id = 'my-tenant'
- my_id = 'my-id'
- args = ['--lb-method', lb_method,
- '--name', name,
- '--protocol', protocol,
- '--subnet-id', subnet_id,
- '--tenant-id', tenant_id]
- position_names = ['admin_state_up', 'lb_method', 'name',
- 'protocol', 'subnet_id', 'tenant_id']
- position_values = [True, lb_method, name,
- protocol, subnet_id, tenant_id]
- self._test_create_resource(resource, cmd, name, my_id, args,
- position_names, position_values)
-
- def test_create_pool_with_all_params(self):
- """lb-pool-create with all params set."""
- resource = 'pool'
- cmd = pool.CreatePool(test_cli20.MyApp(sys.stdout), None)
- name = 'my-name'
- description = 'my-desc'
- lb_method = 'round-robin'
- protocol = 'http'
- subnet_id = 'subnet-id'
- tenant_id = 'my-tenant'
- my_id = 'my-id'
- args = ['--admin-state-down',
- '--description', description,
- '--lb-method', lb_method,
- '--name', name,
- '--protocol', protocol,
- '--subnet-id', subnet_id,
- '--tenant-id', tenant_id]
- position_names = ['admin_state_up', 'description', 'lb_method', 'name',
- 'protocol', 'subnet_id', 'tenant_id']
- position_values = [False, description, lb_method, name,
- protocol, subnet_id, tenant_id]
- self._test_create_resource(resource, cmd, name, my_id, args,
- position_names, position_values)
-
- def test_list_pools(self):
- """lb-pool-list."""
- resources = "pools"
- cmd = pool.ListPool(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd, True)
-
- def test_list_pools_pagination(self):
- """lb-pool-list."""
- resources = "pools"
- cmd = pool.ListPool(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources_with_pagination(resources, cmd)
-
- def test_list_pools_sort(self):
- """lb-pool-list --sort-key name --sort-key id --sort-key asc
- --sort-key desc
- """
- resources = "pools"
- cmd = pool.ListPool(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd,
- sort_key=["name", "id"],
- sort_dir=["asc", "desc"])
-
- def test_list_pools_limit(self):
- """lb-pool-list -P."""
- resources = "pools"
- cmd = pool.ListPool(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd, page_size=1000)
-
- def test_show_pool_id(self):
- """lb-pool-show test_id."""
- resource = 'pool'
- cmd = pool.ShowPool(test_cli20.MyApp(sys.stdout), None)
- args = ['--fields', 'id', self.test_id]
- self._test_show_resource(resource, cmd, self.test_id, args, ['id'])
-
- def test_show_pool_id_name(self):
- """lb-pool-show."""
- resource = 'pool'
- cmd = pool.ShowPool(test_cli20.MyApp(sys.stdout), None)
- args = ['--fields', 'id', '--fields', 'name', self.test_id]
- self._test_show_resource(resource, cmd, self.test_id,
- args, ['id', 'name'])
-
- def test_update_pool(self):
- """lb-pool-update myid --name newname --tags a b."""
- resource = 'pool'
- cmd = pool.UpdatePool(test_cli20.MyApp(sys.stdout), None)
- self._test_update_resource(resource, cmd, 'myid',
- ['myid', '--name', 'newname'],
- {'name': 'newname', })
-
- def test_delete_pool(self):
- """lb-pool-delete my-id."""
- resource = 'pool'
- cmd = pool.DeletePool(test_cli20.MyApp(sys.stdout), None)
- my_id = 'my-id'
- args = [my_id]
- self._test_delete_resource(resource, cmd, my_id, args)
-
- def test_retrieve_pool_stats(self):
- """lb-pool-stats test_id."""
- resource = 'pool'
- cmd = pool.RetrievePoolStats(test_cli20.MyApp(sys.stdout), None)
- my_id = self.test_id
- fields = ['bytes_in', 'bytes_out']
- args = ['--fields', 'bytes_in', '--fields', 'bytes_out', my_id]
-
- self.mox.StubOutWithMock(cmd, "get_client")
- self.mox.StubOutWithMock(self.client.httpclient, "request")
- cmd.get_client().MultipleTimes().AndReturn(self.client)
- query = "&".join(["fields=%s" % field for field in fields])
- expected_res = {'stats': {'bytes_in': '1234', 'bytes_out': '4321'}}
- resstr = self.client.serialize(expected_res)
- path = getattr(self.client, "pool_path_stats")
- return_tup = (test_cli20.MyResp(200), resstr)
- self.client.httpclient.request(
- test_cli20.end_url(path % my_id, query), 'GET',
- body=None,
- headers=mox.ContainsKeyValue(
- 'X-Auth-Token', test_cli20.TOKEN)).AndReturn(return_tup)
- self.mox.ReplayAll()
-
- cmd_parser = cmd.get_parser("test_" + resource)
- parsed_args = cmd_parser.parse_args(args)
- cmd.run(parsed_args)
-
- self.mox.VerifyAll()
- self.mox.UnsetStubs()
- _str = self.fake_stdout.make_string()
- self.assertTrue('bytes_in' in _str)
- self.assertTrue('bytes_out' in _str)
-
-
-class CLITestV20LbPoolXML(CLITestV20LbPoolJSON):
- format = 'xml'
diff --git a/tests/unit/lb/test_cli20_vip.py b/tests/unit/lb/test_cli20_vip.py
deleted file mode 100644
index 4358cfb..0000000
--- a/tests/unit/lb/test_cli20_vip.py
+++ /dev/null
@@ -1,214 +0,0 @@
-# Copyright 2013 Mirantis Inc.
-# All Rights Reserved
-#
-# 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.
-#
-# @author: Ilya Shakhat, Mirantis Inc.
-#
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-import sys
-
-from quantumclient.quantum.v2_0.lb import vip
-from tests.unit import test_cli20
-
-
-class CLITestV20LbVipJSON(test_cli20.CLITestV20Base):
- def setUp(self):
- super(CLITestV20LbVipJSON, self).setUp(plurals={'tags': 'tag'})
-
- def test_create_vip_with_mandatory_params(self):
- """lb-vip-create with all mandatory params."""
- resource = 'vip'
- cmd = vip.CreateVip(test_cli20.MyApp(sys.stdout), None)
- pool_id = 'my-pool-id'
- name = 'my-name'
- subnet_id = 'subnet-id'
- protocol_port = '1000'
- protocol = 'tcp'
- tenant_id = 'my-tenant'
- my_id = 'my-id'
- args = ['--name', name,
- '--protocol-port', protocol_port,
- '--protocol', protocol,
- '--subnet-id', subnet_id,
- '--tenant-id', tenant_id,
- pool_id]
- position_names = ['pool_id', 'name', 'protocol_port', 'protocol',
- 'subnet_id', 'tenant_id']
- position_values = [pool_id, name, protocol_port, protocol,
- subnet_id, tenant_id]
- self._test_create_resource(resource, cmd, name, my_id, args,
- position_names, position_values,
- admin_state_up=True)
-
- def test_create_vip_with_all_params(self):
- """lb-vip-create with all params."""
- resource = 'vip'
- cmd = vip.CreateVip(test_cli20.MyApp(sys.stdout), None)
- pool_id = 'my-pool-id'
- name = 'my-name'
- description = 'my-desc'
- address = '10.0.0.2'
- admin_state = False
- connection_limit = '1000'
- subnet_id = 'subnet-id'
- protocol_port = '80'
- protocol = 'tcp'
- tenant_id = 'my-tenant'
- my_id = 'my-id'
- args = ['--name', name,
- '--description', description,
- '--address', address,
- '--admin-state-down',
- '--connection-limit', connection_limit,
- '--protocol-port', protocol_port,
- '--protocol', protocol,
- '--subnet-id', subnet_id,
- '--tenant-id', tenant_id,
- pool_id]
- position_names = ['pool_id', 'name', 'description', 'address',
- 'admin_state_up', 'connection_limit',
- 'protocol_port', 'protocol', 'subnet_id',
- 'tenant_id']
- position_values = [pool_id, name, description, address,
- admin_state, connection_limit, protocol_port,
- protocol, subnet_id,
- tenant_id]
- self._test_create_resource(resource, cmd, name, my_id, args,
- position_names, position_values)
-
- def test_create_vip_with_session_persistence_params(self):
- """lb-vip-create with mandatory and session-persistence params."""
- resource = 'vip'
- cmd = vip.CreateVip(test_cli20.MyApp(sys.stdout), None)
- pool_id = 'my-pool-id'
- name = 'my-name'
- subnet_id = 'subnet-id'
- protocol_port = '1000'
- protocol = 'tcp'
- tenant_id = 'my-tenant'
- my_id = 'my-id'
- args = ['--name', name,
- '--protocol-port', protocol_port,
- '--protocol', protocol,
- '--subnet-id', subnet_id,
- '--tenant-id', tenant_id,
- pool_id,
- '--session-persistence', 'type=dict',
- 'type=cookie,cookie_name=pie',
- '--optional-param', 'any']
- position_names = ['pool_id', 'name', 'protocol_port', 'protocol',
- 'subnet_id', 'tenant_id', 'optional_param']
- position_values = [pool_id, name, protocol_port, protocol,
- subnet_id, tenant_id, 'any']
- extra_body = {
- 'session_persistence': {
- 'type': 'cookie',
- 'cookie_name': 'pie',
- },
- }
- self._test_create_resource(resource, cmd, name, my_id, args,
- position_names, position_values,
- admin_state_up=True, extra_body=extra_body)
-
- def test_list_vips(self):
- """lb-vip-list."""
- resources = "vips"
- cmd = vip.ListVip(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd, True)
-
- def test_list_vips_pagination(self):
- """lb-vip-list."""
- resources = "vips"
- cmd = vip.ListVip(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources_with_pagination(resources, cmd)
-
- def test_list_vips_sort(self):
- """lb-vip-list --sort-key name --sort-key id --sort-key asc
- --sort-key desc
- """
- resources = "vips"
- cmd = vip.ListVip(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd,
- sort_key=["name", "id"],
- sort_dir=["asc", "desc"])
-
- def test_list_vips_limit(self):
- """lb-vip-list -P."""
- resources = "vips"
- cmd = vip.ListVip(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd, page_size=1000)
-
- def test_show_vip_id(self):
- """lb-vip-show test_id."""
- resource = 'vip'
- cmd = vip.ShowVip(test_cli20.MyApp(sys.stdout), None)
- args = ['--fields', 'id', self.test_id]
- self._test_show_resource(resource, cmd, self.test_id, args, ['id'])
-
- def test_show_vip_id_name(self):
- """lb-vip-show."""
- resource = 'vip'
- cmd = vip.ShowVip(test_cli20.MyApp(sys.stdout), None)
- args = ['--fields', 'id', '--fields', 'name', self.test_id]
- self._test_show_resource(resource, cmd, self.test_id,
- args, ['id', 'name'])
-
- def test_update_vip(self):
- """lb-vip-update myid --name myname --tags a b."""
- resource = 'vip'
- cmd = vip.UpdateVip(test_cli20.MyApp(sys.stdout), None)
- self._test_update_resource(resource, cmd, 'myid',
- ['myid', '--name', 'myname',
- '--tags', 'a', 'b'],
- {'name': 'myname', 'tags': ['a', 'b'], })
-
- def test_update_vip_with_session_persistence(self):
- resource = 'vip'
- cmd = vip.UpdateVip(test_cli20.MyApp(sys.stdout), None)
- body = {
- 'session_persistence': {
- 'type': 'source',
- },
- }
- args = ['myid', '--session-persistence', 'type=dict',
- 'type=source']
- self._test_update_resource(resource, cmd, 'myid', args, body)
-
- def test_update_vip_with_session_persistence_and_name(self):
- resource = 'vip'
- cmd = vip.UpdateVip(test_cli20.MyApp(sys.stdout), None)
- body = {
- 'name': 'newname',
- 'session_persistence': {
- 'type': 'cookie',
- 'cookie_name': 'pie',
- },
- }
- args = ['myid', '--name', 'newname',
- '--session-persistence', 'type=dict',
- 'type=cookie,cookie_name=pie']
- self._test_update_resource(resource, cmd, 'myid', args, body)
-
- def test_delete_vip(self):
- """lb-vip-delete my-id."""
- resource = 'vip'
- cmd = vip.DeleteVip(test_cli20.MyApp(sys.stdout), None)
- my_id = 'my-id'
- args = [my_id]
- self._test_delete_resource(resource, cmd, my_id, args)
-
-
-class CLITestV20LbVipXML(CLITestV20LbVipJSON):
- format = 'xml'
diff --git a/tests/unit/test_auth.py b/tests/unit/test_auth.py
deleted file mode 100644
index 06cc2de..0000000
--- a/tests/unit/test_auth.py
+++ /dev/null
@@ -1,320 +0,0 @@
-# Copyright 2012 NEC Corporation
-# All Rights Reserved
-#
-# 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.
-#
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-import copy
-import httplib2
-import json
-import uuid
-
-import mox
-import testtools
-
-from quantumclient import client
-from quantumclient.common import exceptions
-
-
-USERNAME = 'testuser'
-TENANT_NAME = 'testtenant'
-PASSWORD = 'password'
-AUTH_URL = 'authurl'
-ENDPOINT_URL = 'localurl'
-TOKEN = 'tokentoken'
-REGION = 'RegionTest'
-
-KS_TOKEN_RESULT = {
- 'access': {
- 'token': {'id': TOKEN,
- 'expires': '2012-08-11T07:49:01Z',
- 'tenant': {'id': str(uuid.uuid1())}},
- 'user': {'id': str(uuid.uuid1())},
- 'serviceCatalog': [
- {'endpoints_links': [],
- 'endpoints': [{'adminURL': ENDPOINT_URL,
- 'internalURL': ENDPOINT_URL,
- 'publicURL': ENDPOINT_URL,
- 'region': REGION}],
- 'type': 'network',
- 'name': 'Quantum Service'}
- ]
- }
-}
-
-ENDPOINTS_RESULT = {
- 'endpoints': [{
- 'type': 'network',
- 'name': 'Quantum Service',
- 'region': REGION,
- 'adminURL': ENDPOINT_URL,
- 'internalURL': ENDPOINT_URL,
- 'publicURL': ENDPOINT_URL
- }]
-}
-
-
-class CLITestAuthKeystone(testtools.TestCase):
-
- def setUp(self):
- """Prepare the test environment."""
- super(CLITestAuthKeystone, self).setUp()
- self.mox = mox.Mox()
- self.client = client.HTTPClient(username=USERNAME,
- tenant_name=TENANT_NAME,
- password=PASSWORD,
- auth_url=AUTH_URL,
- region_name=REGION)
- self.addCleanup(self.mox.VerifyAll)
- self.addCleanup(self.mox.UnsetStubs)
-
- def test_get_token(self):
- self.mox.StubOutWithMock(self.client, "request")
-
- res200 = self.mox.CreateMock(httplib2.Response)
- res200.status = 200
-
- self.client.request(
- AUTH_URL + '/tokens', 'POST',
- body=mox.IsA(str), headers=mox.IsA(dict)
- ).AndReturn((res200, json.dumps(KS_TOKEN_RESULT)))
- self.client.request(
- mox.StrContains(ENDPOINT_URL + '/resource'), 'GET',
- headers=mox.ContainsKeyValue('X-Auth-Token', TOKEN)
- ).AndReturn((res200, ''))
- self.mox.ReplayAll()
-
- self.client.do_request('/resource', 'GET')
- self.assertEqual(self.client.endpoint_url, ENDPOINT_URL)
- self.assertEqual(self.client.auth_token, TOKEN)
-
- def test_refresh_token(self):
- self.mox.StubOutWithMock(self.client, "request")
-
- self.client.auth_token = TOKEN
- self.client.endpoint_url = ENDPOINT_URL
-
- res200 = self.mox.CreateMock(httplib2.Response)
- res200.status = 200
- res401 = self.mox.CreateMock(httplib2.Response)
- res401.status = 401
-
- # If a token is expired, quantum server retruns 401
- self.client.request(
- mox.StrContains(ENDPOINT_URL + '/resource'), 'GET',
- headers=mox.ContainsKeyValue('X-Auth-Token', TOKEN)
- ).AndReturn((res401, ''))
- self.client.request(
- AUTH_URL + '/tokens', 'POST',
- body=mox.IsA(str), headers=mox.IsA(dict)
- ).AndReturn((res200, json.dumps(KS_TOKEN_RESULT)))
- self.client.request(
- mox.StrContains(ENDPOINT_URL + '/resource'), 'GET',
- headers=mox.ContainsKeyValue('X-Auth-Token', TOKEN)
- ).AndReturn((res200, ''))
- self.mox.ReplayAll()
- self.client.do_request('/resource', 'GET')
-
- def test_get_endpoint_url(self):
- self.mox.StubOutWithMock(self.client, "request")
-
- self.client.auth_token = TOKEN
-
- res200 = self.mox.CreateMock(httplib2.Response)
- res200.status = 200
-
- self.client.request(
- mox.StrContains(AUTH_URL + '/tokens/%s/endpoints' % TOKEN), 'GET',
- headers=mox.IsA(dict)
- ).AndReturn((res200, json.dumps(ENDPOINTS_RESULT)))
- self.client.request(
- mox.StrContains(ENDPOINT_URL + '/resource'), 'GET',
- headers=mox.ContainsKeyValue('X-Auth-Token', TOKEN)
- ).AndReturn((res200, ''))
- self.mox.ReplayAll()
- self.client.do_request('/resource', 'GET')
-
- def test_get_endpoint_url_other(self):
- self.client = client.HTTPClient(
- username=USERNAME, tenant_name=TENANT_NAME, password=PASSWORD,
- auth_url=AUTH_URL, region_name=REGION, endpoint_type='otherURL')
- self.mox.StubOutWithMock(self.client, "request")
-
- self.client.auth_token = TOKEN
-
- res200 = self.mox.CreateMock(httplib2.Response)
- res200.status = 200
-
- self.client.request(
- mox.StrContains(AUTH_URL + '/tokens/%s/endpoints' % TOKEN), 'GET',
- headers=mox.IsA(dict)
- ).AndReturn((res200, json.dumps(ENDPOINTS_RESULT)))
- self.mox.ReplayAll()
- self.assertRaises(exceptions.EndpointTypeNotFound,
- self.client.do_request,
- '/resource',
- 'GET')
-
- def test_get_endpoint_url_failed(self):
- self.mox.StubOutWithMock(self.client, "request")
-
- self.client.auth_token = TOKEN
-
- res200 = self.mox.CreateMock(httplib2.Response)
- res200.status = 200
- res401 = self.mox.CreateMock(httplib2.Response)
- res401.status = 401
-
- self.client.request(
- mox.StrContains(AUTH_URL + '/tokens/%s/endpoints' % TOKEN), 'GET',
- headers=mox.IsA(dict)
- ).AndReturn((res401, ''))
- self.client.request(
- AUTH_URL + '/tokens', 'POST',
- body=mox.IsA(str), headers=mox.IsA(dict)
- ).AndReturn((res200, json.dumps(KS_TOKEN_RESULT)))
- self.client.request(
- mox.StrContains(ENDPOINT_URL + '/resource'), 'GET',
- headers=mox.ContainsKeyValue('X-Auth-Token', TOKEN)
- ).AndReturn((res200, ''))
- self.mox.ReplayAll()
- self.client.do_request('/resource', 'GET')
-
- def test_url_for(self):
- resources = copy.deepcopy(KS_TOKEN_RESULT)
-
- endpoints = resources['access']['serviceCatalog'][0]['endpoints'][0]
- endpoints['publicURL'] = 'public'
- endpoints['internalURL'] = 'internal'
- endpoints['adminURL'] = 'admin'
- catalog = client.ServiceCatalog(resources)
-
- # endpoint_type not specified
- url = catalog.url_for(attr='region',
- filter_value=REGION)
- self.assertEqual('public', url)
-
- # endpoint type specified (3 cases)
- url = catalog.url_for(attr='region',
- filter_value=REGION,
- endpoint_type='adminURL')
- self.assertEqual('admin', url)
-
- url = catalog.url_for(attr='region',
- filter_value=REGION,
- endpoint_type='publicURL')
- self.assertEqual('public', url)
-
- url = catalog.url_for(attr='region',
- filter_value=REGION,
- endpoint_type='internalURL')
- self.assertEqual('internal', url)
-
- # endpoint_type requested does not exist.
- self.assertRaises(exceptions.EndpointTypeNotFound,
- catalog.url_for,
- attr='region',
- filter_value=REGION,
- endpoint_type='privateURL')
-
- # Test scenario with url_for when the service catalog only has publicURL.
- def test_url_for_only_public_url(self):
- resources = copy.deepcopy(KS_TOKEN_RESULT)
- catalog = client.ServiceCatalog(resources)
-
- # Remove endpoints from the catalog.
- endpoints = resources['access']['serviceCatalog'][0]['endpoints'][0]
- del endpoints['internalURL']
- del endpoints['adminURL']
- endpoints['publicURL'] = 'public'
-
- # Use publicURL when specified explicitly.
- url = catalog.url_for(attr='region',
- filter_value=REGION,
- endpoint_type='publicURL')
- self.assertEqual('public', url)
-
- # Use publicURL when specified explicitly.
- url = catalog.url_for(attr='region',
- filter_value=REGION)
- self.assertEqual('public', url)
-
- # Test scenario with url_for when the service catalog only has adminURL.
- def test_url_for_only_admin_url(self):
- resources = copy.deepcopy(KS_TOKEN_RESULT)
- catalog = client.ServiceCatalog(resources)
- endpoints = resources['access']['serviceCatalog'][0]['endpoints'][0]
- del endpoints['internalURL']
- del endpoints['publicURL']
- endpoints['adminURL'] = 'admin'
-
- # Use publicURL when specified explicitly.
- url = catalog.url_for(attr='region',
- filter_value=REGION,
- endpoint_type='adminURL')
- self.assertEqual('admin', url)
-
- # But not when nothing is specified.
- self.assertRaises(exceptions.EndpointTypeNotFound,
- catalog.url_for,
- attr='region',
- filter_value=REGION)
-
- def test_endpoint_type(self):
- resources = copy.deepcopy(KS_TOKEN_RESULT)
- endpoints = resources['access']['serviceCatalog'][0]['endpoints'][0]
- endpoints['internalURL'] = 'internal'
- endpoints['adminURL'] = 'admin'
- endpoints['publicURL'] = 'public'
-
- # Test default behavior is to choose public.
- self.client = client.HTTPClient(
- username=USERNAME, tenant_name=TENANT_NAME, password=PASSWORD,
- auth_url=AUTH_URL, region_name=REGION)
-
- self.client._extract_service_catalog(resources)
- self.assertEqual(self.client.endpoint_url, 'public')
-
- # Test admin url
- self.client = client.HTTPClient(
- username=USERNAME, tenant_name=TENANT_NAME, password=PASSWORD,
- auth_url=AUTH_URL, region_name=REGION, endpoint_type='adminURL')
-
- self.client._extract_service_catalog(resources)
- self.assertEqual(self.client.endpoint_url, 'admin')
-
- # Test public url
- self.client = client.HTTPClient(
- username=USERNAME, tenant_name=TENANT_NAME, password=PASSWORD,
- auth_url=AUTH_URL, region_name=REGION, endpoint_type='publicURL')
-
- self.client._extract_service_catalog(resources)
- self.assertEqual(self.client.endpoint_url, 'public')
-
- # Test internal url
- self.client = client.HTTPClient(
- username=USERNAME, tenant_name=TENANT_NAME, password=PASSWORD,
- auth_url=AUTH_URL, region_name=REGION, endpoint_type='internalURL')
-
- self.client._extract_service_catalog(resources)
- self.assertEqual(self.client.endpoint_url, 'internal')
-
- # Test url that isn't found in the service catalog
- self.client = client.HTTPClient(
- username=USERNAME, tenant_name=TENANT_NAME, password=PASSWORD,
- auth_url=AUTH_URL, region_name=REGION, endpoint_type='privateURL')
-
- self.assertRaises(exceptions.EndpointTypeNotFound,
- self.client._extract_service_catalog,
- resources)
diff --git a/tests/unit/test_casual_args.py b/tests/unit/test_casual_args.py
deleted file mode 100644
index fcfff93..0000000
--- a/tests/unit/test_casual_args.py
+++ /dev/null
@@ -1,99 +0,0 @@
-# Copyright 2012 OpenStack LLC.
-# All Rights Reserved
-#
-# 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.
-#
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-import testtools
-
-from quantumclient.common import exceptions
-from quantumclient.quantum import v2_0 as quantumV20
-
-
-class CLITestArgs(testtools.TestCase):
-
- def test_empty(self):
- _mydict = quantumV20.parse_args_to_dict([])
- self.assertEqual({}, _mydict)
-
- def test_default_bool(self):
- _specs = ['--my_bool', '--arg1', 'value1']
- _mydict = quantumV20.parse_args_to_dict(_specs)
- self.assertTrue(_mydict['my_bool'])
-
- def test_bool_true(self):
- _specs = ['--my-bool', 'type=bool', 'true', '--arg1', 'value1']
- _mydict = quantumV20.parse_args_to_dict(_specs)
- self.assertTrue(_mydict['my_bool'])
-
- def test_bool_false(self):
- _specs = ['--my_bool', 'type=bool', 'false', '--arg1', 'value1']
- _mydict = quantumV20.parse_args_to_dict(_specs)
- self.assertFalse(_mydict['my_bool'])
-
- def test_nargs(self):
- _specs = ['--tag', 'x', 'y', '--arg1', 'value1']
- _mydict = quantumV20.parse_args_to_dict(_specs)
- self.assertTrue('x' in _mydict['tag'])
- self.assertTrue('y' in _mydict['tag'])
-
- def test_badarg(self):
- _specs = ['--tag=t', 'x', 'y', '--arg1', 'value1']
- self.assertRaises(exceptions.CommandError,
- quantumV20.parse_args_to_dict, _specs)
-
- def test_badarg_with_minus(self):
- _specs = ['--arg1', 'value1', '-D']
- self.assertRaises(exceptions.CommandError,
- quantumV20.parse_args_to_dict, _specs)
-
- def test_goodarg_with_minus_number(self):
- _specs = ['--arg1', 'value1', '-1', '-1.0']
- _mydict = quantumV20.parse_args_to_dict(_specs)
- self.assertEqual(['value1', '-1', '-1.0'],
- _mydict['arg1'])
-
- def test_badarg_duplicate(self):
- _specs = ['--tag=t', '--arg1', 'value1', '--arg1', 'value1']
- self.assertRaises(exceptions.CommandError,
- quantumV20.parse_args_to_dict, _specs)
-
- def test_badarg_early_type_specification(self):
- _specs = ['type=dict', 'key=value']
- self.assertRaises(exceptions.CommandError,
- quantumV20.parse_args_to_dict, _specs)
-
- def test_arg(self):
- _specs = ['--tag=t', '--arg1', 'value1']
- self.assertEqual('value1',
- quantumV20.parse_args_to_dict(_specs)['arg1'])
-
- def test_dict_arg(self):
- _specs = ['--tag=t', '--arg1', 'type=dict', 'key1=value1,key2=value2']
- arg1 = quantumV20.parse_args_to_dict(_specs)['arg1']
- self.assertEqual('value1', arg1['key1'])
- self.assertEqual('value2', arg1['key2'])
-
- def test_dict_arg_with_attribute_named_type(self):
- _specs = ['--tag=t', '--arg1', 'type=dict', 'type=value1,key2=value2']
- arg1 = quantumV20.parse_args_to_dict(_specs)['arg1']
- self.assertEqual('value1', arg1['type'])
- self.assertEqual('value2', arg1['key2'])
-
- def test_list_of_dict_arg(self):
- _specs = ['--tag=t', '--arg1', 'type=dict',
- 'list=true', 'key1=value1,key2=value2']
- arg1 = quantumV20.parse_args_to_dict(_specs)['arg1']
- self.assertEqual('value1', arg1[0]['key1'])
- self.assertEqual('value2', arg1[0]['key2'])
diff --git a/tests/unit/test_cli20.py b/tests/unit/test_cli20.py
deleted file mode 100644
index c89cdfa..0000000
--- a/tests/unit/test_cli20.py
+++ /dev/null
@@ -1,504 +0,0 @@
-# Copyright 2012 OpenStack LLC.
-# All Rights Reserved
-#
-# 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.
-#
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-import urllib
-
-import fixtures
-import mox
-import testtools
-
-from quantumclient.common import constants
-from quantumclient import shell
-from quantumclient.v2_0 import client
-
-
-API_VERSION = "2.0"
-FORMAT = 'json'
-TOKEN = 'testtoken'
-ENDURL = 'localurl'
-
-
-class FakeStdout:
-
- def __init__(self):
- self.content = []
-
- def write(self, text):
- self.content.append(text)
-
- def make_string(self):
- result = ''
- for line in self.content:
- result = result + line
- return result
-
-
-class MyResp(object):
- def __init__(self, status):
- self.status = status
-
-
-class MyApp(object):
- def __init__(self, _stdout):
- self.stdout = _stdout
-
-
-def end_url(path, query=None, format=FORMAT):
- _url_str = ENDURL + "/v" + API_VERSION + path + "." + format
- return query and _url_str + "?" + query or _url_str
-
-
-class MyUrlComparator(mox.Comparator):
- def __init__(self, lhs, client):
- self.lhs = lhs
- self.client = client
-
- def equals(self, rhs):
- return str(self) == rhs
-
- def __str__(self):
- if self.client and self.client.format != FORMAT:
- lhs_parts = self.lhs.split("?", 1)
- if len(lhs_parts) == 2:
- lhs = ("%s.%s?%s" % (lhs_parts[0][:-4],
- self.client.format,
- lhs_parts[1]))
- else:
- lhs = ("%s.%s" % (lhs_parts[0][:-4],
- self.client.format))
- return lhs
- return self.lhs
-
- def __repr__(self):
- return str(self)
-
-
-class MyComparator(mox.Comparator):
- def __init__(self, lhs, client):
- self.lhs = lhs
- self.client = client
-
- def _com_dict(self, lhs, rhs):
- if len(lhs) != len(rhs):
- return False
- for key, value in lhs.iteritems():
- if key not in rhs:
- return False
- rhs_value = rhs[key]
- if not self._com(value, rhs_value):
- return False
- return True
-
- def _com_list(self, lhs, rhs):
- if len(lhs) != len(rhs):
- return False
- for lhs_value in lhs:
- if lhs_value not in rhs:
- return False
- return True
-
- def _com(self, lhs, rhs):
- if lhs is None:
- return rhs is None
- if isinstance(lhs, dict):
- if not isinstance(rhs, dict):
- return False
- return self._com_dict(lhs, rhs)
- if isinstance(lhs, list):
- if not isinstance(rhs, list):
- return False
- return self._com_list(lhs, rhs)
- if isinstance(lhs, tuple):
- if not isinstance(rhs, tuple):
- return False
- return self._com_list(lhs, rhs)
- return lhs == rhs
-
- def equals(self, rhs):
- if self.client:
- rhs = self.client.deserialize(rhs, 200)
- return self._com(self.lhs, rhs)
-
- def __repr__(self):
- if self.client:
- return self.client.serialize(self.lhs)
- return str(self.lhs)
-
-
-class CLITestV20Base(testtools.TestCase):
-
- format = 'json'
- test_id = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'
- id_field = 'id'
-
- def _find_resourceid(self, client, resource, name_or_id):
- return name_or_id
-
- def _get_attr_metadata(self):
- return self.metadata
- client.Client.EXTED_PLURALS.update(constants.PLURALS)
- client.Client.EXTED_PLURALS.update({'tags': 'tag'})
- return {'plurals': client.Client.EXTED_PLURALS,
- 'xmlns': constants.XML_NS_V20,
- constants.EXT_NS: {'prefix': 'http://xxxx.yy.com'}}
-
- def setUp(self, plurals={}):
- """Prepare the test environment."""
- super(CLITestV20Base, self).setUp()
- client.Client.EXTED_PLURALS.update(constants.PLURALS)
- client.Client.EXTED_PLURALS.update(plurals)
- self.metadata = {'plurals': client.Client.EXTED_PLURALS,
- 'xmlns': constants.XML_NS_V20,
- constants.EXT_NS: {'prefix':
- 'http://xxxx.yy.com'}}
- self.mox = mox.Mox()
- self.endurl = ENDURL
- self.fake_stdout = FakeStdout()
- self.useFixture(fixtures.MonkeyPatch('sys.stdout', self.fake_stdout))
- self.useFixture(fixtures.MonkeyPatch(
- 'quantumclient.quantum.v2_0.find_resourceid_by_name_or_id',
- self._find_resourceid))
- self.useFixture(fixtures.MonkeyPatch(
- 'quantumclient.v2_0.client.Client.get_attr_metadata',
- self._get_attr_metadata))
- self.client = client.Client(token=TOKEN, endpoint_url=self.endurl)
-
- def _test_create_resource(self, resource, cmd,
- name, myid, args,
- position_names, position_values, tenant_id=None,
- tags=None, admin_state_up=True, shared=False,
- extra_body=None, **kwargs):
- self.mox.StubOutWithMock(cmd, "get_client")
- self.mox.StubOutWithMock(self.client.httpclient, "request")
- cmd.get_client().MultipleTimes().AndReturn(self.client)
- non_admin_status_resources = ['subnet', 'floatingip', 'security_group',
- 'security_group_rule', 'qos_queue',
- 'network_gateway']
- if (resource in non_admin_status_resources):
- body = {resource: {}, }
- else:
- body = {resource: {'admin_state_up': admin_state_up, }, }
- if tenant_id:
- body[resource].update({'tenant_id': tenant_id})
- if tags:
- body[resource].update({'tags': tags})
- if shared:
- body[resource].update({'shared': shared})
- if extra_body:
- body[resource].update(extra_body)
- body[resource].update(kwargs)
-
- for i in xrange(len(position_names)):
- body[resource].update({position_names[i]: position_values[i]})
- ress = {resource:
- {self.id_field: myid}, }
- if name:
- ress[resource].update({'name': name})
- self.client.format = self.format
- resstr = self.client.serialize(ress)
- # url method body
- path = getattr(self.client, resource + "s_path")
- self.client.httpclient.request(
- end_url(path, format=self.format), 'POST',
- body=MyComparator(body, self.client),
- headers=mox.ContainsKeyValue(
- 'X-Auth-Token', TOKEN)).AndReturn((MyResp(200), resstr))
- args.extend(['--request-format', self.format])
- self.mox.ReplayAll()
- cmd_parser = cmd.get_parser('create_' + resource)
- shell.run_command(cmd, cmd_parser, args)
- self.mox.VerifyAll()
- self.mox.UnsetStubs()
- _str = self.fake_stdout.make_string()
- self.assertTrue(myid in _str)
- if name:
- self.assertTrue(name in _str)
-
- def _test_list_columns(self, cmd, resources_collection,
- resources_out, args=['-f', 'json']):
- self.mox.StubOutWithMock(cmd, "get_client")
- self.mox.StubOutWithMock(self.client.httpclient, "request")
- cmd.get_client().MultipleTimes().AndReturn(self.client)
- self.client.format = self.format
- resstr = self.client.serialize(resources_out)
-
- path = getattr(self.client, resources_collection + "_path")
- self.client.httpclient.request(
- end_url(path, format=self.format), 'GET',
- body=None,
- headers=mox.ContainsKeyValue(
- 'X-Auth-Token', TOKEN)).AndReturn((MyResp(200), resstr))
- args.extend(['--request-format', self.format])
- self.mox.ReplayAll()
- cmd_parser = cmd.get_parser("list_" + resources_collection)
- shell.run_command(cmd, cmd_parser, args)
- self.mox.VerifyAll()
- self.mox.UnsetStubs()
-
- def _test_list_resources(self, resources, cmd, detail=False, tags=[],
- fields_1=[], fields_2=[], page_size=None,
- sort_key=[], sort_dir=[], response_contents=None):
- self.mox.StubOutWithMock(cmd, "get_client")
- self.mox.StubOutWithMock(self.client.httpclient, "request")
- cmd.get_client().MultipleTimes().AndReturn(self.client)
- if response_contents is None:
- contents = [{self.id_field: 'myid1', },
- {self.id_field: 'myid2', }, ]
- else:
- contents = response_contents
- reses = {resources: contents}
- self.client.format = self.format
- resstr = self.client.serialize(reses)
- # url method body
- query = ""
- args = detail and ['-D', ] or []
- args.extend(['--request-format', self.format])
- if fields_1:
- for field in fields_1:
- args.append('--fields')
- args.append(field)
-
- if tags:
- args.append('--')
- args.append("--tag")
- for tag in tags:
- args.append(tag)
- if isinstance(tag, unicode):
- tag = urllib.quote(tag.encode('utf-8'))
- if query:
- query += "&tag=" + tag
- else:
- query = "tag=" + tag
- if (not tags) and fields_2:
- args.append('--')
- if fields_2:
- args.append("--fields")
- for field in fields_2:
- args.append(field)
- if detail:
- query = query and query + '&verbose=True' or 'verbose=True'
- fields_1.extend(fields_2)
- for field in fields_1:
- if query:
- query += "&fields=" + field
- else:
- query = "fields=" + field
- if page_size:
- args.append("--page-size")
- args.append(str(page_size))
- if query:
- query += "&limit=%s" % page_size
- else:
- query = "limit=%s" % page_size
- if sort_key:
- for key in sort_key:
- args.append('--sort-key')
- args.append(key)
- if query:
- query += '&'
- query += 'sort_key=%s' % key
- if sort_dir:
- len_diff = len(sort_key) - len(sort_dir)
- if len_diff > 0:
- sort_dir += ['asc'] * len_diff
- elif len_diff < 0:
- sort_dir = sort_dir[:len(sort_key)]
- for dir in sort_dir:
- args.append('--sort-dir')
- args.append(dir)
- if query:
- query += '&'
- query += 'sort_dir=%s' % dir
- path = getattr(self.client, resources + "_path")
- self.client.httpclient.request(
- MyUrlComparator(end_url(path, query, format=self.format),
- self.client),
- 'GET',
- body=None,
- headers=mox.ContainsKeyValue(
- 'X-Auth-Token', TOKEN)).AndReturn((MyResp(200), resstr))
- self.mox.ReplayAll()
- cmd_parser = cmd.get_parser("list_" + resources)
- shell.run_command(cmd, cmd_parser, args)
- self.mox.VerifyAll()
- self.mox.UnsetStubs()
- _str = self.fake_stdout.make_string()
- if response_contents is None:
- self.assertTrue('myid1' in _str)
- return _str
-
- def _test_list_resources_with_pagination(self, resources, cmd):
- self.mox.StubOutWithMock(cmd, "get_client")
- self.mox.StubOutWithMock(self.client.httpclient, "request")
- cmd.get_client().MultipleTimes().AndReturn(self.client)
- path = getattr(self.client, resources + "_path")
- fake_query = "marker=myid2&limit=2"
- reses1 = {resources: [{'id': 'myid1', },
- {'id': 'myid2', }],
- '%s_links' % resources: [{'href': end_url(path, fake_query),
- 'rel': 'next'}]}
- reses2 = {resources: [{'id': 'myid3', },
- {'id': 'myid4', }]}
- self.client.format = self.format
- resstr1 = self.client.serialize(reses1)
- resstr2 = self.client.serialize(reses2)
- self.client.httpclient.request(
- end_url(path, "", format=self.format), 'GET',
- body=None,
- headers=mox.ContainsKeyValue(
- 'X-Auth-Token', TOKEN)).AndReturn((MyResp(200), resstr1))
- self.client.httpclient.request(
- end_url(path, fake_query, format=self.format), 'GET',
- body=None,
- headers=mox.ContainsKeyValue(
- 'X-Auth-Token', TOKEN)).AndReturn((MyResp(200), resstr2))
- self.mox.ReplayAll()
- cmd_parser = cmd.get_parser("list_" + resources)
- args = ['--request-format', self.format]
- shell.run_command(cmd, cmd_parser, args)
- #parsed_args = cmd_parser.parse_args("")
- #cmd.run(parsed_args)
- self.mox.VerifyAll()
- self.mox.UnsetStubs()
-
- def _test_update_resource(self, resource, cmd, myid, args, extrafields):
- self.mox.StubOutWithMock(cmd, "get_client")
- self.mox.StubOutWithMock(self.client.httpclient, "request")
- cmd.get_client().MultipleTimes().AndReturn(self.client)
- body = {resource: extrafields}
- path = getattr(self.client, resource + "_path")
- self.client.httpclient.request(
- MyUrlComparator(end_url(path % myid, format=self.format),
- self.client),
- 'PUT',
- body=MyComparator(body, self.client),
- headers=mox.ContainsKeyValue(
- 'X-Auth-Token', TOKEN)).AndReturn((MyResp(204), None))
- args.extend(['--request-format', self.format])
- self.mox.ReplayAll()
- cmd_parser = cmd.get_parser("update_" + resource)
- shell.run_command(cmd, cmd_parser, args)
- self.mox.VerifyAll()
- self.mox.UnsetStubs()
- _str = self.fake_stdout.make_string()
- self.assertTrue(myid in _str)
-
- def _test_show_resource(self, resource, cmd, myid, args, fields=[]):
- self.mox.StubOutWithMock(cmd, "get_client")
- self.mox.StubOutWithMock(self.client.httpclient, "request")
- cmd.get_client().MultipleTimes().AndReturn(self.client)
- query = "&".join(["fields=%s" % field for field in fields])
- expected_res = {resource:
- {self.id_field: myid,
- 'name': 'myname', }, }
- self.client.format = self.format
- resstr = self.client.serialize(expected_res)
- path = getattr(self.client, resource + "_path")
- self.client.httpclient.request(
- end_url(path % myid, query, format=self.format), 'GET',
- body=None,
- headers=mox.ContainsKeyValue(
- 'X-Auth-Token', TOKEN)).AndReturn((MyResp(200), resstr))
- args.extend(['--request-format', self.format])
- self.mox.ReplayAll()
- cmd_parser = cmd.get_parser("show_" + resource)
- shell.run_command(cmd, cmd_parser, args)
- self.mox.VerifyAll()
- self.mox.UnsetStubs()
- _str = self.fake_stdout.make_string()
- self.assertTrue(myid in _str)
- self.assertTrue('myname' in _str)
-
- def _test_delete_resource(self, resource, cmd, myid, args):
- self.mox.StubOutWithMock(cmd, "get_client")
- self.mox.StubOutWithMock(self.client.httpclient, "request")
- cmd.get_client().MultipleTimes().AndReturn(self.client)
- path = getattr(self.client, resource + "_path")
- self.client.httpclient.request(
- end_url(path % myid, format=self.format), 'DELETE',
- body=None,
- headers=mox.ContainsKeyValue(
- 'X-Auth-Token', TOKEN)).AndReturn((MyResp(204), None))
- args.extend(['--request-format', self.format])
- self.mox.ReplayAll()
- cmd_parser = cmd.get_parser("delete_" + resource)
- shell.run_command(cmd, cmd_parser, args)
- self.mox.VerifyAll()
- self.mox.UnsetStubs()
- _str = self.fake_stdout.make_string()
- self.assertTrue(myid in _str)
-
- def _test_update_resource_action(self, resource, cmd, myid, action, args,
- body, retval=None):
- self.mox.StubOutWithMock(cmd, "get_client")
- self.mox.StubOutWithMock(self.client.httpclient, "request")
- cmd.get_client().MultipleTimes().AndReturn(self.client)
- path = getattr(self.client, resource + "_path")
- path_action = '%s/%s' % (myid, action)
- self.client.httpclient.request(
- end_url(path % path_action, format=self.format), 'PUT',
- body=MyComparator(body, self.client),
- headers=mox.ContainsKeyValue(
- 'X-Auth-Token', TOKEN)).AndReturn((MyResp(204), retval))
- args.extend(['--request-format', self.format])
- self.mox.ReplayAll()
- cmd_parser = cmd.get_parser("delete_" + resource)
- shell.run_command(cmd, cmd_parser, args)
- self.mox.VerifyAll()
- self.mox.UnsetStubs()
- _str = self.fake_stdout.make_string()
- self.assertTrue(myid in _str)
-
-
-class ClientV2UnicodeTestJson(CLITestV20Base):
- def test_do_request(self):
- self.client.format = self.format
- self.mox.StubOutWithMock(self.client.httpclient, "request")
- unicode_text = u'\u7f51\u7edc'
- # url with unicode
- action = u'/test'
- expected_action = action.encode('utf-8')
- # query string with unicode
- params = {'test': unicode_text}
- expect_query = urllib.urlencode({'test':
- unicode_text.encode('utf-8')})
- # request body with unicode
- body = params
- expect_body = self.client.serialize(body)
- # headers with unicode
- self.client.httpclient.auth_token = unicode_text
- expected_auth_token = unicode_text.encode('utf-8')
-
- self.client.httpclient.request(
- end_url(expected_action, query=expect_query, format=self.format),
- 'PUT', body=expect_body,
- headers=mox.ContainsKeyValue(
- 'X-Auth-Token',
- expected_auth_token)).AndReturn((MyResp(200), expect_body))
-
- self.mox.ReplayAll()
- res_body = self.client.do_request('PUT', action, body=body,
- params=params)
- self.mox.VerifyAll()
- self.mox.UnsetStubs()
-
- # test response with unicode
- self.assertEqual(res_body, body)
-
-
-class ClientV2UnicodeTestXML(ClientV2UnicodeTestJson):
- format = 'xml'
diff --git a/tests/unit/test_cli20_extensions.py b/tests/unit/test_cli20_extensions.py
deleted file mode 100644
index 874de77..0000000
--- a/tests/unit/test_cli20_extensions.py
+++ /dev/null
@@ -1,49 +0,0 @@
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-# Copyright 2013 NEC Corporation
-# All Rights Reserved.
-#
-# 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 sys
-
-from quantumclient.quantum.v2_0.extension import ListExt
-from quantumclient.quantum.v2_0.extension import ShowExt
-from tests.unit.test_cli20 import CLITestV20Base
-from tests.unit.test_cli20 import MyApp
-
-
-class CLITestV20Extension(CLITestV20Base):
- id_field = 'alias'
-
- def test_list_extensions(self):
- resources = 'extensions'
- cmd = ListExt(MyApp(sys.stdout), None)
- contents = [{'alias': 'ext1', 'name': 'name1', 'other': 'other1'},
- {'alias': 'ext2', 'name': 'name2', 'other': 'other2'}]
- ret = self._test_list_resources(resources, cmd,
- response_contents=contents)
- ret_words = set(ret.split())
- # Check only the default columns are shown.
- self.assertTrue('name' in ret_words)
- self.assertTrue('alias' in ret_words)
- self.assertFalse('other' in ret_words)
-
- def test_show_extension(self):
- # -F option does not work for ext-show at the moment, so -F option
- # is not passed in the commandline args as other tests do.
- resource = 'extension'
- cmd = ShowExt(MyApp(sys.stdout), None)
- args = [self.test_id]
- ext_alias = self.test_id
- self._test_show_resource(resource, cmd, ext_alias, args, fields=[])
diff --git a/tests/unit/test_cli20_floatingips.py b/tests/unit/test_cli20_floatingips.py
deleted file mode 100644
index b640112..0000000
--- a/tests/unit/test_cli20_floatingips.py
+++ /dev/null
@@ -1,139 +0,0 @@
-#!/usr/bin/env python
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-# Copyright 2012 Red Hat
-# All Rights Reserved.
-#
-# 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 sys
-
-from quantumclient.quantum.v2_0 import floatingip as fip
-from tests.unit import test_cli20
-
-
-class CLITestV20FloatingIpsJSON(test_cli20.CLITestV20Base):
- def test_create_floatingip(self):
- """Create floatingip: fip1."""
- resource = 'floatingip'
- cmd = fip.CreateFloatingIP(test_cli20.MyApp(sys.stdout), None)
- name = 'fip1'
- myid = 'myid'
- args = [name]
- position_names = ['floating_network_id']
- position_values = [name]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values)
-
- def test_create_floatingip_and_port(self):
- """Create floatingip: fip1."""
- resource = 'floatingip'
- cmd = fip.CreateFloatingIP(test_cli20.MyApp(sys.stdout), None)
- name = 'fip1'
- myid = 'myid'
- pid = 'mypid'
- args = [name, '--port_id', pid]
- position_names = ['floating_network_id', 'port_id']
- position_values = [name, pid]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values)
-
- # Test dashed options
- args = [name, '--port-id', pid]
- position_names = ['floating_network_id', 'port_id']
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values)
-
- def test_create_floatingip_and_port_and_address(self):
- """Create floatingip: fip1 with a given port and address."""
- resource = 'floatingip'
- cmd = fip.CreateFloatingIP(test_cli20.MyApp(sys.stdout), None)
- name = 'fip1'
- myid = 'myid'
- pid = 'mypid'
- addr = '10.0.0.99'
- args = [name, '--port_id', pid, '--fixed_ip_address', addr]
- position_names = ['floating_network_id', 'port_id', 'fixed_ip_address']
- position_values = [name, pid, addr]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values)
- # Test dashed options
- args = [name, '--port-id', pid, '--fixed-ip-address', addr]
- position_names = ['floating_network_id', 'port_id', 'fixed_ip_address']
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values)
-
- def test_list_floatingips(self):
- """list floatingips: -D."""
- resources = 'floatingips'
- cmd = fip.ListFloatingIP(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd, True)
-
- def test_list_floatingips_pagination(self):
- resources = 'floatingips'
- cmd = fip.ListFloatingIP(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources_with_pagination(resources, cmd)
-
- def test_list_floatingips_sort(self):
- """list floatingips: --sort-key name --sort-key id --sort-key asc
- --sort-key desc
- """
- resources = 'floatingips'
- cmd = fip.ListFloatingIP(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd,
- sort_key=["name", "id"],
- sort_dir=["asc", "desc"])
-
- def test_list_floatingips_limit(self):
- """list floatingips: -P."""
- resources = 'floatingips'
- cmd = fip.ListFloatingIP(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd, page_size=1000)
-
- def test_delete_floatingip(self):
- """Delete floatingip: fip1."""
- resource = 'floatingip'
- cmd = fip.DeleteFloatingIP(test_cli20.MyApp(sys.stdout), None)
- myid = 'myid'
- args = [myid]
- self._test_delete_resource(resource, cmd, myid, args)
-
- def test_show_floatingip(self):
- """Show floatingip: --fields id."""
- resource = 'floatingip'
- cmd = fip.ShowFloatingIP(test_cli20.MyApp(sys.stdout), None)
- args = ['--fields', 'id', self.test_id]
- self._test_show_resource(resource, cmd, self.test_id,
- args, ['id'])
-
- def test_disassociate_ip(self):
- """Disassociate floating IP: myid."""
- resource = 'floatingip'
- cmd = fip.DisassociateFloatingIP(test_cli20.MyApp(sys.stdout), None)
- args = ['myid']
- self._test_update_resource(resource, cmd, 'myid',
- args, {"port_id": None}
- )
-
- def test_associate_ip(self):
- """Associate floating IP: myid portid."""
- resource = 'floatingip'
- cmd = fip.AssociateFloatingIP(test_cli20.MyApp(sys.stdout), None)
- args = ['myid', 'portid']
- self._test_update_resource(resource, cmd, 'myid',
- args, {"port_id": "portid"}
- )
-
-
-class CLITestV20FloatingIpsXML(CLITestV20FloatingIpsJSON):
- format = 'xml'
diff --git a/tests/unit/test_cli20_network.py b/tests/unit/test_cli20_network.py
deleted file mode 100644
index 7df54ff..0000000
--- a/tests/unit/test_cli20_network.py
+++ /dev/null
@@ -1,533 +0,0 @@
-# All Rights Reserved
-#
-# 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.
-#
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-import sys
-
-import mox
-
-from quantumclient.common import exceptions
-from quantumclient.common import utils
-from quantumclient.quantum.v2_0 import network
-from quantumclient import shell
-from tests.unit import test_cli20
-
-
-class CLITestV20NetworkJSON(test_cli20.CLITestV20Base):
- def setUp(self):
- super(CLITestV20NetworkJSON, self).setUp(plurals={'tags': 'tag'})
-
- def test_create_network(self):
- """Create net: myname."""
- resource = 'network'
- cmd = network.CreateNetwork(test_cli20.MyApp(sys.stdout), None)
- name = 'myname'
- myid = 'myid'
- args = [name, ]
- position_names = ['name', ]
- position_values = [name, ]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values)
-
- def test_create_network_with_unicode(self):
- """Create net: u'\u7f51\u7edc'."""
- resource = 'network'
- cmd = network.CreateNetwork(test_cli20.MyApp(sys.stdout), None)
- name = u'\u7f51\u7edc'
- myid = 'myid'
- args = [name, ]
- position_names = ['name', ]
- position_values = [name, ]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values)
-
- def test_create_network_tenant(self):
- """Create net: --tenant_id tenantid myname."""
- resource = 'network'
- cmd = network.CreateNetwork(test_cli20.MyApp(sys.stdout), None)
- name = 'myname'
- myid = 'myid'
- args = ['--tenant_id', 'tenantid', name]
- position_names = ['name', ]
- position_values = [name, ]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values,
- tenant_id='tenantid')
-
- # Test dashed options
- args = ['--tenant-id', 'tenantid', name]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values,
- tenant_id='tenantid')
-
- def test_create_network_tags(self):
- """Create net: myname --tags a b."""
- resource = 'network'
- cmd = network.CreateNetwork(test_cli20.MyApp(sys.stdout), None)
- name = 'myname'
- myid = 'myid'
- args = [name, '--tags', 'a', 'b']
- position_names = ['name', ]
- position_values = [name, ]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values,
- tags=['a', 'b'])
-
- def test_create_network_state(self):
- """Create net: --admin_state_down myname."""
- resource = 'network'
- cmd = network.CreateNetwork(test_cli20.MyApp(sys.stdout), None)
- name = 'myname'
- myid = 'myid'
- args = ['--admin_state_down', name, ]
- position_names = ['name', ]
- position_values = [name, ]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values,
- admin_state_up=False)
-
- # Test dashed options
- args = ['--admin-state-down', name, ]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values,
- admin_state_up=False)
-
- def test_list_nets_empty_with_column(self):
- resources = "networks"
- cmd = network.ListNetwork(test_cli20.MyApp(sys.stdout), None)
- self.mox.StubOutWithMock(cmd, "get_client")
- self.mox.StubOutWithMock(self.client.httpclient, "request")
- self.mox.StubOutWithMock(network.ListNetwork, "extend_list")
- network.ListNetwork.extend_list(mox.IsA(list), mox.IgnoreArg())
- cmd.get_client().MultipleTimes().AndReturn(self.client)
- reses = {resources: []}
- resstr = self.client.serialize(reses)
- # url method body
- query = "id=myfakeid"
- args = ['-c', 'id', '--', '--id', 'myfakeid']
- path = getattr(self.client, resources + "_path")
- self.client.httpclient.request(
- test_cli20.end_url(path, query), 'GET',
- body=None,
- headers=mox.ContainsKeyValue(
- 'X-Auth-Token',
- test_cli20.TOKEN)).AndReturn(
- (test_cli20.MyResp(200), resstr))
- self.mox.ReplayAll()
- cmd_parser = cmd.get_parser("list_" + resources)
- shell.run_command(cmd, cmd_parser, args)
- self.mox.VerifyAll()
- self.mox.UnsetStubs()
- _str = self.fake_stdout.make_string()
- self.assertEquals('\n', _str)
-
- def _test_list_networks(self, cmd, detail=False, tags=[],
- fields_1=[], fields_2=[], page_size=None,
- sort_key=[], sort_dir=[]):
- resources = "networks"
- self.mox.StubOutWithMock(network.ListNetwork, "extend_list")
- network.ListNetwork.extend_list(mox.IsA(list), mox.IgnoreArg())
- self._test_list_resources(resources, cmd, detail, tags,
- fields_1, fields_2, page_size=page_size,
- sort_key=sort_key, sort_dir=sort_dir)
-
- def test_list_nets_pagination(self):
- cmd = network.ListNetwork(test_cli20.MyApp(sys.stdout), None)
- self.mox.StubOutWithMock(network.ListNetwork, "extend_list")
- network.ListNetwork.extend_list(mox.IsA(list), mox.IgnoreArg())
- self._test_list_resources_with_pagination("networks", cmd)
-
- def test_list_nets_sort(self):
- """list nets: --sort-key name --sort-key id --sort-dir asc
- --sort-dir desc
- """
- cmd = network.ListNetwork(test_cli20.MyApp(sys.stdout), None)
- self._test_list_networks(cmd, sort_key=['name', 'id'],
- sort_dir=['asc', 'desc'])
-
- def test_list_nets_sort_with_keys_more_than_dirs(self):
- """list nets: --sort-key name --sort-key id --sort-dir desc
- """
- cmd = network.ListNetwork(test_cli20.MyApp(sys.stdout), None)
- self._test_list_networks(cmd, sort_key=['name', 'id'],
- sort_dir=['desc'])
-
- def test_list_nets_sort_with_dirs_more_than_keys(self):
- """list nets: --sort-key name --sort-dir desc --sort-dir asc
- """
- cmd = network.ListNetwork(test_cli20.MyApp(sys.stdout), None)
- self._test_list_networks(cmd, sort_key=['name'],
- sort_dir=['desc', 'asc'])
-
- def test_list_nets_limit(self):
- """list nets: -P."""
- cmd = network.ListNetwork(test_cli20.MyApp(sys.stdout), None)
- self._test_list_networks(cmd, page_size=1000)
-
- def test_list_nets_detail(self):
- """list nets: -D."""
- cmd = network.ListNetwork(test_cli20.MyApp(sys.stdout), None)
- self._test_list_networks(cmd, True)
-
- def test_list_nets_tags(self):
- """List nets: -- --tags a b."""
- cmd = network.ListNetwork(test_cli20.MyApp(sys.stdout), None)
- self._test_list_networks(cmd, tags=['a', 'b'])
-
- def test_list_nets_tags_with_unicode(self):
- """List nets: -- --tags u'\u7f51\u7edc'."""
- cmd = network.ListNetwork(test_cli20.MyApp(sys.stdout), None)
- self._test_list_networks(cmd, tags=[u'\u7f51\u7edc'])
-
- def test_list_nets_detail_tags(self):
- """List nets: -D -- --tags a b."""
- cmd = network.ListNetwork(test_cli20.MyApp(sys.stdout), None)
- self._test_list_networks(cmd, detail=True, tags=['a', 'b'])
-
- def _test_list_nets_extend_subnets(self, data, expected):
- def setup_list_stub(resources, data, query):
- reses = {resources: data}
- resstr = self.client.serialize(reses)
- resp = (test_cli20.MyResp(200), resstr)
- path = getattr(self.client, resources + '_path')
- self.client.httpclient.request(
- test_cli20.end_url(path, query), 'GET',
- body=None,
- headers=mox.ContainsKeyValue(
- 'X-Auth-Token', test_cli20.TOKEN)).AndReturn(resp)
-
- cmd = network.ListNetwork(test_cli20.MyApp(sys.stdout), None)
- self.mox.StubOutWithMock(cmd, 'get_client')
- self.mox.StubOutWithMock(self.client.httpclient, 'request')
- cmd.get_client().AndReturn(self.client)
- setup_list_stub('networks', data, '')
- cmd.get_client().AndReturn(self.client)
- filters = ''
- for n in data:
- for s in n['subnets']:
- filters = filters + "&id=%s" % s
- setup_list_stub('subnets',
- [{'id': 'mysubid1', 'cidr': '192.168.1.0/24'},
- {'id': 'mysubid2', 'cidr': '172.16.0.0/24'},
- {'id': 'mysubid3', 'cidr': '10.1.1.0/24'}],
- query='fields=id&fields=cidr' + filters)
- self.mox.ReplayAll()
-
- args = []
- cmd_parser = cmd.get_parser('list_networks')
- parsed_args = cmd_parser.parse_args(args)
- result = cmd.get_data(parsed_args)
- self.mox.VerifyAll()
- self.mox.UnsetStubs()
- _result = [x for x in result[1]]
- self.assertEqual(len(_result), len(expected))
- for res, exp in zip(_result, expected):
- self.assertEqual(len(res), len(exp))
- for a, b in zip(res, exp):
- self.assertEqual(a, b)
-
- def test_list_nets_extend_subnets(self):
- data = [{'id': 'netid1', 'name': 'net1', 'subnets': ['mysubid1']},
- {'id': 'netid2', 'name': 'net2', 'subnets': ['mysubid2',
- 'mysubid3']}]
- # id, name, subnets
- expected = [('netid1', 'net1', 'mysubid1 192.168.1.0/24'),
- ('netid2', 'net2',
- 'mysubid2 172.16.0.0/24\nmysubid3 10.1.1.0/24')]
- self._test_list_nets_extend_subnets(data, expected)
-
- def test_list_nets_extend_subnets_no_subnet(self):
- data = [{'id': 'netid1', 'name': 'net1', 'subnets': ['mysubid1']},
- {'id': 'netid2', 'name': 'net2', 'subnets': ['mysubid4']}]
- # id, name, subnets
- expected = [('netid1', 'net1', 'mysubid1 192.168.1.0/24'),
- ('netid2', 'net2', 'mysubid4 ')]
- self._test_list_nets_extend_subnets(data, expected)
-
- def test_list_nets_fields(self):
- """List nets: --fields a --fields b -- --fields c d."""
- cmd = network.ListNetwork(test_cli20.MyApp(sys.stdout), None)
- self._test_list_networks(cmd,
- fields_1=['a', 'b'], fields_2=['c', 'd'])
-
- def _test_list_nets_columns(self, cmd, returned_body,
- args=['-f', 'json']):
- resources = 'networks'
- self.mox.StubOutWithMock(network.ListNetwork, "extend_list")
- network.ListNetwork.extend_list(mox.IsA(list), mox.IgnoreArg())
- self._test_list_columns(cmd, resources, returned_body, args=args)
-
- def test_list_nets_defined_column(self):
- cmd = network.ListNetwork(test_cli20.MyApp(sys.stdout), None)
- returned_body = {"networks": [{"name": "buildname3",
- "id": "id3",
- "tenant_id": "tenant_3",
- "subnets": []}]}
- self._test_list_nets_columns(cmd, returned_body,
- args=['-f', 'json', '-c', 'id'])
- _str = self.fake_stdout.make_string()
- returned_networks = utils.loads(_str)
- self.assertEquals(1, len(returned_networks))
- net = returned_networks[0]
- self.assertEquals(1, len(net))
- self.assertEquals("id", net.keys()[0])
-
- def test_list_nets_with_default_column(self):
- cmd = network.ListNetwork(test_cli20.MyApp(sys.stdout), None)
- returned_body = {"networks": [{"name": "buildname3",
- "id": "id3",
- "tenant_id": "tenant_3",
- "subnets": []}]}
- self._test_list_nets_columns(cmd, returned_body)
- _str = self.fake_stdout.make_string()
- returned_networks = utils.loads(_str)
- self.assertEquals(1, len(returned_networks))
- net = returned_networks[0]
- self.assertEquals(3, len(net))
- self.assertEquals(0, len(set(net) ^ set(cmd.list_columns)))
-
- def test_list_external_nets_empty_with_column(self):
- resources = "networks"
- cmd = network.ListExternalNetwork(test_cli20.MyApp(sys.stdout), None)
- self.mox.StubOutWithMock(cmd, "get_client")
- self.mox.StubOutWithMock(self.client.httpclient, "request")
- self.mox.StubOutWithMock(network.ListNetwork, "extend_list")
- network.ListNetwork.extend_list(mox.IsA(list), mox.IgnoreArg())
- cmd.get_client().MultipleTimes().AndReturn(self.client)
- reses = {resources: []}
- resstr = self.client.serialize(reses)
- # url method body
- query = "router%3Aexternal=True&id=myfakeid"
- args = ['-c', 'id', '--', '--id', 'myfakeid']
- path = getattr(self.client, resources + "_path")
- self.client.httpclient.request(
- test_cli20.end_url(path, query), 'GET',
- body=None,
- headers=mox.ContainsKeyValue(
- 'X-Auth-Token',
- test_cli20.TOKEN)).AndReturn(
- (test_cli20.MyResp(200), resstr))
- self.mox.ReplayAll()
- cmd_parser = cmd.get_parser("list_" + resources)
- shell.run_command(cmd, cmd_parser, args)
- self.mox.VerifyAll()
- self.mox.UnsetStubs()
- _str = self.fake_stdout.make_string()
- self.assertEquals('\n', _str)
-
- def _test_list_external_nets(self, resources, cmd,
- detail=False, tags=[],
- fields_1=[], fields_2=[]):
- self.mox.StubOutWithMock(cmd, "get_client")
- self.mox.StubOutWithMock(self.client.httpclient, "request")
- self.mox.StubOutWithMock(network.ListNetwork, "extend_list")
- network.ListNetwork.extend_list(mox.IsA(list), mox.IgnoreArg())
- cmd.get_client().MultipleTimes().AndReturn(self.client)
- reses = {resources: [{'id': 'myid1', },
- {'id': 'myid2', }, ], }
-
- resstr = self.client.serialize(reses)
-
- # url method body
- query = ""
- args = detail and ['-D', ] or []
- if fields_1:
- for field in fields_1:
- args.append('--fields')
- args.append(field)
- if tags:
- args.append('--')
- args.append("--tag")
- for tag in tags:
- args.append(tag)
- if (not tags) and fields_2:
- args.append('--')
- if fields_2:
- args.append("--fields")
- for field in fields_2:
- args.append(field)
- fields_1.extend(fields_2)
- for field in fields_1:
- if query:
- query += "&fields=" + field
- else:
- query = "fields=" + field
- if query:
- query += '&router%3Aexternal=True'
- else:
- query += 'router%3Aexternal=True'
- for tag in tags:
- if query:
- query += "&tag=" + tag
- else:
- query = "tag=" + tag
- if detail:
- query = query and query + '&verbose=True' or 'verbose=True'
- path = getattr(self.client, resources + "_path")
-
- self.client.httpclient.request(
- test_cli20.end_url(path, query), 'GET',
- body=None,
- headers=mox.ContainsKeyValue('X-Auth-Token', test_cli20.TOKEN)
- ).AndReturn((test_cli20.MyResp(200), resstr))
- self.mox.ReplayAll()
- cmd_parser = cmd.get_parser("list_" + resources)
- shell.run_command(cmd, cmd_parser, args)
- self.mox.VerifyAll()
- self.mox.UnsetStubs()
- _str = self.fake_stdout.make_string()
-
- self.assertTrue('myid1' in _str)
-
- def test_list_external_nets_detail(self):
- """list external nets: -D."""
- resources = "networks"
- cmd = network.ListExternalNetwork(test_cli20.MyApp(sys.stdout), None)
- self._test_list_external_nets(resources, cmd, True)
-
- def test_list_external_nets_tags(self):
- """List external nets: -- --tags a b."""
- resources = "networks"
- cmd = network.ListExternalNetwork(test_cli20.MyApp(sys.stdout), None)
- self._test_list_external_nets(resources,
- cmd, tags=['a', 'b'])
-
- def test_list_external_nets_detail_tags(self):
- """List external nets: -D -- --tags a b."""
- resources = "networks"
- cmd = network.ListExternalNetwork(test_cli20.MyApp(sys.stdout), None)
- self._test_list_external_nets(resources, cmd,
- detail=True, tags=['a', 'b'])
-
- def test_list_externel_nets_fields(self):
- """List external nets: --fields a --fields b -- --fields c d."""
- resources = "networks"
- cmd = network.ListExternalNetwork(test_cli20.MyApp(sys.stdout), None)
- self._test_list_external_nets(resources, cmd,
- fields_1=['a', 'b'],
- fields_2=['c', 'd'])
-
- def test_update_network_exception(self):
- """Update net: myid."""
- resource = 'network'
- cmd = network.UpdateNetwork(test_cli20.MyApp(sys.stdout), None)
- self.assertRaises(exceptions.CommandError, self._test_update_resource,
- resource, cmd, 'myid', ['myid'], {})
-
- def test_update_network(self):
- """Update net: myid --name myname --tags a b."""
- resource = 'network'
- cmd = network.UpdateNetwork(test_cli20.MyApp(sys.stdout), None)
- self._test_update_resource(resource, cmd, 'myid',
- ['myid', '--name', 'myname',
- '--tags', 'a', 'b'],
- {'name': 'myname', 'tags': ['a', 'b'], }
- )
-
- def test_update_network_with_unicode(self):
- """Update net: myid --name u'\u7f51\u7edc' --tags a b."""
- resource = 'network'
- cmd = network.UpdateNetwork(test_cli20.MyApp(sys.stdout), None)
- self._test_update_resource(resource, cmd, 'myid',
- ['myid', '--name', u'\u7f51\u7edc',
- '--tags', 'a', 'b'],
- {'name': u'\u7f51\u7edc',
- 'tags': ['a', 'b'], }
- )
-
- def test_show_network(self):
- """Show net: --fields id --fields name myid."""
- resource = 'network'
- cmd = network.ShowNetwork(test_cli20.MyApp(sys.stdout), None)
- args = ['--fields', 'id', '--fields', 'name', self.test_id]
- self._test_show_resource(resource, cmd, self.test_id, args,
- ['id', 'name'])
-
- def test_delete_network(self):
- """Delete net: myid."""
- resource = 'network'
- cmd = network.DeleteNetwork(test_cli20.MyApp(sys.stdout), None)
- myid = 'myid'
- args = [myid]
- self._test_delete_resource(resource, cmd, myid, args)
-
- def _test_extend_list(self, mox_calls):
- data = [{'id': 'netid%d' % i, 'name': 'net%d' % i,
- 'subnets': ['mysubid%d' % i]}
- for i in range(0, 10)]
- self.mox.StubOutWithMock(self.client.httpclient, "request")
- path = getattr(self.client, 'subnets_path')
- cmd = network.ListNetwork(test_cli20.MyApp(sys.stdout), None)
- self.mox.StubOutWithMock(cmd, "get_client")
- cmd.get_client().MultipleTimes().AndReturn(self.client)
- mox_calls(path, data)
- self.mox.ReplayAll()
- known_args, _vs = cmd.get_parser('create_subnets').parse_known_args()
- cmd.extend_list(data, known_args)
- self.mox.VerifyAll()
-
- def _build_test_data(self, data):
- subnet_ids = []
- response = []
- filters = ""
- for n in data:
- if 'subnets' in n:
- subnet_ids.extend(n['subnets'])
- for subnet_id in n['subnets']:
- filters = "%s&id=%s" % (filters, subnet_id)
- response.append({'id': subnet_id,
- 'cidr': '192.168.0.0/16'})
- resp_str = self.client.serialize({'subnets': response})
- resp = (test_cli20.MyResp(200), resp_str)
- return filters, resp
-
- def test_extend_list(self):
- def mox_calls(path, data):
- filters, response = self._build_test_data(data)
- self.client.httpclient.request(
- test_cli20.end_url(path, 'fields=id&fields=cidr' + filters),
- 'GET',
- body=None,
- headers=mox.ContainsKeyValue(
- 'X-Auth-Token', test_cli20.TOKEN)).AndReturn(response)
-
- self._test_extend_list(mox_calls)
-
- def test_extend_list_exceed_max_uri_len(self):
- def mox_calls(path, data):
- sub_data_lists = [data[:len(data) - 1], data[len(data) - 1:]]
- filters, response = self._build_test_data(data)
-
- # 1 char of extra URI len will cause a split in 2 requests
- self.mox.StubOutWithMock(self.client, "_check_uri_length")
- self.client._check_uri_length(mox.IgnoreArg()).AndRaise(
- exceptions.RequestURITooLong(excess=1))
-
- for data in sub_data_lists:
- filters, response = self._build_test_data(data)
- self.client._check_uri_length(mox.IgnoreArg()).AndReturn(None)
- self.client.httpclient.request(
- test_cli20.end_url(path,
- 'fields=id&fields=cidr%s' % filters),
- 'GET',
- body=None,
- headers=mox.ContainsKeyValue(
- 'X-Auth-Token', test_cli20.TOKEN)).AndReturn(response)
-
- self._test_extend_list(mox_calls)
-
-
-class CLITestV20NetworkXML(CLITestV20NetworkJSON):
- format = 'xml'
diff --git a/tests/unit/test_cli20_nvp_queue.py b/tests/unit/test_cli20_nvp_queue.py
deleted file mode 100644
index d84d9df..0000000
--- a/tests/unit/test_cli20_nvp_queue.py
+++ /dev/null
@@ -1,88 +0,0 @@
-#!/usr/bin/env python
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-# Copyright 2013 Nicira Inc.
-# All Rights Reserved.
-#
-# 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 sys
-
-from quantumclient.quantum.v2_0 import nvp_qos_queue as qos
-from tests.unit import test_cli20
-
-
-class CLITestV20NvpQosQueueJSON(test_cli20.CLITestV20Base):
- def setUp(self):
- super(CLITestV20NvpQosQueueJSON, self).setUp(
- plurals={'qos_queues': 'qos_queue'})
-
- def test_create_qos_queue(self):
- """Create a qos queue."""
- resource = 'qos_queue'
- cmd = qos.CreateQoSQueue(
- test_cli20.MyApp(sys.stdout), None)
- myid = 'myid'
- name = 'my_queue'
- default = False
- args = ['--default', default, name]
- position_names = ['name', 'default']
- position_values = [name, default]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values)
-
- def test_create_qos_queue_all_values(self):
- """Create a qos queue."""
- resource = 'qos_queue'
- cmd = qos.CreateQoSQueue(
- test_cli20.MyApp(sys.stdout), None)
- myid = 'myid'
- name = 'my_queue'
- default = False
- min = '10'
- max = '40'
- qos_marking = 'untrusted'
- dscp = '0'
- args = ['--default', default, '--min', min, '--max', max,
- '--qos-marking', qos_marking, '--dscp', dscp, name]
- position_names = ['name', 'default', 'min', 'max', 'qos_marking',
- 'dscp']
- position_values = [name, default, min, max, qos_marking, dscp]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values)
-
- def test_list_qos_queue(self):
- resources = "qos_queues"
- cmd = qos.ListQoSQueue(
- test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd, True)
-
- def test_show_qos_queue_id(self):
- resource = 'qos_queue'
- cmd = qos.ShowQoSQueue(
- test_cli20.MyApp(sys.stdout), None)
- args = ['--fields', 'id', self.test_id]
- self._test_show_resource(resource, cmd, self.test_id,
- args, ['id'])
-
- def test_delete_qos_queue(self):
- resource = 'qos_queue'
- cmd = qos.DeleteQoSQueue(
- test_cli20.MyApp(sys.stdout), None)
- myid = 'myid'
- args = [myid]
- self._test_delete_resource(resource, cmd, myid, args)
-
-
-class CLITestV20NvpQosQueueXML(CLITestV20NvpQosQueueJSON):
- format = 'xml'
diff --git a/tests/unit/test_cli20_nvpnetworkgateway.py b/tests/unit/test_cli20_nvpnetworkgateway.py
deleted file mode 100644
index ebf730a..0000000
--- a/tests/unit/test_cli20_nvpnetworkgateway.py
+++ /dev/null
@@ -1,114 +0,0 @@
-# Copyright 2012 Nicira, Inc
-# All Rights Reserved
-#
-# 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.
-#
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-import sys
-
-from quantumclient.quantum.v2_0 import nvpnetworkgateway as nwgw
-from tests.unit import test_cli20
-
-
-class CLITestV20NetworkGatewayJSON(test_cli20.CLITestV20Base):
-
- resource = "network_gateway"
-
- def setUp(self):
- super(CLITestV20NetworkGatewayJSON, self).setUp(
- plurals={'devices': 'device',
- 'network_gateways': 'network_gateway'})
-
- def test_create_gateway(self):
- cmd = nwgw.CreateNetworkGateway(test_cli20.MyApp(sys.stdout), None)
- name = 'gw-test'
- myid = 'myid'
- args = [name, ]
- position_names = ['name', ]
- position_values = [name, ]
- self._test_create_resource(self.resource, cmd, name, myid, args,
- position_names, position_values)
-
- def test_create_gateway_with_tenant(self):
- cmd = nwgw.CreateNetworkGateway(test_cli20.MyApp(sys.stdout), None)
- name = 'gw-test'
- myid = 'myid'
- args = ['--tenant_id', 'tenantid', name]
- position_names = ['name', ]
- position_values = [name, ]
- self._test_create_resource(self.resource, cmd, name, myid, args,
- position_names, position_values,
- tenant_id='tenantid')
-
- def test_create_gateway_with_device(self):
- cmd = nwgw.CreateNetworkGateway(test_cli20.MyApp(sys.stdout), None)
- name = 'gw-test'
- myid = 'myid'
- args = ['--device', 'device_id=test', name, ]
- position_names = ['name', ]
- position_values = [name, ]
- self._test_create_resource(self.resource, cmd, name, myid, args,
- position_names, position_values,
- devices=[{'device_id': 'test'}])
-
- def test_list_gateways(self):
- resources = '%ss' % self.resource
- cmd = nwgw.ListNetworkGateway(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd, True)
-
- def test_update_gateway(self):
- cmd = nwgw.UpdateNetworkGateway(test_cli20.MyApp(sys.stdout), None)
- self._test_update_resource(self.resource, cmd, 'myid',
- ['myid', '--name', 'cavani'],
- {'name': 'cavani'})
-
- def test_delete_gateway(self):
- cmd = nwgw.DeleteNetworkGateway(test_cli20.MyApp(sys.stdout), None)
- myid = 'myid'
- args = [myid]
- self._test_delete_resource(self.resource, cmd, myid, args)
-
- def test_show_gateway(self):
- cmd = nwgw.ShowNetworkGateway(test_cli20.MyApp(sys.stdout), None)
- args = ['--fields', 'id', '--fields', 'name', self.test_id]
- self._test_show_resource(self.resource, cmd, self.test_id, args,
- ['id', 'name'])
-
- def test_connect_network_to_gateway(self):
- cmd = nwgw.ConnectNetworkGateway(test_cli20.MyApp(sys.stdout), None)
- args = ['gw_id', 'net_id',
- '--segmentation-type', 'edi',
- '--segmentation-id', '7']
- self._test_update_resource_action(self.resource, cmd, 'gw_id',
- 'connect_network',
- args,
- {'network_id': 'net_id',
- 'segmentation_type': 'edi',
- 'segmentation_id': '7'})
-
- def test_disconnect_network_from_gateway(self):
- cmd = nwgw.DisconnectNetworkGateway(test_cli20.MyApp(sys.stdout), None)
- args = ['gw_id', 'net_id',
- '--segmentation-type', 'edi',
- '--segmentation-id', '7']
- self._test_update_resource_action(self.resource, cmd, 'gw_id',
- 'disconnect_network',
- args,
- {'network_id': 'net_id',
- 'segmentation_type': 'edi',
- 'segmentation_id': '7'})
-
-
-class CLITestV20NetworkGatewayXML(CLITestV20NetworkGatewayJSON):
- format = 'xml'
diff --git a/tests/unit/test_cli20_port.py b/tests/unit/test_cli20_port.py
deleted file mode 100644
index 545f35b..0000000
--- a/tests/unit/test_cli20_port.py
+++ /dev/null
@@ -1,304 +0,0 @@
-# Copyright 2012 OpenStack LLC.
-# All Rights Reserved
-#
-# 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.
-#
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-import sys
-
-import mox
-
-from quantumclient.quantum.v2_0 import port
-from quantumclient import shell
-from tests.unit import test_cli20
-
-
-class CLITestV20PortJSON(test_cli20.CLITestV20Base):
- def setUp(self):
- super(CLITestV20PortJSON, self).setUp(plurals={'tags': 'tag'})
-
- def test_create_port(self):
- """Create port: netid."""
- resource = 'port'
- cmd = port.CreatePort(test_cli20.MyApp(sys.stdout), None)
- name = 'myname'
- myid = 'myid'
- netid = 'netid'
- args = [netid]
- position_names = ['network_id']
- position_values = []
- position_values.extend([netid])
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values)
-
- def test_create_port_full(self):
- """Create port: --mac_address mac --device_id deviceid netid."""
- resource = 'port'
- cmd = port.CreatePort(test_cli20.MyApp(sys.stdout), None)
- name = 'myname'
- myid = 'myid'
- netid = 'netid'
- args = ['--mac_address', 'mac', '--device_id', 'deviceid', netid]
- position_names = ['network_id', 'mac_address', 'device_id']
- position_values = [netid, 'mac', 'deviceid']
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values)
-
- # Test dashed options
- args = ['--mac-address', 'mac', '--device-id', 'deviceid', netid]
- position_names = ['network_id', 'mac_address', 'device_id']
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values)
-
- def test_create_port_tenant(self):
- """Create port: --tenant_id tenantid netid."""
- resource = 'port'
- cmd = port.CreatePort(test_cli20.MyApp(sys.stdout), None)
- name = 'myname'
- myid = 'myid'
- netid = 'netid'
- args = ['--tenant_id', 'tenantid', netid, ]
- position_names = ['network_id']
- position_values = []
- position_values.extend([netid])
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values,
- tenant_id='tenantid')
-
- # Test dashed options
- args = ['--tenant-id', 'tenantid', netid, ]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values,
- tenant_id='tenantid')
-
- def test_create_port_tags(self):
- """Create port: netid mac_address device_id --tags a b."""
- resource = 'port'
- cmd = port.CreatePort(test_cli20.MyApp(sys.stdout), None)
- name = 'myname'
- myid = 'myid'
- netid = 'netid'
- args = [netid, '--tags', 'a', 'b']
- position_names = ['network_id']
- position_values = []
- position_values.extend([netid])
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values,
- tags=['a', 'b'])
-
- def test_create_port_secgroup(self):
- """Create port: --security-group sg1_id netid."""
- resource = 'port'
- cmd = port.CreatePort(test_cli20.MyApp(sys.stdout), None)
- name = 'myname'
- myid = 'myid'
- netid = 'netid'
- args = ['--security-group', 'sg1_id', netid]
- position_names = ['network_id', 'security_groups']
- position_values = [netid, ['sg1_id']]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values)
-
- def test_create_port_secgroups(self):
- """Create port: <security_groups> netid
-
- The <security_groups> are
- --security-group sg1_id --security-group sg2_id
- """
- resource = 'port'
- cmd = port.CreatePort(test_cli20.MyApp(sys.stdout), None)
- name = 'myname'
- myid = 'myid'
- netid = 'netid'
- args = ['--security-group', 'sg1_id',
- '--security-group', 'sg2_id',
- netid]
- position_names = ['network_id', 'security_groups']
- position_values = [netid, ['sg1_id', 'sg2_id']]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values)
-
- def test_list_ports(self):
- """List ports: -D."""
- resources = "ports"
- cmd = port.ListPort(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd, True)
-
- def test_list_ports_pagination(self):
- resources = "ports"
- cmd = port.ListPort(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources_with_pagination(resources, cmd)
-
- def test_list_ports_sort(self):
- """list ports: --sort-key name --sort-key id --sort-key asc
- --sort-key desc
- """
- resources = "ports"
- cmd = port.ListPort(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd,
- sort_key=["name", "id"],
- sort_dir=["asc", "desc"])
-
- def test_list_ports_limit(self):
- """list ports: -P."""
- resources = "ports"
- cmd = port.ListPort(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd, page_size=1000)
-
- def test_list_ports_tags(self):
- """List ports: -- --tags a b."""
- resources = "ports"
- cmd = port.ListPort(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd, tags=['a', 'b'])
-
- def test_list_ports_detail_tags(self):
- """List ports: -D -- --tags a b."""
- resources = "ports"
- cmd = port.ListPort(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd, detail=True, tags=['a', 'b'])
-
- def test_list_ports_fields(self):
- """List ports: --fields a --fields b -- --fields c d."""
- resources = "ports"
- cmd = port.ListPort(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd,
- fields_1=['a', 'b'], fields_2=['c', 'd'])
-
- def _test_list_router_port(self, resources, cmd,
- myid, detail=False, tags=[],
- fields_1=[], fields_2=[]):
- self.mox.StubOutWithMock(cmd, "get_client")
- self.mox.StubOutWithMock(self.client.httpclient, "request")
- cmd.get_client().MultipleTimes().AndReturn(self.client)
- reses = {resources: [{'id': 'myid1', },
- {'id': 'myid2', }, ], }
-
- resstr = self.client.serialize(reses)
-
- # url method body
- query = ""
- args = detail and ['-D', ] or []
-
- if fields_1:
- for field in fields_1:
- args.append('--fields')
- args.append(field)
- args.append(myid)
- if tags:
- args.append('--')
- args.append("--tag")
- for tag in tags:
- args.append(tag)
- if (not tags) and fields_2:
- args.append('--')
- if fields_2:
- args.append("--fields")
- for field in fields_2:
- args.append(field)
- fields_1.extend(fields_2)
- for field in fields_1:
- if query:
- query += "&fields=" + field
- else:
- query = "fields=" + field
-
- for tag in tags:
- if query:
- query += "&tag=" + tag
- else:
- query = "tag=" + tag
- if detail:
- query = query and query + '&verbose=True' or 'verbose=True'
- query = query and query + '&device_id=%s' or 'device_id=%s'
- path = getattr(self.client, resources + "_path")
- self.client.httpclient.request(
- test_cli20.end_url(path, query % myid), 'GET',
- body=None,
- headers=mox.ContainsKeyValue('X-Auth-Token', test_cli20.TOKEN)
- ).AndReturn((test_cli20.MyResp(200), resstr))
- self.mox.ReplayAll()
- cmd_parser = cmd.get_parser("list_" + resources)
- shell.run_command(cmd, cmd_parser, args)
- self.mox.VerifyAll()
- self.mox.UnsetStubs()
- _str = self.fake_stdout.make_string()
-
- self.assertTrue('myid1' in _str)
-
- def test_list_router_ports(self):
- """List router ports: -D."""
- resources = "ports"
- cmd = port.ListRouterPort(test_cli20.MyApp(sys.stdout), None)
- self._test_list_router_port(resources, cmd,
- self.test_id, True)
-
- def test_list_router_ports_tags(self):
- """List router ports: -- --tags a b."""
- resources = "ports"
- cmd = port.ListRouterPort(test_cli20.MyApp(sys.stdout), None)
- self._test_list_router_port(resources, cmd,
- self.test_id, tags=['a', 'b'])
-
- def test_list_router_ports_detail_tags(self):
- """List router ports: -D -- --tags a b."""
- resources = "ports"
- cmd = port.ListRouterPort(test_cli20.MyApp(sys.stdout), None)
- self._test_list_router_port(resources, cmd, self.test_id,
- detail=True, tags=['a', 'b'])
-
- def test_list_router_ports_fields(self):
- """List ports: --fields a --fields b -- --fields c d."""
- resources = "ports"
- cmd = port.ListRouterPort(test_cli20.MyApp(sys.stdout), None)
- self._test_list_router_port(resources, cmd, self.test_id,
- fields_1=['a', 'b'],
- fields_2=['c', 'd'])
-
- def test_update_port(self):
- """Update port: myid --name myname --tags a b."""
- resource = 'port'
- cmd = port.UpdatePort(test_cli20.MyApp(sys.stdout), None)
- self._test_update_resource(resource, cmd, 'myid',
- ['myid', '--name', 'myname',
- '--tags', 'a', 'b'],
- {'name': 'myname', 'tags': ['a', 'b'], }
- )
-
- def test_update_port_security_group_off(self):
- """Update port: --no-security-groups myid."""
- resource = 'port'
- cmd = port.UpdatePort(test_cli20.MyApp(sys.stdout), None)
- self._test_update_resource(resource, cmd, 'myid',
- ['--no-security-groups', 'myid'],
- {'security_groups': None})
-
- def test_show_port(self):
- """Show port: --fields id --fields name myid."""
- resource = 'port'
- cmd = port.ShowPort(test_cli20.MyApp(sys.stdout), None)
- args = ['--fields', 'id', '--fields', 'name', self.test_id]
- self._test_show_resource(resource, cmd, self.test_id,
- args, ['id', 'name'])
-
- def test_delete_port(self):
- """Delete port: myid."""
- resource = 'port'
- cmd = port.DeletePort(test_cli20.MyApp(sys.stdout), None)
- myid = 'myid'
- args = [myid]
- self._test_delete_resource(resource, cmd, myid, args)
-
-
-class CLITestV20PortXML(CLITestV20PortJSON):
- format = 'xml'
diff --git a/tests/unit/test_cli20_router.py b/tests/unit/test_cli20_router.py
deleted file mode 100644
index 35ac577..0000000
--- a/tests/unit/test_cli20_router.py
+++ /dev/null
@@ -1,197 +0,0 @@
-# Copyright 2012 Nicira, Inc
-# All Rights Reserved
-#
-# 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.
-#
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-import sys
-
-from quantumclient.common import exceptions
-from quantumclient.quantum.v2_0 import router
-from tests.unit import test_cli20
-
-
-class CLITestV20RouterJSON(test_cli20.CLITestV20Base):
- def test_create_router(self):
- """Create router: router1."""
- resource = 'router'
- cmd = router.CreateRouter(test_cli20.MyApp(sys.stdout), None)
- name = 'router1'
- myid = 'myid'
- args = [name, ]
- position_names = ['name', ]
- position_values = [name, ]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values)
-
- def test_create_router_tenant(self):
- """Create router: --tenant_id tenantid myname."""
- resource = 'router'
- cmd = router.CreateRouter(test_cli20.MyApp(sys.stdout), None)
- name = 'myname'
- myid = 'myid'
- args = ['--tenant_id', 'tenantid', name]
- position_names = ['name', ]
- position_values = [name, ]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values,
- tenant_id='tenantid')
-
- def test_create_router_admin_state(self):
- """Create router: --admin_state_down myname."""
- resource = 'router'
- cmd = router.CreateRouter(test_cli20.MyApp(sys.stdout), None)
- name = 'myname'
- myid = 'myid'
- args = ['--admin_state_down', name, ]
- position_names = ['name', ]
- position_values = [name, ]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values,
- admin_state_up=False)
-
- def test_list_routers_detail(self):
- """list routers: -D."""
- resources = "routers"
- cmd = router.ListRouter(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd, True)
-
- def test_list_routers_pagination(self):
- resources = "routers"
- cmd = router.ListRouter(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources_with_pagination(resources, cmd)
-
- def test_list_routers_sort(self):
- """list routers: --sort-key name --sort-key id --sort-key asc
- --sort-key desc
- """
- resources = "routers"
- cmd = router.ListRouter(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd,
- sort_key=["name", "id"],
- sort_dir=["asc", "desc"])
-
- def test_list_routers_limit(self):
- """list routers: -P."""
- resources = "routers"
- cmd = router.ListRouter(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd, page_size=1000)
-
- def test_update_router_exception(self):
- """Update router: myid."""
- resource = 'router'
- cmd = router.UpdateRouter(test_cli20.MyApp(sys.stdout), None)
- self.assertRaises(exceptions.CommandError, self._test_update_resource,
- resource, cmd, 'myid', ['myid'], {})
-
- def test_update_router(self):
- """Update router: myid --name myname --tags a b."""
- resource = 'router'
- cmd = router.UpdateRouter(test_cli20.MyApp(sys.stdout), None)
- self._test_update_resource(resource, cmd, 'myid',
- ['myid', '--name', 'myname'],
- {'name': 'myname'}
- )
-
- def test_delete_router(self):
- """Delete router: myid."""
- resource = 'router'
- cmd = router.DeleteRouter(test_cli20.MyApp(sys.stdout), None)
- myid = 'myid'
- args = [myid]
- self._test_delete_resource(resource, cmd, myid, args)
-
- def test_show_router(self):
- """Show router: myid."""
- resource = 'router'
- cmd = router.ShowRouter(test_cli20.MyApp(sys.stdout), None)
- args = ['--fields', 'id', '--fields', 'name', self.test_id]
- self._test_show_resource(resource, cmd, self.test_id, args,
- ['id', 'name'])
-
- def _test_add_remove_interface(self, action, mode, cmd, args):
- resource = 'router'
- subcmd = '%s_router_interface' % action
- if mode == 'port':
- body = {'port_id': 'portid'}
- else:
- body = {'subnet_id': 'subnetid'}
- if action == 'add':
- retval = {'subnet_id': 'subnetid', 'port_id': 'portid'}
- else:
- retval = None
- self._test_update_resource_action(resource, cmd, 'myid',
- subcmd, args,
- body, retval)
-
- def test_add_interface_compat(self):
- """Add interface to router: myid subnetid."""
- cmd = router.AddInterfaceRouter(test_cli20.MyApp(sys.stdout), None)
- args = ['myid', 'subnetid']
- self._test_add_remove_interface('add', 'subnet', cmd, args)
-
- def test_add_interface_by_subnet(self):
- """Add interface to router: myid subnet=subnetid."""
- cmd = router.AddInterfaceRouter(test_cli20.MyApp(sys.stdout), None)
- args = ['myid', 'subnet=subnetid']
- self._test_add_remove_interface('add', 'subnet', cmd, args)
-
- def test_add_interface_by_port(self):
- """Add interface to router: myid port=portid."""
- cmd = router.AddInterfaceRouter(test_cli20.MyApp(sys.stdout), None)
- args = ['myid', 'port=portid']
- self._test_add_remove_interface('add', 'port', cmd, args)
-
- def test_del_interface_compat(self):
- """Delete interface from router: myid subnetid."""
- cmd = router.RemoveInterfaceRouter(test_cli20.MyApp(sys.stdout), None)
- args = ['myid', 'subnetid']
- self._test_add_remove_interface('remove', 'subnet', cmd, args)
-
- def test_del_interface_by_subnet(self):
- """Delete interface from router: myid subnet=subnetid."""
- cmd = router.RemoveInterfaceRouter(test_cli20.MyApp(sys.stdout), None)
- args = ['myid', 'subnet=subnetid']
- self._test_add_remove_interface('remove', 'subnet', cmd, args)
-
- def test_del_interface_by_port(self):
- """Delete interface from router: myid port=portid."""
- cmd = router.RemoveInterfaceRouter(test_cli20.MyApp(sys.stdout), None)
- args = ['myid', 'port=portid']
- self._test_add_remove_interface('remove', 'port', cmd, args)
-
- def test_set_gateway(self):
- """Set external gateway for router: myid externalid."""
- resource = 'router'
- cmd = router.SetGatewayRouter(test_cli20.MyApp(sys.stdout), None)
- args = ['myid', 'externalid']
- self._test_update_resource(resource, cmd, 'myid',
- args,
- {"external_gateway_info":
- {"network_id": "externalid",
- "enable_snat": True}}
- )
-
- def test_remove_gateway(self):
- """Remove external gateway from router: externalid."""
- resource = 'router'
- cmd = router.RemoveGatewayRouter(test_cli20.MyApp(sys.stdout), None)
- args = ['externalid']
- self._test_update_resource(resource, cmd, 'externalid',
- args, {"external_gateway_info": {}}
- )
-
-
-class CLITestV20RouterXML(CLITestV20RouterJSON):
- format = 'xml'
diff --git a/tests/unit/test_cli20_securitygroup.py b/tests/unit/test_cli20_securitygroup.py
deleted file mode 100644
index 4302093..0000000
--- a/tests/unit/test_cli20_securitygroup.py
+++ /dev/null
@@ -1,333 +0,0 @@
-#!/usr/bin/env python
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-# Copyright 2012 Red Hat
-# All Rights Reserved.
-#
-# 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 sys
-
-import mox
-
-from quantumclient.quantum.v2_0 import securitygroup
-from tests.unit import test_cli20
-
-
-class CLITestV20SecurityGroupsJSON(test_cli20.CLITestV20Base):
- def test_create_security_group(self):
- """Create security group: webservers."""
- resource = 'security_group'
- cmd = securitygroup.CreateSecurityGroup(
- test_cli20.MyApp(sys.stdout), None)
- name = 'webservers'
- myid = 'myid'
- args = [name, ]
- position_names = ['name']
- position_values = [name]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values)
-
- def test_create_security_group_tenant(self):
- """Create security group: webservers."""
- resource = 'security_group'
- cmd = securitygroup.CreateSecurityGroup(
- test_cli20.MyApp(sys.stdout), None)
- name = 'webservers'
- description = 'my webservers'
- myid = 'myid'
- args = ['--tenant_id', 'tenant_id', '--description', description, name]
- position_names = ['name', 'description']
- position_values = [name, description]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values,
- tenant_id='tenant_id')
-
- def test_create_security_group_with_description(self):
- """Create security group: webservers."""
- resource = 'security_group'
- cmd = securitygroup.CreateSecurityGroup(
- test_cli20.MyApp(sys.stdout), None)
- name = 'webservers'
- description = 'my webservers'
- myid = 'myid'
- args = [name, '--description', description]
- position_names = ['name', 'description']
- position_values = [name, description]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values)
-
- def test_list_security_groups(self):
- resources = "security_groups"
- cmd = securitygroup.ListSecurityGroup(
- test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd, True)
-
- def test_list_security_groups_pagination(self):
- resources = "security_groups"
- cmd = securitygroup.ListSecurityGroup(
- test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources_with_pagination(resources, cmd)
-
- def test_list_security_groups_sort(self):
- resources = "security_groups"
- cmd = securitygroup.ListSecurityGroup(
- test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd,
- sort_key=["name", "id"],
- sort_dir=["asc", "desc"])
-
- def test_list_security_groups_limit(self):
- resources = "security_groups"
- cmd = securitygroup.ListSecurityGroup(
- test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd, page_size=1000)
-
- def test_show_security_group_id(self):
- resource = 'security_group'
- cmd = securitygroup.ShowSecurityGroup(
- test_cli20.MyApp(sys.stdout), None)
- args = ['--fields', 'id', self.test_id]
- self._test_show_resource(resource, cmd, self.test_id,
- args, ['id'])
-
- def test_show_security_group_id_name(self):
- resource = 'security_group'
- cmd = securitygroup.ShowSecurityGroup(
- test_cli20.MyApp(sys.stdout), None)
- args = ['--fields', 'id', '--fields', 'name', self.test_id]
- self._test_show_resource(resource, cmd, self.test_id,
- args, ['id', 'name'])
-
- def test_delete_security_group(self):
- """Delete security group: myid."""
- resource = 'security_group'
- cmd = securitygroup.DeleteSecurityGroup(
- test_cli20.MyApp(sys.stdout), None)
- myid = 'myid'
- args = [myid]
- self._test_delete_resource(resource, cmd, myid, args)
-
- def test_update_security_group(self):
- """Update security group: myid --name myname --description desc."""
- resource = 'security_group'
- cmd = securitygroup.UpdateSecurityGroup(
- test_cli20.MyApp(sys.stdout), None)
- self._test_update_resource(resource, cmd, 'myid',
- ['myid', '--name', 'myname',
- '--description', 'mydescription'],
- {'name': 'myname',
- 'description': 'mydescription'}
- )
-
- def test_update_security_group_with_unicode(self):
- resource = 'security_group'
- cmd = securitygroup.UpdateSecurityGroup(
- test_cli20.MyApp(sys.stdout), None)
- self._test_update_resource(resource, cmd, 'myid',
- ['myid', '--name', u'\u7f51\u7edc',
- '--description', u'\u7f51\u7edc'],
- {'name': u'\u7f51\u7edc',
- 'description': u'\u7f51\u7edc'}
- )
-
- def test_create_security_group_rule_full(self):
- """Create security group rule."""
- resource = 'security_group_rule'
- cmd = securitygroup.CreateSecurityGroupRule(
- test_cli20.MyApp(sys.stdout), None)
- myid = 'myid'
- direction = 'ingress'
- ethertype = 'IPv4'
- protocol = 'tcp'
- port_range_min = '22'
- port_range_max = '22'
- remote_ip_prefix = '10.0.0.0/24'
- security_group_id = '1'
- remote_group_id = '1'
- args = ['--remote_ip_prefix', remote_ip_prefix, '--direction',
- direction, '--ethertype', ethertype, '--protocol', protocol,
- '--port_range_min', port_range_min, '--port_range_max',
- port_range_max, '--remote_group_id', remote_group_id,
- security_group_id]
- position_names = ['remote_ip_prefix', 'direction', 'ethertype',
- 'protocol', 'port_range_min', 'port_range_max',
- 'remote_group_id', 'security_group_id']
- position_values = [remote_ip_prefix, direction, ethertype, protocol,
- port_range_min, port_range_max, remote_group_id,
- security_group_id]
- self._test_create_resource(resource, cmd, None, myid, args,
- position_names, position_values)
-
- def test_delete_security_group_rule(self):
- """Delete security group rule: myid."""
- resource = 'security_group_rule'
- cmd = securitygroup.DeleteSecurityGroupRule(
- test_cli20.MyApp(sys.stdout), None)
- myid = 'myid'
- args = [myid]
- self._test_delete_resource(resource, cmd, myid, args)
-
- def test_list_security_group_rules(self):
- resources = "security_group_rules"
- cmd = securitygroup.ListSecurityGroupRule(
- test_cli20.MyApp(sys.stdout), None)
- self.mox.StubOutWithMock(securitygroup.ListSecurityGroupRule,
- "extend_list")
- securitygroup.ListSecurityGroupRule.extend_list(mox.IsA(list),
- mox.IgnoreArg())
- self._test_list_resources(resources, cmd, True)
-
- def test_list_security_group_rules_pagination(self):
- resources = "security_group_rules"
- cmd = securitygroup.ListSecurityGroupRule(
- test_cli20.MyApp(sys.stdout), None)
- self.mox.StubOutWithMock(securitygroup.ListSecurityGroupRule,
- "extend_list")
- securitygroup.ListSecurityGroupRule.extend_list(mox.IsA(list),
- mox.IgnoreArg())
- self._test_list_resources_with_pagination(resources, cmd)
-
- def test_list_security_group_rules_sort(self):
- resources = "security_group_rules"
- cmd = securitygroup.ListSecurityGroupRule(
- test_cli20.MyApp(sys.stdout), None)
- self.mox.StubOutWithMock(securitygroup.ListSecurityGroupRule,
- "extend_list")
- securitygroup.ListSecurityGroupRule.extend_list(mox.IsA(list),
- mox.IgnoreArg())
- self._test_list_resources(resources, cmd,
- sort_key=["name", "id"],
- sort_dir=["asc", "desc"])
-
- def test_list_security_group_rules_limit(self):
- resources = "security_group_rules"
- cmd = securitygroup.ListSecurityGroupRule(
- test_cli20.MyApp(sys.stdout), None)
- self.mox.StubOutWithMock(securitygroup.ListSecurityGroupRule,
- "extend_list")
- securitygroup.ListSecurityGroupRule.extend_list(mox.IsA(list),
- mox.IgnoreArg())
- self._test_list_resources(resources, cmd, page_size=1000)
-
- def test_show_security_group_rule(self):
- resource = 'security_group_rule'
- cmd = securitygroup.ShowSecurityGroupRule(
- test_cli20.MyApp(sys.stdout), None)
- args = ['--fields', 'id', self.test_id]
- self._test_show_resource(resource, cmd, self.test_id,
- args, ['id'])
-
- def _test_list_security_group_rules_extend(self, data=None, expected=None,
- args=[], conv=True,
- query_field=False):
- def setup_list_stub(resources, data, query):
- reses = {resources: data}
- resstr = self.client.serialize(reses)
- resp = (test_cli20.MyResp(200), resstr)
- path = getattr(self.client, resources + '_path')
- self.client.httpclient.request(
- test_cli20.end_url(path, query), 'GET',
- body=None,
- headers=mox.ContainsKeyValue(
- 'X-Auth-Token', test_cli20.TOKEN)).AndReturn(resp)
-
- # Setup the default data
- _data = {'cols': ['id', 'security_group_id', 'remote_group_id'],
- 'data': [('ruleid1', 'myid1', 'myid1'),
- ('ruleid2', 'myid2', 'myid3'),
- ('ruleid3', 'myid2', 'myid2')]}
- _expected = {'cols': ['id', 'security_group', 'remote_group'],
- 'data': [('ruleid1', 'group1', 'group1'),
- ('ruleid2', 'group2', 'group3'),
- ('ruleid3', 'group2', 'group2')]}
- if data is None:
- data = _data
- list_data = [dict(zip(data['cols'], d)) for d in data['data']]
- if expected is None:
- expected = {}
- expected['cols'] = expected.get('cols', _expected['cols'])
- expected['data'] = expected.get('data', _expected['data'])
-
- cmd = securitygroup.ListSecurityGroupRule(
- test_cli20.MyApp(sys.stdout), None)
- self.mox.StubOutWithMock(cmd, 'get_client')
- self.mox.StubOutWithMock(self.client.httpclient, 'request')
- cmd.get_client().AndReturn(self.client)
- query = ''
- if query_field:
- query = '&'.join(['fields=' + f for f in data['cols']])
- setup_list_stub('security_group_rules', list_data, query)
- if conv:
- cmd.get_client().AndReturn(self.client)
- sec_ids = set()
- for n in data['data']:
- sec_ids.add(n[1])
- sec_ids.add(n[2])
- filters = ''
- for id in sec_ids:
- filters = filters + "&id=%s" % id
- setup_list_stub('security_groups',
- [{'id': 'myid1', 'name': 'group1'},
- {'id': 'myid2', 'name': 'group2'},
- {'id': 'myid3', 'name': 'group3'}],
- query='fields=id&fields=name' + filters)
- self.mox.ReplayAll()
-
- cmd_parser = cmd.get_parser('list_security_group_rules')
- parsed_args = cmd_parser.parse_args(args)
- result = cmd.get_data(parsed_args)
- self.mox.VerifyAll()
- self.mox.UnsetStubs()
- # Check columns
- self.assertEqual(result[0], expected['cols'])
- # Check data
- _result = [x for x in result[1]]
- self.assertEqual(len(_result), len(expected['data']))
- for res, exp in zip(_result, expected['data']):
- self.assertEqual(len(res), len(exp))
- self.assertEqual(res, exp)
-
- def test_list_security_group_rules_extend_source_id(self):
- self._test_list_security_group_rules_extend()
-
- def test_list_security_group_rules_extend_no_nameconv(self):
- expected = {'cols': ['id', 'security_group_id', 'remote_group_id'],
- 'data': [('ruleid1', 'myid1', 'myid1'),
- ('ruleid2', 'myid2', 'myid3'),
- ('ruleid3', 'myid2', 'myid2')]}
- args = ['--no-nameconv']
- self._test_list_security_group_rules_extend(expected=expected,
- args=args, conv=False)
-
- def test_list_security_group_rules_extend_with_columns(self):
- args = '-c id -c security_group_id -c remote_group_id'.split()
- self._test_list_security_group_rules_extend(args=args)
-
- def test_list_security_group_rules_extend_with_columns_no_id(self):
- args = '-c id -c security_group -c remote_group'.split()
- self._test_list_security_group_rules_extend(args=args)
-
- def test_list_security_group_rules_extend_with_fields(self):
- args = '-F id -F security_group_id -F remote_group_id'.split()
- self._test_list_security_group_rules_extend(args=args,
- query_field=True)
-
- def test_list_security_group_rules_extend_with_fields_no_id(self):
- args = '-F id -F security_group -F remote_group'.split()
- self._test_list_security_group_rules_extend(args=args,
- query_field=True)
-
-
-class CLITestV20SecurityGroupsXML(CLITestV20SecurityGroupsJSON):
- format = 'xml'
diff --git a/tests/unit/test_cli20_subnet.py b/tests/unit/test_cli20_subnet.py
deleted file mode 100644
index 5c048a6..0000000
--- a/tests/unit/test_cli20_subnet.py
+++ /dev/null
@@ -1,402 +0,0 @@
-# Copyright 2012 OpenStack LLC.
-# All Rights Reserved
-#
-# 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.
-#
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-import sys
-
-from quantumclient.quantum.v2_0 import subnet
-from tests.unit import test_cli20
-
-
-class CLITestV20SubnetJSON(test_cli20.CLITestV20Base):
- def setUp(self):
- super(CLITestV20SubnetJSON, self).setUp(plurals={'tags': 'tag'})
-
- def test_create_subnet(self):
- """Create subnet: --gateway gateway netid cidr."""
- resource = 'subnet'
- cmd = subnet.CreateSubnet(test_cli20.MyApp(sys.stdout), None)
- name = 'myname'
- myid = 'myid'
- netid = 'netid'
- cidr = 'cidrvalue'
- gateway = 'gatewayvalue'
- args = ['--gateway', gateway, netid, cidr]
- position_names = ['ip_version', 'network_id', 'cidr', 'gateway_ip']
- position_values = [4, netid, cidr, gateway]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values)
-
- def test_create_subnet_with_no_gateway(self):
- """Create subnet: --no-gateway netid cidr."""
- resource = 'subnet'
- cmd = subnet.CreateSubnet(test_cli20.MyApp(sys.stdout), None)
- name = 'myname'
- myid = 'myid'
- netid = 'netid'
- cidr = 'cidrvalue'
- args = ['--no-gateway', netid, cidr]
- position_names = ['ip_version', 'network_id', 'cidr', 'gateway_ip']
- position_values = [4, netid, cidr, None]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values)
-
- def test_create_subnet_with_bad_gateway_option(self):
- """Create sbunet: --no-gateway netid cidr."""
- resource = 'subnet'
- cmd = subnet.CreateSubnet(test_cli20.MyApp(sys.stdout), None)
- name = 'myname'
- myid = 'myid'
- netid = 'netid'
- cidr = 'cidrvalue'
- gateway = 'gatewayvalue'
- args = ['--gateway', gateway, '--no-gateway', netid, cidr]
- position_names = ['ip_version', 'network_id', 'cidr', 'gateway_ip']
- position_values = [4, netid, cidr, None]
- try:
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values)
- except Exception:
- return
- self.fail('No exception for bad gateway option')
-
- def test_create_subnet_tenant(self):
- """Create subnet: --tenant_id tenantid netid cidr."""
- resource = 'subnet'
- cmd = subnet.CreateSubnet(test_cli20.MyApp(sys.stdout), None)
- name = 'myname'
- myid = 'myid'
- netid = 'netid'
- cidr = 'prefixvalue'
- args = ['--tenant_id', 'tenantid', netid, cidr]
- position_names = ['ip_version', 'network_id', 'cidr']
- position_values = [4, netid, cidr]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values,
- tenant_id='tenantid')
-
- def test_create_subnet_tags(self):
- """Create subnet: netid cidr --tags a b."""
- resource = 'subnet'
- cmd = subnet.CreateSubnet(test_cli20.MyApp(sys.stdout), None)
- name = 'myname'
- myid = 'myid'
- netid = 'netid'
- cidr = 'prefixvalue'
- args = [netid, cidr, '--tags', 'a', 'b']
- position_names = ['ip_version', 'network_id', 'cidr']
- position_values = [4, netid, cidr]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values,
- tags=['a', 'b'])
-
- def test_create_subnet_allocation_pool(self):
- """Create subnet: --tenant_id tenantid <allocation_pool> netid cidr.
- The <allocation_pool> is --allocation_pool start=1.1.1.10,end=1.1.1.20
- """
- resource = 'subnet'
- cmd = subnet.CreateSubnet(test_cli20.MyApp(sys.stdout), None)
- name = 'myname'
- myid = 'myid'
- netid = 'netid'
- cidr = 'prefixvalue'
- args = ['--tenant_id', 'tenantid',
- '--allocation_pool', 'start=1.1.1.10,end=1.1.1.20',
- netid, cidr]
- position_names = ['ip_version', 'allocation_pools', 'network_id',
- 'cidr']
- pool = [{'start': '1.1.1.10', 'end': '1.1.1.20'}]
- position_values = [4, pool, netid, cidr]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values,
- tenant_id='tenantid')
-
- def test_create_subnet_allocation_pools(self):
- """Create subnet: --tenant-id tenantid <pools> netid cidr.
- The <pools> are --allocation_pool start=1.1.1.10,end=1.1.1.20 and
- --allocation_pool start=1.1.1.30,end=1.1.1.40
- """
- resource = 'subnet'
- cmd = subnet.CreateSubnet(test_cli20.MyApp(sys.stdout), None)
- name = 'myname'
- myid = 'myid'
- netid = 'netid'
- cidr = 'prefixvalue'
- args = ['--tenant_id', 'tenantid',
- '--allocation_pool', 'start=1.1.1.10,end=1.1.1.20',
- '--allocation_pool', 'start=1.1.1.30,end=1.1.1.40',
- netid, cidr]
- position_names = ['ip_version', 'allocation_pools', 'network_id',
- 'cidr']
- pools = [{'start': '1.1.1.10', 'end': '1.1.1.20'},
- {'start': '1.1.1.30', 'end': '1.1.1.40'}]
- position_values = [4, pools, netid, cidr]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values,
- tenant_id='tenantid')
-
- def test_create_subnet_host_route(self):
- """Create subnet: --tenant_id tenantid <host_route> netid cidr.
- The <host_route> is
- --host-route destination=172.16.1.0/24,nexthop=1.1.1.20
- """
- resource = 'subnet'
- cmd = subnet.CreateSubnet(test_cli20.MyApp(sys.stdout), None)
- name = 'myname'
- myid = 'myid'
- netid = 'netid'
- cidr = 'prefixvalue'
- args = ['--tenant_id', 'tenantid',
- '--host-route', 'destination=172.16.1.0/24,nexthop=1.1.1.20',
- netid, cidr]
- position_names = ['ip_version', 'host_routes', 'network_id',
- 'cidr']
- route = [{'destination': '172.16.1.0/24', 'nexthop': '1.1.1.20'}]
- position_values = [4, route, netid, cidr]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values,
- tenant_id='tenantid')
-
- def test_create_subnet_host_routes(self):
- """Create subnet: --tenant-id tenantid <host_routes> netid cidr.
- The <host_routes> are
- --host-route destination=172.16.1.0/24,nexthop=1.1.1.20 and
- --host-route destination=172.17.7.0/24,nexthop=1.1.1.40
- """
- resource = 'subnet'
- cmd = subnet.CreateSubnet(test_cli20.MyApp(sys.stdout), None)
- name = 'myname'
- myid = 'myid'
- netid = 'netid'
- cidr = 'prefixvalue'
- args = ['--tenant_id', 'tenantid',
- '--host-route', 'destination=172.16.1.0/24,nexthop=1.1.1.20',
- '--host-route', 'destination=172.17.7.0/24,nexthop=1.1.1.40',
- netid, cidr]
- position_names = ['ip_version', 'host_routes', 'network_id',
- 'cidr']
- routes = [{'destination': '172.16.1.0/24', 'nexthop': '1.1.1.20'},
- {'destination': '172.17.7.0/24', 'nexthop': '1.1.1.40'}]
- position_values = [4, routes, netid, cidr]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values,
- tenant_id='tenantid')
-
- def test_create_subnet_dns_nameservers(self):
- """Create subnet: --tenant-id tenantid <dns-nameservers> netid cidr.
- The <dns-nameservers> are
- --dns-nameserver 1.1.1.20 and --dns-nameserver 1.1.1.40
- """
- resource = 'subnet'
- cmd = subnet.CreateSubnet(test_cli20.MyApp(sys.stdout), None)
- name = 'myname'
- myid = 'myid'
- netid = 'netid'
- cidr = 'prefixvalue'
- args = ['--tenant_id', 'tenantid',
- '--dns-nameserver', '1.1.1.20',
- '--dns-nameserver', '1.1.1.40',
- netid, cidr]
- position_names = ['ip_version', 'dns_nameservers', 'network_id',
- 'cidr']
- nameservers = ['1.1.1.20', '1.1.1.40']
- position_values = [4, nameservers, netid, cidr]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values,
- tenant_id='tenantid')
-
- def test_create_subnet_with_disable_dhcp(self):
- """Create subnet: --tenant-id tenantid --disable-dhcp netid cidr."""
- resource = 'subnet'
- cmd = subnet.CreateSubnet(test_cli20.MyApp(sys.stdout), None)
- name = 'myname'
- myid = 'myid'
- netid = 'netid'
- cidr = 'prefixvalue'
- args = ['--tenant_id', 'tenantid',
- '--disable-dhcp',
- netid, cidr]
- position_names = ['ip_version', 'enable_dhcp', 'network_id',
- 'cidr']
- position_values = [4, False, netid, cidr]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values,
- tenant_id='tenantid')
-
- def test_create_subnet_merge_single_plurar(self):
- resource = 'subnet'
- cmd = subnet.CreateSubnet(test_cli20.MyApp(sys.stdout), None)
- name = 'myname'
- myid = 'myid'
- netid = 'netid'
- cidr = 'prefixvalue'
- args = ['--tenant_id', 'tenantid',
- '--allocation-pool', 'start=1.1.1.10,end=1.1.1.20',
- netid, cidr,
- '--allocation-pools', 'list=true', 'type=dict',
- 'start=1.1.1.30,end=1.1.1.40']
- position_names = ['ip_version', 'allocation_pools', 'network_id',
- 'cidr']
- pools = [{'start': '1.1.1.10', 'end': '1.1.1.20'},
- {'start': '1.1.1.30', 'end': '1.1.1.40'}]
- position_values = [4, pools, netid, cidr]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values,
- tenant_id='tenantid')
-
- def test_create_subnet_merge_plurar(self):
- resource = 'subnet'
- cmd = subnet.CreateSubnet(test_cli20.MyApp(sys.stdout), None)
- name = 'myname'
- myid = 'myid'
- netid = 'netid'
- cidr = 'prefixvalue'
- args = ['--tenant_id', 'tenantid',
- netid, cidr,
- '--allocation-pools', 'list=true', 'type=dict',
- 'start=1.1.1.30,end=1.1.1.40']
- position_names = ['ip_version', 'allocation_pools', 'network_id',
- 'cidr']
- pools = [{'start': '1.1.1.30', 'end': '1.1.1.40'}]
- position_values = [4, pools, netid, cidr]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values,
- tenant_id='tenantid')
-
- def test_create_subnet_merge_single_single(self):
- resource = 'subnet'
- cmd = subnet.CreateSubnet(test_cli20.MyApp(sys.stdout), None)
- name = 'myname'
- myid = 'myid'
- netid = 'netid'
- cidr = 'prefixvalue'
- args = ['--tenant_id', 'tenantid',
- '--allocation-pool', 'start=1.1.1.10,end=1.1.1.20',
- netid, cidr,
- '--allocation-pool',
- 'start=1.1.1.30,end=1.1.1.40']
- position_names = ['ip_version', 'allocation_pools', 'network_id',
- 'cidr']
- pools = [{'start': '1.1.1.10', 'end': '1.1.1.20'},
- {'start': '1.1.1.30', 'end': '1.1.1.40'}]
- position_values = [4, pools, netid, cidr]
- self._test_create_resource(resource, cmd, name, myid, args,
- position_names, position_values,
- tenant_id='tenantid')
-
- def test_list_subnets_detail(self):
- """List subnets: -D."""
- resources = "subnets"
- cmd = subnet.ListSubnet(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd, True)
-
- def test_list_subnets_tags(self):
- """List subnets: -- --tags a b."""
- resources = "subnets"
- cmd = subnet.ListSubnet(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd, tags=['a', 'b'])
-
- def test_list_subnets_known_option_after_unknown(self):
- """List subnets: -- --tags a b --request-format xml."""
- resources = "subnets"
- cmd = subnet.ListSubnet(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd, tags=['a', 'b'])
-
- def test_list_subnets_detail_tags(self):
- """List subnets: -D -- --tags a b."""
- resources = "subnets"
- cmd = subnet.ListSubnet(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd, detail=True, tags=['a', 'b'])
-
- def test_list_subnets_fields(self):
- """List subnets: --fields a --fields b -- --fields c d."""
- resources = "subnets"
- cmd = subnet.ListSubnet(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd,
- fields_1=['a', 'b'], fields_2=['c', 'd'])
-
- def test_list_subnets_pagination(self):
- resources = "subnets"
- cmd = subnet.ListSubnet(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources_with_pagination(resources, cmd)
-
- def test_list_subnets_sort(self):
- """List subnets: --sort-key name --sort-key id --sort-key asc
- --sort-key desc
- """
- resources = "subnets"
- cmd = subnet.ListSubnet(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd,
- sort_key=["name", "id"],
- sort_dir=["asc", "desc"])
-
- def test_list_subnets_limit(self):
- """List subnets: -P."""
- resources = "subnets"
- cmd = subnet.ListSubnet(test_cli20.MyApp(sys.stdout), None)
- self._test_list_resources(resources, cmd, page_size=1000)
-
- def test_update_subnet(self):
- """Update subnet: myid --name myname --tags a b."""
- resource = 'subnet'
- cmd = subnet.UpdateSubnet(test_cli20.MyApp(sys.stdout), None)
- self._test_update_resource(resource, cmd, 'myid',
- ['myid', '--name', 'myname',
- '--tags', 'a', 'b'],
- {'name': 'myname', 'tags': ['a', 'b'], }
- )
-
- def test_update_subnet_known_option_before_id(self):
- """Update subnet: --request-format json myid --name myname."""
- # --request-format xml is known option
- resource = 'subnet'
- cmd = subnet.UpdateSubnet(test_cli20.MyApp(sys.stdout), None)
- self._test_update_resource(resource, cmd, 'myid',
- ['--request-format', 'json',
- 'myid', '--name', 'myname'],
- {'name': 'myname', }
- )
-
- def test_update_subnet_known_option_after_id(self):
- """Update subnet: myid --name myname --request-format json."""
- # --request-format xml is known option
- resource = 'subnet'
- cmd = subnet.UpdateSubnet(test_cli20.MyApp(sys.stdout), None)
- self._test_update_resource(resource, cmd, 'myid',
- ['myid', '--name', 'myname',
- '--request-format', 'json'],
- {'name': 'myname', }
- )
-
- def test_show_subnet(self):
- """Show subnet: --fields id --fields name myid."""
- resource = 'subnet'
- cmd = subnet.ShowSubnet(test_cli20.MyApp(sys.stdout), None)
- args = ['--fields', 'id', '--fields', 'name', self.test_id]
- self._test_show_resource(resource, cmd, self.test_id,
- args, ['id', 'name'])
-
- def test_delete_subnet(self):
- """Delete subnet: subnetid."""
- resource = 'subnet'
- cmd = subnet.DeleteSubnet(test_cli20.MyApp(sys.stdout), None)
- myid = 'myid'
- args = [myid]
- self._test_delete_resource(resource, cmd, myid, args)
-
-
-class CLITestV20SubnetXML(CLITestV20SubnetJSON):
- format = 'xml'
diff --git a/tests/unit/test_name_or_id.py b/tests/unit/test_name_or_id.py
deleted file mode 100644
index fc3bdac..0000000
--- a/tests/unit/test_name_or_id.py
+++ /dev/null
@@ -1,132 +0,0 @@
-# Copyright 2012 OpenStack LLC.
-# All Rights Reserved
-#
-# 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.
-#
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-import uuid
-
-import mox
-import testtools
-
-from quantumclient.common import exceptions
-from quantumclient.quantum import v2_0 as quantumv20
-from quantumclient.v2_0 import client
-from tests.unit import test_cli20
-
-
-class CLITestNameorID(testtools.TestCase):
-
- def setUp(self):
- """Prepare the test environment."""
- super(CLITestNameorID, self).setUp()
- self.mox = mox.Mox()
- self.endurl = test_cli20.ENDURL
- self.client = client.Client(token=test_cli20.TOKEN,
- endpoint_url=self.endurl)
- self.addCleanup(self.mox.VerifyAll)
- self.addCleanup(self.mox.UnsetStubs)
-
- def test_get_id_from_id(self):
- _id = str(uuid.uuid4())
- reses = {'networks': [{'id': _id, }, ], }
- resstr = self.client.serialize(reses)
- self.mox.StubOutWithMock(self.client.httpclient, "request")
- path = getattr(self.client, "networks_path")
- self.client.httpclient.request(
- test_cli20.end_url(path, "fields=id&id=" + _id), 'GET',
- body=None,
- headers=mox.ContainsKeyValue('X-Auth-Token', test_cli20.TOKEN)
- ).AndReturn((test_cli20.MyResp(200), resstr))
- self.mox.ReplayAll()
- returned_id = quantumv20.find_resourceid_by_name_or_id(
- self.client, 'network', _id)
- self.assertEqual(_id, returned_id)
-
- def test_get_id_from_id_then_name_empty(self):
- _id = str(uuid.uuid4())
- reses = {'networks': [{'id': _id, }, ], }
- resstr = self.client.serialize(reses)
- resstr1 = self.client.serialize({'networks': []})
- self.mox.StubOutWithMock(self.client.httpclient, "request")
- path = getattr(self.client, "networks_path")
- self.client.httpclient.request(
- test_cli20.end_url(path, "fields=id&id=" + _id), 'GET',
- body=None,
- headers=mox.ContainsKeyValue('X-Auth-Token', test_cli20.TOKEN)
- ).AndReturn((test_cli20.MyResp(200), resstr1))
- self.client.httpclient.request(
- test_cli20.end_url(path, "fields=id&name=" + _id), 'GET',
- body=None,
- headers=mox.ContainsKeyValue('X-Auth-Token', test_cli20.TOKEN)
- ).AndReturn((test_cli20.MyResp(200), resstr))
- self.mox.ReplayAll()
- returned_id = quantumv20.find_resourceid_by_name_or_id(
- self.client, 'network', _id)
- self.assertEqual(_id, returned_id)
-
- def test_get_id_from_name(self):
- name = 'myname'
- _id = str(uuid.uuid4())
- reses = {'networks': [{'id': _id, }, ], }
- resstr = self.client.serialize(reses)
- self.mox.StubOutWithMock(self.client.httpclient, "request")
- path = getattr(self.client, "networks_path")
- self.client.httpclient.request(
- test_cli20.end_url(path, "fields=id&name=" + name), 'GET',
- body=None,
- headers=mox.ContainsKeyValue('X-Auth-Token', test_cli20.TOKEN)
- ).AndReturn((test_cli20.MyResp(200), resstr))
- self.mox.ReplayAll()
- returned_id = quantumv20.find_resourceid_by_name_or_id(
- self.client, 'network', name)
- self.assertEqual(_id, returned_id)
-
- def test_get_id_from_name_multiple(self):
- name = 'myname'
- reses = {'networks': [{'id': str(uuid.uuid4())},
- {'id': str(uuid.uuid4())}]}
- resstr = self.client.serialize(reses)
- self.mox.StubOutWithMock(self.client.httpclient, "request")
- path = getattr(self.client, "networks_path")
- self.client.httpclient.request(
- test_cli20.end_url(path, "fields=id&name=" + name), 'GET',
- body=None,
- headers=mox.ContainsKeyValue('X-Auth-Token', test_cli20.TOKEN)
- ).AndReturn((test_cli20.MyResp(200), resstr))
- self.mox.ReplayAll()
- try:
- quantumv20.find_resourceid_by_name_or_id(
- self.client, 'network', name)
- except exceptions.QuantumClientException as ex:
- self.assertTrue('Multiple' in ex.message)
-
- def test_get_id_from_name_notfound(self):
- name = 'myname'
- reses = {'networks': []}
- resstr = self.client.serialize(reses)
- self.mox.StubOutWithMock(self.client.httpclient, "request")
- path = getattr(self.client, "networks_path")
- self.client.httpclient.request(
- test_cli20.end_url(path, "fields=id&name=" + name), 'GET',
- body=None,
- headers=mox.ContainsKeyValue('X-Auth-Token', test_cli20.TOKEN)
- ).AndReturn((test_cli20.MyResp(200), resstr))
- self.mox.ReplayAll()
- try:
- quantumv20.find_resourceid_by_name_or_id(
- self.client, 'network', name)
- except exceptions.QuantumClientException as ex:
- self.assertTrue('Unable to find' in ex.message)
- self.assertEqual(404, ex.status_code)
diff --git a/tests/unit/test_quota.py b/tests/unit/test_quota.py
deleted file mode 100644
index 633ff62..0000000
--- a/tests/unit/test_quota.py
+++ /dev/null
@@ -1,45 +0,0 @@
-#!/usr/bin/env python
-# Copyright (C) 2013 Yahoo! Inc.
-# All Rights Reserved.
-#
-# 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.
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-import sys
-
-from quantumclient.common import exceptions
-from quantumclient.quantum.v2_0 import quota as test_quota
-from tests.unit import test_cli20
-
-
-class CLITestV20Quota(test_cli20.CLITestV20Base):
- def test_show_quota(self):
- resource = 'quota'
- cmd = test_quota.ShowQuota(
- test_cli20.MyApp(sys.stdout), None)
- args = ['--tenant-id', self.test_id]
- self._test_show_resource(resource, cmd, self.test_id, args)
-
- def test_update_quota(self):
- resource = 'quota'
- cmd = test_quota.UpdateQuota(
- test_cli20.MyApp(sys.stdout), None)
- args = ['--tenant-id', self.test_id, '--network', 'test']
- self.assertRaises(
- exceptions.QuantumClientException, self._test_update_resource,
- resource, cmd, self.test_id, args=args,
- extrafields={'network': 'new'})
-
- def test_delete_quota_get_parser(self):
- cmd = test_cli20.MyApp(sys.stdout)
- test_quota.DeleteQuota(cmd, None).get_parser(cmd)
diff --git a/tests/unit/test_shell.py b/tests/unit/test_shell.py
deleted file mode 100644
index b6a8e14..0000000
--- a/tests/unit/test_shell.py
+++ /dev/null
@@ -1,173 +0,0 @@
-# Copyright (C) 2013 Yahoo! Inc.
-# All Rights Reserved.
-#
-# 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.
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-import argparse
-import cStringIO
-import os
-import re
-import sys
-
-import fixtures
-import mox
-import testtools
-from testtools import matchers
-
-from quantumclient.common import exceptions
-from quantumclient import shell as openstack_shell
-
-
-DEFAULT_USERNAME = 'username'
-DEFAULT_PASSWORD = 'password'
-DEFAULT_TENANT_ID = 'tenant_id'
-DEFAULT_TENANT_NAME = 'tenant_name'
-DEFAULT_AUTH_URL = 'http://127.0.0.1:5000/v2.0/'
-DEFAULT_TOKEN = '3bcc3d3a03f44e3d8377f9247b0ad155'
-DEFAULT_URL = 'http://quantum.example.org:9696/'
-
-
-class NoExitArgumentParser(argparse.ArgumentParser):
- def error(self, message):
- raise exceptions.CommandError(message)
-
-
-class ShellTest(testtools.TestCase):
-
- FAKE_ENV = {
- 'OS_USERNAME': DEFAULT_USERNAME,
- 'OS_PASSWORD': DEFAULT_PASSWORD,
- 'OS_TENANT_ID': DEFAULT_TENANT_ID,
- 'OS_TENANT_NAME': DEFAULT_TENANT_NAME,
- 'OS_AUTH_URL': DEFAULT_AUTH_URL}
-
- def _tolerant_shell(self, cmd):
- t_shell = openstack_shell.QuantumShell('2.0')
- t_shell.run(cmd.split())
-
- # Patch os.environ to avoid required auth info.
- def setUp(self):
- super(ShellTest, self).setUp()
- self.mox = mox.Mox()
- for var in self.FAKE_ENV:
- self.useFixture(
- fixtures.EnvironmentVariable(
- var, self.FAKE_ENV[var]))
-
- # Make a fake shell object, a helping wrapper to call it, and a quick
- # way of asserting that certain API calls were made.
- global shell, _shell, assert_called, assert_called_anytime
- _shell = openstack_shell.QuantumShell('2.0')
- shell = lambda cmd: _shell.run(cmd.split())
-
- def shell(self, argstr):
- orig = sys.stdout
- clean_env = {}
- _old_env, os.environ = os.environ, clean_env.copy()
- try:
- sys.stdout = cStringIO.StringIO()
- _shell = openstack_shell.QuantumShell('2.0')
- _shell.run(argstr.split())
- except SystemExit:
- exc_type, exc_value, exc_traceback = sys.exc_info()
- self.assertEqual(exc_value.code, 0)
- finally:
- out = sys.stdout.getvalue()
- sys.stdout.close()
- sys.stdout = orig
- os.environ = _old_env
- return out
-
- def test_run_unknown_command(self):
- openstack_shell.QuantumShell('2.0').run('fake')
-
- def test_help(self):
- required = 'usage:'
- help_text = self.shell('help')
- self.assertThat(
- help_text,
- matchers.MatchesRegex(required))
-
- def test_help_on_subcommand(self):
- required = [
- '.*?^usage: .* quota-list']
- stdout = self.shell('help quota-list')
- for r in required:
- self.assertThat(
- stdout,
- matchers.MatchesRegex(r, re.DOTALL | re.MULTILINE))
-
- def test_help_command(self):
- required = 'usage:'
- help_text = self.shell('help network-create')
- self.assertThat(
- help_text,
- matchers.MatchesRegex(required))
-
- def test_unknown_auth_strategy(self):
- self.shell('--os-auth-strategy fake quota-list')
-
- def test_auth(self):
- self.shell(' --os-username test'
- ' --os-password test'
- ' --os-tenant-name test'
- ' --os-auth-url http://127.0.0.1:5000/'
- ' --os-auth-strategy keystone quota-list')
-
- def test_build_option_parser(self):
- quant_shell = openstack_shell.QuantumShell('2.0')
- result = quant_shell.build_option_parser('descr', '2.0')
- self.assertEqual(True, isinstance(result, argparse.ArgumentParser))
-
- def test_main_with_unicode(self):
- self.mox.StubOutClassWithMocks(openstack_shell, 'QuantumShell')
- qshell_mock = openstack_shell.QuantumShell('2.0')
- #self.mox.StubOutWithMock(qshell_mock, 'run')
- unicode_text = u'\u7f51\u7edc'
- argv = ['net-list', unicode_text, unicode_text.encode('utf-8')]
- qshell_mock.run([u'net-list', unicode_text,
- unicode_text]).AndReturn(0)
- self.mox.ReplayAll()
- ret = openstack_shell.main(argv=argv)
- self.mox.VerifyAll()
- self.mox.UnsetStubs()
- self.assertEqual(ret, 0)
-
- def test_endpoint_option(self):
- shell = openstack_shell.QuantumShell('2.0')
- parser = shell.build_option_parser('descr', '2.0')
-
- # Neither $OS_ENDPOINT_TYPE nor --endpoint-type
- namespace = parser.parse_args([])
- self.assertEqual('publicURL', namespace.endpoint_type)
-
- # --endpoint-type but not $OS_ENDPOINT_TYPE
- namespace = parser.parse_args(['--endpoint-type=admin'])
- self.assertEqual('admin', namespace.endpoint_type)
-
- def test_endpoint_environment_variable(self):
- fixture = fixtures.EnvironmentVariable("OS_ENDPOINT_TYPE",
- "public")
- self.useFixture(fixture)
-
- shell = openstack_shell.QuantumShell('2.0')
- parser = shell.build_option_parser('descr', '2.0')
-
- # $OS_ENDPOINT_TYPE but not --endpoint-type
- namespace = parser.parse_args([])
- self.assertEqual("public", namespace.endpoint_type)
-
- # --endpoint-type and $OS_ENDPOINT_TYPE
- namespace = parser.parse_args(['--endpoint-type=admin'])
- self.assertEqual('admin', namespace.endpoint_type)
diff --git a/tests/unit/test_utils.py b/tests/unit/test_utils.py
deleted file mode 100644
index d7d1520..0000000
--- a/tests/unit/test_utils.py
+++ /dev/null
@@ -1,190 +0,0 @@
-# Copyright (C) 2013 Yahoo! Inc.
-# All Rights Reserved.
-#
-# 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.
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-
-import datetime
-import sys
-
-import testtools
-
-from quantumclient.common import exceptions
-from quantumclient.common import utils
-
-
-class TestUtils(testtools.TestCase):
- def test_string_to_bool_true(self):
- self.assertTrue(utils.str2bool('true'))
-
- def test_string_to_bool_false(self):
- self.assertFalse(utils.str2bool('false'))
-
- def test_string_to_bool_None(self):
- self.assertIsNone(utils.str2bool(None))
-
- def test_string_to_dictionary(self):
- input_str = 'key1=value1,key2=value2'
- expected = {'key1': 'value1', 'key2': 'value2'}
- self.assertEqual(expected, utils.str2dict(input_str))
-
- def test_get_dict_item_properties(self):
- item = {'name': 'test_name', 'id': 'test_id'}
- fields = ('name', 'id')
- actual = utils.get_item_properties(item=item, fields=fields)
- self.assertEqual(('test_name', 'test_id'), actual)
-
- def test_get_object_item_properties_mixed_case_fields(self):
- class Fake(object):
- def __init__(self):
- self.id = 'test_id'
- self.name = 'test_name'
- self.test_user = 'test'
-
- fields = ('name', 'id', 'test user')
- mixed_fields = ('test user', 'ID')
- item = Fake()
- actual = utils.get_item_properties(item, fields, mixed_fields)
- self.assertEqual(('test_name', 'test_id', 'test'), actual)
-
- def test_get_object_item_desired_fields_differ_from_item(self):
- class Fake(object):
- def __init__(self):
- self.id = 'test_id_1'
- self.name = 'test_name'
- self.test_user = 'test'
-
- fields = ('name', 'id', 'test user')
- item = Fake()
- actual = utils.get_item_properties(item, fields)
- self.assertNotEqual(('test_name', 'test_id', 'test'), actual)
-
- def test_get_object_item_desired_fields_is_empty(self):
- class Fake(object):
- def __init__(self):
- self.id = 'test_id_1'
- self.name = 'test_name'
- self.test_user = 'test'
-
- fields = []
- item = Fake()
- actual = utils.get_item_properties(item, fields)
- self.assertEqual((), actual)
-
- def test_get_object_item_with_formatters(self):
- class Fake(object):
- def __init__(self):
- self.id = 'test_id'
- self.name = 'test_name'
- self.test_user = 'test'
-
- class FakeCallable(object):
- def __call__(self, *args, **kwargs):
- return 'pass'
-
- fields = ('name', 'id', 'test user', 'is_public')
- formatters = {'is_public': FakeCallable()}
- item = Fake()
- act = utils.get_item_properties(item, fields, formatters=formatters)
- self.assertEqual(('test_name', 'test_id', 'test', 'pass'), act)
-
-
-class JSONUtilsTestCase(testtools.TestCase):
- def test_dumps(self):
- self.assertEqual(utils.dumps({'a': 'b'}), '{"a": "b"}')
-
- def test_dumps_dict_with_date_value(self):
- x = datetime.datetime(1920, 2, 3, 4, 5, 6, 7)
- res = utils.dumps({1: 'a', 2: x})
- expected = '{"1": "a", "2": "1920-02-03 04:05:06.000007"}'
- self.assertEqual(expected, res)
-
- def test_dumps_dict_with_spaces(self):
- x = datetime.datetime(1920, 2, 3, 4, 5, 6, 7)
- res = utils.dumps({1: 'a ', 2: x})
- expected = '{"1": "a ", "2": "1920-02-03 04:05:06.000007"}'
- self.assertEqual(expected, res)
-
- def test_loads(self):
- self.assertEqual(utils.loads('{"a": "b"}'), {'a': 'b'})
-
-
-class ToPrimitiveTestCase(testtools.TestCase):
- def test_list(self):
- self.assertEqual(utils.to_primitive([1, 2, 3]), [1, 2, 3])
-
- def test_empty_list(self):
- self.assertEqual(utils.to_primitive([]), [])
-
- def test_tuple(self):
- self.assertEqual(utils.to_primitive((1, 2, 3)), [1, 2, 3])
-
- def test_empty_tuple(self):
- self.assertEqual(utils.to_primitive(()), [])
-
- def test_dict(self):
- self.assertEqual(
- utils.to_primitive(dict(a=1, b=2, c=3)),
- dict(a=1, b=2, c=3))
-
- def test_empty_dict(self):
- self.assertEqual(utils.to_primitive({}), {})
-
- def test_datetime(self):
- x = datetime.datetime(1920, 2, 3, 4, 5, 6, 7)
- self.assertEqual(
- utils.to_primitive(x),
- '1920-02-03 04:05:06.000007')
-
- def test_iter(self):
- x = xrange(1, 6)
- self.assertEqual(utils.to_primitive(x), [1, 2, 3, 4, 5])
-
- def test_iteritems(self):
- d = {'a': 1, 'b': 2, 'c': 3}
-
- class IterItemsClass(object):
- def iteritems(self):
- return d.iteritems()
-
- x = IterItemsClass()
- p = utils.to_primitive(x)
- self.assertEqual(p, {'a': 1, 'b': 2, 'c': 3})
-
- def test_nasties(self):
- def foo():
- pass
- x = [datetime, foo, dir]
- ret = utils.to_primitive(x)
- self.assertEqual(len(ret), 3)
-
- def test_to_primitive_dict_with_date_value(self):
- x = datetime.datetime(1920, 2, 3, 4, 5, 6, 7)
- res = utils.to_primitive({'a': x})
- self.assertEqual({'a': '1920-02-03 04:05:06.000007'}, res)
-
-
-class ImportClassTestCase(testtools.TestCase):
- def test_import_class(self):
- dt = utils.import_class('datetime.datetime')
- self.assertTrue(sys.modules['datetime'].datetime is dt)
-
- def test_import_bad_class(self):
- self.assertRaises(
- ImportError, utils.import_class,
- 'lol.u_mad.brah')
-
- def test_get_client_class_invalid_version(self):
- self.assertRaises(
- exceptions.UnsupportedVersion,
- utils.get_client_class, 'image', '2', {'image': '2'})
View Lorry Controller Status