Merge "decouple versioned writes from COPY"

swift/common/middleware/versioned_writes.py

@@ -117,16 +117,19 @@ Disable versioning from a container (x is any value except empty)::
 
 import calendar
 import json
-import six
 from six.moves.urllib.parse import quote, unquote
 import time
+
 from swift.common.utils import get_logger, Timestamp, \
-    register_swift_info, config_true_value
+    register_swift_info, config_true_value, close_if_possible, FileLikeIter
-from swift.common.request_helpers import get_sys_meta_prefix
+from swift.common.request_helpers import get_sys_meta_prefix, \
+    copy_header_subset
 from swift.common.wsgi import WSGIContext, make_pre_authed_request
-from swift.common.swob import Request, HTTPException
+from swift.common.swob import (
+    Request, HTTPException, HTTPRequestEntityTooLarge)
 from swift.common.constraints import (
-    check_account_format, check_container_format, check_destination_header)
+    check_account_format, check_container_format, check_destination_header,
+    MAX_FILE_SIZE)
 from swift.proxy.controllers.base import get_container_info
 from swift.common.http import (
     is_success, is_client_error, HTTP_NOT_FOUND)
@@ -254,87 +257,122 @@ class VersionedWritesContext(WSGIContext):
             marker = last_item
             yield sublisting
 
-    def handle_obj_versions_put(self, req, object_versions,
+    def _get_source_object(self, req, path_info):
-                                object_name, policy_index):
+        # make a GET request to check object versions
-        ret = None
-
-        # do a HEAD request to check object versions
         _headers = {'X-Newest': 'True',
-                    'X-Backend-Storage-Policy-Index': policy_index,
                     'x-auth-token': req.headers.get('x-auth-token')}
 
         # make a pre_auth request in case the user has write access
         # to container, but not READ. This was allowed in previous version
         # (i.e., before middleware) so keeping the same behavior here
-        head_req = make_pre_authed_request(
+        get_req = make_pre_authed_request(
-            req.environ, path=req.path_info,
+            req.environ, path=path_info,
-            headers=_headers, method='HEAD', swift_source='VW')
+            headers=_headers, method='GET', swift_source='VW')
-        hresp = head_req.get_response(self.app)
+        source_resp = get_req.get_response(self.app)
 
-        is_dlo_manifest = 'X-Object-Manifest' in req.headers or \
+        if source_resp.content_length is None or \
-                          'X-Object-Manifest' in hresp.headers
+                source_resp.content_length > MAX_FILE_SIZE:
+            return HTTPRequestEntityTooLarge(request=req)
+
+        return source_resp
+
+    def _put_versioned_obj(self, req, put_path_info, source_resp):
+        # Create a new Request object to PUT to the versions container, copying
+        # all headers from the source object apart from x-timestamp.
+        put_req = make_pre_authed_request(
+            req.environ, path=put_path_info, method='PUT',
+            swift_source='VW')
+        copy_header_subset(source_resp, put_req,
+                           lambda k: k.lower() != 'x-timestamp')
+        put_req.headers['x-auth-token'] = req.headers.get('x-auth-token')
+        put_req.environ['wsgi.input'] = FileLikeIter(source_resp.app_iter)
+        return put_req.get_response(self.app)
+
+    def _check_response_error(self, req, resp):
+        """
+        Raise Error Response in case of error
+        """
+        if is_success(resp.status_int):
+            return
+        if is_client_error(resp.status_int):
+            # missing container or bad permissions
+            raise HTTPPreconditionFailed(request=req)
+        # could not version the data, bail
+        raise HTTPServiceUnavailable(request=req)
+
+    def handle_obj_versions_put(self, req, versions_cont, api_version,
+                                account_name, object_name):
+        """
+        Copy current version of object to versions_container before proceding
+        with original request.
+
+        :param req: original request.
+        :param versions_cont: container where previous versions of the object
+                              are stored.
+        :param api_version: api version.
+        :param account_name: account name.
+        :param object_name: name of object of original request
+        """
+        if 'X-Object-Manifest' in req.headers:
+            # do not version DLO manifest, proceed with original request
+            return self.app
+
+        get_resp = self._get_source_object(req, req.path_info)
+
+        if 'X-Object-Manifest' in get_resp.headers:
+            # do not version DLO manifest, proceed with original request
+            close_if_possible(get_resp.app_iter)
+            return self.app
+        if get_resp.status_int == HTTP_NOT_FOUND:
+            # nothing to version, proceed with original request
+            close_if_possible(get_resp.app_iter)
+            return self.app
+
+        # check for any other errors
+        self._check_response_error(req, get_resp)
 
         # if there's an existing object, then copy it to
         # X-Versions-Location
-        if is_success(hresp.status_int) and not is_dlo_manifest:
+        prefix_len = '%03x' % len(object_name)
-            lcontainer = object_versions.split('/')[0]
+        lprefix = prefix_len + object_name + '/'
-            prefix_len = '%03x' % len(object_name)
+        ts_source = get_resp.headers.get(
-            lprefix = prefix_len + object_name + '/'
+            'x-timestamp',
-            ts_source = hresp.environ.get('swift_x_timestamp')
+            calendar.timegm(time.strptime(
-            if ts_source is None:
+                get_resp.headers['last-modified'],
-                ts_source = calendar.timegm(time.strptime(
+                '%a, %d %b %Y %H:%M:%S GMT')))
-                                            hresp.headers['last-modified'],
+        vers_obj_name = lprefix + Timestamp(ts_source).internal
-                                            '%a, %d %b %Y %H:%M:%S GMT'))
-            new_ts = Timestamp(ts_source).internal
-            vers_obj_name = lprefix + new_ts
-            copy_headers = {
-                'Destination': '%s/%s' % (lcontainer, vers_obj_name),
-                'x-auth-token': req.headers.get('x-auth-token')}
 
-            # COPY implementation sets X-Newest to True when it internally
+        put_path_info = "/%s/%s/%s/%s" % (
-            # does a GET on source object. So, we don't have to explicity
+            api_version, account_name, versions_cont, vers_obj_name)
-            # set it in request headers here.
+        put_resp = self._put_versioned_obj(req, put_path_info, get_resp)
-            copy_req = make_pre_authed_request(
-                req.environ, path=req.path_info,
-                headers=copy_headers, method='COPY', swift_source='VW')
-            copy_resp = copy_req.get_response(self.app)
 
-            if is_success(copy_resp.status_int):
+        self._check_response_error(req, put_resp)
-                # success versioning previous existing object
+        return self.app
-                # return None and handle original request
-                ret = None
-            else:
-                if is_client_error(copy_resp.status_int):
-                    # missing container or bad permissions
-                    ret = HTTPPreconditionFailed(request=req)
-                else:
-                    # could not copy the data, bail
-                    ret = HTTPServiceUnavailable(request=req)
 
-        else:
+    def handle_obj_versions_delete(self, req, versions_cont, api_version,
-            if hresp.status_int == HTTP_NOT_FOUND or is_dlo_manifest:
-                # nothing to version
-                # return None and handle original request
-                ret = None
-            else:
-                # if not HTTP_NOT_FOUND, return error immediately
-                ret = hresp
-
-        return ret
-
-    def handle_obj_versions_delete(self, req, object_versions,
                                    account_name, container_name, object_name):
-        lcontainer = object_versions.split('/')[0]
+        """
+        Delete current version of object and pop previous version in its place.
+
+        :param req: original request.
+        :param versions_cont: container where previous versions of the object
+                              are stored.
+        :param api_version: api version.
+        :param account_name: account name.
+        :param container_name: container name.
+        :param object_name: object name.
+        """
         prefix_len = '%03x' % len(object_name)
         lprefix = prefix_len + object_name + '/'
 
-        item_iter = self._listing_iter(account_name, lcontainer, lprefix, req)
+        item_iter = self._listing_iter(account_name, versions_cont, lprefix,
+                                       req)
 
         authed = False
         for previous_version in item_iter:
             if not authed:
-                # we're about to start making COPY requests - need to
+                # validate the write access to the versioned container before
-                # validate the write access to the versioned container
+                # making any backend requests
                 if 'swift.authorize' in req.environ:
                     container_info = get_container_info(
                         req.environ, self.app)
@@ -348,35 +386,29 @@ class VersionedWritesContext(WSGIContext):
             # current object and delete the previous version
             prev_obj_name = previous_version['name'].encode('utf-8')
 
-            copy_path = '/v1/' + account_name + '/' + \
+            get_path = "/%s/%s/%s/%s" % (
-                        lcontainer + '/' + prev_obj_name
+                api_version, account_name, versions_cont, prev_obj_name)
 
-            copy_headers = {'X-Newest': 'True',
+            get_resp = self._get_source_object(req, get_path)
-                            'Destination': container_name + '/' + object_name,
-                            'x-auth-token': req.headers.get('x-auth-token')}
-
-            copy_req = make_pre_authed_request(
-                req.environ, path=copy_path,
-                headers=copy_headers, method='COPY', swift_source='VW')
-            copy_resp = copy_req.get_response(self.app)
 
             # if the version isn't there, keep trying with previous version
-            if copy_resp.status_int == HTTP_NOT_FOUND:
+            if get_resp.status_int == HTTP_NOT_FOUND:
                 continue
 
-            if not is_success(copy_resp.status_int):
+            self._check_response_error(req, get_resp)
-                if is_client_error(copy_resp.status_int):
-                    # some user error, maybe permissions
-                    return HTTPPreconditionFailed(request=req)
-                else:
-                    # could not copy the data, bail
-                    return HTTPServiceUnavailable(request=req)
 
-            # reset these because the COPY changed them
+            put_path_info = "/%s/%s/%s/%s" % (
-            new_del_req = make_pre_authed_request(
+                api_version, account_name, container_name, object_name)
-                req.environ, path=copy_path, method='DELETE',
+            put_resp = self._put_versioned_obj(req, put_path_info, get_resp)
+
+            self._check_response_error(req, put_resp)
+
+            # redirect the original DELETE to the source of the reinstated
+            # version object - we already auth'd original req so make a
+            # pre-authed request
+            req = make_pre_authed_request(
+                req.environ, path=get_path, method='DELETE',
                 swift_source='VW')
-            req = new_del_req
 
             # remove 'X-If-Delete-At', since it is not for the older copy
             if 'X-If-Delete-At' in req.headers:
@@ -438,7 +470,7 @@ class VersionedWritesMiddleware(object):
                 req.headers['X-Versions-Location'] = ''
 
                 # if both headers are in the same request
-                # adding location takes precendence over removing
+                # adding location takes precedence over removing
                 if 'X-Remove-Versions-Location' in req.headers:
                     del req.headers['X-Remove-Versions-Location']
             else:
@@ -456,7 +488,7 @@ class VersionedWritesMiddleware(object):
         vw_ctx = VersionedWritesContext(self.app, self.logger)
         return vw_ctx.handle_container_request(req.environ, start_response)
 
-    def object_request(self, req, version, account, container, obj,
+    def object_request(self, req, api_version, account, container, obj,
                        allow_versioned_writes):
         account_name = unquote(account)
         container_name = unquote(container)
@@ -473,7 +505,7 @@ class VersionedWritesMiddleware(object):
                 account_name = check_account_format(req, account_name)
             container_name, object_name = check_destination_header(req)
             req.environ['PATH_INFO'] = "/%s/%s/%s/%s" % (
-                version, account_name, container_name, object_name)
+                api_version, account_name, container_name, object_name)
             container_info = get_container_info(
                 req.environ, self.app)
 
@@ -485,30 +517,26 @@ class VersionedWritesMiddleware(object):
         # If stored as sysmeta, check if middleware is enabled. If sysmeta
         # is not set, but versions property is set in container_info, then
         # for backwards compatibility feature is enabled.
-        object_versions = container_info.get(
+        versions_cont = container_info.get(
             'sysmeta', {}).get('versions-location')
-        if object_versions and isinstance(object_versions, six.text_type):
+        if not versions_cont:
-            object_versions = object_versions.encode('utf-8')
+            versions_cont = container_info.get('versions')
-        elif not object_versions:
-            object_versions = container_info.get('versions')
             # if allow_versioned_writes is not set in the configuration files
             # but 'versions' is configured, enable feature to maintain
             # backwards compatibility
-            if not allow_versioned_writes and object_versions:
+            if not allow_versioned_writes and versions_cont:
                 is_enabled = True
 
-        if is_enabled and object_versions:
+        if is_enabled and versions_cont:
-            object_versions = unquote(object_versions)
+            versions_cont = unquote(versions_cont).split('/')[0]
             vw_ctx = VersionedWritesContext(self.app, self.logger)
             if req.method in ('PUT', 'COPY'):
-                policy_idx = req.headers.get(
-                    'X-Backend-Storage-Policy-Index',
-                    container_info['storage_policy'])
                 resp = vw_ctx.handle_obj_versions_put(
-                    req, object_versions, object_name, policy_idx)
+                    req, versions_cont, api_version, account_name,
+                    object_name)
             else:  # handle DELETE
                 resp = vw_ctx.handle_obj_versions_delete(
-                    req, object_versions, account_name,
+                    req, versions_cont, api_version, account_name,
                     container_name, object_name)
 
         if resp:
@@ -522,7 +550,7 @@ class VersionedWritesMiddleware(object):
         # versioned container
         req = Request(env.copy())
         try:
-            (version, account, container, obj) = req.split_path(3, 4, True)
+            (api_version, account, container, obj) = req.split_path(3, 4, True)
         except ValueError:
             return self.app(env, start_response)
 
@@ -551,7 +579,7 @@ class VersionedWritesMiddleware(object):
         elif obj and req.method in ('PUT', 'COPY', 'DELETE'):
             try:
                 return self.object_request(
-                    req, version, account, container, obj,
+                    req, api_version, account, container, obj,
                     allow_versioned_writes)(env, start_response)
             except HTTPException as error_response:
                 return error_response(env, start_response)

test/functional/tests.py

@@ -3586,10 +3586,23 @@ class TestObjectVersioning(Base):
         obj_name = Utils.create_name()
 
         versioned_obj = container.file(obj_name)
-        versioned_obj.write("aaaaa", hdrs={'Content-Type': 'text/jibberish01'})
+        put_headers = {'Content-Type': 'text/jibberish01',
+                       'Content-Encoding': 'gzip',
+                       'Content-Disposition': 'attachment; filename=myfile'}
+        versioned_obj.write("aaaaa", hdrs=put_headers)
         obj_info = versioned_obj.info()
         self.assertEqual('text/jibberish01', obj_info['content_type'])
 
+        # the allowed headers are configurable in object server, so we cannot
+        # assert that content-encoding or content-disposition get *copied* to
+        # the object version unless they were set on the original PUT, so
+        # populate expected_headers by making a HEAD on the original object
+        resp_headers = dict(versioned_obj.conn.response.getheaders())
+        expected_headers = {}
+        for k, v in put_headers.items():
+            if k.lower() in resp_headers:
+                expected_headers[k] = v
+
         self.assertEqual(0, versions_container.info()['object_count'])
         versioned_obj.write("bbbbb", hdrs={'Content-Type': 'text/jibberish02',
                             'X-Object-Meta-Foo': 'Bar'})
@@ -3605,6 +3618,11 @@ class TestObjectVersioning(Base):
         self.assertEqual("aaaaa", prev_version.read())
         self.assertEqual(prev_version.content_type, 'text/jibberish01')
 
+        resp_headers = dict(prev_version.conn.response.getheaders())
+        for k, v in expected_headers.items():
+            self.assertIn(k.lower(), resp_headers)
+            self.assertEqual(v, resp_headers[k.lower()])
+
         # make sure the new obj metadata did not leak to the prev. version
         self.assertTrue('foo' not in prev_version.metadata)
 
@@ -3653,6 +3671,15 @@ class TestObjectVersioning(Base):
         versioned_obj.delete()
         self.assertEqual("aaaaa", versioned_obj.read())
         self.assertEqual(0, versions_container.info()['object_count'])
+
+        # verify that all the original object headers have been copied back
+        obj_info = versioned_obj.info()
+        self.assertEqual('text/jibberish01', obj_info['content_type'])
+        resp_headers = dict(versioned_obj.conn.response.getheaders())
+        for k, v in expected_headers.items():
+            self.assertIn(k.lower(), resp_headers)
+            self.assertEqual(v, resp_headers[k.lower()])
+
         versioned_obj.delete()
         self.assertRaises(ResponseError, versioned_obj.read)
 
@@ -3816,6 +3843,107 @@ class TestCrossPolicyObjectVersioning(TestObjectVersioning):
                                 self.env.versioning_enabled,))
 
 
+class TestSloWithVersioning(Base):
+
+    def setUp(self):
+        if 'slo' not in cluster_info:
+            raise SkipTest("SLO not enabled")
+
+        self.conn = Connection(tf.config)
+        self.conn.authenticate()
+        self.account = Account(
+            self.conn, tf.config.get('account', tf.config['username']))
+        self.account.delete_containers()
+
+        # create a container with versioning
+        self.versions_container = self.account.container(Utils.create_name())
+        self.container = self.account.container(Utils.create_name())
+        self.segments_container = self.account.container(Utils.create_name())
+        if not self.container.create(
+                hdrs={'X-Versions-Location': self.versions_container.name}):
+            raise ResponseError(self.conn.response)
+        if 'versions' not in self.container.info():
+            raise SkipTest("Object versioning not enabled")
+
+        for cont in (self.versions_container, self.segments_container):
+            if not cont.create():
+                raise ResponseError(self.conn.response)
+
+        # create some segments
+        self.seg_info = {}
+        for letter, size in (('a', 1024 * 1024),
+                             ('b', 1024 * 1024)):
+            seg_name = letter
+            file_item = self.segments_container.file(seg_name)
+            file_item.write(letter * size)
+            self.seg_info[seg_name] = {
+                'size_bytes': size,
+                'etag': file_item.md5,
+                'path': '/%s/%s' % (self.segments_container.name, seg_name)}
+
+    def _create_manifest(self, seg_name):
+        # create a manifest in the versioning container
+        file_item = self.container.file("my-slo-manifest")
+        file_item.write(
+            json.dumps([self.seg_info[seg_name]]),
+            parms={'multipart-manifest': 'put'})
+        return file_item
+
+    def _assert_is_manifest(self, file_item, seg_name):
+        manifest_body = file_item.read(parms={'multipart-manifest': 'get'})
+        resp_headers = dict(file_item.conn.response.getheaders())
+        self.assertIn('x-static-large-object', resp_headers)
+        self.assertEqual('application/json; charset=utf-8',
+                         file_item.content_type)
+        try:
+            manifest = json.loads(manifest_body)
+        except ValueError:
+            self.fail("GET with multipart-manifest=get got invalid json")
+
+        self.assertEqual(1, len(manifest))
+        key_map = {'etag': 'hash', 'size_bytes': 'bytes', 'path': 'name'}
+        for k_client, k_slo in key_map.items():
+            self.assertEqual(self.seg_info[seg_name][k_client],
+                             manifest[0][k_slo])
+
+    def _assert_is_object(self, file_item, seg_name):
+        file_contents = file_item.read()
+        self.assertEqual(1024 * 1024, len(file_contents))
+        self.assertEqual(seg_name, file_contents[0])
+        self.assertEqual(seg_name, file_contents[-1])
+
+    def tearDown(self):
+        # remove versioning to allow simple container delete
+        self.container.update_metadata(hdrs={'X-Versions-Location': ''})
+        self.account.delete_containers()
+
+    def test_slo_manifest_version(self):
+        file_item = self._create_manifest('a')
+        # sanity check: read the manifest, then the large object
+        self._assert_is_manifest(file_item, 'a')
+        self._assert_is_object(file_item, 'a')
+
+        # upload new manifest
+        file_item = self._create_manifest('b')
+        # sanity check: read the manifest, then the large object
+        self._assert_is_manifest(file_item, 'b')
+        self._assert_is_object(file_item, 'b')
+
+        versions_list = self.versions_container.files()
+        self.assertEqual(1, len(versions_list))
+        version_file = self.versions_container.file(versions_list[0])
+        # check the version is still a manifest
+        self._assert_is_manifest(version_file, 'a')
+        self._assert_is_object(version_file, 'a')
+
+        # delete the newest manifest
+        file_item.delete()
+
+        # expect the original manifest file to be restored
+        self._assert_is_manifest(file_item, 'a')
+        self._assert_is_object(file_item, 'a')
+
+
 class TestTempurlEnv(object):
     tempurl_enabled = None  # tri-state: None initially, then True/False
 

test/unit/common/middleware/helpers.py

@@ -49,6 +49,7 @@ class FakeSwift(object):
         self._unclosed_req_paths = defaultdict(int)
         self.req_method_paths = []
         self.swift_sources = []
+        self.txn_ids = []
         self.uploaded = {}
         # mapping of (method, path) --> (response class, headers, body)
         self._responses = {}
@@ -83,6 +84,7 @@ class FakeSwift(object):
 
         req_headers = swob.Request(env).headers
         self.swift_sources.append(env.get('swift.source'))
+        self.txn_ids.append(env.get('swift.trans_id'))
 
         try:
             resp_class, raw_headers, body = self._find_response(method, path)

test/unit/common/middleware/test_versioned_writes.py

@@ -137,9 +137,8 @@ class VersionedWritesTestCase(VersionedWritesBaseTestCase):
             status, headers, body = self.call_vw(req)
             self.assertEqual(status, "412 Precondition Failed")
 
-        # GET/HEAD performs as normal
+        # GET performs as normal
         self.app.register('GET', '/v1/a/c', swob.HTTPOk, {}, 'passed')
-        self.app.register('HEAD', '/v1/a/c', swob.HTTPOk, {}, 'passed')
 
         for method in ('GET', 'HEAD'):
             req = Request.blank('/v1/a/c',
@@ -162,7 +161,31 @@ class VersionedWritesTestCase(VersionedWritesBaseTestCase):
         self.assertEqual('POST', method)
         self.assertEqual('/v1/a/c', path)
         self.assertTrue('x-container-sysmeta-versions-location' in req_headers)
+        self.assertEqual('',
+                         req_headers['x-container-sysmeta-versions-location'])
         self.assertTrue('x-versions-location' in req_headers)
+        self.assertEqual('', req_headers['x-versions-location'])
+        self.assertEqual(len(self.authorized), 1)
+        self.assertRequestEqual(req, self.authorized[0])
+
+    def test_empty_versions_location(self):
+        self.app.register('POST', '/v1/a/c', swob.HTTPOk, {}, 'passed')
+        req = Request.blank('/v1/a/c',
+                            headers={'X-Versions-Location': ''},
+                            environ={'REQUEST_METHOD': 'POST'})
+        status, headers, body = self.call_vw(req)
+        self.assertEqual(status, '200 OK')
+
+        # check for sysmeta header
+        calls = self.app.calls_with_headers
+        method, path, req_headers = calls[0]
+        self.assertEqual('POST', method)
+        self.assertEqual('/v1/a/c', path)
+        self.assertTrue('x-container-sysmeta-versions-location' in req_headers)
+        self.assertEqual('',
+                         req_headers['x-container-sysmeta-versions-location'])
+        self.assertTrue('x-versions-location' in req_headers)
+        self.assertEqual('', req_headers['x-versions-location'])
         self.assertEqual(len(self.authorized), 1)
         self.assertRequestEqual(req, self.authorized[0])
 
@@ -240,51 +263,27 @@ class VersionedWritesTestCase(VersionedWritesBaseTestCase):
         self.app.register(
             'PUT', '/v1/a/c/o', swob.HTTPOk, {}, 'passed')
         self.app.register(
-            'HEAD', '/v1/a/c/o', swob.HTTPNotFound, {}, None)
+            'GET', '/v1/a/c/o', swob.HTTPNotFound, {}, None)
 
         cache = FakeCache({'sysmeta': {'versions-location': 'ver_cont'}})
         req = Request.blank(
             '/v1/a/c/o',
             environ={'REQUEST_METHOD': 'PUT', 'swift.cache': cache,
-                     'CONTENT_LENGTH': '100'})
+                     'CONTENT_LENGTH': '100',
+                     'swift.trans_id': 'fake_trans_id'})
         status, headers, body = self.call_vw(req)
         self.assertEqual(status, '200 OK')
         self.assertEqual(len(self.authorized), 1)
         self.assertRequestEqual(req, self.authorized[0])
-
+        self.assertEqual(2, self.app.call_count)
-    def test_PUT_versioning_with_nonzero_default_policy(self):
+        self.assertEqual(['VW', None], self.app.swift_sources)
-        self.app.register(
+        self.assertEqual({'fake_trans_id'}, set(self.app.txn_ids))
-            'PUT', '/v1/a/c/o', swob.HTTPOk, {}, 'passed')
-        self.app.register(
-            'HEAD', '/v1/a/c/o', swob.HTTPNotFound, {}, None)
-
-        cache = FakeCache({'versions': 'ver_cont', 'storage_policy': '2'})
-        req = Request.blank(
-            '/v1/a/c/o',
-            environ={'REQUEST_METHOD': 'PUT', 'swift.cache': cache,
-                     'CONTENT_LENGTH': '100'})
-        status, headers, body = self.call_vw(req)
-        self.assertEqual(status, '200 OK')
-
-        # check for 'X-Backend-Storage-Policy-Index' in HEAD request
-        calls = self.app.calls_with_headers
-        method, path, req_headers = calls[0]
-        self.assertEqual('HEAD', method)
-        self.assertEqual('/v1/a/c/o', path)
-        self.assertTrue('X-Backend-Storage-Policy-Index' in req_headers)
-        self.assertEqual('2',
-                         req_headers.get('X-Backend-Storage-Policy-Index'))
-        self.assertEqual(len(self.authorized), 1)
-        self.assertRequestEqual(req, self.authorized[0])
 
     def test_put_object_no_versioning_with_container_config_true(self):
-        # set False to versions_write obsously and expect no COPY occurred
+        # set False to versions_write and expect no GET occurred
         self.vw.conf = {'allow_versioned_writes': 'false'}
         self.app.register(
             'PUT', '/v1/a/c/o', swob.HTTPCreated, {}, 'passed')
-        self.app.register(
-            'HEAD', '/v1/a/c/o', swob.HTTPOk,
-            {'last-modified': 'Wed, 19 Nov 2014 18:19:02 GMT'}, 'passed')
         cache = FakeCache({'versions': 'ver_cont'})
         req = Request.blank(
             '/v1/a/c/o',
@@ -295,11 +294,46 @@ class VersionedWritesTestCase(VersionedWritesBaseTestCase):
         self.assertEqual(len(self.authorized), 1)
         self.assertRequestEqual(req, self.authorized[0])
         called_method = [method for (method, path, hdrs) in self.app._calls]
-        self.assertTrue('COPY' not in called_method)
+        self.assertTrue('GET' not in called_method)
+
+    def test_put_request_is_dlo_manifest_with_container_config_true(self):
+        # set x-object-manifest on request and expect no versioning occurred
+        # only the PUT for the original client request
+        self.app.register(
+            'PUT', '/v1/a/c/o', swob.HTTPCreated, {}, 'passed')
+        cache = FakeCache({'versions': 'ver_cont'})
+        req = Request.blank(
+            '/v1/a/c/o',
+            environ={'REQUEST_METHOD': 'PUT', 'swift.cache': cache,
+                     'CONTENT_LENGTH': '100'})
+        req.headers['X-Object-Manifest'] = 'req/manifest'
+        status, headers, body = self.call_vw(req)
+        self.assertEqual(status, '201 Created')
+        self.assertEqual(len(self.authorized), 1)
+        self.assertRequestEqual(req, self.authorized[0])
+        self.assertEqual(1, self.app.call_count)
+
+    def test_put_version_is_dlo_manifest_with_container_config_true(self):
+        # set x-object-manifest on response and expect no versioning occurred
+        # only initial GET on source object ok followed by PUT
+        self.app.register('GET', '/v1/a/c/o', swob.HTTPOk,
+                          {'X-Object-Manifest': 'resp/manifest'}, 'passed')
+        self.app.register(
+            'PUT', '/v1/a/c/o', swob.HTTPCreated, {}, 'passed')
+        cache = FakeCache({'versions': 'ver_cont'})
+        req = Request.blank(
+            '/v1/a/c/o',
+            environ={'REQUEST_METHOD': 'PUT', 'swift.cache': cache,
+                     'CONTENT_LENGTH': '100'})
+        status, headers, body = self.call_vw(req)
+        self.assertEqual(status, '201 Created')
+        self.assertEqual(len(self.authorized), 1)
+        self.assertRequestEqual(req, self.authorized[0])
+        self.assertEqual(2, self.app.call_count)
 
     def test_delete_object_no_versioning_with_container_config_true(self):
         # set False to versions_write obviously and expect no GET versioning
-        # container and COPY called (just delete object as normal)
+        # container and PUT called (just delete object as normal)
         self.vw.conf = {'allow_versioned_writes': 'false'}
         self.app.register(
             'DELETE', '/v1/a/c/o', swob.HTTPNoContent, {}, 'passed')
@@ -313,8 +347,9 @@ class VersionedWritesTestCase(VersionedWritesBaseTestCase):
         self.assertRequestEqual(req, self.authorized[0])
         called_method = \
             [method for (method, path, rheaders) in self.app._calls]
-        self.assertTrue('COPY' not in called_method)
+        self.assertTrue('PUT' not in called_method)
         self.assertTrue('GET' not in called_method)
+        self.assertEqual(1, self.app.call_count)
 
     def test_copy_object_no_versioning_with_container_config_true(self):
         # set False to versions_write obviously and expect no extra
@@ -337,31 +372,90 @@ class VersionedWritesTestCase(VersionedWritesBaseTestCase):
 
     def test_new_version_success(self):
         self.app.register(
-            'PUT', '/v1/a/c/o', swob.HTTPOk, {}, 'passed')
+            'PUT', '/v1/a/c/o', swob.HTTPCreated, {}, 'passed')
         self.app.register(
-            'HEAD', '/v1/a/c/o', swob.HTTPOk,
+            'GET', '/v1/a/c/o', swob.HTTPOk,
-            {'last-modified': 'Wed, 19 Nov 2014 18:19:02 GMT'}, 'passed')
+            {'last-modified': 'Thu, 1 Jan 1970 00:00:01 GMT'}, 'passed')
         self.app.register(
-            'COPY', '/v1/a/c/o', swob.HTTPCreated, {}, None)
+            'PUT', '/v1/a/ver_cont/001o/0000000001.00000', swob.HTTPCreated,
+            {}, None)
+        cache = FakeCache({'sysmeta': {'versions-location': 'ver_cont'}})
+        req = Request.blank(
+            '/v1/a/c/o',
+            environ={'REQUEST_METHOD': 'PUT', 'swift.cache': cache,
+                     'CONTENT_LENGTH': '100',
+                     'swift.trans_id': 'fake_trans_id'})
+        status, headers, body = self.call_vw(req)
+        self.assertEqual(status, '201 Created')
+        self.assertEqual(len(self.authorized), 1)
+        self.assertRequestEqual(req, self.authorized[0])
+        self.assertEqual(['VW', 'VW', None], self.app.swift_sources)
+        self.assertEqual({'fake_trans_id'}, set(self.app.txn_ids))
+
+    def test_new_version_get_errors(self):
+        # GET on source fails, expect client error response,
+        # no PUT should happen
+        self.app.register(
+            'GET', '/v1/a/c/o', swob.HTTPBadRequest, {}, None)
+        cache = FakeCache({'versions': 'ver_cont'})
+        req = Request.blank(
+            '/v1/a/c/o',
+            environ={'REQUEST_METHOD': 'PUT', 'swift.cache': cache,
+                     'CONTENT_LENGTH': '100'})
+        status, headers, body = self.call_vw(req)
+        self.assertEqual(status, '412 Precondition Failed')
+        self.assertEqual(1, self.app.call_count)
+
+        # GET on source fails, expect server error response
+        self.app.register(
+            'GET', '/v1/a/c/o', swob.HTTPBadGateway, {}, None)
+        req = Request.blank(
+            '/v1/a/c/o',
+            environ={'REQUEST_METHOD': 'PUT', 'swift.cache': cache,
+                     'CONTENT_LENGTH': '100'})
+        status, headers, body = self.call_vw(req)
+        self.assertEqual(status, '503 Service Unavailable')
+        self.assertEqual(2, self.app.call_count)
+
+    def test_new_version_put_errors(self):
+        # PUT of version fails, expect client error response
+        self.app.register(
+            'GET', '/v1/a/c/o', swob.HTTPOk,
+            {'last-modified': 'Thu, 1 Jan 1970 00:00:01 GMT'}, 'passed')
+        self.app.register(
+            'PUT', '/v1/a/ver_cont/001o/0000000001.00000',
+            swob.HTTPUnauthorized, {}, None)
         cache = FakeCache({'sysmeta': {'versions-location': 'ver_cont'}})
         req = Request.blank(
             '/v1/a/c/o',
             environ={'REQUEST_METHOD': 'PUT', 'swift.cache': cache,
                      'CONTENT_LENGTH': '100'})
         status, headers, body = self.call_vw(req)
-        self.assertEqual(status, '200 OK')
+        self.assertEqual(status, '412 Precondition Failed')
-        self.assertEqual(len(self.authorized), 1)
+        self.assertEqual(2, self.app.call_count)
-        self.assertRequestEqual(req, self.authorized[0])
+
+        # PUT of version fails, expect server error response
+        self.app.register(
+            'PUT', '/v1/a/ver_cont/001o/0000000001.00000', swob.HTTPBadGateway,
+            {}, None)
+        req = Request.blank(
+            '/v1/a/c/o',
+            environ={'REQUEST_METHOD': 'PUT', 'swift.cache': cache,
+                     'CONTENT_LENGTH': '100'})
+        status, headers, body = self.call_vw(req)
+        self.assertEqual(status, '503 Service Unavailable')
+        self.assertEqual(4, self.app.call_count)
 
     @local_tz
     def test_new_version_sysmeta_precedence(self):
         self.app.register(
             'PUT', '/v1/a/c/o', swob.HTTPOk, {}, 'passed')
         self.app.register(
-            'HEAD', '/v1/a/c/o', swob.HTTPOk,
+            'GET', '/v1/a/c/o', swob.HTTPOk,
             {'last-modified': 'Thu, 1 Jan 1970 00:00:00 GMT'}, 'passed')
         self.app.register(
-            'COPY', '/v1/a/c/o', swob.HTTPCreated, {}, None)
+            'PUT', '/v1/a/ver_cont/001o/0000000000.00000', swob.HTTPOk,
+            {}, None)
 
         # fill cache with two different values for versions location
         # new middleware should use sysmeta first
@@ -379,16 +473,14 @@ class VersionedWritesTestCase(VersionedWritesBaseTestCase):
         # check that sysmeta header was used
         calls = self.app.calls_with_headers
         method, path, req_headers = calls[1]
-        self.assertEqual('COPY', method)
+        self.assertEqual('PUT', method)
-        self.assertEqual('/v1/a/c/o', path)
+        self.assertEqual('/v1/a/ver_cont/001o/0000000000.00000', path)
-        self.assertEqual('ver_cont/001o/0000000000.00000',
-                         req_headers['Destination'])
 
     def test_copy_first_version(self):
         self.app.register(
             'COPY', '/v1/a/src_cont/src_obj', swob.HTTPOk, {}, 'passed')
         self.app.register(
-            'HEAD', '/v1/a/tgt_cont/tgt_obj', swob.HTTPNotFound, {}, None)
+            'GET', '/v1/a/tgt_cont/tgt_obj', swob.HTTPNotFound, {}, None)
         cache = FakeCache({'sysmeta': {'versions-location': 'ver_cont'}})
         req = Request.blank(
             '/v1/a/src_cont/src_obj',
@@ -399,15 +491,17 @@ class VersionedWritesTestCase(VersionedWritesBaseTestCase):
         self.assertEqual(status, '200 OK')
         self.assertEqual(len(self.authorized), 1)
         self.assertRequestEqual(req, self.authorized[0])
+        self.assertEqual(2, self.app.call_count)
 
     def test_copy_new_version(self):
         self.app.register(
             'COPY', '/v1/a/src_cont/src_obj', swob.HTTPOk, {}, 'passed')
         self.app.register(
-            'HEAD', '/v1/a/tgt_cont/tgt_obj', swob.HTTPOk,
+            'GET', '/v1/a/tgt_cont/tgt_obj', swob.HTTPOk,
-            {'last-modified': 'Wed, 19 Nov 2014 18:19:02 GMT'}, 'passed')
+            {'last-modified': 'Thu, 1 Jan 1970 00:00:01 GMT'}, 'passed')
         self.app.register(
-            'COPY', '/v1/a/tgt_cont/tgt_obj', swob.HTTPCreated, {}, None)
+            'PUT', '/v1/a/ver_cont/007tgt_obj/0000000001.00000', swob.HTTPOk,
+            {}, None)
         cache = FakeCache({'sysmeta': {'versions-location': 'ver_cont'}})
         req = Request.blank(
             '/v1/a/src_cont/src_obj',
@@ -418,15 +512,17 @@ class VersionedWritesTestCase(VersionedWritesBaseTestCase):
         self.assertEqual(status, '200 OK')
         self.assertEqual(len(self.authorized), 1)
         self.assertRequestEqual(req, self.authorized[0])
+        self.assertEqual(3, self.app.call_count)
 
     def test_copy_new_version_different_account(self):
         self.app.register(
             'COPY', '/v1/src_a/src_cont/src_obj', swob.HTTPOk, {}, 'passed')
         self.app.register(
-            'HEAD', '/v1/tgt_a/tgt_cont/tgt_obj', swob.HTTPOk,
+            'GET', '/v1/tgt_a/tgt_cont/tgt_obj', swob.HTTPOk,
-            {'last-modified': 'Wed, 19 Nov 2014 18:19:02 GMT'}, 'passed')
+            {'last-modified': 'Thu, 1 Jan 1970 00:00:01 GMT'}, 'passed')
         self.app.register(
-            'COPY', '/v1/tgt_a/tgt_cont/tgt_obj', swob.HTTPCreated, {}, None)
+            'PUT', '/v1/tgt_a/ver_cont/007tgt_obj/0000000001.00000',
+            swob.HTTPOk, {}, None)
         cache = FakeCache({'sysmeta': {'versions-location': 'ver_cont'}})
         req = Request.blank(
             '/v1/src_a/src_cont/src_obj',
@@ -438,6 +534,7 @@ class VersionedWritesTestCase(VersionedWritesBaseTestCase):
         self.assertEqual(status, '200 OK')
         self.assertEqual(len(self.authorized), 1)
         self.assertRequestEqual(req, self.authorized[0])
+        self.assertEqual(3, self.app.call_count)
 
     def test_copy_new_version_bogus_account(self):
         cache = FakeCache({'sysmeta': {'versions-location': 'ver_cont'}})
@@ -462,11 +559,14 @@ class VersionedWritesTestCase(VersionedWritesBaseTestCase):
         req = Request.blank(
             '/v1/a/c/o',
             environ={'REQUEST_METHOD': 'DELETE', 'swift.cache': cache,
-                     'CONTENT_LENGTH': '0'})
+                     'CONTENT_LENGTH': '0', 'swift.trans_id': 'fake_trans_id'})
         status, headers, body = self.call_vw(req)
         self.assertEqual(status, '200 OK')
         self.assertEqual(len(self.authorized), 1)
         self.assertRequestEqual(req, self.authorized[0])
+        self.assertEqual(2, self.app.call_count)
+        self.assertEqual(['VW', None], self.app.swift_sources)
+        self.assertEqual({'fake_trans_id'}, set(self.app.txn_ids))
 
         prefix_listing_prefix = '/v1/a/ver_cont?format=json&prefix=001o/&'
         self.assertEqual(self.app.calls, [
@@ -475,8 +575,6 @@ class VersionedWritesTestCase(VersionedWritesBaseTestCase):
         ])
 
     def test_delete_latest_version_success(self):
-        self.app.register(
-            'DELETE', '/v1/a/c/o', swob.HTTPOk, {}, 'passed')
         self.app.register(
             'GET',
             '/v1/a/ver_cont?format=json&prefix=001o/&marker=&reverse=on',
@@ -492,8 +590,10 @@ class VersionedWritesTestCase(VersionedWritesBaseTestCase):
             '"name": "001o/1", '
             '"content_type": "text/plain"}]')
         self.app.register(
-            'COPY', '/v1/a/ver_cont/001o/2', swob.HTTPCreated,
+            'GET', '/v1/a/ver_cont/001o/2', swob.HTTPCreated,
-            {}, None)
+            {'content-length': '3'}, None)
+        self.app.register(
+            'PUT', '/v1/a/c/o', swob.HTTPCreated, {}, None)
         self.app.register(
             'DELETE', '/v1/a/ver_cont/001o/2', swob.HTTPOk,
             {}, None)
@@ -503,11 +603,14 @@ class VersionedWritesTestCase(VersionedWritesBaseTestCase):
             '/v1/a/c/o',
             headers={'X-If-Delete-At': 1},
             environ={'REQUEST_METHOD': 'DELETE', 'swift.cache': cache,
-                     'CONTENT_LENGTH': '0'})
+                     'CONTENT_LENGTH': '0', 'swift.trans_id': 'fake_trans_id'})
         status, headers, body = self.call_vw(req)
         self.assertEqual(status, '200 OK')
         self.assertEqual(len(self.authorized), 1)
         self.assertRequestEqual(req, self.authorized[0])
+        self.assertEqual(4, self.app.call_count)
+        self.assertEqual(['VW', 'VW', 'VW', 'VW'], self.app.swift_sources)
+        self.assertEqual({'fake_trans_id'}, set(self.app.txn_ids))
 
         # check that X-If-Delete-At was removed from DELETE request
         req_headers = self.app.headers[-1]
@@ -516,7 +619,8 @@ class VersionedWritesTestCase(VersionedWritesBaseTestCase):
         prefix_listing_prefix = '/v1/a/ver_cont?format=json&prefix=001o/&'
         self.assertEqual(self.app.calls, [
             ('GET', prefix_listing_prefix + 'marker=&reverse=on'),
-            ('COPY', '/v1/a/ver_cont/001o/2'),
+            ('GET', '/v1/a/ver_cont/001o/2'),
+            ('PUT', '/v1/a/c/o'),
             ('DELETE', '/v1/a/ver_cont/001o/2'),
         ])
 
@@ -535,8 +639,10 @@ class VersionedWritesTestCase(VersionedWritesBaseTestCase):
             '"name": "001o/1", '
             '"content_type": "text/plain"}]')
         self.app.register(
-            'COPY', '/v1/a/ver_cont/001o/1', swob.HTTPCreated,
+            'GET', '/v1/a/ver_cont/001o/1', swob.HTTPOk,
-            {}, None)
+            {'content-length': '3'}, None)
+        self.app.register(
+            'PUT', '/v1/a/c/o', swob.HTTPCreated, {}, None)
         self.app.register(
             'DELETE', '/v1/a/ver_cont/001o/1', swob.HTTPOk,
             {}, None)
@@ -554,7 +660,8 @@ class VersionedWritesTestCase(VersionedWritesBaseTestCase):
         prefix_listing_prefix = '/v1/a/ver_cont?format=json&prefix=001o/&'
         self.assertEqual(self.app.calls, [
             ('GET', prefix_listing_prefix + 'marker=&reverse=on'),
-            ('COPY', '/v1/a/ver_cont/001o/1'),
+            ('GET', '/v1/a/ver_cont/001o/1'),
+            ('PUT', '/v1/a/c/o'),
             ('DELETE', '/v1/a/ver_cont/001o/1'),
         ])
 
@@ -576,11 +683,13 @@ class VersionedWritesTestCase(VersionedWritesBaseTestCase):
 
         # expired object
         self.app.register(
-            'COPY', '/v1/a/ver_cont/001o/2', swob.HTTPNotFound,
+            'GET', '/v1/a/ver_cont/001o/2', swob.HTTPNotFound,
             {}, None)
         self.app.register(
-            'COPY', '/v1/a/ver_cont/001o/1', swob.HTTPCreated,
+            'GET', '/v1/a/ver_cont/001o/1', swob.HTTPCreated,
-            {}, None)
+            {'content-length': '3'}, None)
+        self.app.register(
+            'PUT', '/v1/a/c/o', swob.HTTPOk, {}, None)
         self.app.register(
             'DELETE', '/v1/a/ver_cont/001o/1', swob.HTTPOk,
             {}, None)
@@ -594,19 +703,19 @@ class VersionedWritesTestCase(VersionedWritesBaseTestCase):
         self.assertEqual(status, '200 OK')
         self.assertEqual(len(self.authorized), 1)
         self.assertRequestEqual(req, self.authorized[0])
+        self.assertEqual(5, self.app.call_count)
 
         prefix_listing_prefix = '/v1/a/ver_cont?format=json&prefix=001o/&'
         self.assertEqual(self.app.calls, [
             ('GET', prefix_listing_prefix + 'marker=&reverse=on'),
-            ('COPY', '/v1/a/ver_cont/001o/2'),
+            ('GET', '/v1/a/ver_cont/001o/2'),
-            ('COPY', '/v1/a/ver_cont/001o/1'),
+            ('GET', '/v1/a/ver_cont/001o/1'),
+            ('PUT', '/v1/a/c/o'),
             ('DELETE', '/v1/a/ver_cont/001o/1'),
         ])
 
     def test_denied_DELETE_of_versioned_object(self):
         authorize_call = []
-        self.app.register(
-            'DELETE', '/v1/a/c/o', swob.HTTPOk, {}, 'passed')
         self.app.register(
             'GET',
             '/v1/a/ver_cont?format=json&prefix=001o/&marker=&reverse=on',
@@ -621,11 +730,9 @@ class VersionedWritesTestCase(VersionedWritesBaseTestCase):
             '"bytes": 3, '
             '"name": "001o/1", '
             '"content_type": "text/plain"}]')
-        self.app.register(
-            'DELETE', '/v1/a/c/o', swob.HTTPForbidden,
-            {}, None)
 
         def fake_authorize(req):
+            # the container GET is pre-auth'd so here we deny the object DELETE
             authorize_call.append(req)
             return swob.HTTPForbidden()
 
@@ -669,8 +776,10 @@ class VersionedWritesOldContainersTestCase(VersionedWritesBaseTestCase):
             '&marker=001o/2',
             swob.HTTPNotFound, {}, None)
         self.app.register(
-            'COPY', '/v1/a/ver_cont/001o/2', swob.HTTPCreated,
+            'GET', '/v1/a/ver_cont/001o/2', swob.HTTPCreated,
-            {}, None)
+            {'content-length': '3'}, None)
+        self.app.register(
+            'PUT', '/v1/a/c/o', swob.HTTPCreated, {}, None)
         self.app.register(
             'DELETE', '/v1/a/ver_cont/001o/2', swob.HTTPOk,
             {}, None)
@@ -680,11 +789,15 @@ class VersionedWritesOldContainersTestCase(VersionedWritesBaseTestCase):
             '/v1/a/c/o',
             headers={'X-If-Delete-At': 1},
             environ={'REQUEST_METHOD': 'DELETE', 'swift.cache': cache,
-                     'CONTENT_LENGTH': '0'})
+                     'CONTENT_LENGTH': '0', 'swift.trans_id': 'fake_trans_id'})
         status, headers, body = self.call_vw(req)
         self.assertEqual(status, '200 OK')
         self.assertEqual(len(self.authorized), 1)
         self.assertRequestEqual(req, self.authorized[0])
+        self.assertEqual(5, self.app.call_count)
+        self.assertEqual(['VW', 'VW', 'VW', 'VW', 'VW'],
+                         self.app.swift_sources)
+        self.assertEqual({'fake_trans_id'}, set(self.app.txn_ids))
 
         # check that X-If-Delete-At was removed from DELETE request
         req_headers = self.app.headers[-1]
@@ -694,7 +807,8 @@ class VersionedWritesOldContainersTestCase(VersionedWritesBaseTestCase):
         self.assertEqual(self.app.calls, [
             ('GET', prefix_listing_prefix + 'marker=&reverse=on'),
             ('GET', prefix_listing_prefix + 'marker=001o/2'),
-            ('COPY', '/v1/a/ver_cont/001o/2'),
+            ('GET', '/v1/a/ver_cont/001o/2'),
+            ('PUT', '/v1/a/c/o'),
             ('DELETE', '/v1/a/ver_cont/001o/2'),
         ])
 
@@ -720,11 +834,13 @@ class VersionedWritesOldContainersTestCase(VersionedWritesBaseTestCase):
 
         # expired object
         self.app.register(
-            'COPY', '/v1/a/ver_cont/001o/2', swob.HTTPNotFound,
+            'GET', '/v1/a/ver_cont/001o/2', swob.HTTPNotFound,
             {}, None)
         self.app.register(
-            'COPY', '/v1/a/ver_cont/001o/1', swob.HTTPCreated,
+            'GET', '/v1/a/ver_cont/001o/1', swob.HTTPCreated,
-            {}, None)
+            {'content-length': '3'}, None)
+        self.app.register(
+            'PUT', '/v1/a/c/o', swob.HTTPOk, {}, None)
         self.app.register(
             'DELETE', '/v1/a/ver_cont/001o/1', swob.HTTPOk,
             {}, None)
@@ -738,13 +854,15 @@ class VersionedWritesOldContainersTestCase(VersionedWritesBaseTestCase):
         self.assertEqual(status, '200 OK')
         self.assertEqual(len(self.authorized), 1)
         self.assertRequestEqual(req, self.authorized[0])
+        self.assertEqual(6, self.app.call_count)
 
         prefix_listing_prefix = '/v1/a/ver_cont?format=json&prefix=001o/&'
         self.assertEqual(self.app.calls, [
             ('GET', prefix_listing_prefix + 'marker=&reverse=on'),
             ('GET', prefix_listing_prefix + 'marker=001o/2'),
-            ('COPY', '/v1/a/ver_cont/001o/2'),
+            ('GET', '/v1/a/ver_cont/001o/2'),
-            ('COPY', '/v1/a/ver_cont/001o/1'),
+            ('GET', '/v1/a/ver_cont/001o/1'),
+            ('PUT', '/v1/a/c/o'),
             ('DELETE', '/v1/a/ver_cont/001o/1'),
         ])
 
@@ -810,13 +928,13 @@ class VersionedWritesOldContainersTestCase(VersionedWritesBaseTestCase):
             swob.HTTPOk, {}, json.dumps(list(reversed(old_versions[2:]))))
         # but all objects are already gone
         self.app.register(
-            'COPY', '/v1/a/ver_cont/001o/4', swob.HTTPNotFound,
+            'GET', '/v1/a/ver_cont/001o/4', swob.HTTPNotFound,
             {}, None)
         self.app.register(
-            'COPY', '/v1/a/ver_cont/001o/3', swob.HTTPNotFound,
+            'GET', '/v1/a/ver_cont/001o/3', swob.HTTPNotFound,
             {}, None)
         self.app.register(
-            'COPY', '/v1/a/ver_cont/001o/2', swob.HTTPNotFound,
+            'GET', '/v1/a/ver_cont/001o/2', swob.HTTPNotFound,
             {}, None)
 
         # second container server can't reverse
@@ -839,8 +957,10 @@ class VersionedWritesOldContainersTestCase(VersionedWritesBaseTestCase):
             'marker=001o/1&end_marker=001o/2',
             swob.HTTPOk, {}, '[]')
         self.app.register(
-            'COPY', '/v1/a/ver_cont/001o/1', swob.HTTPOk,
+            'GET', '/v1/a/ver_cont/001o/1', swob.HTTPOk,
-            {}, None)
+            {'content-length': '3'}, None)
+        self.app.register(
+            'PUT', '/v1/a/c/o', swob.HTTPCreated, {}, None)
         self.app.register(
             'DELETE', '/v1/a/ver_cont/001o/1', swob.HTTPNoContent,
             {}, None)
@@ -855,14 +975,15 @@ class VersionedWritesOldContainersTestCase(VersionedWritesBaseTestCase):
         prefix_listing_prefix = '/v1/a/ver_cont?format=json&prefix=001o/&'
         self.assertEqual(self.app.calls, [
             ('GET', prefix_listing_prefix + 'marker=&reverse=on'),
-            ('COPY', '/v1/a/ver_cont/001o/4'),
+            ('GET', '/v1/a/ver_cont/001o/4'),
-            ('COPY', '/v1/a/ver_cont/001o/3'),
+            ('GET', '/v1/a/ver_cont/001o/3'),
-            ('COPY', '/v1/a/ver_cont/001o/2'),
+            ('GET', '/v1/a/ver_cont/001o/2'),
             ('GET', prefix_listing_prefix + 'marker=001o/2&reverse=on'),
             ('GET', prefix_listing_prefix + 'marker=&end_marker=001o/2'),
             ('GET', prefix_listing_prefix + 'marker=001o/0&end_marker=001o/2'),
             ('GET', prefix_listing_prefix + 'marker=001o/1&end_marker=001o/2'),
-            ('COPY', '/v1/a/ver_cont/001o/1'),
+            ('GET', '/v1/a/ver_cont/001o/1'),
+            ('PUT', '/v1/a/c/o'),
             ('DELETE', '/v1/a/ver_cont/001o/1'),
         ])
 
@@ -882,10 +1003,10 @@ class VersionedWritesOldContainersTestCase(VersionedWritesBaseTestCase):
             swob.HTTPOk, {}, json.dumps(list(reversed(old_versions[-2:]))))
         # but both objects are already gone
         self.app.register(
-            'COPY', '/v1/a/ver_cont/001o/4', swob.HTTPNotFound,
+            'GET', '/v1/a/ver_cont/001o/4', swob.HTTPNotFound,
             {}, None)
         self.app.register(
-            'COPY', '/v1/a/ver_cont/001o/3', swob.HTTPNotFound,
+            'GET', '/v1/a/ver_cont/001o/3', swob.HTTPNotFound,
             {}, None)
 
         # second container server can't reverse
@@ -908,8 +1029,10 @@ class VersionedWritesOldContainersTestCase(VersionedWritesBaseTestCase):
             'marker=001o/2&end_marker=001o/3',
             swob.HTTPOk, {}, '[]')
         self.app.register(
-            'COPY', '/v1/a/ver_cont/001o/2', swob.HTTPOk,
+            'GET', '/v1/a/ver_cont/001o/2', swob.HTTPOk,
-            {}, None)
+            {'content-length': '3'}, None)
+        self.app.register(
+            'PUT', '/v1/a/c/o', swob.HTTPCreated, {}, None)
         self.app.register(
             'DELETE', '/v1/a/ver_cont/001o/2', swob.HTTPNoContent,
             {}, None)
@@ -924,12 +1047,13 @@ class VersionedWritesOldContainersTestCase(VersionedWritesBaseTestCase):
         prefix_listing_prefix = '/v1/a/ver_cont?format=json&prefix=001o/&'
         self.assertEqual(self.app.calls, [
             ('GET', prefix_listing_prefix + 'marker=&reverse=on'),
-            ('COPY', '/v1/a/ver_cont/001o/4'),
+            ('GET', '/v1/a/ver_cont/001o/4'),
-            ('COPY', '/v1/a/ver_cont/001o/3'),
+            ('GET', '/v1/a/ver_cont/001o/3'),
             ('GET', prefix_listing_prefix + 'marker=001o/3&reverse=on'),
             ('GET', prefix_listing_prefix + 'marker=&end_marker=001o/3'),
             ('GET', prefix_listing_prefix + 'marker=001o/1&end_marker=001o/3'),
             ('GET', prefix_listing_prefix + 'marker=001o/2&end_marker=001o/3'),
-            ('COPY', '/v1/a/ver_cont/001o/2'),
+            ('GET', '/v1/a/ver_cont/001o/2'),
+            ('PUT', '/v1/a/c/o'),
             ('DELETE', '/v1/a/ver_cont/001o/2'),
         ])