def test_check_object_creation_copy(self):
        headers = {'Content-Length': '0',
                   'X-Copy-From': 'c/o2',
                   'Content-Type': 'text/plain'}
        self.assertEquals(constraints.check_object_creation(Request.blank(
            '/', headers=headers), 'object_name'), None)

        headers = {'Content-Length': '1',
                   'X-Copy-From': 'c/o2',
                   'Content-Type': 'text/plain'}
        self.assertEquals(constraints.check_object_creation(Request.blank(
            '/', headers=headers), 'object_name').status_int,
            HTTP_BAD_REQUEST)

        headers = {'Transfer-Encoding': 'chunked',
                   'X-Copy-From': 'c/o2',
                   'Content-Type': 'text/plain'}
        self.assertEquals(constraints.check_object_creation(Request.blank(
            '/', headers=headers), 'object_name'), None)

        # a content-length header is always required
        headers = {'X-Copy-From': 'c/o2',
                   'Content-Type': 'text/plain'}
        self.assertEquals(constraints.check_object_creation(Request.blank(
            '/', headers=headers), 'object_name').status_int,
            HTTP_LENGTH_REQUIRED)

 def test_check_object_creation_content_type(self):
     headers = {"Transfer-Encoding": "chunked", "Content-Type": "text/plain"}
     self.assertEquals(constraints.check_object_creation(Request.blank("/", headers=headers), "object_name"), None)
     headers = {"Transfer-Encoding": "chunked"}
     self.assertEquals(
         constraints.check_object_creation(Request.blank("/", headers=headers), "object_name").status_int,
         HTTP_BAD_REQUEST,
     )

 def test_check_object_creation_name_length(self):
     headers = {"Transfer-Encoding": "chunked", "Content-Type": "text/plain"}
     name = "o" * constraints.MAX_OBJECT_NAME_LENGTH
     self.assertEquals(constraints.check_object_creation(Request.blank("/", headers=headers), name), None)
     name = "o" * (constraints.MAX_OBJECT_NAME_LENGTH + 1)
     self.assertEquals(
         constraints.check_object_creation(Request.blank("/", headers=headers), name).status_int, HTTP_BAD_REQUEST
     )

 def test_check_object_creation_content_type(self):
     headers = {'Transfer-Encoding': 'chunked',
                'Content-Type': 'text/plain'}
     self.assertEquals(constraints.check_object_creation(Request.blank(
         '/', headers=headers), 'object_name'), None)
     headers = {'Transfer-Encoding': 'chunked'}
     self.assertEquals(constraints.check_object_creation(
         Request.blank('/', headers=headers), 'object_name').status_int,
         HTTP_BAD_REQUEST)

 def test_check_object_creation_name_length(self):
     headers = {'Transfer-Encoding': 'chunked',
                'Content-Type': 'text/plain'}
     name = 'o' * constraints.MAX_OBJECT_NAME_LENGTH
     self.assertEquals(constraints.check_object_creation(Request.blank(
         '/', headers=headers), name), None)
     name = 'o' * (constraints.MAX_OBJECT_NAME_LENGTH + 1)
     self.assertEquals(constraints.check_object_creation(
         Request.blank('/', headers=headers), name).status_int,
         HTTP_BAD_REQUEST)

    def test_check_object_creation_content_type(self):
        headers = {'Transfer-Encoding': 'chunked',
                   'Content-Type': 'text/plain'}
        self.assertIsNone(constraints.check_object_creation(Request.blank(
            '/', headers=headers), 'object_name'))

        headers = {'Transfer-Encoding': 'chunked'}
        resp = constraints.check_object_creation(
            Request.blank('/', headers=headers), 'object_name')
        self.assertEqual(resp.status_int, HTTP_BAD_REQUEST)
        self.assertIn('No content type', resp.body)

    def test_check_object_creation_bad_delete_headers(self):
        headers = {"Transfer-Encoding": "chunked", "Content-Type": "text/plain", "X-Delete-After": "abc"}
        resp = constraints.check_object_creation(Request.blank("/", headers=headers), "object_name")
        self.assertEquals(resp.status_int, HTTP_BAD_REQUEST)
        self.assertTrue("Non-integer X-Delete-After" in resp.body)

        t = str(int(time.time() - 60))
        headers = {"Transfer-Encoding": "chunked", "Content-Type": "text/plain", "X-Delete-At": t}
        resp = constraints.check_object_creation(Request.blank("/", headers=headers), "object_name")
        self.assertEquals(resp.status_int, HTTP_BAD_REQUEST)
        self.assertTrue("X-Delete-At in past" in resp.body)

    def test_check_object_creation_name_length(self):
        headers = {'Transfer-Encoding': 'chunked',
                   'Content-Type': 'text/plain'}
        name = 'o' * constraints.MAX_OBJECT_NAME_LENGTH
        self.assertIsNone(constraints.check_object_creation(Request.blank(
            '/', headers=headers), name))

        name = 'o' * (MAX_OBJECT_NAME_LENGTH + 1)
        resp = constraints.check_object_creation(
            Request.blank('/', headers=headers), name)
        self.assertEqual(resp.status_int, HTTP_BAD_REQUEST)
        self.assertIn('Object name length of %d longer than %d' %
                      (MAX_OBJECT_NAME_LENGTH + 1, MAX_OBJECT_NAME_LENGTH),
                      resp.body)

 def test_check_object_creation_bad_content_type(self):
     headers = {'Transfer-Encoding': 'chunked',
                'Content-Type': '\xff\xff'}
     resp = constraints.check_object_creation(
         Request.blank('/', headers=headers), 'object_name')
     self.assertEquals(resp.status_int, HTTP_BAD_REQUEST)
     self.assert_('Content-Type' in resp.body)

    def PUT(self, req):
        """HTTP PUT request handler."""
        container_info = self.container_info(
            self.account_name, self.container_name, req)

        req.acl = container_info['write_acl']
        req.environ['swift_sync_key'] = container_info['sync_key']

        # is request authorized
        if 'swift.authorize' in req.environ:
            aresp = req.environ['swift.authorize'](req)
            if aresp:
                return aresp

        old_slo_manifest = None
        # If versioning is disabled, we must check if the object exists.
        # If it's a SLO, we will have to delete the parts if the current
        # operation is a success.
        if (self.app.delete_slo_parts and
                not container_info['sysmeta'].get('versions-location', None)):
            try:
                dest_info = get_object_info(req.environ, self.app)
                if 'slo-size' in dest_info['sysmeta']:
                    manifest_env = req.environ.copy()
                    manifest_env['QUERY_STRING'] = 'multipart-manifest=get'
                    manifest_req = make_subrequest(manifest_env, 'GET')
                    manifest_resp = manifest_req.get_response(self.app)
                    old_slo_manifest = json.loads(manifest_resp.body)
            except Exception as exc:
                self.app.logger.warn(('Failed to check existence of %s. If '
                                      'overwriting a SLO, old parts may '
                                      'remain. Error was: %s') %
                                     (req.path, exc))

        self._update_content_type(req)

        self._update_x_timestamp(req)

        # check constraints on object name and request headers
        error_response = check_object_creation(req, self.object_name) or \
            check_content_type(req)
        if error_response:
            return error_response

        if req.headers.get('Oio-Copy-From'):
            return self._link_object(req)

        data_source = req.environ['wsgi.input']
        if req.content_length:
            data_source = ExpectedSizeReader(data_source, req.content_length)

        headers = self._prepare_headers(req)
        with closing_if_possible(data_source):
            resp = self._store_object(req, data_source, headers)
        if old_slo_manifest and resp.is_success:
            self.app.logger.debug(
                'Previous object %s was a SLO, deleting parts',
                req.path)
            self._delete_slo_parts(req, old_slo_manifest)
        return resp

 def test_check_object_creation_bad_content_type(self):
     headers = {'Transfer-Encoding': 'chunked',
                'Content-Type': '\xff\xff'}
     resp = constraints.check_object_creation(
         Request.blank('/', headers=headers), 'object_name')
     self.assert_(isinstance(resp, HTTPBadRequest))
     self.assert_('Content-Type' in resp.body)

    def test_check_object_creation_bad_delete_headers(self):
        headers = {'Transfer-Encoding': 'chunked',
                   'Content-Type': 'text/plain',
                   'X-Delete-After': 'abc'}
        resp = constraints.check_object_creation(
            Request.blank('/', headers=headers), 'object_name')
        self.assertEquals(resp.status_int, HTTP_BAD_REQUEST)
        self.assert_('Non-integer X-Delete-After' in resp.body)

        t = str(int(time.time() - 60))
        headers = {'Transfer-Encoding': 'chunked',
                   'Content-Type': 'text/plain',
                   'X-Delete-At': t}
        resp = constraints.check_object_creation(
            Request.blank('/', headers=headers), 'object_name')
        self.assertEquals(resp.status_int, HTTP_BAD_REQUEST)
        self.assert_('X-Delete-At in past' in resp.body)

 def test_check_object_creation_content_length(self):
     headers = {'Content-Length': str(constraints.MAX_FILE_SIZE),
                'Content-Type': 'text/plain'}
     self.assertEquals(constraints.check_object_creation(Request.blank(
         '/', headers=headers), 'object_name'), None)
     headers = {'Content-Length': str(constraints.MAX_FILE_SIZE + 1),
                'Content-Type': 'text/plain'}
     self.assertEquals(constraints.check_object_creation(
         Request.blank('/', headers=headers), 'object_name').status_int,
         HTTP_REQUEST_ENTITY_TOO_LARGE)
     headers = {'Transfer-Encoding': 'chunked',
                'Content-Type': 'text/plain'}
     self.assertEquals(constraints.check_object_creation(Request.blank(
         '/', headers=headers), 'object_name'), None)
     headers = {'Content-Type': 'text/plain'}
     self.assertEquals(constraints.check_object_creation(
         Request.blank('/', headers=headers), 'object_name').status_int,
         HTTP_LENGTH_REQUIRED)

 def test_check_object_creation_content_length(self):
     headers = {'Content-Length': str(constraints.MAX_FILE_SIZE),
                'Content-Type': 'text/plain'}
     self.assertEquals(constraints.check_object_creation(Request.blank('/',
         headers=headers), 'object_name'), None)
     headers = {'Content-Length': str(constraints.MAX_FILE_SIZE + 1),
                'Content-Type': 'text/plain'}
     self.assert_(isinstance(constraints.check_object_creation(
         Request.blank('/', headers=headers), 'object_name'),
         HTTPRequestEntityTooLarge))
     headers = {'Transfer-Encoding': 'chunked',
                'Content-Type': 'text/plain'}
     self.assertEquals(constraints.check_object_creation(Request.blank('/',
         headers=headers), 'object_name'), None)
     headers = {'Content-Type': 'text/plain'}
     self.assert_(isinstance(constraints.check_object_creation(
         Request.blank('/', headers=headers), 'object_name'),
         HTTPLengthRequired))

    def test_check_object_creation_copy(self):
        headers = {"Content-Length": "0", "X-Copy-From": "c/o2", "Content-Type": "text/plain"}
        self.assertEquals(constraints.check_object_creation(Request.blank("/", headers=headers), "object_name"), None)

        headers = {"Content-Length": "1", "X-Copy-From": "c/o2", "Content-Type": "text/plain"}
        self.assertEquals(
            constraints.check_object_creation(Request.blank("/", headers=headers), "object_name").status_int,
            HTTP_BAD_REQUEST,
        )

        headers = {"Transfer-Encoding": "chunked", "X-Copy-From": "c/o2", "Content-Type": "text/plain"}
        self.assertEquals(constraints.check_object_creation(Request.blank("/", headers=headers), "object_name"), None)

        # a content-length header is always required
        headers = {"X-Copy-From": "c/o2", "Content-Type": "text/plain"}
        self.assertEquals(
            constraints.check_object_creation(Request.blank("/", headers=headers), "object_name").status_int,
            HTTP_LENGTH_REQUIRED,
        )

 def test_check_object_manifest_header(self):
     resp = constraints.check_object_creation(Request.blank('/',
         headers={'X-Object-Manifest': 'container/prefix', 'Content-Length':
         '0', 'Content-Type': 'text/plain'}), 'manifest')
     self.assert_(not resp)
     resp = constraints.check_object_creation(Request.blank('/',
         headers={'X-Object-Manifest': 'container', 'Content-Length': '0',
         'Content-Type': 'text/plain'}), 'manifest')
     self.assertEquals(resp.status_int, HTTP_BAD_REQUEST)
     resp = constraints.check_object_creation(Request.blank('/',
         headers={'X-Object-Manifest': '/container/prefix',
         'Content-Length': '0', 'Content-Type': 'text/plain'}), 'manifest')
     self.assertEquals(resp.status_int, HTTP_BAD_REQUEST)
     resp = constraints.check_object_creation(Request.blank('/',
         headers={'X-Object-Manifest': 'container/prefix?query=param',
         'Content-Length': '0', 'Content-Type': 'text/plain'}), 'manifest')
     self.assertEquals(resp.status_int, HTTP_BAD_REQUEST)
     resp = constraints.check_object_creation(Request.blank('/',
         headers={'X-Object-Manifest': 'container/prefix&query=param',
         'Content-Length': '0', 'Content-Type': 'text/plain'}), 'manifest')
     self.assertEquals(resp.status_int, HTTP_BAD_REQUEST)
     resp = constraints.check_object_creation(Request.blank('/',
         headers={'X-Object-Manifest': 'http://host/container/prefix',
         'Content-Length': '0', 'Content-Type': 'text/plain'}), 'manifest')
     self.assertEquals(resp.status_int, HTTP_BAD_REQUEST)

    def test_check_object_creation_content_length(self):
        headers = {'Content-Length': str(constraints.MAX_FILE_SIZE),
                   'Content-Type': 'text/plain',
                   'X-Timestamp': str(time.time())}
        self.assertIsNone(constraints.check_object_creation(Request.blank(
            '/', headers=headers), 'object_name'))

        headers = {'Content-Length': str(constraints.MAX_FILE_SIZE + 1),
                   'Content-Type': 'text/plain',
                   'X-Timestamp': str(time.time())}
        resp = constraints.check_object_creation(
            Request.blank('/', headers=headers), 'object_name')
        self.assertEqual(resp.status_int, HTTP_REQUEST_ENTITY_TOO_LARGE)

        headers = {'Transfer-Encoding': 'chunked',
                   'Content-Type': 'text/plain',
                   'X-Timestamp': str(time.time())}
        self.assertIsNone(constraints.check_object_creation(Request.blank(
            '/', headers=headers), 'object_name'))

        headers = {'Transfer-Encoding': 'gzip',
                   'Content-Type': 'text/plain',
                   'X-Timestamp': str(time.time())}
        resp = constraints.check_object_creation(Request.blank(
            '/', headers=headers), 'object_name')
        self.assertEqual(resp.status_int, HTTP_BAD_REQUEST)
        self.assertIn(b'Invalid Transfer-Encoding header value', resp.body)

        headers = {'Content-Type': 'text/plain',
                   'X-Timestamp': str(time.time())}
        resp = constraints.check_object_creation(
            Request.blank('/', headers=headers), 'object_name')
        self.assertEqual(resp.status_int, HTTP_LENGTH_REQUIRED)

        headers = {'Content-Length': 'abc',
                   'Content-Type': 'text/plain',
                   'X-Timestamp': str(time.time())}
        resp = constraints.check_object_creation(Request.blank(
            '/', headers=headers), 'object_name')
        self.assertEqual(resp.status_int, HTTP_BAD_REQUEST)
        self.assertIn(b'Invalid Content-Length header value', resp.body)

        headers = {'Transfer-Encoding': 'gzip,chunked',
                   'Content-Type': 'text/plain',
                   'X-Timestamp': str(time.time())}
        resp = constraints.check_object_creation(Request.blank(
            '/', headers=headers), 'object_name')
        self.assertEqual(resp.status_int, HTTP_NOT_IMPLEMENTED)

    def test_check_object_creation_content_length(self):
        headers = {"Content-Length": str(constraints.MAX_FILE_SIZE), "Content-Type": "text/plain"}
        self.assertEquals(constraints.check_object_creation(Request.blank("/", headers=headers), "object_name"), None)

        headers = {"Content-Length": str(constraints.MAX_FILE_SIZE + 1), "Content-Type": "text/plain"}
        self.assertEquals(
            constraints.check_object_creation(Request.blank("/", headers=headers), "object_name").status_int,
            HTTP_REQUEST_ENTITY_TOO_LARGE,
        )

        headers = {"Transfer-Encoding": "chunked", "Content-Type": "text/plain"}
        self.assertEquals(constraints.check_object_creation(Request.blank("/", headers=headers), "object_name"), None)

        headers = {"Transfer-Encoding": "gzip", "Content-Type": "text/plain"}
        self.assertEquals(
            constraints.check_object_creation(Request.blank("/", headers=headers), "object_name").status_int,
            HTTP_BAD_REQUEST,
        )

        headers = {"Content-Type": "text/plain"}
        self.assertEquals(
            constraints.check_object_creation(Request.blank("/", headers=headers), "object_name").status_int,
            HTTP_LENGTH_REQUIRED,
        )

        headers = {"Content-Length": "abc", "Content-Type": "text/plain"}
        self.assertEquals(
            constraints.check_object_creation(Request.blank("/", headers=headers), "object_name").status_int,
            HTTP_BAD_REQUEST,
        )

        headers = {"Transfer-Encoding": "gzip,chunked", "Content-Type": "text/plain"}
        self.assertEquals(
            constraints.check_object_creation(Request.blank("/", headers=headers), "object_name").status_int,
            HTTP_NOT_IMPLEMENTED,
        )

    def PUT(self, req):
        """HTTP PUT request handler."""
        if req.if_none_match is not None and '*' not in req.if_none_match:
            # Sending an etag with if-none-match isn't currently supported
            return HTTPBadRequest(request=req, content_type='text/plain',
                                  body='If-None-Match only supports *')
        if 'swift.authorize' in req.environ:
            aresp = req.environ['swift.authorize'](req)
            if aresp:
                return aresp

        # Sometimes the 'content-type' header exists, but is set to None.
        content_type_manually_set = True
        detect_content_type = \
            config_true_value(req.headers.get('x-detect-content-type'))
        if detect_content_type or not req.headers.get('content-type'):
            guessed_type, _junk = mimetypes.guess_type(req.path_info)
            req.headers['Content-Type'] = guessed_type or \
                                          'application/octet-stream'
            if detect_content_type:
                req.headers.pop('x-detect-content-type')
            else:
                content_type_manually_set = False

        error_response = check_object_creation(req, self.object_name) or \
                         check_content_type(req)
        if error_response:
            return error_response

        req.headers['X-Timestamp'] = Timestamp(time.time()).internal

        stream = req.environ['wsgi.input']
        source_header = req.headers.get('X-Copy-From')
        source_resp = None
        if source_header:
            if req.environ.get('swift.orig_req_method', req.method) != 'POST':
                req.environ.setdefault('swift.log_info', []).append(
                    'x-copy-from:%s' % source_header)
            ver, acct, _rest = req.split_path(2, 3, True)
            src_account_name = req.headers.get('X-Copy-From-Account', None)
            if src_account_name:
                src_account_name = check_account_format(req, src_account_name)
            else:
                src_account_name = acct
            src_container_name, src_obj_name = check_copy_from_header(req)
            source_header = '/%s/%s/%s/%s' % (ver, src_account_name,
                                              src_container_name, src_obj_name)
            source_req = req.copy_get()

            # make sure the source request uses it's container_info
            source_req.headers.pop('X-Backend-Storage-Policy-Index', None)
            source_req.path_info = source_header
            source_req.headers['X-Newest'] = 'true'
            orig_obj_name = self.object_name
            orig_container_name = self.container_name
            orig_account_name = self.account_name
            self.object_name = src_obj_name
            self.container_name = src_container_name
            self.account_name = src_account_name
            sink_req = Request.blank(req.path_info,
                                     environ=req.environ, headers=req.headers)
            source_resp = self.GET(source_req)

            # This gives middlewares a way to change the source; for example,
            # this lets you COPY a SLO manifest and have the new object be the
            # concatenation of the segments (like what a GET request gives
            # the client), not a copy of the manifest file.
            hook = req.environ.get(
                'swift.copy_hook',
                (lambda source_req, source_resp, sink_req: source_resp))
            source_resp = hook(source_req, source_resp, sink_req)

            if source_resp.status_int >= HTTP_MULTIPLE_CHOICES:
                return source_resp
            self.object_name = orig_obj_name
            self.container_name = orig_container_name
            self.account_name = orig_account_name
            stream = IterO(source_resp.app_iter)
            sink_req.content_length = source_resp.content_length
            if sink_req.content_length is None:
                # This indicates a transfer-encoding: chunked source object,
                # which currently only happens because there are more than
                # CONTAINER_LISTING_LIMIT segments in a segmented object. In
                # this case, we're going to refuse to do the server-side copy.
                return HTTPRequestEntityTooLarge(request=req)
            if sink_req.content_length > constraints.MAX_FILE_SIZE:
                return HTTPRequestEntityTooLarge(request=req)
            sink_req.etag = source_resp.etag

            # we no longer need the X-Copy-From header
            del sink_req.headers['X-Copy-From']
            if 'X-Copy-From-Account' in sink_req.headers:
                del sink_req.headers['X-Copy-From-Account']
            if not content_type_manually_set:
                sink_req.headers['Content-Type'] = \
                    source_resp.headers['Content-Type']
            if config_true_value(
                    sink_req.headers.get('x-fresh-metadata', 'false')):
                # post-as-copy: ignore new sysmeta, copy existing sysmeta
                condition = lambda k: is_sys_meta('object', k)
#.........这里部分代码省略.........

    def PUT(self, req):
        """HTTP PUT request handler."""
        if req.if_none_match is not None and '*' not in req.if_none_match:
            # Sending an etag with if-none-match isn't currently supported
            return HTTPBadRequest(request=req, content_type='text/plain',
                                  body='If-None-Match only supports *')
        container_info = self.container_info(
            self.account_name, self.container_name, req)
        policy_index = req.headers.get('X-Backend-Storage-Policy-Index',
                                       container_info['storage_policy'])
        obj_ring = self.app.get_object_ring(policy_index)

        # pass the policy index to storage nodes via req header
        req.headers['X-Backend-Storage-Policy-Index'] = policy_index
        container_partition = container_info['partition']
        containers = container_info['nodes']
        req.acl = container_info['write_acl']
        req.environ['swift_sync_key'] = container_info['sync_key']
        object_versions = container_info['versions']
        if 'swift.authorize' in req.environ:
            aresp = req.environ['swift.authorize'](req)
            if aresp:
                return aresp

        if not containers:
            return HTTPNotFound(request=req)

        # Sometimes the 'content-type' header exists, but is set to None.
        content_type_manually_set = True
        detect_content_type = \
            config_true_value(req.headers.get('x-detect-content-type'))
        if detect_content_type or not req.headers.get('content-type'):
            guessed_type, _junk = mimetypes.guess_type(req.path_info)
            req.headers['Content-Type'] = guessed_type or \
                'application/octet-stream'
            if detect_content_type:
                req.headers.pop('x-detect-content-type')
            else:
                content_type_manually_set = False

        error_response = check_object_creation(req, self.object_name) or \
            check_content_type(req)
        if error_response:
            return error_response

        partition, nodes = obj_ring.get_nodes(
            self.account_name, self.container_name, self.object_name)

        # do a HEAD request for container sync and checking object versions
        if 'x-timestamp' in req.headers or \
                (object_versions and not
                 req.environ.get('swift_versioned_copy')):
            # make sure proxy-server uses the right policy index
            _headers = {'X-Backend-Storage-Policy-Index': policy_index,
                        'X-Newest': 'True'}
            hreq = Request.blank(req.path_info, headers=_headers,
                                 environ={'REQUEST_METHOD': 'HEAD'})
            hresp = self.GETorHEAD_base(
                hreq, _('Object'), obj_ring, partition,
                hreq.swift_entity_path)

        # Used by container sync feature
        if 'x-timestamp' in req.headers:
            try:
                req_timestamp = Timestamp(req.headers['X-Timestamp'])
                if hresp.environ and 'swift_x_timestamp' in hresp.environ and \
                        hresp.environ['swift_x_timestamp'] >= req_timestamp:
                    return HTTPAccepted(request=req)
            except ValueError:
                return HTTPBadRequest(
                    request=req, content_type='text/plain',
                    body='X-Timestamp should be a UNIX timestamp float value; '
                         'was %r' % req.headers['x-timestamp'])
            req.headers['X-Timestamp'] = req_timestamp.internal
        else:
            req.headers['X-Timestamp'] = Timestamp(time.time()).internal

        if object_versions and not req.environ.get('swift_versioned_copy'):
            if hresp.status_int != HTTP_NOT_FOUND:
                # This is a version manifest and needs to be handled
                # differently. First copy the existing data to a new object,
                # then write the data from this request to the version manifest
                # object.
                lcontainer = object_versions.split('/')[0]
                prefix_len = '%03x' % len(self.object_name)
                lprefix = prefix_len + self.object_name + '/'
                ts_source = hresp.environ.get('swift_x_timestamp')
                if ts_source is None:
                    ts_source = time.mktime(time.strptime(
                                            hresp.headers['last-modified'],
                                            '%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)}
                copy_environ = {'REQUEST_METHOD': 'COPY',
                                'swift_versioned_copy': True
                                }
                copy_req = Request.blank(req.path_info, headers=copy_headers,
                                         environ=copy_environ)
#.........这里部分代码省略.........