8b42f8a40c
This patch forces swiftclient to encode to utf-8
all url and headers arguments, to avoid the
UnicodeDecodeError which is raised by '\r\n'.join([])
invoked in htplib.py.
Currently the affected projects are Horizon(upload file
with unicode name) and swiftclient CLI('swift post' with
unicode filename as header)
This is also a follow-up of this review:
https://review.openstack.org/#/c/14216/
I'd still want to hear what the Swift core devs
think of it. Is it better to create a new
AutoEncodingHTTPConnection? Or to handle the connection
creation and make sure there are no unicode and utf-8
string at the same time. If these unicode checks have to
be added in the calling code(Dashboard, CLI), there are
so many places to be added, and also in all new commands
that might be exposed from the API.
Fixes bug 1008940
Change-Id: Ice2aa29024429d3e6f569a88d5cf8b4202537827
652 lines
23 KiB
Python
652 lines
23 KiB
Python
# 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.
|
|
|
|
# TODO: More tests
|
|
import socket
|
|
import StringIO
|
|
import unittest
|
|
from urlparse import urlparse
|
|
|
|
# TODO: mock http connection class with more control over headers
|
|
from utils import fake_http_connect, fake_get_keystoneclient_2_0
|
|
|
|
from swiftclient import client as c
|
|
|
|
|
|
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.assertTrue('something failed' in str(exc))
|
|
test_kwargs = (
|
|
'scheme',
|
|
'host',
|
|
'port',
|
|
'path',
|
|
'query',
|
|
'status',
|
|
'reason',
|
|
'device',
|
|
)
|
|
for value in test_kwargs:
|
|
kwargs = {
|
|
'http_%s' % value: value,
|
|
}
|
|
exc = c.ClientException('test', **kwargs)
|
|
self.assertTrue(value in str(exc))
|
|
|
|
|
|
class TestJsonImport(unittest.TestCase):
|
|
|
|
def tearDown(self):
|
|
try:
|
|
import json
|
|
except ImportError:
|
|
pass
|
|
else:
|
|
reload(json)
|
|
|
|
try:
|
|
import simplejson
|
|
except ImportError:
|
|
pass
|
|
else:
|
|
reload(simplejson)
|
|
|
|
def test_any(self):
|
|
self.assertTrue(hasattr(c, 'json_loads'))
|
|
|
|
def test_no_simplejson(self):
|
|
# break simplejson
|
|
try:
|
|
import simplejson
|
|
except ImportError:
|
|
# not installed, so we don't have to break it for these tests
|
|
pass
|
|
else:
|
|
delattr(simplejson, 'loads')
|
|
reload(c)
|
|
|
|
try:
|
|
from json import loads
|
|
except ImportError:
|
|
# this case is stested in _no_json
|
|
pass
|
|
else:
|
|
self.assertEquals(loads, c.json_loads)
|
|
|
|
|
|
class MockHttpTest(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
def fake_http_connection(*args, **kwargs):
|
|
_orig_http_connection = c.http_connection
|
|
return_read = kwargs.get('return_read')
|
|
|
|
def wrapper(url, proxy=None):
|
|
parsed, _conn = _orig_http_connection(url, proxy=proxy)
|
|
conn = fake_http_connect(*args, **kwargs)()
|
|
|
|
def request(*args, **kwargs):
|
|
return
|
|
conn.request = request
|
|
|
|
conn.has_been_read = False
|
|
_orig_read = conn.read
|
|
|
|
def read(*args, **kwargs):
|
|
conn.has_been_read = True
|
|
return _orig_read(*args, **kwargs)
|
|
conn.read = return_read or read
|
|
|
|
return parsed, conn
|
|
return wrapper
|
|
self.fake_http_connection = fake_http_connection
|
|
|
|
def tearDown(self):
|
|
reload(c)
|
|
|
|
|
|
class MockHttpResponse():
|
|
def __init__(self):
|
|
self.status = 200
|
|
self.buffer = []
|
|
|
|
def read(self):
|
|
return ""
|
|
|
|
def getheader(self, name, default):
|
|
return ""
|
|
|
|
def fake_response(self):
|
|
return MockHttpResponse()
|
|
|
|
def fake_send(self, msg):
|
|
self.buffer.append(msg)
|
|
|
|
|
|
class TestHttpHelpers(MockHttpTest):
|
|
|
|
def test_quote(self):
|
|
value = 'standard string'
|
|
self.assertEquals('standard%20string', c.quote(value))
|
|
value = u'\u0075nicode string'
|
|
self.assertEquals('unicode%20string', c.quote(value))
|
|
|
|
def test_http_connection(self):
|
|
url = 'http://www.test.com'
|
|
_junk, conn = c.http_connection(url)
|
|
self.assertTrue(isinstance(conn, c.HTTPConnection))
|
|
url = 'https://www.test.com'
|
|
_junk, conn = c.http_connection(url)
|
|
self.assertTrue(isinstance(conn, c.HTTPSConnection))
|
|
url = 'ftp://www.test.com'
|
|
self.assertRaises(c.ClientException, c.http_connection, url)
|
|
|
|
def test_json_request(self):
|
|
def read(*args, **kwargs):
|
|
body = {'a': '1',
|
|
'b': '2'}
|
|
return c.json_dumps(body)
|
|
c.http_connection = self.fake_http_connection(200, return_read=read)
|
|
url = 'http://www.test.com'
|
|
_junk, conn = c.json_request('GET', url, body={'username': 'user1',
|
|
'password': 'secure'})
|
|
self.assertTrue(type(conn) is dict)
|
|
|
|
# TODO: following tests are placeholders, need more tests, better coverage
|
|
|
|
|
|
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.assertEquals(url, None)
|
|
self.assertEquals(token, None)
|
|
|
|
def test_invalid_auth(self):
|
|
c.http_connection = self.fake_http_connection(200)
|
|
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)
|
|
url, token = c.get_auth('http://www.test.com', 'asdf', 'asdf',
|
|
auth_version="1.0")
|
|
self.assertEquals(url, None)
|
|
self.assertEquals(token, None)
|
|
|
|
def test_auth_v2(self):
|
|
os_options={'tenant_name': 'asdf'}
|
|
c.get_keystoneclient_2_0 = fake_get_keystoneclient_2_0(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_tenant_name(self):
|
|
c.get_keystoneclient_2_0 = fake_get_keystoneclient_2_0({})
|
|
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_user_in_user(self):
|
|
tenant_option = {'tenant_name': 'foo'}
|
|
c.get_keystoneclient_2_0 = fake_get_keystoneclient_2_0(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'}
|
|
c.get_keystoneclient_2_0 = fake_get_keystoneclient_2_0(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'}
|
|
c.get_keystoneclient_2_0 = fake_get_keystoneclient_2_0(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'}
|
|
c.get_keystoneclient_2_0 = fake_get_keystoneclient_2_0(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'}
|
|
c.get_keystoneclient_2_0 = fake_get_keystoneclient_2_0(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'}
|
|
c.get_keystoneclient_2_0 = fake_get_keystoneclient_2_0(
|
|
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):
|
|
c.get_keystoneclient_2_0 = fake_get_keystoneclient_2_0(
|
|
{},
|
|
c.ClientException)
|
|
self.assertRaises(c.ClientException, c.get_auth,
|
|
'http://www.tests.com', 'asdf', 'asdf',
|
|
os_options={},
|
|
auth_version='2.0')
|
|
|
|
class TestGetAccount(MockHttpTest):
|
|
|
|
def test_no_content(self):
|
|
c.http_connection = self.fake_http_connection(204)
|
|
value = c.get_account('http://www.test.com', 'asdf')[1]
|
|
self.assertEquals(value, [])
|
|
|
|
|
|
class TestHeadAccount(MockHttpTest):
|
|
|
|
def test_ok(self):
|
|
c.http_connection = self.fake_http_connection(200)
|
|
value = c.head_account('http://www.tests.com', 'asdf')
|
|
# TODO: Hmm. This doesn't really test too much as it uses a fake that
|
|
# always returns the same dict. I guess it "exercises" the code, so
|
|
# I'll leave it for now.
|
|
self.assertEquals(type(value), dict)
|
|
|
|
def test_server_error(self):
|
|
body = 'c' * 65
|
|
c.http_connection = self.fake_http_connection(500, body=body)
|
|
self.assertRaises(c.ClientException, c.head_account,
|
|
'http://www.tests.com', 'asdf')
|
|
try:
|
|
value = c.head_account('http://www.tests.com', 'asdf')
|
|
except c.ClientException as e:
|
|
new_body = "[first 60 chars of response] " + body[0:60]
|
|
self.assertEquals(e.__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', 'asdf', 'asdf')[1]
|
|
self.assertEquals(value, [])
|
|
|
|
|
|
class TestHeadContainer(MockHttpTest):
|
|
|
|
def test_server_error(self):
|
|
body = 'c' * 60
|
|
c.http_connection = self.fake_http_connection(500, body=body)
|
|
self.assertRaises(c.ClientException, c.head_container,
|
|
'http://www.test.com', 'asdf', 'asdf',
|
|
)
|
|
try:
|
|
value = c.head_container('http://www.test.com', 'asdf', 'asdf')
|
|
except c.ClientException as e:
|
|
self.assertEquals(e.http_response_content, body)
|
|
|
|
|
|
class TestPutContainer(MockHttpTest):
|
|
|
|
def test_ok(self):
|
|
c.http_connection = self.fake_http_connection(200)
|
|
value = c.put_container('http://www.test.com', 'asdf', 'asdf')
|
|
self.assertEquals(value, None)
|
|
|
|
def test_server_error(self):
|
|
body = 'c' * 60
|
|
c.http_connection = self.fake_http_connection(500, body=body)
|
|
self.assertRaises(c.ClientException, c.put_container,
|
|
'http://www.test.com', 'asdf', 'asdf',
|
|
)
|
|
try:
|
|
value = c.put_container('http://www.test.com', 'asdf', 'asdf')
|
|
except c.ClientException as e:
|
|
self.assertEquals(e.http_response_content, body)
|
|
|
|
|
|
class TestDeleteContainer(MockHttpTest):
|
|
|
|
def test_ok(self):
|
|
c.http_connection = self.fake_http_connection(200)
|
|
value = c.delete_container('http://www.test.com', 'asdf', 'asdf')
|
|
self.assertEquals(value, None)
|
|
|
|
|
|
class TestGetObject(MockHttpTest):
|
|
|
|
def test_server_error(self):
|
|
c.http_connection = self.fake_http_connection(500)
|
|
self.assertRaises(c.ClientException, c.get_object,
|
|
'http://www.test.com', 'asdf', 'asdf', 'asdf')
|
|
|
|
|
|
class TestHeadObject(MockHttpTest):
|
|
|
|
def test_server_error(self):
|
|
c.http_connection = self.fake_http_connection(500)
|
|
self.assertRaises(c.ClientException, c.head_object,
|
|
'http://www.test.com', 'asdf', 'asdf', 'asdf')
|
|
|
|
|
|
class TestPutObject(MockHttpTest):
|
|
|
|
def test_ok(self):
|
|
c.http_connection = self.fake_http_connection(200)
|
|
args = ('http://www.test.com', 'asdf', 'asdf', 'asdf', 'asdf')
|
|
value = c.put_object(*args)
|
|
self.assertTrue(isinstance(value, basestring))
|
|
|
|
def test_unicode_ok(self):
|
|
conn = c.http_connection(u'http://www.test.com/')
|
|
file = StringIO.StringIO(u'\u5929\u7a7a\u4e2d\u7684\u4e4c\u4e91')
|
|
args = (u'\u5929\u7a7a\u4e2d\u7684\u4e4c\u4e91',
|
|
'\u5929\u7a7a\u4e2d\u7684\u4e4c\u4e91',
|
|
u'\u5929\u7a7a\u4e2d\u7684\u4e4c\u4e91',
|
|
u'\u5929\u7a7a\u4e2d\u7684\u4e4c\u4e91',
|
|
file)
|
|
headers = {'X-Header1': u'\u5929\u7a7a\u4e2d\u7684\u4e4c\u4e91',
|
|
'X-2': 1, 'X-3': {'a': 'b'}, 'a-b': '.x:yz mn:fg:lp'}
|
|
|
|
resp = MockHttpResponse()
|
|
conn[1].getresponse = resp.fake_response
|
|
conn[1].send = resp.fake_send
|
|
value = c.put_object(*args, headers=headers, http_conn=conn)
|
|
self.assertTrue(isinstance(value, basestring))
|
|
# Test for RFC-2616 encoded symbols
|
|
self.assertTrue("a-b: .x:yz mn:fg:lp" in resp.buffer[0],
|
|
"[a-b: .x:yz mn:fg:lp] header is missing")
|
|
|
|
def test_server_error(self):
|
|
body = 'c' * 60
|
|
c.http_connection = self.fake_http_connection(500, body=body)
|
|
args = ('http://www.test.com', 'asdf', 'asdf', 'asdf', 'asdf')
|
|
self.assertRaises(c.ClientException, c.put_object, *args)
|
|
try:
|
|
value = c.put_object(*args)
|
|
except c.ClientException as e:
|
|
self.assertEquals(e.http_response_content, body)
|
|
|
|
|
|
class TestPostObject(MockHttpTest):
|
|
|
|
def test_ok(self):
|
|
c.http_connection = self.fake_http_connection(200)
|
|
args = ('http://www.test.com', 'asdf', 'asdf', 'asdf', {})
|
|
value = c.post_object(*args)
|
|
|
|
def test_unicode_ok(self):
|
|
conn = c.http_connection(u'http://www.test.com/')
|
|
args = (u'\u5929\u7a7a\u4e2d\u7684\u4e4c\u4e91',
|
|
'\u5929\u7a7a\u4e2d\u7684\u4e4c\u4e91',
|
|
u'\u5929\u7a7a\u4e2d\u7684\u4e4c\u4e91',
|
|
u'\u5929\u7a7a\u4e2d\u7684\u4e4c\u4e91')
|
|
headers = {'X-Header1': u'\u5929\u7a7a\u4e2d\u7684\u4e4c\u4e91',
|
|
'X-2': 1, 'X-3': {'a': 'b'}, 'a-b': '.x:yz mn:kl:qr'}
|
|
|
|
resp = MockHttpResponse()
|
|
conn[1].getresponse = resp.fake_response
|
|
conn[1].send = resp.fake_send
|
|
c.post_object(*args, headers=headers, http_conn=conn)
|
|
# Test for RFC-2616 encoded symbols
|
|
self.assertTrue("a-b: .x:yz mn:kl:qr" in resp.buffer[0],
|
|
"[a-b: .x:yz mn:kl:qr] header is missing")
|
|
|
|
def test_server_error(self):
|
|
body = 'c' * 60
|
|
c.http_connection = self.fake_http_connection(500, body=body)
|
|
args = ('http://www.test.com', 'asdf', 'asdf', 'asdf', {})
|
|
self.assertRaises(c.ClientException, c.post_object, *args)
|
|
try:
|
|
value = c.post_object(*args)
|
|
except c.ClientException as e:
|
|
self.assertEquals(e.http_response_content, body)
|
|
|
|
|
|
class TestDeleteObject(MockHttpTest):
|
|
|
|
def test_ok(self):
|
|
c.http_connection = self.fake_http_connection(200)
|
|
value = c.delete_object('http://www.test.com', 'asdf', 'asdf', 'asdf')
|
|
|
|
def test_server_error(self):
|
|
c.http_connection = self.fake_http_connection(500)
|
|
self.assertRaises(c.ClientException, c.delete_object,
|
|
'http://www.test.com', 'asdf', 'asdf', 'asdf')
|
|
|
|
|
|
class TestConnection(MockHttpTest):
|
|
|
|
def test_instance(self):
|
|
conn = c.Connection('http://www.test.com', 'asdf', 'asdf')
|
|
self.assertEquals(conn.retries, 5)
|
|
|
|
def test_instance_kwargs(self):
|
|
args = {'user': 'ausername',
|
|
'key': 'secretpass',
|
|
'authurl': 'http://www.test.com',
|
|
'tenant_name': 'atenant'}
|
|
conn = c.Connection(**args)
|
|
self.assertEquals(type(conn), c.Connection)
|
|
|
|
def test_instance_kwargs_token(self):
|
|
args = {'preauthtoken': 'atoken123',
|
|
'preauthurl': 'http://www.test.com:8080/v1/AUTH_123456'}
|
|
conn = c.Connection(**args)
|
|
self.assertEquals(type(conn), c.Connection)
|
|
|
|
def test_retry(self):
|
|
c.http_connection = self.fake_http_connection(500)
|
|
|
|
def quick_sleep(*args):
|
|
pass
|
|
c.sleep = quick_sleep
|
|
conn = c.Connection('http://www.test.com', 'asdf', 'asdf')
|
|
self.assertRaises(c.ClientException, conn.head_account)
|
|
self.assertEquals(conn.attempts, conn.retries + 1)
|
|
|
|
def test_resp_read_on_server_error(self):
|
|
c.http_connection = self.fake_http_connection(500)
|
|
conn = c.Connection('http://www.test.com', 'asdf', 'asdf', retries=0)
|
|
|
|
def get_auth(*args, **kwargs):
|
|
return 'http://www.new.com', 'new'
|
|
conn.get_auth = get_auth
|
|
self.url, self.token = conn.get_auth()
|
|
|
|
method_signatures = (
|
|
(conn.head_account, []),
|
|
(conn.get_account, []),
|
|
(conn.head_container, ('asdf',)),
|
|
(conn.get_container, ('asdf',)),
|
|
(conn.put_container, ('asdf',)),
|
|
(conn.delete_container, ('asdf',)),
|
|
(conn.head_object, ('asdf', 'asdf')),
|
|
(conn.get_object, ('asdf', 'asdf')),
|
|
(conn.put_object, ('asdf', 'asdf', 'asdf')),
|
|
(conn.post_object, ('asdf', 'asdf', {})),
|
|
(conn.delete_object, ('asdf', 'asdf')),
|
|
)
|
|
|
|
for method, args in method_signatures:
|
|
self.assertRaises(c.ClientException, method, *args)
|
|
try:
|
|
self.assertTrue(conn.http_conn[1].has_been_read)
|
|
except AssertionError:
|
|
msg = '%s did not read resp on server error' % method.__name__
|
|
self.fail(msg)
|
|
except Exception, e:
|
|
raise e.__class__("%s - %s" % (method.__name__, e))
|
|
|
|
def test_reauth(self):
|
|
c.http_connection = self.fake_http_connection(401)
|
|
|
|
def get_auth(*args, **kwargs):
|
|
return 'http://www.new.com', 'new'
|
|
|
|
def swap_sleep(*args):
|
|
self.swap_sleep_called = True
|
|
c.get_auth = get_auth
|
|
c.http_connection = self.fake_http_connection(200)
|
|
c.sleep = swap_sleep
|
|
self.swap_sleep_called = False
|
|
|
|
conn = c.Connection('http://www.test.com', 'asdf', 'asdf',
|
|
preauthurl='http://www.old.com',
|
|
preauthtoken='old',
|
|
)
|
|
|
|
self.assertEquals(conn.attempts, 0)
|
|
self.assertEquals(conn.url, 'http://www.old.com')
|
|
self.assertEquals(conn.token, 'old')
|
|
|
|
value = conn.head_account()
|
|
|
|
self.assertTrue(self.swap_sleep_called)
|
|
self.assertEquals(conn.attempts, 2)
|
|
self.assertEquals(conn.url, 'http://www.new.com')
|
|
self.assertEquals(conn.token, 'new')
|
|
|
|
def test_reset_stream(self):
|
|
|
|
class LocalContents(object):
|
|
|
|
def __init__(self, tell_value=0):
|
|
self.already_read = False
|
|
self.seeks = []
|
|
self.tell_value = tell_value
|
|
|
|
def tell(self):
|
|
return self.tell_value
|
|
|
|
def seek(self, position):
|
|
self.seeks.append(position)
|
|
self.already_read = False
|
|
|
|
def read(self, size=-1):
|
|
if self.already_read:
|
|
return ''
|
|
else:
|
|
self.already_read = True
|
|
return 'abcdef'
|
|
|
|
class LocalConnection(object):
|
|
|
|
def __init__(self, parsed_url=None):
|
|
self.reason = ""
|
|
if parsed_url:
|
|
self.host = parsed_url.netloc
|
|
self.port = parsed_url.netloc
|
|
|
|
def putrequest(self, *args, **kwargs):
|
|
return
|
|
|
|
def putheader(self, *args, **kwargs):
|
|
return
|
|
|
|
def endheaders(self, *args, **kwargs):
|
|
return
|
|
|
|
def send(self, *args, **kwargs):
|
|
raise socket.error('oops')
|
|
|
|
def request(self, *args, **kwargs):
|
|
return
|
|
|
|
def getresponse(self, *args, **kwargs):
|
|
self.status = 200
|
|
return self
|
|
|
|
def getheader(self, *args, **kwargs):
|
|
return 'header'
|
|
|
|
def read(self, *args, **kwargs):
|
|
return ''
|
|
|
|
def local_http_connection(url, proxy=None):
|
|
parsed = urlparse(url)
|
|
return parsed, LocalConnection()
|
|
|
|
orig_conn = c.http_connection
|
|
try:
|
|
c.http_connection = local_http_connection
|
|
conn = c.Connection('http://www.example.com', 'asdf', 'asdf',
|
|
retries=1, starting_backoff=.0001)
|
|
|
|
contents = LocalContents()
|
|
exc = None
|
|
try:
|
|
conn.put_object('c', 'o', contents)
|
|
except socket.error, err:
|
|
exc = err
|
|
self.assertEquals(contents.seeks, [0])
|
|
self.assertEquals(str(exc), 'oops')
|
|
|
|
contents = LocalContents(tell_value=123)
|
|
exc = None
|
|
try:
|
|
conn.put_object('c', 'o', contents)
|
|
except socket.error, err:
|
|
exc = err
|
|
self.assertEquals(contents.seeks, [123])
|
|
self.assertEquals(str(exc), 'oops')
|
|
|
|
contents = LocalContents()
|
|
contents.tell = None
|
|
exc = None
|
|
try:
|
|
conn.put_object('c', 'o', contents)
|
|
except c.ClientException, err:
|
|
exc = err
|
|
self.assertEquals(contents.seeks, [])
|
|
self.assertEquals(str(exc), "put_object('c', 'o', ...) failure "
|
|
"and no ability to reset contents for reupload.")
|
|
finally:
|
|
c.http_connection = orig_conn
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|