OpenStack Storage (Swift) Client
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

3418 lines
144 KiB

# Copyright (c) 2010-2012 OpenStack, LLC.
#
# 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 gzip
import json
import logging
import mock
import six
import socket
import string
import unittest
import warnings
import tempfile
from hashlib import md5
from six import binary_type
from six.moves.urllib.parse import urlparse
from requests.exceptions import RequestException
from .utils import (MockHttpTest, fake_get_auth_keystone, StubResponse,
FakeKeystone)
from swiftclient.utils import EMPTY_ETAG
from swiftclient.exceptions import ClientException
from swiftclient import client as c
import swiftclient.utils
import swiftclient
class TestClientException(unittest.TestCase):
def test_is_exception(self):
self.assertTrue(issubclass(c.ClientException, Exception))
def test_format(self):
exc = c.ClientException('something failed')
self.assertIn('something failed', str(exc))
test_kwargs = (
'scheme',
'host',
'port',
'path',
'query',
'status',
'reason',
'device',
'response_content',
)
for value in test_kwargs:
kwargs = {
'http_%s' % value: value,
}
exc = c.ClientException('test', **kwargs)
self.assertIn(value, str(exc))
def test_attrs(self):
test_kwargs = (
'scheme',
'host',
'port',
'path',
'query',
'status',
'reason',
'device',
'response_content',
'response_headers',
)
for value in test_kwargs:
key = 'http_%s' % value
kwargs = {key: value}
exc = c.ClientException('test', **kwargs)
self.assertIs(True, hasattr(exc, key))
self.assertEqual(getattr(exc, key), value)
class MockHttpResponse(object):
def __init__(self, status=0, headers=None, verify=False):
self.status = status
self.status_code = status
self.reason = "OK"
self.buffer = []
self.requests_params = None
self.verify = verify
self.md5sum = md5()
self.headers = {'etag': '"%s"' % EMPTY_ETAG}
if headers:
self.headers.update(headers)
self.closed = False
class Raw(object):
def __init__(self, headers):
self.headers = headers
def read(self, **kw):
return ""
def getheader(self, name, default):
return self.headers.get(name, default)
self.raw = Raw(headers)
def read(self):
return ""
def close(self):
self.closed = True
def getheader(self, name, default):
return self.headers.get(name, default)
def getheaders(self):
return dict(self.headers).items()
def fake_response(self):
return self
def _fake_request(self, *arg, **kwarg):
self.status = 200
self.requests_params = kwarg
if self.verify:
for chunk in kwarg['data']:
self.md5sum.update(chunk)
# This simulate previous httplib implementation that would do a
# putrequest() and then use putheader() to send header.
for k, v in kwarg['headers'].items():
self.buffer.append((k, v))
return self.fake_response()
class TestHttpHelpers(MockHttpTest):
def test_quote(self):
value = b'bytes\xff'
self.assertEqual('bytes%FF', c.quote(value))
value = 'native string'
self.assertEqual('native%20string', c.quote(value))
value = u'unicode string'
self.assertEqual('unicode%20string', c.quote(value))
value = u'unicode:\xe9\u20ac'
self.assertEqual('unicode%3A%C3%A9%E2%82%AC', c.quote(value))
def test_parse_header_string(self):
value = b'bytes'
self.assertEqual(u'bytes', c.parse_header_string(value))
value = u'unicode:\xe9\u20ac'
self.assertEqual(u'unicode:\xe9\u20ac', c.parse_header_string(value))
value = 'native%20string'
self.assertEqual(u'native string', c.parse_header_string(value))
value = b'encoded%20bytes%E2%82%AC'
self.assertEqual(u'encoded bytes\u20ac', c.parse_header_string(value))
value = 'encoded%20unicode%E2%82%AC'
self.assertEqual(u'encoded unicode\u20ac',
c.parse_header_string(value))
value = b'bad%20bytes%ff%E2%82%AC'
self.assertEqual(u'bad%20bytes%ff%E2%82%AC',
c.parse_header_string(value))
value = u'bad%20unicode%ff\u20ac'
self.assertEqual(u'bad%20unicode%ff\u20ac',
c.parse_header_string(value))
value = b'really%20bad\xffbytes'
self.assertEqual(u'really%2520bad%FFbytes',
c.parse_header_string(value))
def test_http_connection(self):
url = 'http://www.test.com'
_junk, conn = c.http_connection(url)
self.assertIs(type(conn), c.HTTPConnection)
url = 'https://www.test.com'
_junk, conn = c.http_connection(url)
self.assertIs(type(conn), c.HTTPConnection)
url = 'ftp://www.test.com'
self.assertRaises(c.ClientException, c.http_connection, url)
def test_encode_meta_headers(self):
headers = {'abc': '123',
u'x-container-meta-\u0394': 123,
u'x-account-meta-\u0394': 12.3,
u'x-object-meta-\u0394': True}
r = swiftclient.encode_meta_headers(headers)
self.assertEqual(len(headers), len(r))
# ensure non meta headers are not encoded
self.assertIs(type(r.get('abc')), binary_type)
del r['abc']
for k, v in r.items():
self.assertIs(type(k), binary_type)
self.assertIs(type(v), binary_type)
self.assertIn(v, (b'123', b'12.3', b'True'))
def test_set_user_agent_default(self):
_junk, conn = c.http_connection('http://www.example.com')
req_headers = {}
def my_request_handler(*a, **kw):
req_headers.update(kw.get('headers', {}))
conn._request = my_request_handler
# test the default
conn.request('GET', '/')
ua = req_headers.get('user-agent', 'XXX-MISSING-XXX')
self.assertTrue(ua.startswith('python-swiftclient-'))
def test_set_user_agent_per_request_override(self):
_junk, conn = c.http_connection('http://www.example.com')
req_headers = {}
def my_request_handler(*a, **kw):
req_headers.update(kw.get('headers', {}))
conn._request = my_request_handler
# test if it's actually set
conn.request('GET', '/', headers={'User-Agent': 'Me'})
ua = req_headers.get('user-agent', 'XXX-MISSING-XXX')
self.assertEqual(ua, b'Me', req_headers)
def test_set_user_agent_default_override(self):
_junk, conn = c.http_connection(
'http://www.example.com',
default_user_agent='a-new-default')
req_headers = {}
def my_request_handler(*a, **kw):
req_headers.update(kw.get('headers', {}))
conn._request = my_request_handler
# test setting a default
conn._request = my_request_handler
conn.request('GET', '/')
ua = req_headers.get('user-agent', 'XXX-MISSING-XXX')
self.assertEqual(ua, 'a-new-default')
class TestGetAuth(MockHttpTest):
def test_ok(self):
c.http_connection = self.fake_http_connection(200)
url, token = c.get_auth('http://www.test.com', 'asdf', 'asdf')
self.assertIsNone(url)
self.assertIsNone(token)
def test_invalid_auth(self):
self.assertRaises(c.ClientException, c.get_auth,
'http://www.tests.com', 'asdf', 'asdf',
auth_version="foo")
def test_auth_v1(self):
c.http_connection = self.fake_http_connection(200, auth_v1=True)
url, token = c.get_auth('http://www.test.com', 'asdf', 'asdf',
auth_version="1.0")
self.assertEqual(url, 'storageURL')
self.assertEqual(token, 'someauthtoken')
def test_auth_v1_insecure(self):
c.http_connection = self.fake_http_connection(200, 200, auth_v1=True)
url, token = c.get_auth('http://www.test.com/invalid_cert',
'asdf', 'asdf',
auth_version='1.0',
insecure=True)
self.assertEqual(url, 'storageURL')
self.assertEqual(token, 'someauthtoken')
with self.assertRaises(c.ClientException) as exc_context:
c.get_auth('http://www.test.com/invalid_cert',
'asdf', 'asdf', auth_version='1.0')
# TODO: this test is really on validating the mock and not the
# the full plumbing into the requests's 'verify' option
self.assertIn('invalid_certificate', str(exc_context.exception))
def test_auth_v1_timeout(self):
# this test has some overlap with
# TestConnection.test_timeout_passed_down but is required to check that
# get_auth does the right thing when it is not passed a timeout arg
orig_http_connection = c.http_connection
timeouts = []
def fake_request_handler(*a, **kw):
if 'timeout' in kw:
timeouts.append(kw['timeout'])
else:
timeouts.append(None)
return MockHttpResponse(
status=200,
headers={
'x-auth-token': 'a_token',
'x-storage-url': 'http://files.example.com/v1/AUTH_user'})
def fake_connection(*a, **kw):
url, conn = orig_http_connection(*a, **kw)
conn._request = fake_request_handler
return url, conn
with mock.patch('swiftclient.client.http_connection', fake_connection):
c.get_auth('http://www.test.com', 'asdf', 'asdf',
auth_version="1.0", timeout=42.0)
c.get_auth('http://www.test.com', 'asdf', 'asdf',
auth_version="1.0", timeout=None)
c.get_auth('http://www.test.com', 'asdf', 'asdf',
auth_version="1.0")
self.assertEqual(timeouts, [42.0, None, None])
def test_auth_v2_timeout(self):
# this test has some overlap with
# TestConnection.test_timeout_passed_down but is required to check that
# get_auth does the right thing when it is not passed a timeout arg
fake_ks = FakeKeystone(endpoint='http://some_url', token='secret')
with mock.patch('swiftclient.client.ksclient_v2', fake_ks):
c.get_auth('http://www.test.com', 'asdf', 'asdf',
os_options=dict(tenant_name='tenant'),
auth_version="2.0", timeout=42.0)
c.get_auth('http://www.test.com', 'asdf', 'asdf',
os_options=dict(tenant_name='tenant'),
auth_version="2.0", timeout=None)
c.get_auth('http://www.test.com', 'asdf', 'asdf',
os_options=dict(tenant_name='tenant'),
auth_version="2.0")
self.assertEqual(3, len(fake_ks.calls))
timeouts = [call['timeout'] for call in fake_ks.calls]
self.assertEqual([42.0, None, None], timeouts)
def test_auth_v2_with_tenant_name(self):
os_options = {'tenant_name': 'asdf'}
req_args = {'auth_version': '2.0'}
ks = fake_get_auth_keystone(os_options, required_kwargs=req_args)
with mock.patch('swiftclient.client.get_auth_keystone', ks):
url, token = c.get_auth('http://www.test.com', 'asdf', 'asdf',
os_options=os_options,
auth_version="2.0")
self.assertTrue(url.startswith("http"))
self.assertTrue(token)
def test_auth_v2_with_tenant_id(self):
os_options = {'tenant_id': 'asdf'}
req_args = {'auth_version': '2.0'}
ks = fake_get_auth_keystone(os_options, required_kwargs=req_args)
with mock.patch('swiftclient.client.get_auth_keystone', ks):
url, token = c.get_auth('http://www.test.com', 'asdf', 'asdf',
os_options=os_options,
auth_version="2.0")
self.assertTrue(url.startswith("http"))
self.assertTrue(token)
def test_auth_v2_with_project_name(self):
os_options = {'project_name': 'asdf'}
req_args = {'auth_version': '2.0'}
ks = fake_get_auth_keystone(os_options, required_kwargs=req_args)
with mock.patch('swiftclient.client.get_auth_keystone', ks):
url, token = c.get_auth('http://www.test.com', 'asdf', 'asdf',
os_options=os_options,
auth_version="2.0")
self.assertTrue(url.startswith("http"))
self.assertTrue(token)
def test_auth_v2_with_project_id(self):
os_options = {'project_id': 'asdf'}
req_args = {'auth_version': '2.0'}
ks = fake_get_auth_keystone(os_options, required_kwargs=req_args)
with mock.patch('swiftclient.client.get_auth_keystone', ks):
url, token = c.get_auth('http://www.test.com', 'asdf', 'asdf',
os_options=os_options,
auth_version="2.0")
self.assertTrue(url.startswith("http"))
self.assertTrue(token)
def test_auth_v2_no_tenant_name_or_tenant_id(self):
with mock.patch('swiftclient.client.get_auth_keystone',
fake_get_auth_keystone({})):
self.assertRaises(c.ClientException, c.get_auth,
'http://www.tests.com', 'asdf', 'asdf',
os_options={},
auth_version='2.0')
def test_auth_v2_with_tenant_name_none_and_tenant_id_none(self):
os_options = {'tenant_name': None,
'tenant_id': None}
with mock.patch('swiftclient.client.get_auth_keystone',
fake_get_auth_keystone(os_options)):
self.assertRaises(c.ClientException, c.get_auth,
'http://www.tests.com', 'asdf', 'asdf',
os_options=os_options,
auth_version='2.0')
def test_auth_v2_with_tenant_user_in_user(self):
tenant_option = {'tenant_name': 'foo'}
with mock.patch('swiftclient.client.get_auth_keystone',
fake_get_auth_keystone(tenant_option)):
url, token = c.get_auth('http://www.test.com', 'foo:bar', 'asdf',
os_options={},
auth_version="2.0")
self.assertTrue(url.startswith("http"))
self.assertTrue(token)
def test_auth_v2_tenant_name_no_os_options(self):
tenant_option = {'tenant_name': 'asdf'}
with mock.patch('swiftclient.client.get_auth_keystone',
fake_get_auth_keystone(tenant_option)):
url, token = c.get_auth('http://www.test.com', 'asdf', 'asdf',
tenant_name='asdf',
os_options={},
auth_version="2.0")
self.assertTrue(url.startswith("http"))
self.assertTrue(token)
def test_auth_v2_with_os_options(self):
os_options = {'service_type': 'object-store',
'endpoint_type': 'internalURL',
'tenant_name': 'asdf'}
with mock.patch('swiftclient.client.get_auth_keystone',
fake_get_auth_keystone(os_options)):
url, token = c.get_auth('http://www.test.com', 'asdf', 'asdf',
os_options=os_options,
auth_version="2.0")
self.assertTrue(url.startswith("http"))
self.assertTrue(token)
def test_auth_v2_with_tenant_user_in_user_no_os_options(self):
tenant_option = {'tenant_name': 'foo'}
with mock.patch('swiftclient.client.get_auth_keystone',
fake_get_auth_keystone(tenant_option)):
url, token = c.get_auth('http://www.test.com', 'foo:bar', 'asdf',
auth_version="2.0")
self.assertTrue(url.startswith("http"))
self.assertTrue(token)
def test_auth_v2_with_os_region_name(self):
os_options = {'region_name': 'good-region',
'tenant_name': 'asdf'}
with mock.patch('swiftclient.client.get_auth_keystone',
fake_get_auth_keystone(os_options)):
url, token = c.get_auth('http://www.test.com', 'asdf', 'asdf',
os_options=os_options,
auth_version="2.0")
self.assertTrue(url.startswith("http"))
self.assertTrue(token)
def test_auth_v2_no_endpoint(self):
os_options = {'region_name': 'unknown_region',
'tenant_name': 'asdf'}
with mock.patch('swiftclient.client.get_auth_keystone',
fake_get_auth_keystone(os_options, c.ClientException)):
self.assertRaises(c.ClientException, c.get_auth,
'http://www.tests.com', 'asdf', 'asdf',
os_options=os_options, auth_version='2.0')
def test_auth_v2_ks_exception(self):
with mock.patch('swiftclient.client.get_auth_keystone',
fake_get_auth_keystone({}, c.ClientException)):
self.assertRaises(c.ClientException, c.get_auth,
'http://www.tests.com', 'asdf', 'asdf',
os_options={},
auth_version='2.0')
def test_auth_v2_cacert(self):
os_options = {'tenant_name': 'foo'}
auth_url_secure = 'https://www.tests.com'
auth_url_insecure = 'https://www.tests.com/self-signed-certificate'
with mock.patch('swiftclient.client.get_auth_keystone',
fake_get_auth_keystone(os_options, None)):
url, token = c.get_auth(auth_url_secure, 'asdf', 'asdf',
os_options=os_options, auth_version='2.0',
insecure=False)
self.assertTrue(url.startswith("http"))
self.assertTrue(token)
url, token = c.get_auth(auth_url_insecure, 'asdf', 'asdf',
os_options=os_options, auth_version='2.0',
cacert='ca.pem', insecure=False)
self.assertTrue(url.startswith("http"))
self.assertTrue(token)
self.assertRaises(c.ClientException, c.get_auth,
auth_url_insecure, 'asdf', 'asdf',
os_options=os_options, auth_version='2.0')
self.assertRaises(c.ClientException, c.get_auth,
auth_url_insecure, 'asdf', 'asdf',
os_options=os_options, auth_version='2.0',
insecure=False)
def test_auth_v2_insecure(self):
os_options = {'tenant_name': 'foo'}
auth_url_secure = 'https://www.tests.com'
auth_url_insecure = 'https://www.tests.com/invalid-certificate'
with mock.patch('swiftclient.client.get_auth_keystone',
fake_get_auth_keystone(os_options, None)):
url, token = c.get_auth(auth_url_secure, 'asdf', 'asdf',
os_options=os_options, auth_version='2.0')
self.assertTrue(url.startswith("http"))
self.assertTrue(token)
url, token = c.get_auth(auth_url_insecure, 'asdf', 'asdf',
os_options=os_options, auth_version='2.0',
insecure=True)
self.assertTrue(url.startswith("http"))
self.assertTrue(token)
self.assertRaises(c.ClientException, c.get_auth,
auth_url_insecure, 'asdf', 'asdf',
os_options=os_options, auth_version='2.0')
self.assertRaises(c.ClientException, c.get_auth,
auth_url_insecure, 'asdf', 'asdf',
os_options=os_options, auth_version='2.0',
insecure=False)
def test_auth_v2_cert(self):
os_options = {'tenant_name': 'foo'}
auth_url_no_sslauth = 'https://www.tests.com'
auth_url_sslauth = 'https://www.tests.com/client-certificate'
with mock.patch('swiftclient.client.get_auth_keystone',
fake_get_auth_keystone(os_options, None)):
url, token = c.get_auth(auth_url_no_sslauth, 'asdf', 'asdf',
os_options=os_options, auth_version='2.0')
self.assertTrue(url.startswith("http"))
self.assertTrue(token)
url, token = c.get_auth(auth_url_sslauth, 'asdf', 'asdf',
os_options=os_options, auth_version='2.0',
cert='minnie', cert_key='mickey')
self.assertTrue(url.startswith("http"))
self.assertTrue(token)
self.assertRaises(c.ClientException, c.get_auth,
auth_url_sslauth, 'asdf', 'asdf',
os_options=os_options, auth_version='2.0')
self.assertRaises(c.ClientException, c.get_auth,
auth_url_sslauth, 'asdf', 'asdf',
os_options=os_options, auth_version='2.0',
cert='minnie')
def test_auth_v3_with_tenant_name(self):
# check the correct auth version is passed to get_auth_keystone
os_options = {'tenant_name': 'asdf'}
req_args = {'auth_version': '3'}
ks = fake_get_auth_keystone(os_options, required_kwargs=req_args)
with mock.patch('swiftclient.client.get_auth_keystone', ks):
url, token = c.get_auth('http://www.test.com', 'asdf', 'asdf',
os_options=os_options,
auth_version="3")
self.assertTrue(url.startswith("http"))
self.assertTrue(token)
def test_auth_v3applicationcredential(self):
from keystoneauth1 import exceptions as ksauthexceptions
os_options = {
"auth_type": "v3applicationcredential",
"application_credential_id": "proejct_id",
"application_credential_secret": "secret"}
class FakeEndpointData(object):
catalog_url = 'http://swift.cluster/v1/KEY_project_id'
class FakeKeystoneuth1v3Session(object):
def __init__(self, auth):
self.auth = auth
self.token = 'token'
def get_token(self):
if self.auth.auth_url == 'http://keystone:5000/v3':
return self.token
elif self.auth.auth_url == 'http://keystone:9000/v3':
raise ksauthexceptions.AuthorizationFailure
else:
raise ksauthexceptions.Unauthorized
def get_endpoint_data(self, service_type, endpoint_type, **kwargs):
return FakeEndpointData()
mock_sess = FakeKeystoneuth1v3Session
with mock.patch('keystoneauth1.session.Session', mock_sess):
url, token = c.get_auth('http://keystone:5000', '', '',
os_options=os_options,
auth_version="3")
self.assertTrue(url.startswith("http"))
self.assertEqual(url, 'http://swift.cluster/v1/KEY_project_id')
self.assertEqual(token, 'token')
with mock.patch('keystoneauth1.session.Session', mock_sess):
with self.assertRaises(c.ClientException) as exc_mgr:
url, token = c.get_auth('http://keystone:9000', '', '',
os_options=os_options,
auth_version="3")
body = 'Unauthorized. Check application credential id and secret.'
body = 'Authorization Failure. Cannot authorize API client.'
self.assertEqual(exc_mgr.exception.__str__()[-89:], body)
with mock.patch('keystoneauth1.session.Session', mock_sess):
with self.assertRaises(c.ClientException) as exc_mgr:
url, token = c.get_auth('http://keystone:5000', '', '',
os_options=os_options,
auth_version="2")
body = 'Unauthorized. Check application credential id and secret.'
self.assertEqual(exc_mgr.exception.__str__()[-89:], body)
def test_get_keystone_client_2_0(self):
# check the correct auth version is passed to get_auth_keystone
os_options = {'tenant_name': 'asdf'}
req_args = {'auth_version': '2.0'}
ks = fake_get_auth_keystone(os_options, required_kwargs=req_args)
with mock.patch('swiftclient.client.get_auth_keystone', ks):
url, token = c.get_keystoneclient_2_0('http://www.test.com',
'asdf', 'asdf',
os_options=os_options)
self.assertTrue(url.startswith("http"))
self.assertTrue(token)
def test_get_auth_keystone_versionless(self):
fake_ks = FakeKeystone(endpoint='http://some_url', token='secret')
with mock.patch('swiftclient.client.ksclient_v3', fake_ks):
c.get_auth_keystone('http://authurl', 'user', 'key', {})
self.assertEqual(1, len(fake_ks.calls))
self.assertEqual('http://authurl/v3', fake_ks.calls[0].get('auth_url'))
def test_get_auth_keystone_versionless_auth_version_set(self):
fake_ks = FakeKeystone(endpoint='http://some_url', token='secret')
with mock.patch('swiftclient.client.ksclient_v2', fake_ks):
c.get_auth_keystone('http://auth_url', 'user', 'key',
{}, auth_version='2.0')
self.assertEqual(1, len(fake_ks.calls))
self.assertEqual('http://auth_url/v2.0',
fake_ks.calls[0].get('auth_url'))
def test_get_auth_keystone_versionful(self):
fake_ks = FakeKeystone(endpoint='http://some_url', token='secret')
with mock.patch('swiftclient.client.ksclient_v3', fake_ks):
c.get_auth_keystone('http://auth_url/v3', 'user', 'key',
{}, auth_version='3')
self.assertEqual(1, len(fake_ks.calls))
self.assertEqual('http://auth_url/v3',
fake_ks.calls[0].get('auth_url'))
def test_get_auth_keystone_devstack_versionful(self):
fake_ks = FakeKeystone(
endpoint='http://storage.example.com/v1/AUTH_user', token='secret')
with mock.patch('swiftclient.client.ksclient_v3', fake_ks):
c.get_auth_keystone('https://192.168.8.8/identity/v3',
'user', 'key', {}, auth_version='3')
self.assertEqual(1, len(fake_ks.calls))
self.assertEqual('https://192.168.8.8/identity/v3',
fake_ks.calls[0].get('auth_url'))
def test_get_auth_keystone_devstack_versionless(self):
fake_ks = FakeKeystone(
endpoint='http://storage.example.com/v1/AUTH_user', token='secret')
with mock.patch('swiftclient.client.ksclient_v3', fake_ks):
c.get_auth_keystone('https://192.168.8.8/identity',
'user', 'key', {}, auth_version='3')
self.assertEqual(1, len(fake_ks.calls))
self.assertEqual('https://192.168.8.8/identity/v3',
fake_ks.calls[0].get('auth_url'))
def test_auth_keystone_url_some_junk_nonsense(self):
fake_ks = FakeKeystone(
endpoint='http://storage.example.com/v1/AUTH_user',
token='secret')
with mock.patch('swiftclient.client.ksclient_v3', fake_ks):
c.get_auth_keystone('http://blah.example.com/v2moo',
'user', 'key', {}, auth_version='3')
self.assertEqual(1, len(fake_ks.calls))
# v2 looks sorta version-y, but it's not an exact match, so this is
# probably about just as bad as anything else we might guess at
self.assertEqual('http://blah.example.com/v2moo/v3',
fake_ks.calls[0].get('auth_url'))
def test_auth_with_session(self):
mock_session = mock.MagicMock()
mock_session.get_endpoint.return_value = 'http://storagehost/v1/acct'
mock_session.get_token.return_value = 'token'
url, token = c.get_auth('http://www.test.com', 'asdf', 'asdf',
session=mock_session)
self.assertEqual(url, 'http://storagehost/v1/acct')
self.assertTrue(token)
class TestGetAccount(MockHttpTest):
def test_no_content(self):
c.http_connection = self.fake_http_connection(204)
value = c.get_account('http://www.test.com/v1/acct', 'asdf')[1]
self.assertEqual(value, [])
self.assertRequests([
('GET', '/v1/acct?format=json', '', {
'accept-encoding': 'gzip',
'x-auth-token': 'asdf'}),
])
def test_param_marker(self):
c.http_connection = self.fake_http_connection(
204,
query_string="format=json&marker=marker")
c.get_account('http://www.test.com/v1/acct', 'asdf', marker='marker')
self.assertRequests([
('GET', '/v1/acct?format=json&marker=marker', '', {
'accept-encoding': 'gzip',
'x-auth-token': 'asdf'}),
])
def test_param_limit(self):
c.http_connection = self.fake_http_connection(
204,
query_string="format=json&limit=10")
c.get_account('http://www.test.com/v1/acct', 'asdf', limit=10)
self.assertRequests([
('GET', '/v1/acct?format=json&limit=10', '', {
'accept-encoding': 'gzip',
'x-auth-token': 'asdf'}),
])
def test_param_prefix(self):
c.http_connection = self.fake_http_connection(
204,
query_string="format=json&prefix=asdf/")
c.get_account('http://www.test.com/v1/acct', 'asdf', prefix='asdf/')
self.assertRequests([
('GET', '/v1/acct?format=json&prefix=asdf/', '', {
'accept-encoding': 'gzip',
'x-auth-token': 'asdf'}),
])
def test_param_end_marker(self):
c.http_connection = self.fake_http_connection(
204,
query_string="format=json&end_marker=end_marker")
c.get_account('http://www.test.com/v1/acct', 'asdf',
end_marker='end_marker')
self.assertRequests([
('GET', '/v1/acct?format=json&end_marker=end_marker', '', {
'accept-encoding': 'gzip',
'x-auth-token': 'asdf'}),
])
def test_param_delimiter(self):
c.http_connection = self.fake_http_connection(
204,
query_string="format=json&delimiter=-")
c.get_account('http://www.test.com/v1/acct', 'asdf',
delimiter='-')
self.assertRequests([
('GET', '/v1/acct?format=json&delimiter=-', '', {
'accept-encoding': 'gzip',
'x-auth-token': 'asdf'}),
])
class TestHeadAccount(MockHttpTest):
def test_ok(self):
c.http_connection = self.fake_http_connection(200, headers={
'x-account-meta-color': 'blue',
})
resp_headers = c.head_account('http://www.tests.com', 'asdf')
self.assertEqual(resp_headers['x-account-meta-color'], 'blue')
self.assertRequests([
('HEAD', 'http://www.tests.com', '', {'x-auth-token': 'asdf'})
])
self.assertTrue(self.request_log[-1][-1]._closed)
def test_server_error(self):
body = 'c' * 65
headers = {'foo': 'bar'}
c.http_connection = self.fake_http_connection(
StubResponse(500, body, headers))
with self.assertRaises(c.ClientException) as exc_context:
c.head_account('http://www.tests.com', 'asdf')
e = exc_context.exception
self.assertEqual(e.http_response_content, body)
self.assertEqual(e.http_status, 500)
self.assertRequests([
('HEAD', 'http://www.tests.com', '', {'x-auth-token': 'asdf'})
])
# TODO: this is a fairly brittle test of the __repr__ on the
# ClientException which should probably be in a targeted test
new_body = "[first 60 chars of response] " + body[0:60]
self.assertEqual(e.__str__()[-89:], new_body)
class TestPostAccount(MockHttpTest):
def test_ok(self):
c.http_connection = self.fake_http_connection(200, headers={
'X-Account-Meta-Color': 'blue',
}, body='foo')
headers = {'x-account-meta-shape': 'square'}
resp_headers, body = c.post_account(
'http://www.tests.com/path/to/account', 'asdf',
headers, query_string='bar=baz',
data='some data')
self.assertEqual('blue', resp_headers.get('x-account-meta-color'))
self.assertEqual('foo', body)
self.assertRequests([
('POST', 'http://www.tests.com/path/to/account?bar=baz',
'some data', {'x-auth-token': 'asdf',
'x-account-meta-shape': 'square'})
])
# Check that we didn't mutate the request ehader dict
self.assertEqual(headers, {'x-account-meta-shape': 'square'})
def test_server_error(self):
body = 'c' * 65
c.http_connection = self.fake_http_connection(500, body=body)
with self.assertRaises(c.ClientException) as exc_mgr:
c.post_account('http://www.tests.com', 'asdf', {})
self.assertEqual(exc_mgr.exception.http_response_content, body)
self.assertEqual(exc_mgr.exception.http_status, 500)
self.assertRequests([
('POST', 'http://www.tests.com', None, {'x-auth-token': 'asdf'})
])
# TODO: this is a fairly brittle test of the __repr__ on the
# ClientException which should probably be in a targeted test
new_body = "[first 60 chars of response] " + body[0:60]
self.assertEqual(exc_mgr.exception.__str__()[-89:], new_body)
class TestGetContainer(MockHttpTest):
def test_no_content(self):
c.http_connection = self.fake_http_connection(204)
value = c.get_container('http://www.test.com/v1/acct', 'token',
'container')[1]
self.assertEqual(value, [])
self.assertRequests([
('GET', '/v1/acct/container?format=json', '', {
'accept-encoding': 'gzip',
'x-auth-token': 'token'}),
])
def test_param_marker(self):
c.http_connection = self.fake_http_connection(
204,
query_string="format=json&marker=marker")
c.get_container('http://www.test.com/v1/acct', 'token', 'container',
marker='marker')
self.assertRequests([
('GET', '/v1/acct/container?format=json&marker=marker', '', {
'accept-encoding': 'gzip',
'x-auth-token': 'token'}),
])
def test_param_limit(self):
c.http_connection = self.fake_http_connection(
204,
query_string="format=json&limit=10")
c.get_container('http://www.test.com/v1/acct', 'token', 'container',
limit=10)
self.assertRequests([
('GET', '/v1/acct/container?format=json&limit=10', '', {
'accept-encoding': 'gzip',
'x-auth-token': 'token'}),
])
def test_param_prefix(self):
c.http_connection = self.fake_http_connection(
204,
query_string="format=json&prefix=asdf/")
c.get_container('http://www.test.com/v1/acct', 'token', 'container',
prefix='asdf/')
self.assertRequests([
('GET', '/v1/acct/container?format=json&prefix=asdf/', '', {
'accept-encoding': 'gzip',
'x-auth-token': 'token'}),
])
def test_param_delimiter(self):
c.http_connection = self.fake_http_connection(
204,
query_string="format=json&delimiter=/")
c.get_container('http://www.test.com/v1/acct', 'token', 'container',
delimiter='/')
self.assertRequests([
('GET', '/v1/acct/container?format=json&delimiter=/', '', {
'accept-encoding': 'gzip',
'x-auth-token': 'token'}),
])
def test_param_end_marker(self):
c.http_connection = self.fake_http_connection(
204,
query_string="format=json&end_marker=end_marker")
c.get_container('http://www.test.com/v1/acct', 'token', 'container',
end_marker='end_marker')
self.assertRequests([
('GET', '/v1/acct/container?format=json&end_marker=end_marker',
'', {'x-auth-token': 'token', 'accept-encoding': 'gzip'}),
])
def test_param_path(self):
c.http_connection = self.fake_http_connection(
204,
query_string="format=json&path=asdf")
c.get_container('http://www.test.com/v1/acct', 'token', 'container',
path='asdf')
self.assertRequests([
('GET', '/v1/acct/container?format=json&path=asdf', '', {
'accept-encoding': 'gzip',
'x-auth-token': 'token'}),
])
def test_request_headers(self):
c.http_connection = self.fake_http_connection(
204, query_string="format=json")
conn = c.http_connection('http://www.test.com')
headers = {'x-client-key': 'client key'}
c.get_container('url_is_irrelevant', 'TOKEN', 'container',
http_conn=conn, headers=headers)
self.assertRequests([
('GET', '/container?format=json', '', {
'x-auth-token': 'TOKEN',
'x-client-key': 'client key',
'accept-encoding': 'gzip',
}),
])
def test_query_string(self):
c.http_connection = self.fake_http_connection(
200, query_string="format=json&hello=20", body=b'[]')
c.get_container('http://www.test.com', 'asdf', 'asdf',
query_string="hello=20")
self.assertRequests([
('GET', '/asdf?format=json&hello=20', '', {
'accept-encoding': 'gzip',
'x-auth-token': 'asdf'}),
])
class TestHeadContainer(MockHttpTest):
def test_head_ok(self):
fake_conn = self.fake_http_connection(
200, headers={'x-container-meta-color': 'blue'})
with mock.patch('swiftclient.client.http_connection',
new=fake_conn):
resp = c.head_container('https://example.com/v1/AUTH_test',
'token', 'container')
self.assertEqual(resp['x-container-meta-color'], 'blue')
self.assertRequests([
('HEAD', 'https://example.com/v1/AUTH_test/container', '',
{'x-auth-token': 'token'}),
])
def test_server_error(self):
body = 'c' * 60
headers = {'foo': 'bar'}
c.http_connection = self.fake_http_connection(
StubResponse(500, body, headers))
with self.assertRaises(c.ClientException) as exc_context:
c.head_container('http://www.test.com', 'asdf', 'container')
e = exc_context.exception
self.assertRequests([
('HEAD', '/container', '', {'x-auth-token': 'asdf'}),
])
self.assertEqual(e.http_status, 500)
self.assertEqual(e.http_response_content, body)
self.assertEqual(e.http_response_headers, headers)
class TestPutContainer(MockHttpTest):
def test_ok(self):
c.http_connection = self.fake_http_connection(200)
value = c.put_container('http://www.test.com', 'token', 'container')
self.assertIsNone(value)
self.assertRequests([
('PUT', '/container', '', {
'x-auth-token': 'token',
'content-length': '0'}),
])
def test_server_error(self):
body = 'c' * 60
headers = {'foo': 'bar'}
c.http_connection = self.fake_http_connection(
StubResponse(500, body, headers))
with self.assertRaises(c.ClientException) as exc_context:
c.put_container('http://www.test.com', 'token', 'container')
self.assertEqual(exc_context.exception.http_response_content, body)
self.assertEqual(exc_context.exception.http_response_headers, headers)
self.assertRequests([
('PUT', '/container', '', {
'x-auth-token': 'token',
'content-length': '0'}),
])
def test_query_string(self):
c.http_connection = self.fake_http_connection(200,
query_string="hello=20")
c.put_container('http://www.test.com', 'asdf', 'asdf',
query_string="hello=20")
for req in self.iter_request_log():
self.assertEqual(req['method'], 'PUT')
self.assertEqual(req['parsed_path'].path, '/asdf')
self.assertEqual(req['parsed_path'].query, 'hello=20')
self.assertEqual(req['headers']['x-auth-token'], 'asdf')
class TestDeleteContainer(MockHttpTest):
def test_ok(self):
c.http_connection = self.fake_http_connection(200)
value = c.delete_container('http://www.test.com', 'token', 'container')
self.assertIsNone(value)
self.assertRequests([
('DELETE', '/container', '', {
'x-auth-token': 'token'}),
])
def test_query_string(self):
c.http_connection = self.fake_http_connection(200,
query_string="hello=20")
c.delete_container('http://www.test.com', 'token', 'container',
query_string="hello=20")
self.assertRequests([
('DELETE', 'http://www.test.com/container?hello=20', '', {
'x-auth-token': 'token'})
])
class TestGetObject(MockHttpTest):
def test_server_error(self):
body = 'c' * 60
headers = {'foo': 'bar'}
c.http_connection = self.fake_http_connection(
StubResponse(500, body, headers))
with self.assertRaises(c.ClientException) as exc_context:
c.get_object('http://www.test.com', 'asdf', 'asdf', 'asdf')
self.assertEqual(exc_context.exception.http_response_content, body)
self.assertEqual(exc_context.exception.http_response_headers, headers)
def test_query_string(self):
c.http_connection = self.fake_http_connection(200,
query_string="hello=20")
c.get_object('http://www.test.com', 'asdf', 'asdf', 'asdf',
query_string="hello=20")
self.assertRequests([
('GET', '/asdf/asdf?hello=20', '', {
'x-auth-token': 'asdf'}),
])
def test_get_object_as_string(self):
c.http_connection = self.fake_http_connection(200, body='abcde')
__, resp = c.get_object('http://storage.example.com', 'TOKEN',
'container_name', 'object_name')
self.assertEqual(resp, 'abcde')
def test_request_headers(self):
c.http_connection = self.fake_http_connection(200)
conn = c.http_connection('http://www.test.com')
headers = {'Range': 'bytes=1-2'}
c.get_object('url_is_irrelevant', 'TOKEN', 'container', 'object',
http_conn=conn, headers=headers)
self.assertRequests([
('GET', '/container/object', '', {
'x-auth-token': 'TOKEN',
'range': 'bytes=1-2',
}),
])
def test_response_headers(self):
c.http_connection = self.fake_http_connection(
200, headers={'X-Utf-8-Header': b't%c3%a9st',
'X-Non-Utf-8-Header': b'%ff',
'X-Binary-Header': b'\xff'})
conn = c.http_connection('http://www.test.com')
headers, data = c.get_object('url_is_irrelevant', 'TOKEN',
'container', 'object', http_conn=conn)
self.assertEqual(u't\xe9st', headers.get('x-utf-8-header', ''))
self.assertEqual(u'%ff', headers.get('x-non-utf-8-header', ''))
self.assertEqual(u'%FF', headers.get('x-binary-header', ''))
def test_chunk_size_read_method(self):
conn = c.Connection('http://auth.url/', 'some_user', 'some_key')
with mock.patch('swiftclient.client.get_auth_1_0') as mock_get_auth:
mock_get_auth.return_value = ('http://auth.url/', 'tToken')
c.http_connection = self.fake_http_connection(200, body='abcde')
__, resp = conn.get_object('asdf', 'asdf', resp_chunk_size=3)
self.assertTrue(hasattr(resp, 'read'))
self.assertEqual(resp.read(3), 'abc')
self.assertEqual(resp.read(None), 'de')
self.assertEqual(resp.read(), '')
def test_chunk_size_iter(self):
conn = c.Connection('http://auth.url/', 'some_user', 'some_key')
with mock.patch('swiftclient.client.get_auth_1_0') as mock_get_auth:
mock_get_auth.return_value = ('http://auth.url/', 'tToken')
c.http_connection = self.fake_http_connection(200, body='abcde')
__, resp = conn.get_object('asdf', 'asdf', resp_chunk_size=3)
self.assertTrue(hasattr(resp, 'next'))
self.assertEqual(next(resp), 'abc')
self.assertEqual(next(resp), 'de')
self.assertRaises(StopIteration, next, resp)
def test_chunk_size_read_and_iter(self):
conn = c.Connection('http://auth.url/', 'some_user', 'some_key')
with mock.patch('swiftclient.client.get_auth_1_0') as mock_get_auth:
mock_get_auth.return_value = ('http://auth.url/', 'tToken')
c.http_connection = self.fake_http_connection(200, body='abcdef')
__, resp = conn.get_object('asdf', 'asdf', resp_chunk_size=2)
self.assertTrue(hasattr(resp, 'read'))
self.assertEqual(resp.read(3), 'abc')
self.assertEqual(next(resp), 'de')
self.assertEqual(resp.read(), 'f')
self.assertRaises(StopIteration, next, resp)
self.assertEqual(resp.read(), '')
def test_chunk_size_iter_chunked_no_retry(self):
conn = c.Connection('http://auth.url/', 'some_user', 'some_key')
with mock.patch('swiftclient.client.get_auth_1_0') as mock_get_auth:
mock_get_auth.return_value = ('http://auth.url/', 'tToken')
c.http_connection = self.fake_http_connection(
200, body='abcdef', headers={'Transfer-Encoding': 'chunked'})
__, resp = conn.get_object('asdf', 'asdf', resp_chunk_size=2)
self.assertEqual(next(resp), 'ab')
# simulate a dropped connection
resp.resp.read()
self.assertRaises(StopIteration, next, resp)
def test_chunk_size_iter_retry(self):
conn = c.Connection('http://auth.url/', 'some_user', 'some_key')
with mock.patch('swiftclient.client.get_auth_1_0') as mock_get_auth:
mock_get_auth.return_value = ('http://auth.url', 'tToken')
c.http_connection = self.fake_http_connection(
StubResponse(200, 'abcdef', {'etag': 'some etag',
'content-length': '6'}),
StubResponse(206, 'cdef', {'etag': 'some etag',
'content-length': '4',
'content-range': 'bytes 2-5/6'}),
StubResponse(206, 'ef', {'etag': 'some etag',
'content-length': '2',
'content-range': 'bytes 4-5/6'}),
)
__, resp = conn.get_object('asdf', 'asdf', resp_chunk_size=2)
self.assertEqual(next(resp), 'ab')
self.assertEqual(1, conn.attempts)
# simulate a dropped connection
resp.resp.read()
self.assertEqual(next(resp), 'cd')
self.assertEqual(2, conn.attempts)
# simulate a dropped connection
resp.resp.read()
self.assertEqual(next(resp), 'ef')
self.assertEqual(3, conn.attempts)
self.assertRaises(StopIteration, next, resp)
self.assertRequests([
('GET', '/asdf/asdf', '', {
'x-auth-token': 'tToken',
}),
('GET', '/asdf/asdf', '', {
'range': 'bytes=2-',
'if-match': 'some etag',
'x-auth-token': 'tToken',
}),
('GET', '/asdf/asdf', '', {
'range': 'bytes=4-',
'if-match': 'some etag',
'x-auth-token': 'tToken',
}),
])
def test_chunk_size_iter_retry_no_range_support(self):
conn = c.Connection('http://auth.url/', 'some_user', 'some_key')
with mock.patch('swiftclient.client.get_auth_1_0') as mock_get_auth:
mock_get_auth.return_value = ('http://auth.url', 'tToken')
c.http_connection = self.fake_http_connection(*[
StubResponse(200, 'abcdef', {'etag': 'some etag',
'content-length': '6'})
] * 3)
__, resp = conn.get_object('asdf', 'asdf', resp_chunk_size=2)
self.assertEqual(next(resp), 'ab')
self.assertEqual(1, conn.attempts)
# simulate a dropped connection
resp.resp.read()
self.assertEqual(next(resp), 'cd')
self.assertEqual(2, conn.attempts)
# simulate a dropped connection
resp.resp.read()
self.assertEqual(next(resp), 'ef')
self.assertEqual(3, conn.attempts)
self.assertRaises(StopIteration, next, resp)
self.assertRequests([
('GET', '/asdf/asdf', '', {
'x-auth-token': 'tToken',
}),
('GET', '/asdf/asdf', '', {
'range': 'bytes=2-',
'if-match': 'some etag',
'x-auth-token': 'tToken',
}),
('GET', '/asdf/asdf', '', {
'range': 'bytes=4-',
'if-match': 'some etag',
'x-auth-token': 'tToken',
}),
])
def test_chunk_size_iter_retry_bad_range_response(self):
conn = c.Connection('http://auth.url/', 'some_user', 'some_key')
with mock.patch('swiftclient.client.get_auth_1_0') as mock_get_auth:
mock_get_auth.return_value = ('http://auth.url', 'tToken')
c.http_connection = self.fake_http_connection(
StubResponse(200, 'abcdef', {'etag': 'some etag',
'content-length': '6'}),
StubResponse(206, 'abcdef', {'etag': 'some etag',
'content-length': '6',
'content-range': 'chunk 1-2/3'})
)
__, resp = conn.get_object('asdf', 'asdf', resp_chunk_size=2)
self.assertEqual(next(resp), 'ab')
self.assertEqual(1, conn.attempts)
# simulate a dropped connection
resp.resp.read()
self.assertRaises(c.ClientException, next, resp)
self.assertRequests([
('GET', '/asdf/asdf', '', {
'x-auth-token': 'tToken',
}),
('GET', '/asdf/asdf', '', {
'range': 'bytes=2-',
'if-match': 'some etag',
'x-auth-token': 'tToken',
}),
])
def test_get_object_with_resp_chunk_size_zero(self):
def get_connection(self):
def get_auth():
return 'http://auth.test.com', 'token'
conn = c.Connection('http://www.test.com', 'asdf', 'asdf')
self.assertIs(type(conn), c.Connection)
conn.get_auth = get_auth
self.assertEqual(conn.attempts, 0)
return conn
with mock.patch('swiftclient.client.http_connection',
self.fake_http_connection(200)):
conn = get_connection(self)
conn.get_object('container1', 'obj1', resp_chunk_size=0)
self.assertEqual(conn.attempts, 1)
class TestHeadObject(MockHttpTest):
def test_server_error(self):
body = 'c' * 60
headers = {'foo': 'bar'}
c.http_connection = self.fake_http_connection(
StubResponse(500, body, headers))
with self.assertRaises(c.ClientException) as exc_context:
c.head_object('http://www.test.com', 'asdf', 'asdf', 'asdf')
self.assertEqual(exc_context.exception.http_response_content, body)
self.assertEqual(exc_context.exception.http_response_headers, headers)
def test_request_headers(self):
c.http_connection = self.fake_http_connection(204)
conn = c.http_connection('http://www.test.com')
headers = {'x-client-key': 'client key'}
c.head_object('url_is_irrelevant', 'TOKEN', 'container',
'asdf', http_conn=conn, headers=headers)
self.assertRequests([
('HEAD', '/container/asdf', '', {
'x-auth-token': 'TOKEN',
'x-client-key': 'client key',
}),
])
def test_query_string(self):
c.http_connection = self.fake_http_connection(204)
conn = c.http_connection('http://www.test.com')
query_string = 'foo=bar'
c.head_object('url_is_irrelevant', 'token', 'container', 'key',
http_conn=conn, query_string=query_string)
self.assertRequests([
('HEAD', '/container/key?foo=bar', '', {'x-auth-token': 'token'})
])
class TestPutObject(MockHttpTest):
@mock.patch('swiftclient.requests.__version__', '2.2.0')
def test_ok(self):
c.http_connection = self.fake_http_connection(200)
args = ('http://www.test.com', 'TOKEN', 'container', 'obj', 'body', 4)
value = c.put_object(*args)
self.assertIsInstance(value, six.string_types)
self.assertEqual(value, EMPTY_ETAG)
self.assertRequests([
('PUT', '/container/obj', 'body', {
'x-auth-token': 'TOKEN',
'content-length': '4',
'content-type': ''
}),
])
def test_unicode_ok(self):
conn = c.http_connection(u'http://www.test.com/')
mock_file = six.StringIO(u'\u5929\u7a7a\u4e2d\u7684\u4e4c\u4e91')
args = (u'\u5929\u7a7a\u4e2d\u7684\u4e4c\u4e91',
u'\u5929\u7a7a\u4e2d\u7684\u4e4c\u4e91',
u'\u5929\u7a7a\u4e2d\u7684\u4e4c\u4e91',
u'\u5929\u7a7a\u4e2d\u7684\u4e4c\u4e91',
mock_file)
text = u'\u5929\u7a7a\u4e2d\u7684\u4e4c\u4e91'
headers = {'X-Header1': text,
'X-2': '1', 'X-3': "{'a': 'b'}", 'a-b': '.x:yz mn:fg:lp'}
resp = MockHttpResponse()
conn[1].getresponse = resp.fake_response
conn[1]._request = resp._fake_request
value = c.put_object(*args, headers=headers, http_conn=conn)
self.assertIsInstance(value, six.string_types)
# Test for RFC-2616 encoded symbols
self.assertIn(("a-b", b".x:yz mn:fg:lp"),
resp.buffer)
# Test unicode header
self.assertIn(('x-header1', text.encode('utf8')),
resp.buffer)
def test_chunk_warning(self):
conn = c.http_connection('http://www.test.com/')
mock_file = six.StringIO('asdf')
args = ('asdf', 'asdf', 'asdf', 'asdf', mock_file)
resp = MockHttpResponse()
conn[1].getresponse = resp.fake_response
conn[1]._request = resp._fake_request
with warnings.catch_warnings(record=True) as w:
c.put_object(*args, chunk_size=20, headers={}, http_conn=conn)
self.assertEqual(len(w), 0)
body = 'c' * 60
c.http_connection = self.fake_http_connection(200, body=body)
args = ('http://www.test.com', 'asdf', 'asdf', 'asdf', 'asdf')
with warnings.catch_warnings(record=True) as w:
c.put_object(*args, chunk_size=20)
self.assertEqual(len(w), 1)
self.assertTrue(issubclass(w[-1].category, UserWarning))
@mock.patch('swiftclient.requests.__version__', '2.2.0')
def test_server_error(self):
body = 'c' * 60
headers = {'foo': 'bar'}
c.http_connection = self.fake_http_connection(
StubResponse(500, body, headers))
args = ('http://www.test.com', 'asdf', 'asdf', 'asdf', 'asdf')
with self.assertRaises(c.ClientException) as exc_context:
c.put_object(*args)
e = exc_context.exception
self.assertEqual(e.http_response_content, body)
self.assertEqual(e.http_response_headers, headers)
self.assertEqual(e.http_status, 500)
self.assertRequests([
('PUT', '/asdf/asdf', 'asdf', {
'x-auth-token': 'asdf',
'content-type': ''}),
])
def test_query_string(self):
c.http_connection = self.fake_http_connection(200,
query_string="hello=20")
c.put_object('http://www.test.com', 'asdf', 'asdf', 'asdf',
query_string="hello=20")
for req in self.iter_request_log():
self.assertEqual(req['method'], 'PUT')
self.assertEqual(req['parsed_path'].path, '/asdf/asdf')
self.assertEqual(req['parsed_path'].query, 'hello=20')
self.assertEqual(req['headers']['x-auth-token'], 'asdf')
def test_raw_upload(self):
# Raw upload happens when content_length is passed to put_object
conn = c.http_connection(u'http://www.test.com/')
resp = MockHttpResponse(status=200)
conn[1].getresponse = resp.fake_response
conn[1]._request = resp._fake_request
raw_data = b'asdf' * 256
raw_data_len = len(raw_data)
for kwarg in ({'headers': {'Content-Length': str(raw_data_len)}},
{'content_length': raw_data_len}):
with tempfile.TemporaryFile() as mock_file:
mock_file.write(raw_data)
mock_file.seek(0)
c.put_object(url='http://www.test.com', http_conn=conn,
contents=mock_file, **kwarg)
req_data = resp.requests_params['data']
self.assertIs(type(req_data), swiftclient.utils.LengthWrapper)
self.assertEqual(raw_data_len, len(req_data.read()))
def test_chunk_upload(self):
# Chunked upload happens when no content_length is passed to put_object
conn = c.http_connection(u'http://www.test.com/')
resp = MockHttpResponse(status=200)
conn[1].getresponse = resp.fake_response
conn[1]._request = resp._fake_request
raw_data = b'asdf' * 256
chunk_size = 16
with tempfile.TemporaryFile() as mock_file:
mock_file.write(raw_data)
mock_file.seek(0)
c.put_object(url='http://www.test.com', http_conn=conn,
contents=mock_file, chunk_size=chunk_size)
req_data = resp.requests_params['data']
self.assertTrue(hasattr(req_data, '__iter__'))
data = b''
for chunk in req_data:
self.assertEqual(chunk_size, len(chunk))
data += chunk
self.assertEqual(data, raw_data)
def test_iter_upload(self):
def data():
for chunk in ('foo', '', 'bar'):
yield chunk
conn = c.http_connection(u'http://www.test.com/')
resp = MockHttpResponse(status=200)
conn[1].getresponse = resp.fake_response
conn[1]._request = resp._fake_request
c.put_object(url='http://www.test.com', http_conn=conn,
contents=data())
req_headers = resp.requests_params['headers']
self.assertNotIn('Content-Length', req_headers)
req_data = resp.requests_params['data']
self.assertTrue(hasattr(req_data, '__iter__'))
# If we emit an empty chunk, requests will go ahead and send it,
# causing the server to close the connection. So make sure we don't
# do that.
self.assertEqual(['foo', 'bar'], list(req_data))
def test_md5_mismatch(self):
conn = c.http_connection('http://www.test.com')
resp = MockHttpResponse(status=200, verify=True,
headers={'etag': '"badresponseetag"'})
conn[1].getresponse = resp.fake_response
conn[1]._request = resp._fake_request
raw_data = b'asdf' * 256
raw_data_md5 = md5(raw_data).hexdigest()
chunk_size = 16
with tempfile.TemporaryFile() as mock_file:
mock_file.write(raw_data)
mock_file.seek(0)
contents = swiftclient.utils.ReadableToIterable(mock_file,
md5=True)
etag = c.put_object(url='http://www.test.com',
http_conn=conn,
contents=contents,
chunk_size=chunk_size)
self.assertNotEqual(etag, contents.get_md5sum())
self.assertEqual(etag, 'badresponseetag')
self.assertEqual(raw_data_md5, contents.get_md5sum())
def test_md5_match(self):
conn = c.http_connection('http://www.test.com')
raw_data = b'asdf' * 256
raw_data_md5 = md5(raw_data).hexdigest()
resp = MockHttpResponse(status=200, verify=True,
headers={'etag': '"' + raw_data_md5 + '"'})
conn[1].getresponse = resp.fake_response
conn[1]._request = resp._fake_request
chunk_size = 16
with tempfile.TemporaryFile() as mock_file:
mock_file.write(raw_data)
mock_file.seek(0)
contents = swiftclient.utils.ReadableToIterable(mock_file,
md5=True)
etag = c.put_object(url='http://www.test.com',
http_conn=conn,
contents=contents,
chunk_size=chunk_size)
self.assertEqual(raw_data_md5, contents.get_md5sum())
self.assertEqual(etag, contents.get_md5sum())
def test_params(self):
conn = c.http_connection(u'http://www.test.com/')
resp = MockHttpResponse(status=200)
conn[1].getresponse = resp.fake_response
conn[1]._request = resp._fake_request
c.put_object(url='http://www.test.com', http_conn=conn,
etag='1234-5678', content_type='text/plain')
request_header = resp.requests_params['headers']
self.assertEqual(request_header['etag'], b'1234-5678')
self.assertEqual(request_header['content-type'], b'text/plain')
@mock.patch('swiftclient.requests.__version__', '2.2.0')
def test_no_content_type_old_requests(self):
conn = c.http_connection(u'http://www.test.com/')
resp = MockHttpResponse(status=200)
conn[1].getresponse = resp.fake_response
conn[1]._request = resp._fake_request
c.put_object(url='http://www.test.com', http_conn=conn)
request_header = resp.requests_params['headers']
self.assertEqual(request_header['content-type'], b'')
@mock.patch('swiftclient.requests.__version__', '2.4.0')
def test_no_content_type_new_requests(self):
conn = c.http_connection(u'http://www.test.com/')
resp = MockHttpResponse(status=200)
conn[1].getresponse = resp.fake_response
conn[1]._request = resp._fake_request
c.put_object(url='http://www.test.com', http_conn=conn)
request_header = resp.requests_params['headers']
self.assertNotIn('content-type', request_header)
def test_content_type_in_headers(self):
conn = c.http_connection(u'http://www.test.com/')
resp = MockHttpResponse(status=200)
conn[1].getresponse = resp.fake_response
conn[1]._request = resp._fake_request
# title-case header
hdrs = {'Content-Type': 'text/Plain'}
c.put_object(url='http://www.test.com', http_conn=conn, headers=hdrs)
request_header = resp.requests_params['headers']
self.assertEqual(request_header['content-type'], b'text/Plain')
# method param overrides headers
c.put_object(url='http://www.test.com', http_conn=conn, headers=hdrs,
content_type='image/jpeg')
request_header = resp.requests_params['headers']
self.assertEqual(request_header['content-type'], b'image/jpeg')
class TestPostObject(MockHttpTest):
def test_ok(self):
c.http_connection = self.fake_http_connection(200)
delete_at = 2.1 # not str! we don't know what other devs will use!
args = ('http://www.test.com', 'token', 'container', 'obj',
{'X-Object-Meta-Test': 'mymeta',
'X-Delete-At': delete_at})
c.post_object(*args)
self.assertRequests([
('POST', '/container/obj', '', {
'x-auth-token': 'token',
'X-Object-Meta-Test': 'mymeta',
'X-Delete-At': delete_at}),
])
# Check that the request header dict didn't get mutated
self.assertEqual(args[-1], {
'X-Object-Meta-Test': 'mymeta',
'X-Delete-At': delete_at,
})
def test_unicode_ok(self):
conn = c.http_connection(u'http://www.test.com/')
args = (u'\u5929\u7a7a\u4e2d\u7684\u4e4c\u4e91',
u'\u5929\u7a7a\u4e2d\u7684\u4e4c\u4e91',
u'\u5929\u7a7a\u4e2d\u7684\u4e4c\u4e91',
u'\u5929\u7a7a\u4e2d\u7684\u4e4c\u4e91')
text = u'\u5929\u7a7a\u4e2d\u7684\u4e4c\u4e91'
headers = {'X-Header1': text,
b'X-Header2': 'value',
'X-2': '1', 'X-3': "{'a': 'b'}", 'a-b': '.x:yz mn:kl:qr',
'X-Object-Meta-Header-not-encoded': text,
b'X-Object-Meta-Header-encoded': 'value'}
resp = MockHttpResponse()
conn[1].getresponse = resp.fake_response
conn[1]._request = resp._fake_request
c.post_object(*args, headers=headers, http_conn=conn)
# Test for RFC-2616 encoded symbols
self.assertIn(('a-b', b".x:yz mn:kl:qr"), resp.buffer)
# Test unicode header
self.assertIn(('x-header1', text.encode('utf8')),
resp.buffer)
self.assertIn((b'x-object-meta-header-not-encoded',
text.encode('utf8')), resp.buffer)
self.assertIn((b'x-object-meta-header-encoded', b'value'),
resp.buffer)
self.assertIn((b'x-header2', b'value'), resp.buffer)
def test_server_error(self):
body = 'c' * 60
headers = {'foo': 'bar'}
c.http_connection = self.fake_http_connection(
StubResponse(500, body, headers))
args = ('http://www.test.com', 'token', 'container', 'obj', {})
with self.assertRaises(c.ClientException) as exc_context:
c.post_object(*args)
self.assertEqual(exc_context.exception.http_response_content, body)
self.assertEqual(exc_context.exception.http_response_headers, headers)
self.assertRequests([
('POST', 'http://www.test.com/container/obj', '', {
'x-auth-token': 'token',
}),
])
class TestCopyObject(MockHttpTest):
def test_server_error(self):
c.http_connection = self.fake_http_connection(500)
self.assertRaises(
c.ClientException, c.copy_object,
'http://www.test.com/v1/AUTH', 'asdf', 'asdf', 'asdf')
def test_ok(self):
c.http_connection = self.fake_http_connection(200)
c.copy_object(
'http://www.test.com/v1/AUTH', 'token', 'container', 'obj',
destination='/container2/obj')
self.assertRequests([
('COPY', 'http://www.test.com/v1/AUTH/container/obj', '', {
'X-Auth-Token': 'token',
'Destination': '/container2/obj',
}),
])
def test_service_token(self):
c.http_connection = self.fake_http_connection(200)
c.copy_object('http://www.test.com/v1/AUTH', None, 'container',
'obj', destination='/container2/obj',
service_token="TOKEN")
self.assertRequests([
('COPY', 'http://www.test.com/v1/AUTH/container/obj', '', {
'X-Service-Token': 'TOKEN',
'Destination': '/container2/obj',
}),
])
def test_headers(self):
c.http_connection = self.fake_http_connection(200)
c.copy_object(
'http://www.test.com/v1/AUTH', 'token', 'container', 'obj',
destination='/container2/obj',
headers={'some-hdr': 'a', 'other-hdr': 'b'})
self.assertRequests([
('COPY', 'http://www.test.com/v1/AUTH/container/obj', '', {
'X-Auth-Token': 'token',
'Destination': '/container2/obj',
'some-hdr': 'a',
'other-hdr': 'b',
}),
])
def test_fresh_metadata_default(self):
c.http_connection = self.fake_http_connection(200)
c.copy_object(
'http://www.test.com/v1/AUTH', 'token', 'container', 'obj',
'/container2/obj', {'x-fresh-metadata': 'hdr-value'})
self.assertRequests([
('COPY', 'http://www.test.com/v1/AUTH/container/obj', '', {
'X-Auth-Token': 'token',
'Destination': '/container2/obj',
'X-Fresh-Metadata': 'hdr-value',
}),
])
def test_fresh_metadata_true(self):
c.http_connection = self.fake_http_connection(200)
c.copy_object(
'http://www.test.com/v1/AUTH', 'token', 'container', 'obj',
destination='/container2/obj',
headers={'x-fresh-metadata': 'hdr-value'},
fresh_metadata=True)
self.assertRequests([
('COPY', 'http://www.test.com/v1/AUTH/container/obj', '', {
'X-Auth-Token': 'token',
'Destination': '/container2/obj',
'X-Fresh-Metadata': 'true',
}),
])
def test_fresh_metadata_false(self):
c.http_connection = self.fake_http_connection(200)
c.copy_object(
'http://www.test.com/v1/AUTH', 'token', 'container', 'obj',
destination='/container2/obj',
headers={'x-fresh-metadata': 'hdr-value'},
fresh_metadata=False)
self.assertRequests([
('COPY', 'http://www.test.com/v1/AUTH/container/obj', '', {
'x-auth-token': 'token',
'Destination': '/container2/obj',
'X-Fresh-Metadata': 'false',
}),
])
def test_no_destination(self):
c.http_connection = self.fake_http_connection(200)
c.copy_object(
'http://www.test.com/v1/AUTH', 'token', 'container', 'obj')
self.assertRequests([
('COPY', 'http://www.test.com/v1/AUTH/container/obj', '', {
'x-auth-token': 'token',
'Destination': '/container/obj',
}),
])
class TestDeleteObject(MockHttpTest):
def test_ok(self):
c.http_connection = self.fake_http_connection(200)
c.delete_object('http://www.test.com', 'token', 'container', 'obj')
self.assertRequests([
('DELETE', 'http://www.test.com/container/obj', '', {
'x-auth-token': 'token',
}),
])
def test_server_error(self):
body = 'c' * 60
headers = {'foo': 'bar'}
c.http_connection = self.fake_http_connection(
StubResponse(500, body, headers))
with self.assertRaises(c.ClientException) as exc_context:
c.delete_object('http://www.test.com', 'asdf', 'asdf', 'asdf')
self.assertEqual(exc_context.exception.http_response_content, body)
self.assertEqual(exc_context.exception.http_response_headers, headers)
def test_query_string(self):
c.http_connection = self.fake_http_connection(200,
query_string="hello=20")
c.delete_object('http://www.test.com', 'token', 'container', 'obj',
query_string="hello=20")
self.assertRequests([
('DELETE', 'http://www.test.com/container/obj?hello=20', '', {
'x-auth-token': 'token',
}),
])
class TestGetCapabilities(MockHttpTest):
def test_ok(self):
conn = self.fake_http_connection(200, body=b'{}')
http_conn = conn('http://www.test.com/info')
info = c.get_capabilities(http_conn)
self.assertRequests([
('GET', '/info', '', {'Accept-Encoding': 'gzip'}),
])
self.assertEqual(info, {})
self.assertTrue(http_conn[1].resp.has_been_read)
def test_server_error(self):
body = 'c' * 60
headers = {'foo': 'bar'}
conn = self.fake_http_connection(
StubResponse(500, body, headers))
http_conn = conn('http://www.test.com/info')
with self.assertRaises(c.ClientException) as exc_context:
c.get_capabilities(http_conn)
self.assertEqual(exc_context.exception.http_response_content, body)
self.assertEqual(exc_context.exception.http_response_headers, headers)
def test_conn_get_capabilities_with_auth(self):
auth_headers = {
'x-auth-token': 'token',
'x-storage-url': 'http://storage.example.com/v1/AUTH_test'
}
auth_v1_response = StubResponse(headers=auth_headers)
stub_info = {'swift': {'fake': True}}
info_response = StubResponse(body=b'{"swift":{"fake":true}}')
fake_conn = self.fake_http_connection(auth_v1_response, info_response)
conn = c.Connection('http://auth.example.com/auth/v1.0',
'user', 'key')
with mock.patch('swiftclient.client.http_connection',
new=fake_conn):
info = conn.get_capabilities()
self.assertEqual(info, stub_info)
self.assertRequests([
('GET', '/auth/v1.0', '', {
'x-auth-user': 'user',
'x-auth-key': 'key'}),
('GET', 'http://storage.example.com/info', '', {
'accept-encoding': 'gzip'}),
])
def test_conn_get_capabilities_with_os_auth(self):
fake_keystone = fake_get_auth_keystone(
storage_url='http://storage.example.com/v1/AUTH_test')
stub_info = {'swift': {'fake': True}}
info_response = StubResponse(body=b'{"swift":{"fake":true}}')
fake_conn = self.fake_http_connection(info_response)
os_options = {'project_id': 'test'}
conn = c.Connection('http://keystone.example.com/v3.0',
'user', 'key', os_options=os_options,
auth_version=3)
with mock.patch.multiple('swiftclient.client',
get_auth_keystone=fake_keystone,
http_connection=fake_conn):
info = conn.get_capabilities()
self.assertEqual(info, stub_info)
self.assertRequests([
('GET', 'http://storage.example.com/info'),
])
def test_conn_get_capabilities_with_url_param(self):
stub_info = {'swift': {'fake': True}}
info_response = StubResponse(body=b'{"swift":{"fake":true}}')
fake_conn = self.fake_http_connection(info_response)
conn = c.Connection('http://auth.example.com/auth/v1.0',
'user', 'key')
with mock.patch('swiftclient.client.http_connection',
new=fake_conn):
info = conn.get_capabilities(
'http://other-storage.example.com/info')
self.assertEqual(info, stub_info)
self.assertRequests([
('GET', 'http://other-storage.example.com/info'),
])
def test_conn_get_capabilities_with_preauthurl_param(self):
stub_info = {'swift': {'fake': True}}
info_response = StubResponse(body=b'{"swift":{"fake":true}}')
fake_conn = self.fake_http_connection(info_response)
storage_url = 'http://storage.example.com/v1/AUTH_test'
conn = c.Connection('http://auth.example.com/auth/v1.0',
'user', 'key', preauthurl=storage_url)
with mock.patch('swiftclient.client.http_connection',
new=fake_conn):
info = conn.get_capabilities()
self.assertEqual(info, stub_info)
self.assertRequests([
('GET', 'http://storage.example.com/info'),
])
def test_conn_get_capabilities_with_os_options(self):
stub_info = {'swift': {'fake': True}}
info_response = StubResponse(body=b'{"swift":{"fake":true}}')
fake_conn = self.fake_http_connection(info_response)
storage_url = 'http://storage.example.com/v1/AUTH_test'
os_options = {
'project_id': 'test',
'object_storage_url': storage_url,
}
conn = c.Connection('http://keystone.example.com/v3.0',
'user', 'key', os_options=os_options,
auth_version=3)
with mock.patch('swiftclient.client.http_connection',
new=fake_conn):
info = conn.get_capabilities()
self.assertEqual(info, stub_info)
self.assertRequests([
('GET', 'http://storage.example.com/info'),
])
class TestHTTPConnection(MockHttpTest):
def test_bad_url_scheme(self):
url = u'www.test.com'
with self.assertRaises(c.ClientException) as exc_context:
c.http_connection(url)
exc = exc_context.exception
expected = u'Unsupported scheme "" in url "www.test.com"'
self.assertEqual(expected, str(exc))
url = u'://www.test.com'
with self.assertRaises(c.ClientException) as exc_context:
c.http_connection(url)
exc = exc_context.exception
expected = u'Unsupported scheme "" in url "://www.test.com"'
self.assertEqual(expected, str(exc))
url = u'blah://www.test.com'
with self.assertRaises(c.ClientException) as exc_context:
c.http_connection(url)
exc = exc_context.exception
expected = u'Unsupported scheme "blah" in url "blah://www.test.com"'
self.assertEqual(expected, str(exc))
def test_ok_url_scheme(self):
for scheme in ('http', 'https', 'HTTP', 'HTTPS'):
url = u'%s://www.test.com' % scheme
parsed_url, conn = c.http_connection(url)
self.assertEqual(scheme.lower(), parsed_url.scheme)
self.assertEqual(u'%s://www.test.com' % scheme, conn.url)
def test_ok_proxy(self):
conn = c.http_connection(u'http://www.test.com/',
proxy='http://localhost:8080')
self.assertEqual(conn[1].requests_args['proxies']['http'],
'http://localhost:8080')
def test_bad_proxy(self):
try:
c.http_connection(u'http://www.test.com/', proxy='localhost:8080')
except c.ClientException as e:
self.assertEqual(e.msg, "Proxy's missing scheme")
def test_cacert(self):
conn = c.http_connection(u'http://www.test.com/',
cacert='/dev/urandom')
self.assertEqual(conn[1].requests_args['verify'], '/dev/urandom')
def test_insecure(self):
conn = c.http_connection(u'http://www.test.com/', insecure=True)
self.assertEqual(conn[1].requests_args['verify'], False)
def test_cert(self):
conn = c.http_connection(u'http://www.test.com/', cert='minnie')
self.assertEqual(conn[1].requests_args['cert'], 'minnie')
def test_cert_key(self):
conn = c.http_connection(
u'http://www.test.com/', cert='minnie', cert_key='mickey')
self.assertEqual(conn[1].requests_args['cert'], ('minnie', 'mickey'))
def test_response_connection_released(self):
_parsed_url, conn = c.http_connection(u'http://www.test.com/')
conn.resp = MockHttpResponse()
conn.resp.raw = mock.Mock()
conn.resp.raw.read.side_effect = ["Chunk", ""]
resp = conn.getresponse()
self.assertFalse(resp.closed)
self.assertEqual("Chunk", resp.read())
self.assertFalse(resp.read())
self.assertTrue(resp.closed)
@unittest.skipIf(six.PY3, 'python2 specific test')
def test_response_python2_headers(self):
'''Test utf-8 headers in Python 2.
'''
_, conn = c.http_connection(u'http://www.test.com/')
conn.resp = MockHttpResponse(
status=200,
headers={
'\xd8\xaa-unicode': '\xd8\xaa-value',
'empty-header': ''
}
)
resp = conn.getresponse()
self.assertEqual(
'\xd8\xaa-value', resp.getheader('\xd8\xaa-unicode'))
self.assertEqual(
'\xd8\xaa-value', resp.getheader('\xd8\xaa-UNICODE'))
self.