openstack/python-openstackclient
Code Issues Proposed changes

Test top-to-bottom: object-store containers

Browse Source 
Replicate the object-store container command tests but use requests_mock
to test the entire stack down to the requests module.

These will be useful regressions tests when the existing object-store lib
modules are moved to the low-level API object.

Change-Id: Ibf25be46156eb1009f1b66f02f2073d3913b846d
This commit is contained in:
Dean Troyer 2014-09-18 22:10:03 -05:00
parent 02320a5a24
commit 207c8cf3ef
4 changed files with 353 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
openstackclient/tests/fakes.py
View File

@ -47,12 +47,18 @@ class FakeApp(object):
self.stderr = sys.stderr
class FakeClient(object):
def __init__(self, **kwargs):
self.endpoint = kwargs['endpoint']
self.token = kwargs['token']
class FakeClientManager(object):
def __init__(self):
self.compute = None
self.identity = None
self.image = None
self.object = None
self.object_store = None
self.volume = None
self.network = None
self.session = None

11
openstackclient/tests/object/v1/fakes.py
View File

@ -17,6 +17,9 @@ from openstackclient.tests import fakes
from openstackclient.tests import utils
ACCOUNT_ID = 'tqbfjotld'
ENDPOINT = 'https://0.0.0.0:6482/v1/' + ACCOUNT_ID
container_name = 'bit-bucket'
container_bytes = 1024
container_count = 1
@ -71,17 +74,11 @@ OBJECT_2 = {
}
class FakeObjectv1Client(object):
def __init__(self, **kwargs):
self.endpoint = kwargs['endpoint']
self.token = kwargs['token']
class TestObjectv1(utils.TestCommand):
def setUp(self):
super(TestObjectv1, self).setUp()
self.app.client_manager.object_store = FakeObjectv1Client(
self.app.client_manager.object_store = fakes.FakeClient(
endpoint=fakes.AUTH_URL,
token=fakes.AUTH_TOKEN,
)

341
openstackclient/tests/object/v1/test_container_all.py Normal file
View File

@ -0,0 +1,341 @@
# 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 copy
from requests_mock.contrib import fixture
from keystoneclient import session
from openstackclient.object.v1 import container
from openstackclient.tests.object.v1 import fakes as object_fakes
class TestObjectAll(object_fakes.TestObjectv1):
def setUp(self):
super(TestObjectAll, self).setUp()
self.app.client_manager.session = session.Session()
self.requests_mock = self.useFixture(fixture.Fixture())
# TODO(dtroyer): move this to object_fakes.TestObjectv1
self.app.client_manager.object_store.endpoint = object_fakes.ENDPOINT
class TestContainerCreate(TestObjectAll):
def setUp(self):
super(TestContainerCreate, self).setUp()
# Get the command object to test
self.cmd = container.CreateContainer(self.app, None)
def test_object_create_container_single(self):
self.requests_mock.register_uri(
'PUT',
object_fakes.ENDPOINT + '/ernie',
headers={'x-trans-id': '314159'},
status_code=200,
)
arglist = [
'ernie',
]
verifylist = [(
'containers', ['ernie'],
)]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
# DisplayCommandBase.take_action() returns two tuples
columns, data = self.cmd.take_action(parsed_args)
collist = ('account', 'container', 'x-trans-id')
self.assertEqual(collist, columns)
datalist = [(
object_fakes.ACCOUNT_ID,
'ernie',
'314159',
)]
self.assertEqual(datalist, list(data))
def test_object_create_container_more(self):
self.requests_mock.register_uri(
'PUT',
object_fakes.ENDPOINT + '/ernie',
headers={'x-trans-id': '314159'},
status_code=200,
)
self.requests_mock.register_uri(
'PUT',
object_fakes.ENDPOINT + '/bert',
headers={'x-trans-id': '42'},
status_code=200,
)
arglist = [
'ernie',
'bert',
]
verifylist = [(
'containers', ['ernie', 'bert'],
)]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
# DisplayCommandBase.take_action() returns two tuples
columns, data = self.cmd.take_action(parsed_args)
collist = ('account', 'container', 'x-trans-id')
self.assertEqual(collist, columns)
datalist = [
(
object_fakes.ACCOUNT_ID,
'ernie',
'314159',
),
(
object_fakes.ACCOUNT_ID,
'bert',
'42',
),
]
self.assertEqual(datalist, list(data))
class TestContainerDelete(TestObjectAll):
def setUp(self):
super(TestContainerDelete, self).setUp()
# Get the command object to test
self.cmd = container.DeleteContainer(self.app, None)
def test_object_delete_container_single(self):
self.requests_mock.register_uri(
'DELETE',
object_fakes.ENDPOINT + '/ernie',
status_code=200,
)
arglist = [
'ernie',
]
verifylist = [(
'containers', ['ernie'],
)]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
# Command.take_action() returns None
ret = self.cmd.take_action(parsed_args)
self.assertIsNone(ret)
def test_object_delete_container_more(self):
self.requests_mock.register_uri(
'DELETE',
object_fakes.ENDPOINT + '/ernie',
status_code=200,
)
self.requests_mock.register_uri(
'DELETE',
object_fakes.ENDPOINT + '/bert',
status_code=200,
)
arglist = [
'ernie',
'bert',
]
verifylist = [(
'containers', ['ernie', 'bert'],
)]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
# Command.take_action() returns None
ret = self.cmd.take_action(parsed_args)
self.assertIsNone(ret)
class TestContainerList(TestObjectAll):
def setUp(self):
super(TestContainerList, self).setUp()
# Get the command object to test
self.cmd = container.ListContainer(self.app, None)
def test_object_list_containers_no_options(self):
return_body = [
copy.deepcopy(object_fakes.CONTAINER),
copy.deepcopy(object_fakes.CONTAINER_3),
copy.deepcopy(object_fakes.CONTAINER_2),
]
self.requests_mock.register_uri(
'GET',
object_fakes.ENDPOINT + '?format=json',
json=return_body,
status_code=200,
)
arglist = []
verifylist = []
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
# Lister.take_action() returns two tuples
columns, data = self.cmd.take_action(parsed_args)
collist = ('Name',)
self.assertEqual(collist, columns)
datalist = [
(object_fakes.container_name, ),
(object_fakes.container_name_3, ),
(object_fakes.container_name_2, ),
]
self.assertEqual(datalist, list(data))
def test_object_list_containers_prefix(self):
return_body = [
copy.deepcopy(object_fakes.CONTAINER),
copy.deepcopy(object_fakes.CONTAINER_3),
]
self.requests_mock.register_uri(
'GET',
object_fakes.ENDPOINT + '?format=json&prefix=bit',
json=return_body,
status_code=200,
)
arglist = [
'--prefix', 'bit',
]
verifylist = [
('prefix', 'bit'),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
# Lister.take_action() returns two tuples
columns, data = self.cmd.take_action(parsed_args)
collist = ('Name',)
self.assertEqual(collist, columns)
datalist = [
(object_fakes.container_name, ),
(object_fakes.container_name_3, ),
]
self.assertEqual(datalist, list(data))
class TestContainerSave(TestObjectAll):
def setUp(self):
super(TestContainerSave, self).setUp()
# Get the command object to test
self.cmd = container.SaveContainer(self.app, None)
# TODO(dtroyer): need to mock out object_lib.save_object() to test this
# def test_object_save_container(self):
# return_body = [
# copy.deepcopy(object_fakes.OBJECT),
# copy.deepcopy(object_fakes.OBJECT_2),
# ]
# # Initial container list request
# self.requests_mock.register_uri(
# 'GET',
# object_fakes.ENDPOINT + '/oscar?format=json',
# json=return_body,
# status_code=200,
# )
# # Individual object save requests
# self.requests_mock.register_uri(
# 'GET',
# object_fakes.ENDPOINT + '/oscar/' + object_fakes.object_name_1,
# json=object_fakes.OBJECT,
# status_code=200,
# )
# self.requests_mock.register_uri(
# 'GET',
# object_fakes.ENDPOINT + '/oscar/' + object_fakes.object_name_2,
# json=object_fakes.OBJECT_2,
# status_code=200,
# )
#
# arglist = [
# 'oscar',
# ]
# verifylist = [(
# 'container', 'oscar',
# )]
# parsed_args = self.check_parser(self.cmd, arglist, verifylist)
#
# # Command.take_action() returns None
# ret = self.cmd.take_action(parsed_args)
# self.assertIsNone(ret)
class TestContainerShow(TestObjectAll):
def setUp(self):
super(TestContainerShow, self).setUp()
# Get the command object to test
self.cmd = container.ShowContainer(self.app, None)
def test_object_show_container(self):
headers = {
'x-container-meta-owner': object_fakes.ACCOUNT_ID,
'x-container-object-count': '42',
'x-container-bytes-used': '123',
'x-container-read': 'qaz',
'x-container-write': 'wsx',
'x-container-sync-to': 'edc',
'x-container-sync-key': 'rfv',
}
self.requests_mock.register_uri(
'HEAD',
object_fakes.ENDPOINT + '/ernie',
headers=headers,
status_code=200,
)
arglist = [
'ernie',
]
verifylist = [(
'container', 'ernie',
)]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
# DisplayCommandBase.take_action() returns two tuples
columns, data = self.cmd.take_action(parsed_args)
collist = (
'account',
'bytes_used',
'container',
'object_count',
'read_acl',
'sync_key',
'sync_to',
'write_acl',
)
self.assertEqual(collist, columns)
datalist = [
object_fakes.ACCOUNT_ID,
'123',
'ernie',
'42',
'qaz',
'rfv',
'edc',
'wsx',
]
self.assertEqual(datalist, list(data))

1
test-requirements.txt
View File

@ -8,6 +8,7 @@ discover
fixtures>=0.3.14
mock>=1.0
oslosphinx>=2.2.0 # Apache-2.0
requests-mock>=0.4.0 # Apache-2.0
sphinx>=1.1.2,!=1.2.0,<1.3
testrepository>=0.0.18
testtools>=0.9.34