本文整理汇总了Python中swift.proxy.controllers.base.GetOrHeadHandler类的典型用法代码示例。如果您正苦于以下问题：Python GetOrHeadHandler类的具体用法？Python GetOrHeadHandler怎么用？Python GetOrHeadHandler使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了GetOrHeadHandler类的12个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_client_chunk_size_resuming
def test_client_chunk_size_resuming(self):
class TestSource(object):
def __init__(self, chunks):
self.chunks = list(chunks)
self.status = 200
def read(self, _read_size):
if self.chunks:
chunk = self.chunks.pop(0)
if chunk is None:
raise exceptions.ChunkReadTimeout()
else:
return chunk
else:
return ""
def getheader(self, header):
if header.lower() == "content-length":
return str(sum(len(c) for c in self.chunks if c is not None))
def getheaders(self):
return [("content-length", self.getheader("content-length"))]
node = {"ip": "1.2.3.4", "port": 6000, "device": "sda"}
source1 = TestSource(["abcd", "1234", "abc", None])
source2 = TestSource(["efgh5678"])
req = Request.blank("/v1/a/c/o")
handler = GetOrHeadHandler(self.app, req, "Object", None, None, None, {}, client_chunk_size=8)
app_iter = handler._make_app_iter(req, node, source1)
with patch.object(handler, "_get_source_and_node", lambda: (source2, node)):
client_chunks = list(app_iter)
self.assertEqual(client_chunks, ["abcd1234", "efgh5678"])
开发者ID:helen5haha，项目名称:swift，代码行数:34，代码来源:test_base.py
示例2: test_client_chunk_size
def test_client_chunk_size(self):
class TestSource(object):
def __init__(self, chunks):
self.chunks = list(chunks)
self.status = 200
def read(self, _read_size):
if self.chunks:
return self.chunks.pop(0)
else:
return b''
def getheader(self, header):
if header.lower() == "content-length":
return str(sum(len(c) for c in self.chunks))
def getheaders(self):
return [('content-length', self.getheader('content-length'))]
source = TestSource((
b'abcd', b'1234', b'abc', b'd1', b'234abcd1234abcd1', b'2'))
req = Request.blank('/v1/a/c/o')
node = {}
handler = GetOrHeadHandler(self.app, req, None, None, None, None, {},
client_chunk_size=8)
app_iter = handler._make_app_iter(req, node, source)
client_chunks = list(app_iter)
self.assertEqual(client_chunks, [
b'abcd1234', b'abcd1234', b'abcd1234', b'abcd12'])
开发者ID:mahak，项目名称:swift，代码行数:31，代码来源:test_base.py
示例3: test_client_chunk_size
def test_client_chunk_size(self):
class TestSource(object):
def __init__(self, chunks):
self.chunks = list(chunks)
self.status = 200
def read(self, _read_size):
if self.chunks:
return self.chunks.pop(0)
else:
return ""
def getheader(self, header):
if header.lower() == "content-length":
return str(sum(len(c) for c in self.chunks))
def getheaders(self):
return [("content-length", self.getheader("content-length"))]
source = TestSource(("abcd", "1234", "abc", "d1", "234abcd1234abcd1", "2"))
req = Request.blank("/v1/a/c/o")
node = {}
handler = GetOrHeadHandler(self.app, req, None, None, None, None, {}, client_chunk_size=8)
app_iter = handler._make_app_iter(req, node, source)
client_chunks = list(app_iter)
self.assertEqual(client_chunks, ["abcd1234", "abcd1234", "abcd1234", "abcd12"])
开发者ID:helen5haha，项目名称:swift，代码行数:27，代码来源:test_base.py
示例4: test_client_chunk_size_resuming
def test_client_chunk_size_resuming(self):
class TestSource(object):
def __init__(self, chunks):
self.chunks = list(chunks)
def read(self, _read_size):
if self.chunks:
chunk = self.chunks.pop(0)
if chunk is None:
raise exceptions.ChunkReadTimeout()
else:
return chunk
else:
return ''
node = {'ip': '1.2.3.4', 'port': 6000, 'device': 'sda'}
source1 = TestSource(['abcd', '1234', 'abc', None])
source2 = TestSource(['efgh5678'])
req = Request.blank('/v1/a/c/o')
handler = GetOrHeadHandler(
self.app, req, 'Object', None, None, None, {},
client_chunk_size=8)
app_iter = handler._make_app_iter(req, node, source1)
with patch.object(handler, '_get_source_and_node',
lambda: (source2, node)):
client_chunks = list(app_iter)
self.assertEqual(client_chunks, ['abcd1234', 'efgh5678'])
self.assertEqual(handler.backend_headers['Range'], 'bytes=8-')
开发者ID:kun--hust，项目名称:sdscloud，代码行数:31，代码来源:test_base.py
示例5: test_client_chunk_size_resuming
def test_client_chunk_size_resuming(self):
class TestSource(object):
def __init__(self, chunks):
self.chunks = list(chunks)
self.status = 200
def read(self, _read_size):
if self.chunks:
chunk = self.chunks.pop(0)
if chunk is None:
raise exceptions.ChunkReadTimeout()
else:
return chunk
else:
return b''
def getheader(self, header):
# content-length for the whole object is generated dynamically
# by summing non-None chunks initialized as source1
if header.lower() == "content-length":
return str(sum(len(c) for c in self.chunks
if c is not None))
def getheaders(self):
return [('content-length', self.getheader('content-length'))]
node = {'ip': '1.2.3.4', 'port': 6200, 'device': 'sda'}
source1 = TestSource([b'abcd', b'1234', None,
b'efgh', b'5678', b'lots', b'more', b'data'])
# incomplete reads of client_chunk_size will be re-fetched
source2 = TestSource([b'efgh', b'5678', b'lots', None])
source3 = TestSource([b'lots', b'more', b'data'])
req = Request.blank('/v1/a/c/o')
handler = GetOrHeadHandler(
self.app, req, 'Object', None, None, None, {},
client_chunk_size=8)
range_headers = []
sources = [(source2, node), (source3, node)]
def mock_get_source_and_node():
range_headers.append(handler.backend_headers['Range'])
return sources.pop(0)
app_iter = handler._make_app_iter(req, node, source1)
with mock.patch.object(handler, '_get_source_and_node',
side_effect=mock_get_source_and_node):
client_chunks = list(app_iter)
self.assertEqual(range_headers, ['bytes=8-27', 'bytes=16-27'])
self.assertEqual(client_chunks, [
b'abcd1234', b'efgh5678', b'lotsmore', b'data'])
开发者ID:mahak，项目名称:swift，代码行数:53，代码来源:test_base.py
示例6: test_client_chunk_size_resuming_chunked
def test_client_chunk_size_resuming_chunked(self):
class TestChunkedSource(object):
def __init__(self, chunks):
self.chunks = list(chunks)
self.status = 200
self.headers = {'transfer-encoding': 'chunked',
'content-type': 'text/plain'}
def read(self, _read_size):
if self.chunks:
chunk = self.chunks.pop(0)
if chunk is None:
raise exceptions.ChunkReadTimeout()
else:
return chunk
else:
return b''
def getheader(self, header):
return self.headers.get(header.lower())
def getheaders(self):
return self.headers
node = {'ip': '1.2.3.4', 'port': 6200, 'device': 'sda'}
source1 = TestChunkedSource([b'abcd', b'1234', b'abc', None])
source2 = TestChunkedSource([b'efgh5678'])
req = Request.blank('/v1/a/c/o')
handler = GetOrHeadHandler(
self.app, req, 'Object', None, None, None, {},
client_chunk_size=8)
app_iter = handler._make_app_iter(req, node, source1)
with mock.patch.object(handler, '_get_source_and_node',
lambda: (source2, node)):
client_chunks = list(app_iter)
self.assertEqual(client_chunks, [b'abcd1234', b'efgh5678'])
开发者ID:mahak，项目名称:swift，代码行数:39，代码来源:test_base.py
示例7: test_client_chunk_size_resuming
def test_client_chunk_size_resuming(self):
class TestSource(object):
def __init__(self, chunks):
self.chunks = list(chunks)
self.status = 200
def read(self, _read_size):
if self.chunks:
chunk = self.chunks.pop(0)
if chunk is None:
raise exceptions.ChunkReadTimeout()
else:
return chunk
else:
return ''
def getheader(self, header):
if header.lower() == "content-length":
return str(sum(len(c) for c in self.chunks
if c is not None))
def getheaders(self):
return [('content-length', self.getheader('content-length'))]
node = {'ip': '1.2.3.4', 'port': 6000, 'device': 'sda'}
source1 = TestSource(['abcd', '1234', 'abc', None])
source2 = TestSource(['efgh5678'])
req = Request.blank('/v1/a/c/o')
handler = GetOrHeadHandler(
self.app, req, 'Object', None, None, None, {},
client_chunk_size=8)
app_iter = handler._make_app_iter(req, node, source1)
with patch.object(handler, '_get_source_and_node',
lambda: (source2, node)):
client_chunks = list(app_iter)
self.assertEqual(client_chunks, ['abcd1234', 'efgh5678'])
开发者ID:bouncestorage，项目名称:swift，代码行数:39，代码来源:test_base.py
示例8: test_client_chunk_size
def test_client_chunk_size(self):
class TestSource(object):
def __init__(self, chunks):
self.chunks = list(chunks)
def read(self, _read_size):
if self.chunks:
return self.chunks.pop(0)
else:
return ''
source = TestSource((
'abcd', '1234', 'abc', 'd1', '234abcd1234abcd1', '2'))
req = Request.blank('/v1/a/c/o')
node = {}
handler = GetOrHeadHandler(self.app, req, None, None, None, None, {},
client_chunk_size=8)
app_iter = handler._make_app_iter(req, node, source)
client_chunks = list(app_iter)
self.assertEqual(client_chunks, [
'abcd1234', 'abcd1234', 'abcd1234', 'abcd12'])
开发者ID:kun--hust，项目名称:sdscloud，代码行数:23，代码来源:test_base.py
示例9: test_disconnected_warning
def test_disconnected_warning(self):
self.app.logger = mock.Mock()
req = Request.blank('/v1/a/c/o')
class TestSource(object):
def __init__(self):
self.headers = {'content-type': 'text/plain',
'content-length': len(self.read(-1))}
self.status = 200
def read(self, _read_size):
return b'the cake is a lie'
def getheader(self, header):
return self.headers.get(header.lower())
def getheaders(self):
return self.headers
source = TestSource()
node = {'ip': '1.2.3.4', 'port': 6200, 'device': 'sda'}
handler = GetOrHeadHandler(
self.app, req, 'Object', None, None, None, {})
app_iter = handler._make_app_iter(req, node, source)
app_iter.close()
self.app.logger.warning.assert_called_once_with(
'Client disconnected on read')
self.app.logger = mock.Mock()
node = {'ip': '1.2.3.4', 'port': 6200, 'device': 'sda'}
handler = GetOrHeadHandler(
self.app, req, 'Object', None, None, None, {})
app_iter = handler._make_app_iter(req, node, source)
next(app_iter)
app_iter.close()
self.app.logger.warning.assert_not_called()
开发者ID:mahak，项目名称:swift，代码行数:37，代码来源:test_base.py
示例10: test_range_fast_forward_after_data_timeout
def test_range_fast_forward_after_data_timeout(self):
req = Request.blank('/')
# We get a 200 and learn that it's a 1000-byte object, but receive 0
# bytes of data, so then we get a new node, fast_forward(0), and
# send out a new request. That new request must be for all 1000
# bytes.
handler = GetOrHeadHandler(None, req, None, None, None, None, {})
handler.learn_size_from_content_range(0, 999, 1000)
handler.fast_forward(0)
self.assertEqual(handler.backend_headers['Range'], 'bytes=0-999')
# Same story as above, but a 1-byte object so we can have our byte
# indices be 0.
handler = GetOrHeadHandler(None, req, None, None, None, None, {})
handler.learn_size_from_content_range(0, 0, 1)
handler.fast_forward(0)
self.assertEqual(handler.backend_headers['Range'], 'bytes=0-0')
# last 100 bytes
handler = GetOrHeadHandler(None, req, None, None, None, None,
{'Range': 'bytes=-100'})
handler.learn_size_from_content_range(900, 999, 1000)
handler.fast_forward(0)
self.assertEqual(handler.backend_headers['Range'], 'bytes=900-999')
开发者ID:mahak，项目名称:swift，代码行数:25，代码来源:test_base.py
示例11: test_range_fast_forward
def test_range_fast_forward(self):
req = Request.blank('/')
handler = GetOrHeadHandler(None, req, None, None, None, None, {})
handler.fast_forward(50)
self.assertEqual(handler.backend_headers['Range'], 'bytes=50-')
handler = GetOrHeadHandler(None, req, None, None, None, None,
{'Range': 'bytes=23-50'})
handler.fast_forward(20)
self.assertEqual(handler.backend_headers['Range'], 'bytes=43-50')
self.assertRaises(HTTPException,
handler.fast_forward, 80)
self.assertRaises(exceptions.RangeAlreadyComplete,
handler.fast_forward, 8)
handler = GetOrHeadHandler(None, req, None, None, None, None,
{'Range': 'bytes=23-'})
handler.fast_forward(20)
self.assertEqual(handler.backend_headers['Range'], 'bytes=43-')
handler = GetOrHeadHandler(None, req, None, None, None, None,
{'Range': 'bytes=-100'})
handler.fast_forward(20)
self.assertEqual(handler.backend_headers['Range'], 'bytes=-80')
self.assertRaises(HTTPException,
handler.fast_forward, 100)
self.assertRaises(exceptions.RangeAlreadyComplete,
handler.fast_forward, 80)
handler = GetOrHeadHandler(None, req, None, None, None, None,
{'Range': 'bytes=0-0'})
self.assertRaises(exceptions.RangeAlreadyComplete,
handler.fast_forward, 1)
开发者ID:mahak，项目名称:swift，代码行数:33，代码来源:test_base.py
示例12: test_range_fast_forward
def test_range_fast_forward(self):
req = Request.blank("/")
handler = GetOrHeadHandler(None, req, None, None, None, None, {})
handler.fast_forward(50)
self.assertEquals(handler.backend_headers["Range"], "bytes=50-")
handler = GetOrHeadHandler(None, req, None, None, None, None, {"Range": "bytes=23-50"})
handler.fast_forward(20)
self.assertEquals(handler.backend_headers["Range"], "bytes=43-50")
self.assertRaises(HTTPException, handler.fast_forward, 80)
handler = GetOrHeadHandler(None, req, None, None, None, None, {"Range": "bytes=23-"})
handler.fast_forward(20)
self.assertEquals(handler.backend_headers["Range"], "bytes=43-")
handler = GetOrHeadHandler(None, req, None, None, None, None, {"Range": "bytes=-100"})
handler.fast_forward(20)
self.assertEquals(handler.backend_headers["Range"], "bytes=-80")
开发者ID:helen5haha，项目名称:swift，代码行数:18，代码来源:test_base.py
注：本文中的swift.proxy.controllers.base.GetOrHeadHandler类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论