Rename "tests" directory to be "test" like in the swift repo

In addition to being less confusing for devs, this lets us actually run
tempauth tests in swiftclient dsvm jobs.

The job definition (over in the swift repo) specifies test/sample.conf,
which does not exist in this repo. As a result, those tests would skip
with

    SKIPPING FUNCTIONAL TESTS DUE TO NO CONFIG

Change-Id: I558dbf9a657d442e6e19468e543bbec855129eeb

1 files changed, 2909 insertions, 0 deletions

diff --git a/test/unit/test_service.py b/test/unit/test_service.py
new file mode 100644
index 0000000..ed3a2d6
--- /dev/null
+++ b/test/unit/test_service.py
@@ -0,0 +1,2909 @@
+# -*- coding: utf-8 -*-
+# Copyright (c) 2014 OpenStack Foundation
+#
+# 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.
+from __future__ import unicode_literals
+import contextlib
+import mock
+import os
+import six
+import tempfile
+import unittest
+import time
+
+from concurrent.futures import Future
+from hashlib import md5
+from mock import Mock, PropertyMock
+from six.moves.queue import Queue, Empty as QueueEmptyError
+from six import BytesIO
+from time import sleep
+
+import swiftclient
+import swiftclient.utils as utils
+from swiftclient.client import Connection, ClientException
+from swiftclient.service import (
+    SwiftService, SwiftError, SwiftUploadObject
+)
+
+from test.unit import utils as test_utils
+
+
+clean_os_environ = {}
+environ_prefixes = ('ST_', 'OS_')
+for key in os.environ:
+    if any(key.startswith(m) for m in environ_prefixes):
+        clean_os_environ[key] = ''
+
+
+if six.PY2:
+    import __builtin__ as builtins
+else:
+    import builtins
+
+
+class TestSwiftPostObject(unittest.TestCase):
+
+    def setUp(self):
+        super(TestSwiftPostObject, self).setUp()
+        self.spo = swiftclient.service.SwiftPostObject
+
+    def test_create(self):
+        spo = self.spo('obj_name')
+
+        self.assertEqual(spo.object_name, 'obj_name')
+        self.assertIsNone(spo.options)
+
+    def test_create_with_invalid_name(self):
+        # empty strings are not allowed as names
+        self.assertRaises(SwiftError, self.spo, '')
+
+        # names cannot be anything but strings
+        self.assertRaises(SwiftError, self.spo, 1)
+
+
+class TestSwiftCopyObject(unittest.TestCase):
+
+    def setUp(self):
+        super(TestSwiftCopyObject, self).setUp()
+        self.sco = swiftclient.service.SwiftCopyObject
+
+    def test_create(self):
+        sco = self.sco('obj_name')
+
+        self.assertEqual(sco.object_name, 'obj_name')
+        self.assertIsNone(sco.destination)
+        self.assertFalse(sco.fresh_metadata)
+
+        sco = self.sco('obj_name',
+                       {'destination': '/dest', 'fresh_metadata': True})
+
+        self.assertEqual(sco.object_name, 'obj_name')
+        self.assertEqual(sco.destination, '/dest/obj_name')
+        self.assertTrue(sco.fresh_metadata)
+
+        sco = self.sco('obj_name',
+                       {'destination': '/dest/new_obj/a',
+                        'fresh_metadata': False})
+
+        self.assertEqual(sco.object_name, 'obj_name')
+        self.assertEqual(sco.destination, '/dest/new_obj/a')
+        self.assertFalse(sco.fresh_metadata)
+
+    def test_create_with_invalid_name(self):
+        # empty strings are not allowed as names
+        self.assertRaises(SwiftError, self.sco, '')
+
+        # names cannot be anything but strings
+        self.assertRaises(SwiftError, self.sco, 1)
+
+
+class TestSwiftReader(unittest.TestCase):
+
+    def setUp(self):
+        super(TestSwiftReader, self).setUp()
+        self.sr = swiftclient.service._SwiftReader
+        self.md5_type = type(md5())
+
+    def test_create(self):
+        sr = self.sr('path', 'body', {})
+
+        self.assertEqual(sr._path, 'path')
+        self.assertEqual(sr._body, 'body')
+        self.assertIsNone(sr._content_length)
+        self.assertFalse(sr._expected_md5)
+
+        self.assertIsNone(sr._actual_md5)
+
+    def test_create_with_large_object_headers(self):
+        # md5 should not be initialized if large object headers are present
+        sr = self.sr('path', 'body', {'x-object-manifest': 'test',
+                                      'etag': '"%s"' % ('0' * 32)})
+        self.assertEqual(sr._path, 'path')
+        self.assertEqual(sr._body, 'body')
+        self.assertIsNone(sr._content_length)
+        self.assertFalse(sr._expected_md5)
+        self.assertIsNone(sr._actual_md5)
+
+        sr = self.sr('path', 'body', {'x-static-large-object': 'test',
+                                      'etag': '"%s"' % ('0' * 32)})
+        self.assertEqual(sr._path, 'path')
+        self.assertEqual(sr._body, 'body')
+        self.assertIsNone(sr._content_length)
+        self.assertFalse(sr._expected_md5)
+        self.assertIsNone(sr._actual_md5)
+
+    def test_create_with_content_range_header(self):
+        # md5 should not be initialized if large object headers are present
+        sr = self.sr('path', 'body', {'content-range': 'bytes 0-3/10',
+                                      'etag': '"%s"' % ('0' * 32)})
+        self.assertEqual(sr._path, 'path')
+        self.assertEqual(sr._body, 'body')
+        self.assertIsNone(sr._content_length)
+        self.assertFalse(sr._expected_md5)
+        self.assertIsNone(sr._actual_md5)
+
+    def test_create_with_ignore_checksum(self):
+        # md5 should not be initialized if checksum is False
+        sr = self.sr('path', 'body', {}, False)
+        self.assertEqual(sr._path, 'path')
+        self.assertEqual(sr._body, 'body')
+        self.assertIsNone(sr._content_length)
+        self.assertFalse(sr._expected_md5)
+        self.assertIsNone(sr._actual_md5)
+
+    def test_create_with_content_length(self):
+        sr = self.sr('path', 'body', {'content-length': 5})
+
+        self.assertEqual(sr._path, 'path')
+        self.assertEqual(sr._body, 'body')
+        self.assertEqual(sr._content_length, 5)
+        self.assertFalse(sr._expected_md5)
+
+        self.assertIsNone(sr._actual_md5)
+
+        # Check Contentlength raises error if it isn't an integer
+        self.assertRaises(SwiftError, self.sr, 'path', 'body',
+                          {'content-length': 'notanint'})
+
+    def test_iterator_usage(self):
+        def _consume(sr):
+            for _ in sr:
+                pass
+
+        sr = self.sr('path', BytesIO(b'body'), {})
+        _consume(sr)
+
+        # Check error is raised if expected etag doesn't match calculated md5.
+        # md5 for a SwiftReader that has done nothing is
+        # d41d8cd98f00b204e9800998ecf8427e  i.e md5 of nothing
+        sr = self.sr('path', BytesIO(b'body'),
+                     {'etag': md5(b'doesntmatch').hexdigest()})
+        self.assertRaises(SwiftError, _consume, sr)
+
+        sr = self.sr('path', BytesIO(b'body'),
+                     {'etag': md5(b'body').hexdigest()})
+        _consume(sr)
+
+        # Should still work if etag was quoted
+        sr = self.sr('path', BytesIO(b'body'),
+                     {'etag': '"%s"' % md5(b'body').hexdigest()})
+        _consume(sr)
+
+        # Check error is raised if SwiftReader doesn't read the same length
+        # as the content length it is created with
+        sr = self.sr('path', BytesIO(b'body'), {'content-length': 5})
+        self.assertRaises(SwiftError, _consume, sr)
+
+        sr = self.sr('path', BytesIO(b'body'), {'content-length': 4})
+        _consume(sr)
+
+        # Check that the iterator generates expected length and etag values
+        sr = self.sr('path', ['abc'.encode()] * 3,
+                     {'content-length': 9,
+                      'etag': md5('abc'.encode() * 3).hexdigest()})
+        _consume(sr)
+        self.assertEqual(sr._actual_read, 9)
+        self.assertEqual(sr._actual_md5.hexdigest(),
+                         md5('abc'.encode() * 3).hexdigest())
+
+
+class _TestServiceBase(unittest.TestCase):
+    def _get_mock_connection(self, attempts=2):
+        m = Mock(spec=Connection)
+        type(m).attempts = PropertyMock(return_value=attempts)
+        type(m).auth_end_time = PropertyMock(return_value=4)
+        return m
+
+    def _get_queue(self, q):
+        # Instead of blocking pull items straight from the queue.
+        # expects at least one item otherwise the test will fail.
+        try:
+            return q.get_nowait()
+        except QueueEmptyError:
+            self.fail('Expected item in queue but found none')
+
+    def _get_expected(self, update=None):
+        expected = self.expected.copy()
+        if update:
+            expected.update(update)
+
+        return expected
+
+
+class TestServiceDelete(_TestServiceBase):
+    def setUp(self):
+        super(TestServiceDelete, self).setUp()
+        self.opts = {'leave_segments': False, 'yes_all': False}
+        self.exc = Exception('test_exc')
+        # Base response to be copied and updated to matched the expected
+        # response for each test
+        self.expected = {
+            'action': None,   # Should be string in the form delete_XX
+            'container': 'test_c',
+            'object': 'test_o',
+            'attempts': 2,
+            'response_dict': {},
+            'success': None   # Should be a bool
+        }
+
+    def test_delete_segment(self):
+        mock_q = Queue()
+        mock_conn = self._get_mock_connection()
+        expected_r = self._get_expected({
+            'action': 'delete_segment',
+            'object': 'test_s',
+            'success': True,
+        })
+
+        r = SwiftService._delete_segment(mock_conn, 'test_c', 'test_s', mock_q)
+
+        mock_conn.delete_object.assert_called_once_with(
+            'test_c', 'test_s', response_dict={}
+        )
+        self.assertEqual(expected_r, r)
+        self.assertEqual(expected_r, self._get_queue(mock_q))
+
+    def test_delete_segment_exception(self):
+        mock_q = Queue()
+        mock_conn = self._get_mock_connection()
+        mock_conn.delete_object = Mock(side_effect=self.exc)
+        expected_r = self._get_expected({
+            'action': 'delete_segment',
+            'object': 'test_s',
+            'success': False,
+            'error': self.exc,
+            'traceback': mock.ANY,
+            'error_timestamp': mock.ANY
+        })
+
+        before = time.time()
+        r = SwiftService._delete_segment(mock_conn, 'test_c', 'test_s', mock_q)
+        after = time.time()
+
+        mock_conn.delete_object.assert_called_once_with(
+            'test_c', 'test_s', response_dict={}
+        )
+        self.assertEqual(expected_r, r)
+        self.assertEqual(expected_r, self._get_queue(mock_q))
+        self.assertGreaterEqual(r['error_timestamp'], before)
+        self.assertLessEqual(r['error_timestamp'], after)
+        self.assertIn('Traceback', r['traceback'])
+
+    def test_delete_object(self):
+        mock_q = Queue()
+        mock_conn = self._get_mock_connection()
+        mock_conn.head_object = Mock(return_value={})
+        expected_r = self._get_expected({
+            'action': 'delete_object',
+            'success': True
+        })
+
+        s = SwiftService()
+        r = s._delete_object(mock_conn, 'test_c', 'test_o', self.opts, mock_q)
+
+        mock_conn.head_object.assert_called_once_with(
+            'test_c', 'test_o', query_string='symlink=get', headers={})
+        mock_conn.delete_object.assert_called_once_with(
+            'test_c', 'test_o', query_string=None, response_dict={},
+            headers={}
+        )
+        self.assertEqual(expected_r, r)
+
+    def test_delete_object_with_headers(self):
+        mock_q = Queue()
+        mock_conn = self._get_mock_connection()
+        mock_conn.head_object = Mock(return_value={})
+        expected_r = self._get_expected({
+            'action': 'delete_object',
+            'success': True
+        })
+        opt_c = self.opts.copy()
+        opt_c['header'] = ['Skip-Middleware: Test']
+
+        s = SwiftService()
+        r = s._delete_object(mock_conn, 'test_c', 'test_o', opt_c, mock_q)
+
+        mock_conn.head_object.assert_called_once_with(
+            'test_c', 'test_o', headers={'Skip-Middleware': 'Test'},
+            query_string='symlink=get')
+        mock_conn.delete_object.assert_called_once_with(
+            'test_c', 'test_o', query_string=None, response_dict={},
+            headers={'Skip-Middleware': 'Test'}
+        )
+        self.assertEqual(expected_r, r)
+
+    def test_delete_object_exception(self):
+        mock_q = Queue()
+        mock_conn = self._get_mock_connection()
+        mock_conn.delete_object = Mock(side_effect=self.exc)
+        expected_r = self._get_expected({
+            'action': 'delete_object',
+            'success': False,
+            'error': self.exc,
+            'traceback': mock.ANY,
+            'error_timestamp': mock.ANY
+        })
+        # _delete_object doesn't populate attempts or response dict if it hits
+        # an error. This may not be the correct behaviour.
+        del expected_r['response_dict'], expected_r['attempts']
+
+        before = time.time()
+        s = SwiftService()
+        r = s._delete_object(mock_conn, 'test_c', 'test_o', self.opts, mock_q)
+        after = time.time()
+
+        mock_conn.head_object.assert_called_once_with(
+            'test_c', 'test_o', query_string='symlink=get', headers={})
+        mock_conn.delete_object.assert_called_once_with(
+            'test_c', 'test_o', query_string=None, response_dict={},
+            headers={}
+        )
+        self.assertEqual(expected_r, r)
+        self.assertGreaterEqual(r['error_timestamp'], before)
+        self.assertLessEqual(r['error_timestamp'], after)
+        self.assertIn('Traceback', r['traceback'])
+
+    def test_delete_object_slo_support(self):
+        # If SLO headers are present the delete call should include an
+        # additional query string to cause the right delete server side
+        mock_q = Queue()
+        mock_conn = self._get_mock_connection()
+        mock_conn.head_object = Mock(
+            return_value={'x-static-large-object': True}
+        )
+        expected_r = self._get_expected({
+            'action': 'delete_object',
+            'success': True
+        })
+
+        s = SwiftService()
+        r = s._delete_object(mock_conn, 'test_c', 'test_o', self.opts, mock_q)
+
+        mock_conn.head_object.assert_called_once_with(
+            'test_c', 'test_o', query_string='symlink=get', headers={})
+        mock_conn.delete_object.assert_called_once_with(
+            'test_c', 'test_o',
+            query_string='multipart-manifest=delete',
+            response_dict={},
+            headers={}
+        )
+        self.assertEqual(expected_r, r)
+
+    def test_delete_object_dlo_support(self):
+        mock_q = Queue()
+        s = SwiftService()
+        mock_conn = self._get_mock_connection()
+        expected_r = self._get_expected({
+            'action': 'delete_object',
+            'success': True,
+            'dlo_segments_deleted': True
+        })
+        # A DLO object is determined in _delete_object by heading the object
+        # and checking for the existence of a x-object-manifest header.
+        # Mock that here.
+        mock_conn.head_object = Mock(
+            return_value={'x-object-manifest': 'manifest_c/manifest_p'}
+        )
+        mock_conn.get_container = Mock(
+            side_effect=[(None, [{'name': 'test_seg_1'},
+                                 {'name': 'test_seg_2'}]),
+                         (None, {})]
+        )
+
+        def get_mock_list_conn(options):
+            return mock_conn
+
+        with mock.patch('swiftclient.service.get_conn', get_mock_list_conn):
+            r = s._delete_object(
+                mock_conn, 'test_c', 'test_o', self.opts, mock_q
+            )
+
+        self.assertEqual(expected_r, r)
+        expected = [
+            mock.call('test_c', 'test_o', query_string=None, response_dict={},
+                      headers={}),
+            mock.call('manifest_c', 'test_seg_1', response_dict={}),
+            mock.call('manifest_c', 'test_seg_2', response_dict={})]
+        mock_conn.delete_object.assert_has_calls(expected, any_order=True)
+
+    def test_delete_empty_container(self):
+        mock_conn = self._get_mock_connection()
+        expected_r = self._get_expected({
+            'action': 'delete_container',
+            'success': True,
+            'object': None
+        })
+
+        r = SwiftService._delete_empty_container(mock_conn, 'test_c',
+                                                 self.opts)
+
+        mock_conn.delete_container.assert_called_once_with(
+            'test_c', response_dict={}, headers={}
+        )
+        self.assertEqual(expected_r, r)
+
+    def test_delete_empty_container_with_headers(self):
+        mock_conn = self._get_mock_connection()
+        expected_r = self._get_expected({
+            'action': 'delete_container',
+            'success': True,
+            'object': None
+        })
+        opt_c = self.opts.copy()
+        opt_c['header'] = ['Skip-Middleware: Test']
+
+        r = SwiftService._delete_empty_container(mock_conn, 'test_c', opt_c)
+
+        mock_conn.delete_container.assert_called_once_with(
+            'test_c', response_dict={}, headers={'Skip-Middleware': 'Test'}
+        )
+        self.assertEqual(expected_r, r)
+
+    def test_delete_empty_container_exception(self):
+        mock_conn = self._get_mock_connection()
+        mock_conn.delete_container = Mock(side_effect=self.exc)
+        expected_r = self._get_expected({
+            'action': 'delete_container',
+            'success': False,
+            'object': None,
+            'error': self.exc,
+            'traceback': mock.ANY,
+            'error_timestamp': mock.ANY
+        })
+
+        before = time.time()
+        s = SwiftService()
+        r = s._delete_empty_container(mock_conn, 'test_c', {})
+        after = time.time()
+
+        mock_conn.delete_container.assert_called_once_with(
+            'test_c', response_dict={}, headers={}
+        )
+        self.assertEqual(expected_r, r)
+        self.assertGreaterEqual(r['error_timestamp'], before)
+        self.assertLessEqual(r['error_timestamp'], after)
+        self.assertIn('Traceback', r['traceback'])
+
+    @mock.patch.object(swiftclient.service.SwiftService, 'capabilities',
+                       lambda *a: {'action': 'capabilities',
+                                   'timestamp': time.time(),
+                                   'success': True,
+                                   'capabilities': {
+                                       'bulk_delete':
+                                       {'max_deletes_per_request': 10}}
+                                   })
+    def test_bulk_delete_page_size(self):
+        # make a list of 100 objects
+        obj_list = ['x%02d' % i for i in range(100)]
+        errors = []
+
+        # _bulk_delete_page_size uses 2x the number of threads to determine
+        # if if there are "many" object to delete or not
+
+        # format is: [(thread_count, expected result), ...]
+        obj_threads_exp = [
+            (10, 10),  # something small
+            (49, 10),  # just under the bounds
+            (50, 1),  # cutover point
+            (51, 1),  # just over bounds
+            (100, 1),  # something big
+        ]
+        for thread_count, exp in obj_threads_exp:
+            s = SwiftService(options={'object_dd_threads': thread_count})
+            res = s._bulk_delete_page_size(obj_list)
+            if res != exp:
+                msg = 'failed for thread_count %d: got %r expected %r' % \
+                    (thread_count, res, exp)
+                errors.append(msg)
+        if errors:
+            self.fail('_bulk_delete_page_size() failed\n' + '\n'.join(errors))
+
+
+class TestSwiftError(unittest.TestCase):
+
+    def test_is_exception(self):
+        se = SwiftError(5)
+        self.assertIsInstance(se, Exception)
+
+    def test_empty_swifterror_creation(self):
+        se = SwiftError(5)
+
+        self.assertEqual(se.value, 5)
+        self.assertIsNone(se.container)
+        self.assertIsNone(se.obj)
+        self.assertIsNone(se.segment)
+        self.assertIsNone(se.exception)
+
+        self.assertEqual(str(se), '5')
+
+    def test_swifterror_creation(self):
+        test_exc = Exception('test exc')
+        se = SwiftError(5, 'con', 'obj', 'seg', test_exc)
+
+        self.assertEqual(se.value, 5)
+        self.assertEqual(se.container, 'con')
+        self.assertEqual(se.obj, 'obj')
+        self.assertEqual(se.segment, 'seg')
+        self.assertEqual(se.exception, test_exc)
+
+        self.assertEqual(str(se), '5 container:con object:obj segment:seg')
+
+
+class TestServiceUtils(unittest.TestCase):
+
+    def setUp(self):
+        super(TestServiceUtils, self).setUp()
+        with mock.patch.dict(swiftclient.service.environ, clean_os_environ):
+            swiftclient.service._default_global_options = \
+                swiftclient.service._build_default_global_options()
+        self.opts = swiftclient.service._default_global_options.copy()
+
+    def test_process_options_defaults(self):
+        # The only actions that should be taken on default options set is
+        # to change the auth version to v2.0 and create the os_options dict
+        opt_c = self.opts.copy()
+
+        swiftclient.service.process_options(opt_c)
+
+        self.assertIn('os_options', opt_c)
+        del opt_c['os_options']
+        self.assertEqual(opt_c['auth_version'], '2.0')
+        opt_c['auth_version'] = '1.0'
+
+        self.assertEqual(opt_c, self.opts)
+
+    def test_process_options_auth_version(self):
+        # auth_version should be set to 2.0
+        # if it isn't already set to 3.0
+        # and the v1 command line arguments aren't present
+        opt_c = self.opts.copy()
+
+        # Check v3 isn't changed
+        opt_c['auth_version'] = '3'
+        swiftclient.service.process_options(opt_c)
+        self.assertEqual(opt_c['auth_version'], '3')
+
+        # Check v1 isn't changed if user, key and auth are set
+        opt_c = self.opts.copy()
+        opt_c['auth_version'] = '1'
+        opt_c['auth'] = True
+        opt_c['user'] = True
+        opt_c['key'] = True
+        swiftclient.service.process_options(opt_c)
+        self.assertEqual(opt_c['auth_version'], '1')
+
+    def test_process_options_new_style_args(self):
+        # checks new style args are copied to old style
+        # when old style don't exist
+        opt_c = self.opts.copy()
+
+        opt_c['auth'] = ''
+        opt_c['user'] = ''
+        opt_c['key'] = ''
+        opt_c['os_auth_url'] = 'os_auth'
+        opt_c['os_username'] = 'os_user'
+        opt_c['os_password'] = 'os_pass'
+        swiftclient.service.process_options(opt_c)
+        self.assertEqual(opt_c['auth_version'], '2.0')
+        self.assertEqual(opt_c['auth'], 'os_auth')
+        self.assertEqual(opt_c['user'], 'os_user')
+        self.assertEqual(opt_c['key'], 'os_pass')
+
+        # Check old style args are left alone if they exist
+        opt_c = self.opts.copy()
+        opt_c['auth'] = 'auth'
+        opt_c['user'] = 'user'
+        opt_c['key'] = 'key'
+        opt_c['os_auth_url'] = 'os_auth'
+        opt_c['os_username'] = 'os_user'
+        opt_c['os_password'] = 'os_pass'
+        swiftclient.service.process_options(opt_c)
+        self.assertEqual(opt_c['auth_version'], '1.0')
+        self.assertEqual(opt_c['auth'], 'auth')
+        self.assertEqual(opt_c['user'], 'user')
+        self.assertEqual(opt_c['key'], 'key')
+
+    def test_split_headers(self):
+        mock_headers = ['color:blue', 'SIZE: large']
+        expected = {'Color': 'blue', 'Size': 'large'}
+
+        actual = swiftclient.service.split_headers(mock_headers)
+        self.assertEqual(expected, actual)
+
+    def test_split_headers_prefix(self):
+        mock_headers = ['color:blue', 'size:large']
+        expected = {'Prefix-Color': 'blue', 'Prefix-Size': 'large'}
+
+        actual = swiftclient.service.split_headers(mock_headers, 'prefix-')
+        self.assertEqual(expected, actual)
+
+    def test_split_headers_list_of_tuples(self):
+        mock_headers = [('color', 'blue'), ('size', 'large')]
+        expected = {'Prefix-Color': 'blue', 'Prefix-Size': 'large'}
+
+        actual = swiftclient.service.split_headers(mock_headers, 'prefix-')
+        self.assertEqual(expected, actual)
+
+    def test_split_headers_dict(self):
+        expected = {'Color': 'blue', 'Size': 'large'}
+
+        actual = swiftclient.service.split_headers(expected)
+        self.assertEqual(expected, actual)
+
+    def test_split_headers_error(self):
+        self.assertRaises(SwiftError, swiftclient.service.split_headers,
+                          ['notvalid'])
+        self.assertRaises(SwiftError, swiftclient.service.split_headers,
+                          [('also', 'not', 'valid')])
+
+
+class TestSwiftUploadObject(unittest.TestCase):
+
+    def setUp(self):
+        self.suo = swiftclient.service.SwiftUploadObject
+        super(TestSwiftUploadObject, self).setUp()
+
+    def test_create_with_string(self):
+        suo = self.suo('source')
+        self.assertEqual(suo.source, 'source')
+        self.assertEqual(suo.object_name, 'source')
+        self.assertIsNone(suo.options)
+
+        suo = self.suo('source', 'obj_name')
+        self.assertEqual(suo.source, 'source')
+        self.assertEqual(suo.object_name, 'obj_name')
+        self.assertIsNone(suo.options)
+
+        suo = self.suo('source', 'obj_name', {'opt': '123'})
+        self.assertEqual(suo.source, 'source')
+        self.assertEqual(suo.object_name, 'obj_name')
+        self.assertEqual(suo.options, {'opt': '123'})
+
+    def test_create_with_file(self):
+        with tempfile.TemporaryFile() as mock_file:
+            # Check error is raised if no object name is provided with a
+            # filelike object
+            self.assertRaises(SwiftError, self.suo, mock_file)
+
+            # Check that empty strings are invalid object names
+            self.assertRaises(SwiftError, self.suo, mock_file, '')
+
+            suo = self.suo(mock_file, 'obj_name')
+            self.assertEqual(suo.source, mock_file)
+            self.assertEqual(suo.object_name, 'obj_name')
+            self.assertIsNone(suo.options)
+
+            suo = self.suo(mock_file, 'obj_name', {'opt': '123'})
+            self.assertEqual(suo.source, mock_file)
+            self.assertEqual(suo.object_name, 'obj_name')
+            self.assertEqual(suo.options, {'opt': '123'})
+
+    def test_create_with_no_source(self):
+        suo = self.suo(None, 'obj_name')
+        self.assertIsNone(suo.source)
+        self.assertEqual(suo.object_name, 'obj_name')
+        self.assertIsNone(suo.options)
+
+        # Check error is raised if source is None without an object name
+        self.assertRaises(SwiftError, self.suo, None)
+
+    def test_create_with_invalid_source(self):
+        # Source can only be None, string or filelike object,
+        # check an error is raised with an invalid type.
+        self.assertRaises(SwiftError, self.suo, [])
+
+
+class TestServiceList(_TestServiceBase):
+    def setUp(self):
+        super(TestServiceList, self).setUp()
+        self.opts = {'prefix': None, 'long': False, 'delimiter': ''}
+        self.exc = Exception('test_exc')
+        # Base response to be copied and updated to matched the expected
+        # response for each test
+        self.expected = {
+            'action': None,   # Should be list_X_part (account or container)
+            'container': None,   # Should be a string when listing a container
+            'prefix': None,
+            'success': None   # Should be a bool
+        }
+
+    def test_list_account(self):
+        mock_q = Queue()
+        mock_conn = self._get_mock_connection()
+        get_account_returns = [
+            (None, [{'name': 'test_c'}]),
+            (None, [])
+        ]
+        mock_conn.get_account = Mock(side_effect=get_account_returns)
+
+        expected_r = self._get_expected({
+            'action': 'list_account_part',
+            'success': True,
+            'listing': [{'name': 'test_c'}],
+            'marker': ''
+        })
+
+        SwiftService._list_account_job(
+            mock_conn, self.opts, mock_q
+        )
+        self.assertEqual(expected_r, self._get_queue(mock_q))
+        self.assertIsNone(self._get_queue(mock_q))
+
+        long_opts = dict(self.opts, **{'long': True})
+        mock_conn.head_container = Mock(return_value={'test_m': '1'})
+        get_account_returns = [
+            (None, [{'name': 'test_c'}]),
+            (None, [])
+        ]
+        mock_conn.get_account = Mock(side_effect=get_account_returns)
+
+        expected_r_long = self._get_expected({
+            'action': 'list_account_part',
+            'success': True,
+            'listing': [{'name': 'test_c', 'meta': {'test_m': '1'}}],
+            'marker': '',
+        })
+
+        SwiftService._list_account_job(
+            mock_conn, long_opts, mock_q
+        )
+        self.assertEqual(expected_r_long, self._get_queue(mock_q))
+        self.assertIsNone(self._get_queue(mock_q))
+
+    def test_list_account_with_headers(self):
+        mock_q = Queue()
+        mock_conn = self._get_mock_connection()
+        get_account_returns = [
+            (None, [{'name': 'test_c'}]),
+            (None, [])
+        ]
+        mock_conn.get_account = Mock(side_effect=get_account_returns)
+
+        expected_r = self._get_expected({
+            'action': 'list_account_part',
+            'success': True,
+            'listing': [{'name': 'test_c'}],
+            'marker': ''
+        })
+        opt_c = self.opts.copy()
+        opt_c['header'] = ['Skip-Middleware: True']
+        SwiftService._list_account_job(
+            mock_conn, opt_c, mock_q
+        )
+        self.assertEqual(expected_r, self._get_queue(mock_q))
+        self.assertIsNone(self._get_queue(mock_q))
+        self.assertEqual(mock_conn.get_account.mock_calls, [
+            mock.call(headers={'Skip-Middleware': 'True'}, marker='',
+                      prefix=None),
+            mock.call(headers={'Skip-Middleware': 'True'}, marker='test_c',
+                      prefix=None)])
+
+    def test_list_account_exception(self):
+        mock_q = Queue()
+        mock_conn = self._get_mock_connection()
+        mock_conn.get_account = Mock(side_effect=self.exc)
+        expected_r = self._get_expected({
+            'action': 'list_account_part',
+            'success': False,
+            'error': self.exc,
+            'marker': '',
+            'traceback': mock.ANY,
+            'error_timestamp': mock.ANY
+        })
+
+        SwiftService._list_account_job(
+            mock_conn, self.opts, mock_q)
+
+        mock_conn.get_account.assert_called_once_with(
+            marker='', prefix=None, headers={}
+        )
+        self.assertEqual(expected_r, self._get_queue(mock_q))
+        self.assertIsNone(self._get_queue(mock_q))
+
+    def test_list_container(self):
+        mock_q = Queue()
+        mock_conn = self._get_mock_connection()
+        get_container_returns = [
+            (None, [{'name': 'test_o'}]),
+            (None, [])
+        ]
+        mock_conn.get_container = Mock(side_effect=get_container_returns)
+
+        expected_r = self._get_expected({
+            'action': 'list_container_part',
+            'container': 'test_c',
+            'success': True,
+            'listing': [{'name': 'test_o'}],
+            'marker': ''
+        })
+
+        SwiftService._list_container_job(
+            mock_conn, 'test_c', self.opts, mock_q
+        )
+        self.assertEqual(expected_r, self._get_queue(mock_q))
+        self.assertIsNone(self._get_queue(mock_q))
+
+        long_opts = dict(self.opts, **{'long': True})
+        mock_conn.head_container = Mock(return_value={'test_m': '1'})
+        get_container_returns = [
+            (None, [{'name': 'test_o'}]),
+            (None, [])
+        ]
+        mock_conn.get_container = Mock(side_effect=get_container_returns)
+
+        expected_r_long = self._get_expected({
+            'action': 'list_container_part',
+            'container': 'test_c',
+            'success': True,
+            'listing': [{'name': 'test_o'}],
+            'marker': ''
+        })
+
+        SwiftService._list_container_job(
+            mock_conn, 'test_c', long_opts, mock_q
+        )
+        self.assertEqual(expected_r_long, self._get_queue(mock_q))
+        self.assertIsNone(self._get_queue(mock_q))
+
+    def test_list_container_marker(self):
+        mock_q = Queue()
+        mock_conn = self._get_mock_connection()
+
+        get_container_returns = [
+            (None, [{'name': 'b'}, {'name': 'c'}]),
+            (None, [])
+        ]
+        mock_get_cont = Mock(side_effect=get_container_returns)
+        mock_conn.get_container = mock_get_cont
+
+        expected_r = self._get_expected({
+            'action': 'list_container_part',
+            'container': 'test_c',
+            'success': True,
+            'listing': [{'name': 'b'}, {'name': 'c'}],
+            'marker': 'b'
+        })
+
+        _opts = self.opts.copy()
+        _opts['marker'] = 'b'
+        SwiftService._list_container_job(mock_conn, 'test_c', _opts, mock_q)
+
+        # This does not test if the marker is propagated, because we always
+        # get the final call to the get_container with the final item 'c',
+        # even if marker wasn't set. This test just makes sure the whole
+        # stack works in a sane way.
+        mock_kw = mock_get_cont.call_args[1]
+        self.assertEqual(mock_kw['marker'], 'c')
+
+        # This tests that the lower levels get the marker delivered.
+        self.assertEqual(expected_r, self._get_queue(mock_q))
+
+        self.assertIsNone(self._get_queue(mock_q))
+
+    def test_list_container_with_headers(self):
+        mock_q = Queue()
+        mock_conn = self._get_mock_connection()
+        get_container_returns = [
+            (None, [{'name': 'test_o'}]),
+            (None, [])
+        ]
+        mock_conn.get_container = Mock(side_effect=get_container_returns)
+
+        expected_r = self._get_expected({
+            'action': 'list_container_part',
+            'container': 'test_c',
+            'success': True,
+            'listing': [{'name': 'test_o'}],
+            'marker': ''
+        })
+
+        opt_c = self.opts.copy()
+        opt_c['header'] = ['Skip-Middleware: Test']
+
+        SwiftService._list_container_job(
+            mock_conn, 'test_c', opt_c, mock_q
+        )
+        self.assertEqual(expected_r, self._get_queue(mock_q))
+        self.assertIsNone(self._get_queue(mock_q))
+        self.assertEqual(mock_conn.get_container.mock_calls, [
+            mock.call('test_c', headers={'Skip-Middleware': 'Test'},
+                      delimiter='', marker='', prefix=None),
+            mock.call('test_c', headers={'Skip-Middleware': 'Test'},
+                      delimiter='', marker='test_o', prefix=None)])
+
+    def test_list_container_exception(self):
+        mock_q = Queue()
+        mock_conn = self._get_mock_connection()
+        mock_conn.get_container = Mock(side_effect=self.exc)
+        expected_r = self._get_expected({
+            'action': 'list_container_part',
+            'container': 'test_c',
+            'success': False,
+            'error': self.exc,
+            'marker': '',
+            'error_timestamp': mock.ANY,
+            'traceback': mock.ANY
+        })
+
+        SwiftService._list_container_job(
+            mock_conn, 'test_c', self.opts, mock_q
+        )
+
+        mock_conn.get_container.assert_called_once_with(
+            'test_c', marker='', delimiter='', prefix=None, headers={}
+        )
+        self.assertEqual(expected_r, self._get_queue(mock_q))
+        self.assertIsNone(self._get_queue(mock_q))
+
+    @mock.patch('swiftclient.service.get_conn')
+    def test_list_queue_size(self, mock_get_conn):
+        mock_conn = self._get_mock_connection()
+        # Return more results than should fit in the results queue
+        get_account_returns = [
+            (None, [{'name': 'container1'}]),
+            (None, [{'name': 'container2'}]),
+            (None, [{'name': 'container3'}]),
+            (None, [{'name': 'container4'}]),
+            (None, [{'name': 'container5'}]),
+            (None, [{'name': 'container6'}]),
+            (None, [{'name': 'container7'}]),
+            (None, [{'name': 'container8'}]),
+            (None, [{'name': 'container9'}]),
+            (None, [{'name': 'container10'}]),
+            (None, [{'name': 'container11'}]),
+            (None, [{'name': 'container12'}]),
+            (None, [{'name': 'container13'}]),
+            (None, [{'name': 'container14'}]),
+            (None, [])
+        ]
+        mock_conn.get_account = Mock(side_effect=get_account_returns)
+        mock_get_conn.return_value = mock_conn
+
+        s = SwiftService(options=self.opts)
+        lg = s.list()
+
+        # Start the generator
+        first_list_part = next(lg)
+
+        # Wait for the number of calls to get_account to reach our expected
+        # value, then let it run some more to make sure the value remains
+        # stable
+        count = mock_conn.get_account.call_count
+        stable = 0
+        while mock_conn.get_account.call_count != count or stable < 5:
+            if mock_conn.get_account.call_count == count:
+                stable += 1
+            else:
+                count = mock_conn.get_account.call_count
+                stable = 0
+            # The test requires a small sleep to allow other threads to
+            # execute - in this mocked environment we assume that if the call
+            # count to get_account has not changed in 0.25s then no more calls
+            # will be made.
+            sleep(0.05)
+
+        stable_get_account_call_count = mock_conn.get_account.call_count
+
+        # Collect all remaining results from the generator
+        list_results = [first_list_part] + list(lg)
+
+        # Make sure the stable call count is correct - this should be 12 calls
+        # to get_account;
+        #  1 for first_list_part
+        #  10 for the values on the queue
+        #  1 for the value blocking whilst trying to place onto the queue
+        self.assertEqual(12, stable_get_account_call_count)
+
+        # Make sure all the containers were listed and placed onto the queue
+        self.assertEqual(15, mock_conn.get_account.call_count)
+
+        # Check the results were all returned
+        observed_listing = []
+        for lir in list_results:
+            observed_listing.append(
+                [li['name'] for li in lir['listing']]
+            )
+        expected_listing = []
+        for gar in get_account_returns[:-1]:  # The empty list is not returned
+            expected_listing.append(
+                [li['name'] for li in gar[1]]
+            )
+        self.assertEqual(observed_listing, expected_listing)
+
+
+class TestService(unittest.TestCase):
+
+    def test_upload_with_bad_segment_size(self):
+        for bad in ('ten', '1234X', '100.3'):
+            options = {'segment_size': bad}
+            try:
+                service = SwiftService(options)
+                next(service.upload('c', 'o'))
+                self.fail('Expected SwiftError when segment_size=%s' % bad)
+            except SwiftError as exc:
+                self.assertEqual('Segment size should be an integer value',
+                                 exc.value)
+
+    @mock.patch('swiftclient.service.stat')
+    @mock.patch('swiftclient.service.getmtime', return_value=1.0)
+    @mock.patch('swiftclient.service.getsize', return_value=4)
+    def test_upload_with_relative_path(self, *args, **kwargs):
+        service = SwiftService({})
+        objects = [{'path': "./testobj",
+                    'strt_indx': 2},
+                   {'path': os.path.join(os.getcwd(), "testobj"),
+                    'strt_indx': 1},
+                   {'path': ".\\testobj",
+                    'strt_indx': 2}]
+        for obj in objects:
+            with mock.patch('swiftclient.service.Connection') as mock_conn, \
+                    mock.patch.object(builtins, 'open') as mock_open:
+                mock_open.return_value = six.StringIO('asdf')
+                mock_conn.return_value.head_object.side_effect = \
+                    ClientException('Not Found', http_status=404)
+                mock_conn.return_value.put_object.return_value =\
+                    'd41d8cd98f00b204e9800998ecf8427e'
+                resp_iter = service.upload(
+                    'c', [SwiftUploadObject(obj['path'])])
+                responses = [x for x in resp_iter]
+                for resp in responses:
+                    self.assertIsNone(resp.get('error'))
+                    self.assertIs(True, resp['success'])
+                self.assertEqual(2, len(responses))
+                create_container_resp, upload_obj_resp = responses
+                self.assertEqual(create_container_resp['action'],
+                                 'create_container')
+                self.assertEqual(upload_obj_resp['action'],
+                                 'upload_object')
+                self.assertEqual(upload_obj_resp['object'],
+                                 obj['path'][obj['strt_indx']:])
+                self.assertEqual(upload_obj_resp['path'], obj['path'])
+                self.assertTrue(mock_open.return_value.closed)
+
+    @mock.patch('swiftclient.service.Connection')
+    def test_upload_stream(self, mock_conn):
+        service = SwiftService({})
+
+        stream = test_utils.FakeStream(2048)
+        segment_etag = md5(b'A' * 1024).hexdigest()
+
+        mock_conn.return_value.head_object.side_effect = \
+            ClientException('Not Found', http_status=404)
+        mock_conn.return_value.put_object.return_value = \
+            segment_etag
+        options = {'use_slo': True, 'segment_size': 1024}
+        resp_iter = service.upload(
+            'container',
+            [SwiftUploadObject(stream, object_name='streamed')],
+            options)
+        responses = [x for x in resp_iter]
+        for resp in responses:
+            self.assertFalse('error' in resp)
+            self.assertTrue(resp['success'])
+        self.assertEqual(5, len(responses))
+        container_resp, segment_container_resp = responses[0:2]
+        segment_response = responses[2:4]
+        upload_obj_resp = responses[-1]
+        self.assertEqual(container_resp['action'],
+                         'create_container')
+        self.assertEqual(upload_obj_resp['action'],
+                         'upload_object')
+        self.assertEqual(upload_obj_resp['object'],
+                         'streamed')
+        self.assertTrue(upload_obj_resp['path'] is None)
+        self.assertTrue(upload_obj_resp['large_object'])
+        self.assertIn('manifest_response_dict', upload_obj_resp)
+        self.assertEqual(upload_obj_resp['manifest_response_dict'], {})
+        for i, resp in enumerate(segment_response):
+            self.assertEqual(i, resp['segment_index'])
+            self.assertEqual(1024, resp['segment_size'])
+            self.assertEqual('d47b127bc2de2d687ddc82dac354c415',
+                             resp['segment_etag'])
+            self.assertTrue(resp['segment_location'].endswith(
+                '/0000000%d' % i))
+            self.assertTrue(resp['segment_location'].startswith(
+                '/container_segments/streamed'))
+
+    @mock.patch('swiftclient.service.Connection')
+    def test_upload_stream_fits_in_one_segment(self, mock_conn):
+        service = SwiftService({})
+
+        stream = test_utils.FakeStream(2048)
+        whole_etag = md5(b'A' * 2048).hexdigest()
+
+        mock_conn.return_value.head_object.side_effect = \
+            ClientException('Not Found', http_status=404)
+        mock_conn.return_value.put_object.return_value = \
+            whole_etag
+        options = {'use_slo': True, 'segment_size': 10240}
+        resp_iter = service.upload(
+            'container',
+            [SwiftUploadObject(stream, object_name='streamed')],
+            options)
+        responses = [x for x in resp_iter]
+        for resp in responses:
+            self.assertNotIn('error', resp)
+            self.assertTrue(resp['success'])
+        self.assertEqual(3, len(responses))
+        container_resp, segment_container_resp = responses[0:2]
+        upload_obj_resp = responses[-1]
+        self.assertEqual(container_resp['action'],
+                         'create_container')
+        self.assertEqual(upload_obj_resp['action'],
+                         'upload_object')
+        self.assertEqual(upload_obj_resp['object'],
+                         'streamed')
+        self.assertTrue(upload_obj_resp['path'] is None)
+        self.assertFalse(upload_obj_resp['large_object'])
+        self.assertNotIn('manifest_response_dict', upload_obj_resp)
+
+
+class TestServiceUpload(_TestServiceBase):
+
+    @contextlib.contextmanager
+    def assert_open_results_are_closed(self):
+        opened_files = []
+        builtin_open = builtins.open
+
+        def open_wrapper(*a, **kw):
+            opened_files.append((builtin_open(*a, **kw), a, kw))
+            return opened_files[-1][0]
+
+        with mock.patch.object(builtins, 'open', open_wrapper):
+            yield
+        for fp, args, kwargs in opened_files:
+            formatted_args = [repr(a) for a in args]
+            formatted_args.extend('%s=%r' % kv for kv in kwargs.items())
+            formatted_args = ', '.join(formatted_args)
+            self.assertTrue(fp.closed,
+                            'Failed to close open(%s)' % formatted_args)
+
+    def test_upload_object_job_file_with_unicode_path(self):
+        # Uploading a file results in the file object being wrapped in a
+        # LengthWrapper. This test sets the options in such a way that much
+        # of _upload_object_job is skipped bringing the critical path down
+        # to around 60 lines to ease testing.
+        with tempfile.NamedTemporaryFile() as f:
+            f.write(b'a' * 30)
+            f.flush()
+            expected_r = {
+                'action': 'upload_object',
+                'attempts': 2,
+                'container': 'test_c',
+                'headers': {},
+                'large_object': True,
+                'object': 'テスト/dummy.dat',
+                'manifest_response_dict': {},
+                'segment_results': [{'action': 'upload_segment',
+                                    'success': True}] * 3,
+                'status': 'uploaded',
+                'success': True,
+            }
+            expected_mtime = '%f' % os.path.getmtime(f.name)
+
+            mock_conn = mock.Mock()
+            mock_conn.put_object.return_value = ''
+            type(mock_conn).attempts = mock.PropertyMock(return_value=2)
+
+            s = SwiftService()
+            with mock.patch.object(s, '_upload_segment_job') as mock_job:
+                mock_job.return_value = {
+                    'action': 'upload_segment',
+                    'success': True}
+
+                r = s._upload_object_job(conn=mock_conn,
+                                         container='test_c',
+                                         source=f.name,
+                                         obj='テスト/dummy.dat',
+                                         options=dict(s._options,
+                                                      segment_size=10,
+                                                      leave_segments=True))
+
+            mtime = r['headers']['x-object-meta-mtime']
+            self.assertEqual(expected_mtime, mtime)
+            del r['headers']['x-object-meta-mtime']
+
+            self.assertEqual(
+                'test_c_segments/%E3%83%86%E3%82%B9%E3%83%88/dummy.dat/' +
+                '%s/30/10/' % mtime, r['headers']['x-object-manifest'])
+            del r['headers']['x-object-manifest']
+
+            self.assertEqual(r['path'], f.name)
+            del r['path']
+
+            self.assertEqual(r, expected_r)
+            self.assertEqual(mock_conn.put_object.call_count, 1)
+            mock_conn.put_object.assert_called_with('test_c', 'テスト/dummy.dat',
+                                                    '',
+                                                    content_length=0,
+                                                    headers={},
+                                                    response_dict={})
+
+    def test_upload_segment_job(self):
+        with tempfile.NamedTemporaryFile() as f:
+            f.write(b'a' * 10)
+            f.write(b'b' * 10)
+            f.write(b'c' * 10)
+            f.flush()
+
+            # run read() when put_object is called to calculate md5sum
+            def _consuming_conn(*a, **kw):
+                contents = a[2]
+                contents.read()  # Force md5 calculation
+                return contents.get_md5sum()
+
+            mock_conn = mock.Mock()
+            mock_conn.put_object.side_effect = _consuming_conn
+            type(mock_conn).attempts = mock.PropertyMock(return_value=2)
+            expected_r = {
+                'action': 'upload_segment',
+                'for_container': 'test_c',
+                'for_object': 'test_o',
+                'segment_index': 2,
+                'segment_size': 10,
+                'segment_location': '/test_c_segments/test_s_1',
+                'log_line': 'test_o segment 2',
+                'success': True,
+                'response_dict': {},
+                'segment_etag': md5(b'b' * 10).hexdigest(),
+                'attempts': 2,
+            }
+
+            s = SwiftService()
+            with self.assert_open_results_are_closed():
+                r = s._upload_segment_job(conn=mock_conn,
+                                          path=f.name,
+                                          container='test_c',
+                                          segment_name='test_s_1',
+                                          segment_start=10,
+                                          segment_size=10,
+                                          segment_index=2,
+                                          obj_name='test_o',
+                                          options={'segment_container': None,
+                                                   'checksum': True})
+
+            self.assertEqual(r, expected_r)
+
+            self.assertEqual(mock_conn.put_object.call_count, 1)
+            mock_conn.put_object.assert_called_with(
+                'test_c_segments', 'test_s_1',
+                mock.ANY,
+                content_length=10,
+                content_type='application/swiftclient-segment',
+                response_dict={})
+            contents = mock_conn.put_object.call_args[0][2]
+            self.assertIsInstance(contents, utils.LengthWrapper)
+            self.assertEqual(len(contents), 10)
+
+    def test_upload_stream_segment(self):
+        common_params = {
+            'segment_container': 'segments',
+            'segment_name': 'test_stream_2',
+            'container': 'test_stream',
+            'object': 'stream_object',
+        }
+        tests = [
+            {'test_params': {
+                'segment_size': 1024,
+                'segment_index': 2,
+                'content_size': 1024},
+             'put_object_args': {
+                'container': 'segments',
+                'object': 'test_stream_2'},
+             'expected': {
+                'complete': False,
+                'segment_etag': md5(b'A' * 1024).hexdigest()}},
+            {'test_params': {
+                'segment_size': 2048,
+                'segment_index': 0,
+                'content_size': 512},
+             'put_object_args': {
+                'container': 'test_stream',
+                'object': 'stream_object'},
+             'expected': {
+                'complete': True,
+                'segment_etag': md5(b'A' * 512).hexdigest()}},
+            # 0-sized segment should not be uploaded
+            {'test_params': {
+                'segment_size': 1024,
+                'segment_index': 1,
+                'content_size': 0},
+             'put_object_args': {},
+             'expected': {
+                'complete': True}},
+            # 0-sized objects should be uploaded
+            {'test_params': {
+                'segment_size': 1024,
+                'segment_index': 0,
+                'content_size': 0},
+             'put_object_args': {
+                'container': 'test_stream',
+                'object': 'stream_object'},
+             'expected': {
+                'complete': True,
+                'segment_etag': md5(b'').hexdigest()}},
+            # Test boundary conditions
+            {'test_params': {
+                'segment_size': 1024,
+                'segment_index': 1,
+                'content_size': 1023},
+             'put_object_args': {
+                'container': 'segments',
+                'object': 'test_stream_2'},
+             'expected': {
+                'complete': True,
+                'segment_etag': md5(b'A' * 1023).hexdigest()}},
+            {'test_params': {
+                'segment_size': 2048,
+                'segment_index': 0,
+                'content_size': 2047},
+             'put_object_args': {
+                'container': 'test_stream',
+                'object': 'stream_object'},
+             'expected': {
+                'complete': True,
+                'segment_etag': md5(b'A' * 2047).hexdigest()}},
+            {'test_params': {
+                'segment_size': 1024,
+                'segment_index': 2,
+                'content_size': 1025},
+             'put_object_args': {
+                'container': 'segments',
+                'object': 'test_stream_2'},
+             'expected': {
+                'complete': False,
+                'segment_etag': md5(b'A' * 1024).hexdigest()}},
+        ]
+
+        for test_args in tests:
+            params = test_args['test_params']
+            stream = test_utils.FakeStream(params['content_size'])
+            segment_size = params['segment_size']
+            segment_index = params['segment_index']
+
+            def _fake_put_object(*args, **kwargs):
+                contents = args[2]
+                # Consume and compute md5
+                return md5(contents).hexdigest()
+
+            mock_conn = mock.Mock()
+            mock_conn.put_object.side_effect = _fake_put_object
+
+            s = SwiftService()
+            resp = s._upload_stream_segment(
+                conn=mock_conn,
+                container=common_params['container'],
+                object_name=common_params['object'],
+                segment_container=common_params['segment_container'],
+                segment_name=common_params['segment_name'],
+                segment_size=segment_size,
+                segment_index=segment_index,
+                headers={},
+                fd=stream)
+            expected_args = test_args['expected']
+            put_args = test_args['put_object_args']
+            expected_response = {
+                'segment_size': min(len(stream), segment_size),
+                'complete': expected_args['complete'],
+                'success': True,
+            }
+            if len(stream) or segment_index == 0:
+                segment_location = '/%s/%s' % (put_args['container'],
+                                               put_args['object'])
+                expected_response.update(
+                    {'segment_index': segment_index,
+                     'segment_location': segment_location,
+                     'segment_etag': expected_args['segment_etag'],
+                     'for_object': common_params['object']})
+                mock_conn.put_object.assert_called_once_with(
+                    put_args['container'],
+                    put_args['object'],
+                    mock.ANY,
+                    content_length=min(len(stream), segment_size),
+                    headers={'etag': expected_args['segment_etag']},
+                    response_dict=mock.ANY)
+            else:
+                self.assertEqual([], mock_conn.put_object.mock_calls)
+                expected_response.update(
+                    {'segment_index': None,
+                     'segment_location': None,
+                     'segment_etag': None})
+            self.assertEqual(expected_response, resp)
+
+    def test_etag_mismatch_with_ignore_checksum(self):
+        def _consuming_conn(*a, **kw):
+            contents = a[2]
+            contents.read()  # Force md5 calculation
+            return 'badresponseetag'
+
+        with tempfile.NamedTemporaryFile() as f:
+            f.write(b'a' * 10)
+            f.write(b'b' * 10)
+            f.write(b'c' * 10)
+            f.flush()
+
+            mock_conn = mock.Mock()
+            mock_conn.put_object.side_effect = _consuming_conn
+            type(mock_conn).attempts = mock.PropertyMock(return_value=2)
+
+            s = SwiftService()
+            r = s._upload_segment_job(conn=mock_conn,
+                                      path=f.name,
+                                      container='test_c',
+                                      segment_name='test_s_1',
+                                      segment_start=10,
+                                      segment_size=10,
+                                      segment_index=2,
+                                      obj_name='test_o',
+                                      options={'segment_container': None,
+                                               'checksum': False})
+
+            self.assertIsNone(r.get('error'))
+            self.assertEqual(mock_conn.put_object.call_count, 1)
+            mock_conn.put_object.assert_called_with(
+                'test_c_segments', 'test_s_1',
+                mock.ANY,
+                content_length=10,
+                content_type='application/swiftclient-segment',
+                response_dict={})
+            contents = mock_conn.put_object.call_args[0][2]
+            # Check that md5sum is not calculated.
+            self.assertEqual(contents.get_md5sum(), '')
+
+    def test_upload_segment_job_etag_mismatch(self):
+        def _consuming_conn(*a, **kw):
+            contents = a[2]
+            contents.read()  # Force md5 calculation
+            return 'badresponseetag'
+
+        with tempfile.NamedTemporaryFile() as f:
+            f.write(b'a' * 10)
+            f.write(b'b' * 10)
+            f.write(b'c' * 10)
+            f.flush()
+
+            mock_conn = mock.Mock()
+            mock_conn.put_object.side_effect = _consuming_conn
+            type(mock_conn).attempts = mock.PropertyMock(return_value=2)
+
+            s = SwiftService()
+            with self.assert_open_results_are_closed():
+                r = s._upload_segment_job(conn=mock_conn,
+                                          path=f.name,
+                                          container='test_c',
+                                          segment_name='test_s_1',
+                                          segment_start=10,
+                                          segment_size=10,
+                                          segment_index=2,
+                                          obj_name='test_o',
+                                          options={'segment_container': None,
+                                                   'checksum': True})
+
+            self.assertIn('md5 mismatch', str(r.get('error')))
+
+            self.assertEqual(mock_conn.put_object.call_count, 1)
+            mock_conn.put_object.assert_called_with(
+                'test_c_segments', 'test_s_1',
+                mock.ANY,
+                content_length=10,
+                content_type='application/swiftclient-segment',
+                response_dict={})
+            contents = mock_conn.put_object.call_args[0][2]
+            self.assertEqual(contents.get_md5sum(), md5(b'b' * 10).hexdigest())
+
+    def test_upload_object_job_file(self):
+        # Uploading a file results in the file object being wrapped in a
+        # LengthWrapper. This test sets the options in such a way that much
+        # of _upload_object_job is skipped bringing the critical path down
+        # to around 60 lines to ease testing.
+        with tempfile.NamedTemporaryFile() as f:
+            f.write(b'a' * 30)
+            f.flush()
+            expected_r = {
+                'action': 'upload_object',
+                'attempts': 2,
+                'container': 'test_c',
+                'headers': {},
+                'large_object': False,
+                'object': 'test_o',
+                'response_dict': {},
+                'status': 'uploaded',
+                'success': True,
+            }
+            expected_mtime = '%f' % os.path.getmtime(f.name)
+
+            # run read() when put_object is called to calculate md5sum
+            # md5sum is verified in _upload_object_job.
+            def _consuming_conn(*a, **kw):
+                contents = a[2]
+                contents.read()  # Force md5 calculation
+                return contents.get_md5sum()
+
+            mock_conn = mock.Mock()
+            mock_conn.put_object.side_effect = _consuming_conn
+            type(mock_conn).attempts = mock.PropertyMock(return_value=2)
+
+            s = SwiftService()
+            with self.assert_open_results_are_closed():
+                r = s._upload_object_job(conn=mock_conn,
+                                         container='test_c',
+                                         source=f.name,
+                                         obj='test_o',
+                                         options=dict(s._options,
+                                                      leave_segments=True))
+
+            mtime = r['headers']['x-object-meta-mtime']
+            self.assertEqual(expected_mtime, mtime)
+            del r['headers']['x-object-meta-mtime']
+
+            self.assertEqual(r['path'], f.name)
+            del r['path']
+
+            self.assertEqual(r, expected_r)
+            self.assertEqual(mock_conn.put_object.call_count, 1)
+            mock_conn.put_object.assert_called_with('test_c', 'test_o',
+                                                    mock.ANY,
+                                                    content_length=30,
+                                                    headers={},
+                                                    response_dict={})
+            contents = mock_conn.put_object.call_args[0][2]
+            self.assertIsInstance(contents, utils.LengthWrapper)
+            self.assertEqual(len(contents), 30)
+
+    @mock.patch('swiftclient.service.time', return_value=1400000000)
+    def test_upload_object_job_stream(self, time_mock):
+        # Streams are wrapped as ReadableToIterable
+        with tempfile.TemporaryFile() as f:
+            f.write(b'a' * 30)
+            f.flush()
+            f.seek(0)
+            expected_r = {
+                'action': 'upload_object',
+                'attempts': 2,
+                'container': 'test_c',
+                'headers': {},
+                'large_object': False,
+                'object': 'test_o',
+                'response_dict': {},
+                'status': 'uploaded',
+                'success': True,
+                'path': None,
+            }
+            expected_mtime = 1400000000
+
+            mock_conn = mock.Mock()
+            mock_conn.put_object.return_value = ''
+            type(mock_conn).attempts = mock.PropertyMock(return_value=2)
+
+            s = SwiftService()
+            r = s._upload_object_job(conn=mock_conn,
+                                     container='test_c',
+                                     source=f,
+                                     obj='test_o',
+                                     options=dict(s._options,
+                                                  leave_segments=True))
+
+            mtime = float(r['headers']['x-object-meta-mtime'])
+            self.assertEqual(mtime, expected_mtime)
+            del r['headers']['x-object-meta-mtime']
+
+            self.assertEqual(r, expected_r)
+            self.assertEqual(mock_conn.put_object.call_count, 1)
+            mock_conn.put_object.assert_called_with('test_c', 'test_o',
+                                                    mock.ANY,
+                                                    content_length=None,
+                                                    headers={},
+                                                    response_dict={})
+            contents = mock_conn.put_object.call_args[0][2]
+            self.assertIsInstance(contents, utils.ReadableToIterable)
+            self.assertEqual(contents.chunk_size, 65536)
+            # next retrieves the first chunk of the stream or len(chunk_size)
+            # or less, it also forces the md5 to be calculated.
+            self.assertEqual(next(contents), b'a' * 30)
+            self.assertEqual(contents.get_md5sum(), md5(b'a' * 30).hexdigest())
+
+    def test_upload_object_job_etag_mismatch(self):
+        # The etag test for both streams and files use the same code
+        # so only one test should be needed.
+        def _consuming_conn(*a, **kw):
+            contents = a[2]
+            contents.read()  # Force md5 calculation
+            return 'badresponseetag'
+
+        with tempfile.NamedTemporaryFile() as f:
+            f.write(b'a' * 30)
+            f.flush()
+
+            mock_conn = mock.Mock()
+            mock_conn.put_object.side_effect = _consuming_conn
+            type(mock_conn).attempts = mock.PropertyMock(return_value=2)
+
+            s = SwiftService()
+            r = s._upload_object_job(conn=mock_conn,
+                                     container='test_c',
+                                     source=f.name,
+                                     obj='test_o',
+                                     options=dict(s._options,
+                                                  leave_segments=True))
+
+            self.assertIs(r['success'], False)
+            self.assertIn('md5 mismatch', str(r.get('error')))
+
+            self.assertEqual(mock_conn.put_object.call_count, 1)
+            expected_headers = {'x-object-meta-mtime': mock.ANY}
+            mock_conn.put_object.assert_called_with('test_c', 'test_o',
+                                                    mock.ANY,
+                                                    content_length=30,
+                                                    headers=expected_headers,
+                                                    response_dict={})
+
+            contents = mock_conn.put_object.call_args[0][2]
+            self.assertEqual(contents.get_md5sum(), md5(b'a' * 30).hexdigest())
+
+    def test_upload_object_job_identical_etag(self):
+        with tempfile.NamedTemporaryFile() as f:
+            f.write(b'a' * 30)
+            f.flush()
+
+            mock_conn = mock.Mock()
+            mock_conn.head_object.return_value = {
+                'content-length': 30,
+                'etag': md5(b'a' * 30).hexdigest()}
+            type(mock_conn).attempts = mock.PropertyMock(return_value=2)
+
+            s = SwiftService()
+            r = s._upload_object_job(conn=mock_conn,
+                                     container='test_c',
+                                     source=f.name,
+                                     obj='test_o',
+                                     options={'changed': False,
+                                              'skip_identical': True,
+                                              'leave_segments': True,
+                                              'header': '',
+                                              'segment_size': 0})
+
+            self.assertIsNone(r.get('error'))
+            self.assertIs(True, r['success'])
+            self.assertEqual(r.get('status'), 'skipped-identical')
+            self.assertEqual(mock_conn.put_object.call_count, 0)
+            self.assertEqual(mock_conn.head_object.call_count, 1)
+            mock_conn.head_object.assert_called_with('test_c', 'test_o')
+
+    def test_upload_object_job_identical_slo_with_nesting(self):
+        with tempfile.NamedTemporaryFile() as f:
+            f.write(b'a' * 30)
+            f.flush()
+            seg_etag = md5(b'a' * 10).hexdigest()
+            submanifest = "[%s]" % ",".join(
+                ['{"bytes":10,"hash":"%s"}' % seg_etag] * 2)
+            submanifest_etag = md5(seg_etag.encode('ascii') * 2).hexdigest()
+            manifest = "[%s]" % ",".join([
+                '{"sub_slo":true,"name":"/test_c_segments/test_sub_slo",'
+                '"bytes":20,"hash":"%s"}' % submanifest_etag,
+                '{"bytes":10,"hash":"%s"}' % seg_etag])
+
+            mock_conn = mock.Mock()
+            mock_conn.head_object.return_value = {
+                'x-static-large-object': True,
+                'content-length': 30,
+                'etag': md5(submanifest_etag.encode('ascii') +
+                            seg_etag.encode('ascii')).hexdigest()}
+            mock_conn.get_object.side_effect = [
+                ({}, manifest.encode('ascii')),
+                ({}, submanifest.encode('ascii'))]
+            type(mock_conn).attempts = mock.PropertyMock(return_value=2)
+
+            s = SwiftService()
+            r = s._upload_object_job(conn=mock_conn,
+                                     container='test_c',
+                                     source=f.name,
+                                     obj='test_o',
+                                     options={'changed': False,
+                                              'skip_identical': True,
+                                              'leave_segments': True,
+                                              'header': '',
+                                              'segment_size': 10})
+
+            self.assertIsNone(r.get('error'))
+            self.assertIs(True, r['success'])
+            self.assertEqual('skipped-identical', r.get('status'))
+            self.assertEqual(0, mock_conn.put_object.call_count)
+            self.assertEqual([mock.call('test_c', 'test_o')],
+                             mock_conn.head_object.mock_calls)
+            self.assertEqual([
+                mock.call('test_c', 'test_o',
+                          query_string='multipart-manifest=get'),
+                mock.call('test_c_segments', 'test_sub_slo',
+                          query_string='multipart-manifest=get'),
+            ], mock_conn.get_object.mock_calls)
+
+    def test_upload_object_job_identical_dlo(self):
+        with tempfile.NamedTemporaryFile() as f:
+            f.write(b'a' * 30)
+            f.flush()
+            segment_etag = md5(b'a' * 10).hexdigest()
+
+            mock_conn = mock.Mock()
+            mock_conn.head_object.return_value = {
+                'x-object-manifest': 'test_c_segments/test_o/prefix',
+                'content-length': 30,
+                'etag': md5(segment_etag.encode('ascii') * 3).hexdigest()}
+            mock_conn.get_container.side_effect = [
+                (None, [{"bytes": 10, "hash": segment_etag,
+                         "name": "test_o/prefix/00"},
+                        {"bytes": 10, "hash": segment_etag,
+                         "name": "test_o/prefix/01"}]),
+                (None, [{"bytes": 10, "hash": segment_etag,
+                         "name": "test_o/prefix/02"}]),
+                (None, {})]
+            type(mock_conn).attempts = mock.PropertyMock(return_value=2)
+
+            s = SwiftService()
+            with mock.patch('swiftclient.service.get_conn',
+                            return_value=mock_conn):
+                r = s._upload_object_job(conn=mock_conn,
+                                         container='test_c',
+                                         source=f.name,
+                                         obj='test_o',
+                                         options={'changed': False,
+                                                  'skip_identical': True,
+                                                  'leave_segments': True,
+                                                  'header': '',
+                                                  'segment_size': 10})
+
+            self.assertIsNone(r.get('error'))
+            self.assertIs(True, r['success'])
+            self.assertEqual('skipped-identical', r.get('status'))
+            self.assertEqual(0, mock_conn.put_object.call_count)
+            self.assertEqual(1, mock_conn.head_object.call_count)
+            self.assertEqual(3, mock_conn.get_container.call_count)
+            mock_conn.head_object.assert_called_with('test_c', 'test_o')
+            expected = [
+                mock.call('test_c_segments', prefix='test_o/prefix',
+                          marker='', delimiter=None, headers={}),
+                mock.call('test_c_segments', prefix='test_o/prefix',
+                          marker="test_o/prefix/01", delimiter=None,
+                          headers={}),
+                mock.call('test_c_segments', prefix='test_o/prefix',
+                          marker="test_o/prefix/02", delimiter=None,
+                          headers={}),
+            ]
+            mock_conn.get_container.assert_has_calls(expected)
+
+    def test_make_upload_objects(self):
+        check_names_pseudo_to_expected = {
+            (('/absolute/file/path',), ''): ['absolute/file/path'],
+            (('relative/file/path',), ''): ['relative/file/path'],
+            (('/absolute/file/path',), '/absolute/pseudo/dir'): [
+                'absolute/pseudo/dir/absolute/file/path'],
+            (('/absolute/file/path',), 'relative/pseudo/dir'): [
+                'relative/pseudo/dir/absolute/file/path'],
+            (('relative/file/path',), '/absolute/pseudo/dir'): [
+                'absolute/pseudo/dir/relative/file/path'],
+            (('relative/file/path',), 'relative/pseudo/dir'): [
+                'relative/pseudo/dir/relative/file/path'],
+        }
+        errors = []
+        for (filenames, pseudo_folder), expected in \
+                check_names_pseudo_to_expected.items():
+            actual = SwiftService._make_upload_objects(
+                filenames, pseudo_folder=pseudo_folder)
+            try:
+                self.assertEqual(expected, [o.object_name for o in actual])
+            except AssertionError as e:
+                msg = 'given (%r, %r) expected %r, got %s' % (
+                    filenames, pseudo_folder, expected, e)
+                errors.append(msg)
+        self.assertFalse(errors, "\nERRORS:\n%s" % '\n'.join(errors))
+
+    def test_create_dir_marker_job_unchanged(self):
+        mock_conn = mock.Mock()
+        mock_conn.head_object.return_value = {
+            'content-type': 'application/directory',
+            'content-length': '0',
+            'x-object-meta-mtime': '1.234000',
+            'etag': md5().hexdigest()}
+
+        s = SwiftService()
+        with mock.patch('swiftclient.service.get_conn',
+                        return_value=mock_conn):
+            with mock.patch('swiftclient.service.getmtime',
+                            return_value=1.234):
+                r = s._create_dir_marker_job(conn=mock_conn,
+                                             container='test_c',
+                                             obj='test_o',
+                                             path='test',
+                                             options={'changed': True,
+                                                      'skip_identical': True,
+                                                      'leave_segments': True,
+                                                      'header': '',
+                                                      'segment_size': 10})
+        self.assertEqual({
+            'action': 'create_dir_marker',
+            'container': 'test_c',
+            'object': 'test_o',
+            'path': 'test',
+            'headers': {'x-object-meta-mtime': '1.234000'},
+            # NO response dict!
+            'success': True,
+        }, r)
+        self.assertEqual([], mock_conn.put_object.mock_calls)
+
+    def test_create_dir_marker_job_unchanged_old_type(self):
+        mock_conn = mock.Mock()
+        mock_conn.head_object.return_value = {
+            'content-type': 'text/directory',
+            'content-length': '0',
+            'x-object-meta-mtime': '1.000000',
+            'etag': md5().hexdigest()}
+
+        s = SwiftService()
+        with mock.patch('swiftclient.service.get_conn',
+                        return_value=mock_conn):
+            with mock.patch('swiftclient.service.time',
+                            return_value=1.234):
+                r = s._create_dir_marker_job(conn=mock_conn,
+                                             container='test_c',
+                                             obj='test_o',
+                                             options={'changed': True,
+                                                      'skip_identical': True,
+                                                      'leave_segments': True,
+                                                      'header': '',
+                                                      'segment_size': 10})
+        self.assertEqual({
+            'action': 'create_dir_marker',
+            'container': 'test_c',
+            'object': 'test_o',
+            'path': None,
+            'headers': {'x-object-meta-mtime': '1.000000'},
+            # NO response dict!
+            'success': True,
+        }, r)
+        self.assertEqual([], mock_conn.put_object.mock_calls)
+
+    def test_create_dir_marker_job_overwrites_bad_type(self):
+        mock_conn = mock.Mock()
+        mock_conn.head_object.return_value = {
+            'content-type': 'text/plain',
+            'content-length': '0',
+            'x-object-meta-mtime': '1.000000',
+            'etag': md5().hexdigest()}
+
+        s = SwiftService()
+        with mock.patch('swiftclient.service.get_conn',
+                        return_value=mock_conn):
+            with mock.patch('swiftclient.service.time',
+                            return_value=1.234):
+                r = s._create_dir_marker_job(conn=mock_conn,
+                                             container='test_c',
+                                             obj='test_o',
+                                             options={'changed': True,
+                                                      'skip_identical': True,
+                                                      'leave_segments': True,
+                                                      'header': '',
+                                                      'segment_size': 10})
+        self.assertEqual({
+            'action': 'create_dir_marker',
+            'container': 'test_c',
+            'object': 'test_o',
+            'path': None,
+            'headers': {'x-object-meta-mtime': '1.000000'},
+            'response_dict': {},
+            'success': True,
+        }, r)
+        self.assertEqual([mock.call(
+            'test_c', 'test_o', '',
+            content_length=0,
+            content_type='application/directory',
+            headers={'x-object-meta-mtime': '1.000000'},
+            response_dict={})], mock_conn.put_object.mock_calls)
+
+    def test_create_dir_marker_job_missing(self):
+        mock_conn = mock.Mock()
+        mock_conn.head_object.side_effect = \
+            ClientException('Not Found', http_status=404)
+
+        s = SwiftService()
+        with mock.patch('swiftclient.service.get_conn',
+                        return_value=mock_conn):
+            with mock.patch('swiftclient.service.time',
+                            return_value=1.234):
+                r = s._create_dir_marker_job(conn=mock_conn,
+                                             container='test_c',
+                                             obj='test_o',
+                                             options={'changed': True,
+                                                      'skip_identical': True,
+                                                      'leave_segments': True,
+                                                      'header': '',
+                                                      'segment_size': 10})
+        self.assertEqual({
+            'action': 'create_dir_marker',
+            'container': 'test_c',
+            'object': 'test_o',
+            'path': None,
+            'headers': {'x-object-meta-mtime': '1.000000'},
+            'response_dict': {},
+            'success': True,
+        }, r)
+        self.assertEqual([mock.call(
+            'test_c', 'test_o', '',
+            content_length=0,
+            content_type='application/directory',
+            headers={'x-object-meta-mtime': '1.000000'},
+            response_dict={})], mock_conn.put_object.mock_calls)
+
+
+class TestServiceDownload(_TestServiceBase):
+
+    def setUp(self):
+        super(TestServiceDownload, self).setUp()
+        self.opts = swiftclient.service._default_local_options.copy()
+        self.opts['no_download'] = True
+        self.obj_content = b'c' * 10
+        self.obj_etag = md5(self.obj_content).hexdigest()
+        self.obj_len = len(self.obj_content)
+        self.exc = Exception('test_exc')
+        # Base response to be copied and updated to matched the expected
+        # response for each test
+        self.expected = {
+            'action': 'download_object',   # Should always be download_object
+            'container': 'test_c',
+            'object': 'test_o',
+            'attempts': 2,
+            'response_dict': {},
+            'path': 'test_o',
+            'pseudodir': False,
+            'success': None   # Should be a bool
+        }
+
+    def _readbody(self):
+        yield self.obj_content
+
+    @mock.patch('swiftclient.service.SwiftService.list')
+    @mock.patch('swiftclient.service.SwiftService._submit_page_downloads')
+    @mock.patch('swiftclient.service.interruptable_as_completed')
+    def test_download_container_job(self, as_comp, sub_page, service_list):
+        """
+        Check that paged downloads work correctly
+        """
+        obj_count = [0]
+
+        def make_counting_generator(object_to_yield, total_count):
+            # maintain a counter of objects yielded
+            count = [0]
+
+            def counting_generator():
+                while count[0] < 10:
+                    yield object_to_yield
+                    count[0] += 1
+                    total_count[0] += 1
+            return counting_generator()
+
+        obj_count_on_sub_page_call = []
+        sub_page_call_count = [0]
+
+        def fake_sub_page(*args):
+            # keep a record of obj_count when this function is called
+            obj_count_on_sub_page_call.append(obj_count[0])
+            sub_page_call_count[0] += 1
+            if sub_page_call_count[0] < 3:
+                return range(0, 10)
+            return None
+
+        sub_page.side_effect = fake_sub_page
+
+        r = Mock(spec=Future)
+        r.result.return_value = self._get_expected({
+            'success': True,
+            'start_time': 1,
+            'finish_time': 2,
+            'headers_receipt': 3,
+            'auth_end_time': 4,
+            'read_length': len(b'objcontent'),
+        })
+
+        as_comp.side_effect = [
+            make_counting_generator(r, obj_count),
+            make_counting_generator(r, obj_count)
+        ]
+
+        s = SwiftService()
+        down_gen = s._download_container('test_c', self.opts)
+        results = list(down_gen)
+        self.assertEqual(20, len(results))
+        self.assertEqual(2, as_comp.call_count)
+        self.assertEqual(3, sub_page_call_count[0])
+        self.assertEqual([0, 7, 17], obj_count_on_sub_page_call)
+
+    @mock.patch('swiftclient.service.SwiftService.list')
+    @mock.patch('swiftclient.service.SwiftService._submit_page_downloads')
+    @mock.patch('swiftclient.service.interruptable_as_completed')
+    def test_download_container_job_error(
+            self, as_comp, sub_page, service_list):
+        """
+        Check that paged downloads work correctly
+        """
+        class BoomError(Exception):
+            def __init__(self, value):
+                self.value = value
+
+            def __str__(self):
+                return repr(self.value)
+
+        def _make_result():
+            r = Mock(spec=Future)
+            r.result.return_value = self._get_expected({
+                'success': True,
+                'start_time': 1,
+                'finish_time': 2,
+                'headers_receipt': 3,
+                'auth_end_time': 4,
+                'read_length': len(b'objcontent'),
+            })
+            return r
+
+        as_comp.side_effect = [
+
+        ]
+        # We need Futures here because the error will cause a call to .cancel()
+        sub_page_effects = [
+            [_make_result() for _ in range(0, 10)],
+            BoomError('Go Boom')
+        ]
+        sub_page.side_effect = sub_page_effects
+        # ...but we must also mock the returns to as_completed
+        as_comp.side_effect = [
+            [_make_result() for _ in range(0, 10)]
+        ]
+
+        s = SwiftService()
+        self.assertRaises(
+            BoomError,
+            lambda: list(s._download_container('test_c', self.opts))
+        )
+        # This was an unknown error, so make sure we attempt to cancel futures
+        for spe in sub_page_effects[0]:
+            spe.cancel.assert_called_once_with()
+        self.assertEqual(1, as_comp.call_count)
+
+        # Now test ClientException
+        sub_page_effects = [
+            [_make_result() for _ in range(0, 10)],
+            ClientException('Go Boom')
+        ]
+        sub_page.side_effect = sub_page_effects
+        as_comp.reset_mock()
+        as_comp.side_effect = [
+            [_make_result() for _ in range(0, 10)],
+        ]
+        self.assertRaises(
+            ClientException,
+            lambda: list(s._download_container('test_c', self.opts))
+        )
+        # This was a ClientException, so make sure we don't cancel futures
+        for spe in sub_page_effects[0]:
+            self.assertFalse(spe.cancel.called)
+        self.assertEqual(1, as_comp.call_count)
+
+    def test_download_object_job(self):
+        mock_conn = self._get_mock_connection()
+        objcontent = six.BytesIO(b'objcontent')
+        mock_conn.get_object.side_effect = [
+            ({'content-type': 'text/plain',
+              'etag': '2cbbfe139a744d6abbe695e17f3c1991'},
+             objcontent)
+        ]
+        expected_r = self._get_expected({
+            'success': True,
+            'start_time': 1,
+            'finish_time': 2,
+            'headers_receipt': 3,
+            'auth_end_time': 4,
+            'read_length': len(b'objcontent'),
+        })
+
+        with mock.patch.object(builtins, 'open') as mock_open:
+            written_content = Mock()
+            mock_open.return_value = written_content
+            s = SwiftService()
+            _opts = self.opts.copy()
+            _opts['no_download'] = False
+            actual_r = s._download_object_job(
+                mock_conn, 'test_c', 'test_o', _opts)
+            actual_r = dict(  # Need to override the times we got from the call
+                actual_r,
+                **{
+                    'start_time': 1,
+                    'finish_time': 2,
+                    'headers_receipt': 3
+                }
+            )
+            mock_open.assert_called_once_with('test_o', 'wb', 65536)
+            written_content.write.assert_called_once_with(b'objcontent')
+
+        mock_conn.get_object.assert_called_once_with(
+            'test_c', 'test_o', resp_chunk_size=65536, headers={},
+            response_dict={}
+        )
+        self.assertEqual(expected_r, actual_r)
+
+    def test_download_object_job_with_mtime(self):
+        mock_conn = self._get_mock_connection()
+        objcontent = six.BytesIO(b'objcontent')
+        mock_conn.get_object.side_effect = [
+            ({'content-type': 'text/plain',
+              'etag': '2cbbfe139a744d6abbe695e17f3c1991',
+              'x-object-meta-mtime': '1454113727.682512'},
+             objcontent)
+        ]
+        expected_r = self._get_expected({
+            'success': True,
+            'start_time': 1,
+            'finish_time': 2,
+            'headers_receipt': 3,
+            'auth_end_time': 4,
+            'read_length': len(b'objcontent'),
+        })
+
+        with mock.patch.object(builtins, 'open') as mock_open, \
+                mock.patch('swiftclient.service.utime') as mock_utime:
+            written_content = Mock()
+            mock_open.return_value = written_content
+            s = SwiftService()
+            _opts = self.opts.copy()
+            _opts['no_download'] = False
+            actual_r = s._download_object_job(
+                mock_conn, 'test_c', 'test_o', _opts)
+            actual_r = dict(  # Need to override the times we got from the call
+                actual_r,
+                **{
+                    'start_time': 1,
+                    'finish_time': 2,
+                    'headers_receipt': 3
+                }
+            )
+            mock_open.assert_called_once_with('test_o', 'wb', 65536)
+            mock_utime.assert_called_once_with(
+                'test_o', (1454113727.682512, 1454113727.682512))
+            written_content.write.assert_called_once_with(b'objcontent')
+
+        mock_conn.get_object.assert_called_once_with(
+            'test_c', 'test_o', resp_chunk_size=65536, headers={},
+            response_dict={}
+        )
+        self.assertEqual(expected_r, actual_r)
+
+    def test_download_object_job_bad_mtime(self):
+        mock_conn = self._get_mock_connection()
+        objcontent = six.BytesIO(b'objcontent')
+        mock_conn.get_object.side_effect = [
+            ({'content-type': 'text/plain',
+              'etag': '2cbbfe139a744d6abbe695e17f3c1991',
+              'x-object-meta-mtime': 'foo'},
+             objcontent)
+        ]
+        expected_r = self._get_expected({
+            'success': True,
+            'start_time': 1,
+            'finish_time': 2,
+            'headers_receipt': 3,
+            'auth_end_time': 4,
+            'read_length': len(b'objcontent'),
+        })
+
+        with mock.patch.object(builtins, 'open') as mock_open, \
+                mock.patch('swiftclient.service.utime') as mock_utime:
+            written_content = Mock()
+            mock_open.return_value = written_content
+            s = SwiftService()
+            _opts = self.opts.copy()
+            _opts['no_download'] = False
+            actual_r = s._download_object_job(
+                mock_conn, 'test_c', 'test_o', _opts)
+            actual_r = dict(  # Need to override the times we got from the call
+                actual_r,
+                **{
+                    'start_time': 1,
+                    'finish_time': 2,
+                    'headers_receipt': 3
+                }
+            )
+            mock_open.assert_called_once_with('test_o', 'wb', 65536)
+            self.assertEqual(0, len(mock_utime.mock_calls))
+            written_content.write.assert_called_once_with(b'objcontent')
+
+        mock_conn.get_object.assert_called_once_with(
+            'test_c', 'test_o', resp_chunk_size=65536, headers={},
+            response_dict={}
+        )
+        self.assertEqual(expected_r, actual_r)
+
+    def test_download_object_job_ignore_mtime(self):
+        mock_conn = self._get_mock_connection()
+        objcontent = six.BytesIO(b'objcontent')
+        mock_conn.get_object.side_effect = [
+            ({'content-type': 'text/plain',
+              'etag': '2cbbfe139a744d6abbe695e17f3c1991',
+              'x-object-meta-mtime': '1454113727.682512'},
+             objcontent)
+        ]
+        expected_r = self._get_expected({
+            'success': True,
+            'start_time': 1,
+            'finish_time': 2,
+            'headers_receipt': 3,
+            'auth_end_time': 4,
+            'read_length': len(b'objcontent'),
+        })
+
+        with mock.patch.object(builtins, 'open') as mock_open, \
+                mock.patch('swiftclient.service.utime') as mock_utime:
+            written_content = Mock()
+            mock_open.return_value = written_content
+            s = SwiftService()
+            _opts = self.opts.copy()
+            _opts['no_download'] = False
+            _opts['ignore_mtime'] = True
+            actual_r = s._download_object_job(
+                mock_conn, 'test_c', 'test_o', _opts)
+            actual_r = dict(  # Need to override the times we got from the call
+                actual_r,
+                **{
+                    'start_time': 1,
+                    'finish_time': 2,
+                    'headers_receipt': 3
+                }
+            )
+            mock_open.assert_called_once_with('test_o', 'wb', 65536)
+            self.assertEqual([], mock_utime.mock_calls)
+            written_content.write.assert_called_once_with(b'objcontent')
+
+        mock_conn.get_object.assert_called_once_with(
+            'test_c', 'test_o', resp_chunk_size=65536, headers={},
+            response_dict={}
+        )
+        self.assertEqual(expected_r, actual_r)
+
+    def test_download_object_job_exception(self):
+        mock_conn = self._get_mock_connection()
+        mock_conn.get_object = Mock(side_effect=self.exc)
+        expected_r = self._get_expected({
+            'success': False,
+            'error': self.exc,
+            'error_timestamp': mock.ANY,
+            'traceback': mock.ANY
+        })
+
+        s = SwiftService()
+        actual_r = s._download_object_job(
+            mock_conn, 'test_c', 'test_o', self.opts)
+
+        mock_conn.get_object.assert_called_once_with(
+            'test_c', 'test_o', resp_chunk_size=65536, headers={},
+            response_dict={}
+        )
+        self.assertEqual(expected_r, actual_r)
+
+    def test_download(self):
+        with mock.patch('swiftclient.service.Connection') as mock_conn:
+            header = {'content-length': self.obj_len,
+                      'etag': self.obj_etag}
+            mock_conn.get_object.return_value = header, self._readbody()
+
+            resp = SwiftService()._download_object_job(mock_conn,
+                                                       'c',
+                                                       'test',
+                                                       self.opts)
+
+        self.assertIsNone(resp.get('error'))
+        self.assertIs(True, resp['success'])
+        self.assertEqual(resp['action'], 'download_object')
+        self.assertEqual(resp['object'], 'test')
+        self.assertEqual(resp['path'], 'test')
+
+    @mock.patch('swiftclient.service.interruptable_as_completed')
+    @mock.patch('swiftclient.service.SwiftService._download_container')
+    @mock.patch('swiftclient.service.SwiftService._download_object_job')
+    def test_download_with_objects_empty(self, mock_down_obj,
+                                         mock_down_cont, mock_as_comp):
+        fake_future = Future()
+        fake_future.set_result(1)
+        mock_as_comp.return_value = [fake_future]
+        service = SwiftService()
+        next(service.download('c', [], self.opts), None)
+        mock_down_obj.assert_not_called()
+        mock_down_cont.assert_not_called()
+
+        next(service.download('c', options=self.opts), None)
+        self.assertTrue(mock_down_cont.called)
+
+    def test_download_with_output_dir(self):
+        with mock.patch('swiftclient.service.Connection') as mock_conn:
+            header = {'content-length': self.obj_len,
+                      'etag': self.obj_etag}
+            mock_conn.get_object.return_value = header, self._readbody()
+
+            options = self.opts.copy()
+            options['out_directory'] = 'temp_dir'
+            resp = SwiftService()._download_object_job(mock_conn,
+                                                       'c',
+                                                       'example/test',
+                                                       options)
+
+        self.assertIsNone(resp.get('error'))
+        self.assertIs(True, resp['success'])
+        self.assertEqual(resp['action'], 'download_object')
+        self.assertEqual(resp['object'], 'example/test')
+        self.assertEqual(resp['path'], 'temp_dir/example/test')
+
+    def test_download_with_remove_prefix(self):
+        with mock.patch('swiftclient.service.Connection') as mock_conn:
+            header = {'content-length': self.obj_len,
+                      'etag': self.obj_etag}
+            mock_conn.get_object.return_value = header, self._readbody()
+
+            options = self.opts.copy()
+            options['prefix'] = 'example/'
+            options['remove_prefix'] = True
+            resp = SwiftService()._download_object_job(mock_conn,
+                                                       'c',
+                                                       'example/test',
+                                                       options)
+
+        self.assertIsNone(resp.get('error'))
+        self.assertIs(True, resp['success'])
+        self.assertEqual(resp['action'], 'download_object')
+        self.assertEqual(resp['object'], 'example/test')
+        self.assertEqual(resp['path'], 'test')
+
+    def test_download_with_remove_prefix_and_remove_slashes(self):
+        with mock.patch('swiftclient.service.Connection') as mock_conn:
+            header = {'content-length': self.obj_len,
+                      'etag': self.obj_etag}
+            mock_conn.get_object.return_value = header, self._readbody()
+
+            options = self.opts.copy()
+            options['prefix'] = 'example'
+            options['remove_prefix'] = True
+            resp = SwiftService()._download_object_job(mock_conn,
+                                                       'c',
+                                                       'example/test',
+                                                       options)
+
+        self.assertIsNone(resp.get('error'))
+        self.assertIs(True, resp['success'])
+        self.assertEqual(resp['action'], 'download_object')
+        self.assertEqual(resp['object'], 'example/test')
+        self.assertEqual(resp['path'], 'test')
+
+    def test_download_with_output_dir_and_remove_prefix(self):
+        with mock.patch('swiftclient.service.Connection') as mock_conn:
+            header = {'content-length': self.obj_len,
+                      'etag': self.obj_etag}
+            mock_conn.get_object.return_value = header, self._readbody()
+
+            options = self.opts.copy()
+            options['prefix'] = 'example'
+            options['out_directory'] = 'new/dir'
+            options['remove_prefix'] = True
+            resp = SwiftService()._download_object_job(mock_conn,
+                                                       'c',
+                                                       'example/test',
+                                                       options)
+
+        self.assertIsNone(resp.get('error'))
+        self.assertIs(True, resp['success'])
+        self.assertEqual(resp['action'], 'download_object')
+        self.assertEqual(resp['object'], 'example/test')
+        self.assertEqual(resp['path'], 'new/dir/test')
+
+    def test_download_object_job_skip_identical(self):
+        with tempfile.NamedTemporaryFile() as f:
+            f.write(b'a' * 30)
+            f.flush()
+
+            err = swiftclient.ClientException('Object GET failed',
+                                              http_status=304)
+
+            def fake_get(*args, **kwargs):
+                kwargs['response_dict']['headers'] = {}
+                raise err
+
+            mock_conn = mock.Mock()
+            mock_conn.get_object.side_effect = fake_get
+            type(mock_conn).attempts = mock.PropertyMock(return_value=2)
+            expected_r = {
+                'action': 'download_object',
+                'container': 'test_c',
+                'object': 'test_o',
+                'success': False,
+                'error': err,
+                'response_dict': {'headers': {}},
+                'path': 'test_o',
+                'pseudodir': False,
+                'attempts': 2,
+                'traceback': mock.ANY,
+                'error_timestamp': mock.ANY
+            }
+
+            s = SwiftService()
+            r = s._download_object_job(conn=mock_conn,
+                                       container='test_c',
+                                       obj='test_o',
+                                       options={'out_file': f.name,
+                                                'out_directory': None,
+                                                'prefix': None,
+                                                'remove_prefix': False,
+                                                'header': {},
+                                                'yes_all': False,
+                                                'skip_identical': True})
+            self.assertEqual(r, expected_r)
+
+            self.assertEqual(mock_conn.get_object.call_count, 1)
+            mock_conn.get_object.assert_called_with(
+                'test_c',
+                'test_o',
+                resp_chunk_size=65536,
+                headers={'If-None-Match': md5(b'a' * 30).hexdigest()},
+                query_string='multipart-manifest=get',
+                response_dict=expected_r['response_dict'])
+
+    def test_download_object_job_skip_identical_dlo(self):
+        with tempfile.NamedTemporaryFile() as f:
+            f.write(b'a' * 30)
+            f.flush()
+            on_disk_md5 = md5(b'a' * 30).hexdigest()
+            segment_md5 = md5(b'a' * 10).hexdigest()
+
+            mock_conn = mock.Mock()
+            mock_conn.get_object.return_value = (
+                {'x-object-manifest': 'test_c_segments/test_o/prefix'}, [b''])
+            mock_conn.get_container.side_effect = [
+                (None, [{'name': 'test_o/prefix/1',
+                         'bytes': 10, 'hash': segment_md5},
+                        {'name': 'test_o/prefix/2',
+                         'bytes': 10, 'hash': segment_md5}]),
+                (None, [{'name': 'test_o/prefix/3',
+                         'bytes': 10, 'hash': segment_md5}]),
+                (None, [])]
+
+            type(mock_conn).attempts = mock.PropertyMock(return_value=2)
+            expected_r = {
+                'action': 'download_object',
+                'container': 'test_c',
+                'object': 'test_o',
+                'success': False,
+                'response_dict': {},
+                'path': 'test_o',
+                'pseudodir': False,
+                'attempts': 2,
+                'traceback': mock.ANY,
+                'error_timestamp': mock.ANY
+            }
+
+            s = SwiftService()
+            with mock.patch('swiftclient.service.get_conn',
+                            return_value=mock_conn):
+                r = s._download_object_job(conn=mock_conn,
+                                           container='test_c',
+                                           obj='test_o',
+                                           options={'out_file': f.name,
+                                                    'out_directory': None,
+                                                    'prefix': None,
+                                                    'remove_prefix': False,
+                                                    'header': {},
+                                                    'yes_all': False,
+                                                    'skip_identical': True})
+
+            err = r.pop('error')
+            self.assertEqual("Large object is identical", err.msg)
+            self.assertEqual(304, err.http_status)
+
+            self.assertEqual(r, expected_r)
+
+            self.assertEqual(mock_conn.get_object.call_count, 1)
+            mock_conn.get_object.assert_called_with(
+                'test_c',
+                'test_o',
+                resp_chunk_size=65536,
+                headers={'If-None-Match': on_disk_md5},
+                query_string='multipart-manifest=get',
+                response_dict=expected_r['response_dict'])
+            self.assertEqual(mock_conn.get_container.mock_calls, [
+                mock.call('test_c_segments',
+                          delimiter=None,
+                          prefix='test_o/prefix',
+                          marker='',
+                          headers={}),
+                mock.call('test_c_segments',
+                          delimiter=None,
+                          prefix='test_o/prefix',
+                          marker='test_o/prefix/2',
+                          headers={}),
+                mock.call('test_c_segments',
+                          delimiter=None,
+                          prefix='test_o/prefix',
+                          marker='test_o/prefix/3',
+                          headers={})])
+
+    def test_download_object_job_skip_identical_nested_slo(self):
+        with tempfile.NamedTemporaryFile() as f:
+            f.write(b'a' * 30)
+            f.flush()
+            on_disk_md5 = md5(b'a' * 30).hexdigest()
+
+            seg_etag = md5(b'a' * 10).hexdigest()
+            submanifest = "[%s]" % ",".join(
+                ['{"bytes":10,"hash":"%s"}' % seg_etag] * 2)
+            submanifest_etag = md5(seg_etag.encode('ascii') * 2).hexdigest()
+            manifest = "[%s]" % ",".join([
+                '{"sub_slo":true,"name":"/test_c_segments/test_sub_slo",'
+                '"bytes":20,"hash":"%s"}' % submanifest_etag,
+                '{"bytes":10,"hash":"%s"}' % seg_etag])
+
+            mock_conn = mock.Mock()
+            mock_conn.get_object.side_effect = [
+                ({'x-static-large-object': True,
+                  'content-length': 30,
+                  'etag': md5(submanifest_etag.encode('ascii') +
+                              seg_etag.encode('ascii')).hexdigest()},
+                 [manifest.encode('ascii')]),
+                ({'x-static-large-object': True,
+                  'content-length': 20,
+                  'etag': submanifest_etag},
+                 submanifest.encode('ascii'))]
+
+            type(mock_conn).attempts = mock.PropertyMock(return_value=2)
+            expected_r = {
+                'action': 'download_object',
+                'container': 'test_c',
+                'object': 'test_o',
+                'success': False,
+                'response_dict': {},
+                'path': 'test_o',
+                'pseudodir': False,
+                'attempts': 2,
+                'traceback': mock.ANY,
+                'error_timestamp': mock.ANY
+            }
+
+            s = SwiftService()
+            with mock.patch('swiftclient.service.get_conn',
+                            return_value=mock_conn):
+                r = s._download_object_job(conn=mock_conn,
+                                           container='test_c',
+                                           obj='test_o',
+                                           options={'out_file': f.name,
+                                                    'out_directory': None,
+                                                    'prefix': None,
+                                                    'remove_prefix': False,
+                                                    'header': {},
+                                                    'yes_all': False,
+                                                    'skip_identical': True})
+
+            err = r.pop('error')
+            self.assertEqual("Large object is identical", err.msg)
+            self.assertEqual(304, err.http_status)
+
+            self.assertEqual(r, expected_r)
+            self.assertEqual(mock_conn.get_object.mock_calls, [
+                mock.call('test_c',
+                          'test_o',
+                          resp_chunk_size=65536,
+                          headers={'If-None-Match': on_disk_md5},
+                          query_string='multipart-manifest=get',
+                          response_dict={}),
+                mock.call('test_c_segments',
+                          'test_sub_slo',
+                          query_string='multipart-manifest=get')])
+
+    def test_download_object_job_skip_identical_diff_dlo(self):
+        with tempfile.NamedTemporaryFile() as f:
+            f.write(b'a' * 30)
+            f.write(b'b')
+            f.flush()
+            on_disk_md5 = md5(b'a' * 30 + b'b').hexdigest()
+            segment_md5 = md5(b'a' * 10).hexdigest()
+
+            mock_conn = mock.Mock()
+            mock_conn.get_object.side_effect = [
+                ({'x-object-manifest': 'test_c_segments/test_o/prefix'},
+                 [b'']),
+                ({'x-object-manifest': 'test_c_segments/test_o/prefix'},
+                 [b'a' * 30])]
+            mock_conn.get_container.side_effect = [
+                (None, [{'name': 'test_o/prefix/1',
+                         'bytes': 10, 'hash': segment_md5},
+                        {'name': 'test_o/prefix/2',
+                         'bytes': 10, 'hash': segment_md5}]),
+                (None, [{'name': 'test_o/prefix/3',
+                         'bytes': 10, 'hash': segment_md5}]),
+                (None, [])]
+
+            type(mock_conn).attempts = mock.PropertyMock(return_value=2)
+            type(mock_conn).auth_end_time = mock.PropertyMock(return_value=14)
+            expected_r = {
+                'action': 'download_object',
+                'container': 'test_c',
+                'object': 'test_o',
+                'success': True,
+                'response_dict': {},
+                'path': 'test_o',
+                'pseudodir': False,
+                'read_length': 30,
+                'attempts': 2,
+                'start_time': 0,
+                'headers_receipt': 1,
+                'finish_time': 2,
+                'auth_end_time': mock_conn.auth_end_time,
+            }
+
+            options = self.opts.copy()
+            options['out_file'] = f.name
+            options['skip_identical'] = True
+            s = SwiftService()
+            with mock.patch('swiftclient.service.time', side_effect=range(3)):
+                with mock.patch('swiftclient.service.get_conn',
+                                return_value=mock_conn):
+                    r = s._download_object_job(
+                        conn=mock_conn,
+                        container='test_c',
+                        obj='test_o',
+                        options=options)
+
+            self.assertEqual(r, expected_r)
+
+            self.assertEqual(mock_conn.get_container.mock_calls, [
+                mock.call('test_c_segments',
+                          delimiter=None,
+                          prefix='test_o/prefix',
+                          marker='',
+                          headers={}),
+                mock.call('test_c_segments',
+                          delimiter=None,
+                          prefix='test_o/prefix',
+                          marker='test_o/prefix/2',
+                          headers={}),
+                mock.call('test_c_segments',
+                          delimiter=None,
+                          prefix='test_o/prefix',
+                          marker='test_o/prefix/3',
+                          headers={})])
+            self.assertEqual(mock_conn.get_object.mock_calls, [
+                mock.call('test_c',
+                          'test_o',
+                          resp_chunk_size=65536,
+                          headers={'If-None-Match': on_disk_md5},
+                          query_string='multipart-manifest=get',
+                          response_dict={}),
+                mock.call('test_c',
+                          'test_o',
+                          resp_chunk_size=65536,
+                          headers={'If-None-Match': on_disk_md5},
+                          response_dict={})])
+
+    def test_download_object_job_skip_identical_diff_nested_slo(self):
+        with tempfile.NamedTemporaryFile() as f:
+            f.write(b'a' * 29)
+            f.flush()
+            on_disk_md5 = md5(b'a' * 29).hexdigest()
+
+            seg_etag = md5(b'a' * 10).hexdigest()
+            submanifest = "[%s]" % ",".join(
+                ['{"bytes":10,"hash":"%s"}' % seg_etag] * 2)
+            submanifest_etag = md5(seg_etag.encode('ascii') * 2).hexdigest()
+            manifest = "[%s]" % ",".join([
+                '{"sub_slo":true,"name":"/test_c_segments/test_sub_slo",'
+                '"bytes":20,"hash":"%s"}' % submanifest_etag,
+                '{"bytes":10,"hash":"%s"}' % seg_etag])
+
+            mock_conn = mock.Mock()
+            mock_conn.get_object.side_effect = [
+                ({'x-static-large-object': True,
+                  'content-length': 30,
+                  'etag': md5(submanifest_etag.encode('ascii') +
+                              seg_etag.encode('ascii')).hexdigest()},
+                 [manifest.encode('ascii')]),
+                ({'x-static-large-object': True,
+                  'content-length': 20,
+                  'etag': submanifest_etag},
+                 submanifest.encode('ascii')),
+                ({'x-static-large-object': True,
+                  'content-length': 30,
+                  'etag': md5(submanifest_etag.encode('ascii') +
+                              seg_etag.encode('ascii')).hexdigest()},
+                 [b'a' * 30])]
+
+            type(mock_conn).attempts = mock.PropertyMock(return_value=2)
+            type(mock_conn).auth_end_time = mock.PropertyMock(return_value=14)
+            expected_r = {
+                'action': 'download_object',
+                'container': 'test_c',
+                'object': 'test_o',
+                'success': True,
+                'response_dict': {},
+                'path': 'test_o',
+                'pseudodir': False,
+                'read_length': 30,
+                'attempts': 2,
+                'start_time': 0,
+                'headers_receipt': 1,
+                'finish_time': 2,
+                'auth_end_time': mock_conn.auth_end_time,
+            }
+
+            options = self.opts.copy()
+            options['out_file'] = f.name
+            options['skip_identical'] = True
+            s = SwiftService()
+            with mock.patch('swiftclient.service.time', side_effect=range(3)):
+                with mock.patch('swiftclient.service.get_conn',
+                                return_value=mock_conn):
+                    r = s._download_object_job(
+                        conn=mock_conn,
+                        container='test_c',
+                        obj='test_o',
+                        options=options)
+
+            self.assertEqual(r, expected_r)
+            self.assertEqual(mock_conn.get_object.mock_calls, [
+                mock.call('test_c',
+                          'test_o',
+                          resp_chunk_size=65536,
+                          headers={'If-None-Match': on_disk_md5},
+                          query_string='multipart-manifest=get',
+                          response_dict={}),
+                mock.call('test_c_segments',
+                          'test_sub_slo',
+                          query_string='multipart-manifest=get'),
+                mock.call('test_c',
+                          'test_o',
+                          resp_chunk_size=65536,
+                          headers={'If-None-Match': on_disk_md5},
+                          response_dict={})])
+
+
+class TestServicePost(_TestServiceBase):
+
+    def setUp(self):
+        super(TestServicePost, self).setUp()
+        self.opts = swiftclient.service._default_local_options.copy()
+
+    @mock.patch('swiftclient.service.MultiThreadingManager')
+    @mock.patch('swiftclient.service.ResultsIterator')
+    def test_object_post(self, res_iter, thread_manager):
+        """
+        Check post method translates strings and objects to _post_object_job
+        calls correctly
+        """
+        tm_instance = Mock()
+        thread_manager.return_value = tm_instance
+
+        self.opts.update({'meta': ["meta1:test1"], "header": ["hdr1:test1"]})
+        spo = swiftclient.service.SwiftPostObject(
+            "test_spo",
+            {'meta': ["meta1:test2"], "header": ["hdr1:test2"]})
+
+        SwiftService().post('test_c', ['test_o', spo], self.opts)
+
+        calls = [
+            mock.call(
+                SwiftService._post_object_job, 'test_c', 'test_o',
+                {
+                    "X-Object-Meta-Meta1": "test1",
+                    "Hdr1": "test1"},
+                {}),
+            mock.call(
+                SwiftService._post_object_job, 'test_c', 'test_spo',
+                {
+                    "X-Object-Meta-Meta1": "test2",
+                    "Hdr1": "test2"},
+                {}),
+        ]
+        tm_instance.object_uu_pool.submit.assert_has_calls(calls)
+        self.assertEqual(
+            tm_instance.object_uu_pool.submit.call_count, len(calls))
+
+        res_iter.assert_called_with(
+            [tm_instance.object_uu_pool.submit()] * len(calls))
+
+
+class TestServiceCopy(_TestServiceBase):
+
+    def setUp(self):
+        super(TestServiceCopy, self).setUp()
+        self.opts = swiftclient.service._default_local_options.copy()
+
+    @mock.patch('swiftclient.service.MultiThreadingManager')
+    @mock.patch('swiftclient.service.interruptable_as_completed')
+    def test_object_copy(self, inter_compl, thread_manager):
+        """
+        Check copy method translates strings and objects to _copy_object_job
+        calls correctly
+        """
+        tm_instance = Mock()
+        thread_manager.return_value = tm_instance
+
+        self.opts.update({'meta': ["meta1:test1"], "header": ["hdr1:test1"]})
+        sco = swiftclient.service.SwiftCopyObject(
+            "test_sco",
+            options={'meta': ["meta1:test2"], "header": ["hdr1:test2"],
+                     'destination': "/cont_new/test_sco"})
+
+        res = SwiftService().copy('test_c', ['test_o', sco], self.opts)
+        res = list(res)
+
+        calls = [
+            mock.call(
+                SwiftService._create_container_job, 'cont_new', headers={}),
+        ]
+        tm_instance.container_pool.submit.assert_has_calls(calls,
+                                                           any_order=True)
+        self.assertEqual(
+            tm_instance.container_pool.submit.call_count, len(calls))
+
+        calls = [
+            mock.call(
+                SwiftService._copy_object_job, 'test_c', 'test_o',
+                None,
+                {
+                    "X-Object-Meta-Meta1": "test1",
+                    "Hdr1": "test1"},
+                False),
+            mock.call(
+                SwiftService._copy_object_job, 'test_c', 'test_sco',
+                '/cont_new/test_sco',
+                {
+                    "X-Object-Meta-Meta1": "test2",
+                    "Hdr1": "test2"},
+                False),
+        ]
+        tm_instance.object_uu_pool.submit.assert_has_calls(calls)
+        self.assertEqual(
+            tm_instance.object_uu_pool.submit.call_count, len(calls))
+
+        inter_compl.assert_called_with(
+            [tm_instance.object_uu_pool.submit()] * len(calls))
+
+    def test_object_copy_fail_dest(self):
+        """
+        Destination in incorrect format and destination with object
+        used when multiple objects are copied raises SwiftError
+        """
+        with self.assertRaises(SwiftError):
+            list(SwiftService().copy('test_c', ['test_o'],
+                                     {'destination': 'cont'}))
+        with self.assertRaises(SwiftError):
+            list(SwiftService().copy('test_c', ['test_o', 'test_o2'],
+                                     {'destination': '/cont/obj'}))