bcf4b3caec
* Switch to use io.open() for py3 compatibility and simpler testing. * Open files in 'rb' mode to avoid translation on Windows Previously tests simply relied on files that were present in the repository to run tests using open(). Change the filenames to ensure that no longer happens. requests_mock doesn't have a way to match against the request body for PUT/POST; an attempt to add a new Matcher to do that worked but it needs to subclass the currently private adapter._Matcher class or duplicate most of its functionality. Change-Id: I8c30b41db20af8ecafe67e760e872fc08adec905
340 lines
9.8 KiB
Python
340 lines
9.8 KiB
Python
# 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.
|
|
#
|
|
|
|
"""Object Store v1 API Library Tests"""
|
|
|
|
import mock
|
|
|
|
from requests_mock.contrib import fixture
|
|
|
|
from keystoneclient import session
|
|
from openstackclient.api import object_store_v1 as object_store
|
|
from openstackclient.tests import utils
|
|
|
|
|
|
FAKE_ACCOUNT = 'q12we34r'
|
|
FAKE_AUTH = '11223344556677889900'
|
|
FAKE_URL = 'http://gopher.com/v1/' + FAKE_ACCOUNT
|
|
|
|
FAKE_CONTAINER = 'rainbarrel'
|
|
FAKE_OBJECT = 'spigot'
|
|
|
|
LIST_CONTAINER_RESP = [
|
|
'qaz',
|
|
'fred',
|
|
]
|
|
|
|
LIST_OBJECT_RESP = [
|
|
{'name': 'fred', 'bytes': 1234, 'content_type': 'text'},
|
|
{'name': 'wilma', 'bytes': 5678, 'content_type': 'text'},
|
|
]
|
|
|
|
|
|
class TestObjectAPIv1(utils.TestCase):
|
|
|
|
def setUp(self):
|
|
super(TestObjectAPIv1, self).setUp()
|
|
sess = session.Session()
|
|
self.api = object_store.APIv1(session=sess, endpoint=FAKE_URL)
|
|
self.requests_mock = self.useFixture(fixture.Fixture())
|
|
|
|
|
|
class TestContainer(TestObjectAPIv1):
|
|
|
|
def setUp(self):
|
|
super(TestContainer, self).setUp()
|
|
|
|
def test_container_create(self):
|
|
headers = {
|
|
'x-trans-id': '1qaz2wsx',
|
|
}
|
|
self.requests_mock.register_uri(
|
|
'PUT',
|
|
FAKE_URL + '/qaz',
|
|
headers=headers,
|
|
status_code=201,
|
|
)
|
|
ret = self.api.container_create(container='qaz')
|
|
data = {
|
|
'account': FAKE_ACCOUNT,
|
|
'container': 'qaz',
|
|
'x-trans-id': '1qaz2wsx',
|
|
}
|
|
self.assertEqual(data, ret)
|
|
|
|
def test_container_delete(self):
|
|
self.requests_mock.register_uri(
|
|
'DELETE',
|
|
FAKE_URL + '/qaz',
|
|
status_code=204,
|
|
)
|
|
ret = self.api.container_delete(container='qaz')
|
|
self.assertIsNone(ret)
|
|
|
|
def test_container_list_no_options(self):
|
|
self.requests_mock.register_uri(
|
|
'GET',
|
|
FAKE_URL,
|
|
json=LIST_CONTAINER_RESP,
|
|
status_code=200,
|
|
)
|
|
ret = self.api.container_list()
|
|
self.assertEqual(LIST_CONTAINER_RESP, ret)
|
|
|
|
def test_container_list_prefix(self):
|
|
self.requests_mock.register_uri(
|
|
'GET',
|
|
FAKE_URL + '?prefix=foo%2f&format=json',
|
|
json=LIST_CONTAINER_RESP,
|
|
status_code=200,
|
|
)
|
|
ret = self.api.container_list(
|
|
prefix='foo/',
|
|
)
|
|
self.assertEqual(LIST_CONTAINER_RESP, ret)
|
|
|
|
def test_container_list_marker_limit_end(self):
|
|
self.requests_mock.register_uri(
|
|
'GET',
|
|
FAKE_URL + '?marker=next&limit=2&end_marker=stop&format=json',
|
|
json=LIST_CONTAINER_RESP,
|
|
status_code=200,
|
|
)
|
|
ret = self.api.container_list(
|
|
marker='next',
|
|
limit=2,
|
|
end_marker='stop',
|
|
)
|
|
self.assertEqual(LIST_CONTAINER_RESP, ret)
|
|
|
|
# def test_container_list_full_listing(self):
|
|
# sess = self.app.client_manager.session
|
|
#
|
|
# def side_effect(*args, **kwargs):
|
|
# rv = sess.get().json.return_value
|
|
# sess.get().json.return_value = []
|
|
# sess.get().json.side_effect = None
|
|
# return rv
|
|
#
|
|
# resp = [{'name': 'is-name'}]
|
|
# sess.get().json.return_value = resp
|
|
# sess.get().json.side_effect = side_effect
|
|
#
|
|
# data = lib_container.list_containers(
|
|
# self.app.client_manager.session,
|
|
# fake_url,
|
|
# full_listing=True,
|
|
# )
|
|
#
|
|
# # Check expected values
|
|
# sess.get.assert_called_with(
|
|
# fake_url,
|
|
# params={
|
|
# 'format': 'json',
|
|
# 'marker': 'is-name',
|
|
# }
|
|
# )
|
|
# self.assertEqual(resp, data)
|
|
|
|
def test_container_show(self):
|
|
headers = {
|
|
'X-Container-Meta-Owner': FAKE_ACCOUNT,
|
|
'x-container-object-count': '1',
|
|
'x-container-bytes-used': '577',
|
|
}
|
|
resp = {
|
|
'account': FAKE_ACCOUNT,
|
|
'container': 'qaz',
|
|
'object_count': '1',
|
|
'bytes_used': '577',
|
|
'read_acl': None,
|
|
'write_acl': None,
|
|
'sync_to': None,
|
|
'sync_key': None,
|
|
}
|
|
self.requests_mock.register_uri(
|
|
'HEAD',
|
|
FAKE_URL + '/qaz',
|
|
headers=headers,
|
|
status_code=204,
|
|
)
|
|
ret = self.api.container_show(container='qaz')
|
|
self.assertEqual(resp, ret)
|
|
|
|
|
|
class TestObject(TestObjectAPIv1):
|
|
|
|
def setUp(self):
|
|
super(TestObject, self).setUp()
|
|
|
|
@mock.patch('openstackclient.api.object_store_v1.io.open')
|
|
def base_object_create(self, file_contents, mock_open):
|
|
mock_open.read.return_value = file_contents
|
|
|
|
headers = {
|
|
'etag': 'youreit',
|
|
'x-trans-id': '1qaz2wsx',
|
|
}
|
|
# TODO(dtroyer): When requests_mock gains the ability to
|
|
# match against request.body add this check
|
|
# https://review.openstack.org/127316
|
|
self.requests_mock.register_uri(
|
|
'PUT',
|
|
FAKE_URL + '/qaz/counter.txt',
|
|
headers=headers,
|
|
# body=file_contents,
|
|
status_code=201,
|
|
)
|
|
ret = self.api.object_create(
|
|
container='qaz',
|
|
object='counter.txt',
|
|
)
|
|
data = {
|
|
'account': FAKE_ACCOUNT,
|
|
'container': 'qaz',
|
|
'object': 'counter.txt',
|
|
'etag': 'youreit',
|
|
'x-trans-id': '1qaz2wsx',
|
|
}
|
|
self.assertEqual(data, ret)
|
|
|
|
def test_object_create(self):
|
|
self.base_object_create('111\n222\n333\n')
|
|
self.base_object_create(bytes([0x31, 0x00, 0x0d, 0x0a, 0x7f, 0xff]))
|
|
|
|
def test_object_delete(self):
|
|
self.requests_mock.register_uri(
|
|
'DELETE',
|
|
FAKE_URL + '/qaz/wsx',
|
|
status_code=204,
|
|
)
|
|
ret = self.api.object_delete(
|
|
container='qaz',
|
|
object='wsx',
|
|
)
|
|
self.assertIsNone(ret)
|
|
|
|
def test_object_list_no_options(self):
|
|
self.requests_mock.register_uri(
|
|
'GET',
|
|
FAKE_URL + '/qaz',
|
|
json=LIST_OBJECT_RESP,
|
|
status_code=200,
|
|
)
|
|
ret = self.api.object_list(container='qaz')
|
|
self.assertEqual(LIST_OBJECT_RESP, ret)
|
|
|
|
def test_object_list_delimiter(self):
|
|
self.requests_mock.register_uri(
|
|
'GET',
|
|
FAKE_URL + '/qaz?delimiter=%7C',
|
|
json=LIST_OBJECT_RESP,
|
|
status_code=200,
|
|
)
|
|
ret = self.api.object_list(
|
|
container='qaz',
|
|
delimiter='|',
|
|
)
|
|
self.assertEqual(LIST_OBJECT_RESP, ret)
|
|
|
|
def test_object_list_prefix(self):
|
|
self.requests_mock.register_uri(
|
|
'GET',
|
|
FAKE_URL + '/qaz?prefix=foo%2f',
|
|
json=LIST_OBJECT_RESP,
|
|
status_code=200,
|
|
)
|
|
ret = self.api.object_list(
|
|
container='qaz',
|
|
prefix='foo/',
|
|
)
|
|
self.assertEqual(LIST_OBJECT_RESP, ret)
|
|
|
|
def test_object_list_marker_limit_end(self):
|
|
self.requests_mock.register_uri(
|
|
'GET',
|
|
FAKE_URL + '/qaz?marker=next&limit=2&end_marker=stop',
|
|
json=LIST_CONTAINER_RESP,
|
|
status_code=200,
|
|
)
|
|
ret = self.api.object_list(
|
|
container='qaz',
|
|
marker='next',
|
|
limit=2,
|
|
end_marker='stop',
|
|
)
|
|
self.assertEqual(LIST_CONTAINER_RESP, ret)
|
|
|
|
# def test_list_objects_full_listing(self):
|
|
# sess = self.app.client_manager.session
|
|
#
|
|
# def side_effect(*args, **kwargs):
|
|
# rv = sess.get().json.return_value
|
|
# sess.get().json.return_value = []
|
|
# sess.get().json.side_effect = None
|
|
# return rv
|
|
#
|
|
# resp = [{'name': 'is-name'}]
|
|
# sess.get().json.return_value = resp
|
|
# sess.get().json.side_effect = side_effect
|
|
#
|
|
# data = lib_object.list_objects(
|
|
# sess,
|
|
# fake_url,
|
|
# fake_container,
|
|
# full_listing=True,
|
|
# )
|
|
#
|
|
# # Check expected values
|
|
# sess.get.assert_called_with(
|
|
# fake_url + '/' + fake_container,
|
|
# params={
|
|
# 'format': 'json',
|
|
# 'marker': 'is-name',
|
|
# }
|
|
# )
|
|
# self.assertEqual(resp, data)
|
|
|
|
def test_object_show(self):
|
|
headers = {
|
|
'content-type': 'text/alpha',
|
|
'content-length': '577',
|
|
'last-modified': '20130101',
|
|
'etag': 'qaz',
|
|
'x-container-meta-owner': FAKE_ACCOUNT,
|
|
'x-object-meta-wife': 'Wilma',
|
|
'x-tra-header': 'yabba-dabba-do',
|
|
}
|
|
resp = {
|
|
'account': FAKE_ACCOUNT,
|
|
'container': 'qaz',
|
|
'object': FAKE_OBJECT,
|
|
'content-type': 'text/alpha',
|
|
'content-length': '577',
|
|
'last-modified': '20130101',
|
|
'etag': 'qaz',
|
|
'wife': 'Wilma',
|
|
'x-tra-header': 'yabba-dabba-do',
|
|
}
|
|
self.requests_mock.register_uri(
|
|
'HEAD',
|
|
FAKE_URL + '/qaz/' + FAKE_OBJECT,
|
|
headers=headers,
|
|
status_code=204,
|
|
)
|
|
ret = self.api.object_show(
|
|
container='qaz',
|
|
object=FAKE_OBJECT,
|
|
)
|
|
self.assertEqual(resp, ret)
|