"Fossies" - the Fresh Open Source Software Archive

Member "swift-2.21.0/test/unit/common/middleware/test_copy.py" (25 Mar 2019, 71839 Bytes) of package /linux/misc/openstack/swift-2.21.0.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the latest Fossies "Diffs" side-by-side code changes report for "test_copy.py": 2.19.1_vs_2.21.0.

    1 #!/usr/bin/env python
    2 # Copyright (c) 2015 OpenStack Foundation
    3 #
    4 # Licensed under the Apache License, Version 2.0 (the "License");
    5 # you may not use this file except in compliance with the License.
    6 # You may obtain a copy of the License at
    7 #
    8 #    http://www.apache.org/licenses/LICENSE-2.0
    9 #
   10 # Unless required by applicable law or agreed to in writing, software
   11 # distributed under the License is distributed on an "AS IS" BASIS,
   12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
   13 # implied.
   14 # See the License for the specific language governing permissions and
   15 # limitations under the License.
   16 
   17 import mock
   18 import unittest
   19 from hashlib import md5
   20 from six.moves import urllib
   21 
   22 from swift.common import swob
   23 from swift.common.middleware import copy
   24 from swift.common.storage_policy import POLICIES
   25 from swift.common.swob import Request, HTTPException
   26 from swift.common.utils import closing_if_possible
   27 from test.unit import patch_policies, debug_logger, FakeMemcache, FakeRing
   28 from test.unit.common.middleware.helpers import FakeSwift
   29 from test.unit.proxy.controllers.test_obj import set_http_connect, \
   30     PatchedObjControllerApp
   31 
   32 
   33 class TestCopyConstraints(unittest.TestCase):
   34     def test_validate_copy_from(self):
   35         req = Request.blank(
   36             '/v/a/c/o',
   37             headers={'x-copy-from': 'c/o2'})
   38         src_cont, src_obj = copy._check_copy_from_header(req)
   39         self.assertEqual(src_cont, 'c')
   40         self.assertEqual(src_obj, 'o2')
   41         req = Request.blank(
   42             '/v/a/c/o',
   43             headers={'x-copy-from': 'c/subdir/o2'})
   44         src_cont, src_obj = copy._check_copy_from_header(req)
   45         self.assertEqual(src_cont, 'c')
   46         self.assertEqual(src_obj, 'subdir/o2')
   47         req = Request.blank(
   48             '/v/a/c/o',
   49             headers={'x-copy-from': '/c/o2'})
   50         src_cont, src_obj = copy._check_copy_from_header(req)
   51         self.assertEqual(src_cont, 'c')
   52         self.assertEqual(src_obj, 'o2')
   53 
   54     def test_validate_bad_copy_from(self):
   55         req = Request.blank(
   56             '/v/a/c/o',
   57             headers={'x-copy-from': 'bad_object'})
   58         self.assertRaises(HTTPException,
   59                           copy._check_copy_from_header, req)
   60 
   61     def test_validate_destination(self):
   62         req = Request.blank(
   63             '/v/a/c/o',
   64             headers={'destination': 'c/o2'})
   65         src_cont, src_obj = copy._check_destination_header(req)
   66         self.assertEqual(src_cont, 'c')
   67         self.assertEqual(src_obj, 'o2')
   68         req = Request.blank(
   69             '/v/a/c/o',
   70             headers={'destination': 'c/subdir/o2'})
   71         src_cont, src_obj = copy._check_destination_header(req)
   72         self.assertEqual(src_cont, 'c')
   73         self.assertEqual(src_obj, 'subdir/o2')
   74         req = Request.blank(
   75             '/v/a/c/o',
   76             headers={'destination': '/c/o2'})
   77         src_cont, src_obj = copy._check_destination_header(req)
   78         self.assertEqual(src_cont, 'c')
   79         self.assertEqual(src_obj, 'o2')
   80 
   81     def test_validate_bad_destination(self):
   82         req = Request.blank(
   83             '/v/a/c/o',
   84             headers={'destination': 'bad_object'})
   85         self.assertRaises(HTTPException,
   86                           copy._check_destination_header, req)
   87 
   88 
   89 class TestServerSideCopyMiddleware(unittest.TestCase):
   90     def setUp(self):
   91         self.app = FakeSwift()
   92         self.ssc = copy.filter_factory({})(self.app)
   93         self.ssc.logger = self.app.logger
   94 
   95     def tearDown(self):
   96         self.assertEqual(self.app.unclosed_requests, {})
   97 
   98     def call_app(self, req, app=None, expect_exception=False):
   99         if app is None:
  100             app = self.app
  101 
  102         self.authorized = []
  103 
  104         def authorize(req):
  105             self.authorized.append(req)
  106 
  107         if 'swift.authorize' not in req.environ:
  108             req.environ['swift.authorize'] = authorize
  109 
  110         req.headers.setdefault("User-Agent", "Bruce Wayne")
  111 
  112         status = [None]
  113         headers = [None]
  114 
  115         def start_response(s, h, ei=None):
  116             status[0] = s
  117             headers[0] = h
  118 
  119         body_iter = app(req.environ, start_response)
  120         body = b''
  121         caught_exc = None
  122         try:
  123             # appease the close-checker
  124             with closing_if_possible(body_iter):
  125                 for chunk in body_iter:
  126                     body += chunk
  127         except Exception as exc:
  128             if expect_exception:
  129                 caught_exc = exc
  130             else:
  131                 raise
  132 
  133         if expect_exception:
  134             return status[0], headers[0], body, caught_exc
  135         else:
  136             return status[0], headers[0], body
  137 
  138     def call_ssc(self, req, **kwargs):
  139         return self.call_app(req, app=self.ssc, **kwargs)
  140 
  141     def assertRequestEqual(self, req, other):
  142         self.assertEqual(req.method, other.method)
  143         self.assertEqual(req.path, other.path)
  144 
  145     def test_no_object_in_path_pass_through(self):
  146         self.app.register('PUT', '/v1/a/c', swob.HTTPCreated, {})
  147         req = Request.blank('/v1/a/c', method='PUT')
  148         status, headers, body = self.call_ssc(req)
  149         self.assertEqual(status, '201 Created')
  150         self.assertEqual(len(self.authorized), 1)
  151         self.assertRequestEqual(req, self.authorized[0])
  152 
  153     def test_object_pass_through_methods(self):
  154         for method in ['DELETE', 'GET', 'HEAD', 'REPLICATE']:
  155             self.app.register(method, '/v1/a/c/o', swob.HTTPOk, {})
  156             req = Request.blank('/v1/a/c/o', method=method)
  157             status, headers, body = self.call_ssc(req)
  158             self.assertEqual(status, '200 OK')
  159             self.assertEqual(len(self.authorized), 1)
  160             self.assertRequestEqual(req, self.authorized[0])
  161             self.assertNotIn('swift.orig_req_method', req.environ)
  162 
  163     def test_basic_put_with_x_copy_from(self):
  164         self.app.register('GET', '/v1/a/c/o', swob.HTTPOk, {}, 'passed')
  165         self.app.register('PUT', '/v1/a/c/o2', swob.HTTPCreated, {})
  166         req = Request.blank('/v1/a/c/o2', environ={'REQUEST_METHOD': 'PUT'},
  167                             headers={'Content-Length': '0',
  168                                      'X-Copy-From': 'c/o'})
  169         status, headers, body = self.call_ssc(req)
  170         self.assertEqual(status, '201 Created')
  171         self.assertTrue(('X-Copied-From', 'c/o') in headers)
  172         self.assertEqual(len(self.authorized), 2)
  173         self.assertEqual('GET', self.authorized[0].method)
  174         self.assertEqual('/v1/a/c/o', self.authorized[0].path)
  175         self.assertEqual('PUT', self.authorized[1].method)
  176         self.assertEqual('/v1/a/c/o2', self.authorized[1].path)
  177         self.assertEqual(self.app.swift_sources[0], 'SSC')
  178         self.assertEqual(self.app.swift_sources[1], 'SSC')
  179         # For basic test cases, assert orig_req_method behavior
  180         self.assertNotIn('swift.orig_req_method', req.environ)
  181 
  182     def test_static_large_object_manifest(self):
  183         self.app.register('GET', '/v1/a/c/o', swob.HTTPOk,
  184                           {'X-Static-Large-Object': 'True',
  185                            'Etag': 'should not be sent'}, 'passed')
  186         self.app.register('PUT', '/v1/a/c/o2?multipart-manifest=put',
  187                           swob.HTTPCreated, {})
  188         req = Request.blank('/v1/a/c/o2?multipart-manifest=get',
  189                             environ={'REQUEST_METHOD': 'PUT'},
  190                             headers={'Content-Length': '0',
  191                                      'X-Copy-From': 'c/o'})
  192         status, headers, body = self.call_ssc(req)
  193         self.assertEqual(status, '201 Created')
  194         self.assertTrue(('X-Copied-From', 'c/o') in headers)
  195         self.assertEqual(2, len(self.app.calls))
  196         self.assertEqual('GET', self.app.calls[0][0])
  197         get_path, qs = self.app.calls[0][1].split('?')
  198         params = urllib.parse.parse_qs(qs)
  199         self.assertDictEqual(
  200             {'format': ['raw'], 'multipart-manifest': ['get']}, params)
  201         self.assertEqual(get_path, '/v1/a/c/o')
  202         self.assertEqual(self.app.calls[1],
  203                          ('PUT', '/v1/a/c/o2?multipart-manifest=put'))
  204         req_headers = self.app.headers[1]
  205         self.assertNotIn('X-Static-Large-Object', req_headers)
  206         self.assertNotIn('Etag', req_headers)
  207         self.assertEqual(len(self.authorized), 2)
  208         self.assertEqual('GET', self.authorized[0].method)
  209         self.assertEqual('/v1/a/c/o', self.authorized[0].path)
  210         self.assertEqual('PUT', self.authorized[1].method)
  211         self.assertEqual('/v1/a/c/o2', self.authorized[1].path)
  212 
  213     def test_static_large_object(self):
  214         self.app.register('GET', '/v1/a/c/o', swob.HTTPOk,
  215                           {'X-Static-Large-Object': 'True',
  216                            'Etag': 'should not be sent'}, 'passed')
  217         self.app.register('PUT', '/v1/a/c/o2',
  218                           swob.HTTPCreated, {})
  219         req = Request.blank('/v1/a/c/o2',
  220                             environ={'REQUEST_METHOD': 'PUT'},
  221                             headers={'Content-Length': '0',
  222                                      'X-Copy-From': 'c/o'})
  223         status, headers, body = self.call_ssc(req)
  224         self.assertEqual(status, '201 Created')
  225         self.assertTrue(('X-Copied-From', 'c/o') in headers)
  226         self.assertEqual(self.app.calls, [
  227             ('GET', '/v1/a/c/o'),
  228             ('PUT', '/v1/a/c/o2')])
  229         req_headers = self.app.headers[1]
  230         self.assertNotIn('X-Static-Large-Object', req_headers)
  231         self.assertNotIn('Etag', req_headers)
  232         self.assertEqual(len(self.authorized), 2)
  233         self.assertEqual('GET', self.authorized[0].method)
  234         self.assertEqual('/v1/a/c/o', self.authorized[0].path)
  235         self.assertEqual('PUT', self.authorized[1].method)
  236         self.assertEqual('/v1/a/c/o2', self.authorized[1].path)
  237 
  238     def test_basic_put_with_x_copy_from_across_container(self):
  239         self.app.register('GET', '/v1/a/c1/o1', swob.HTTPOk, {}, 'passed')
  240         self.app.register('PUT', '/v1/a/c2/o2', swob.HTTPCreated, {})
  241         req = Request.blank('/v1/a/c2/o2', environ={'REQUEST_METHOD': 'PUT'},
  242                             headers={'Content-Length': '0',
  243                                      'X-Copy-From': 'c1/o1'})
  244         status, headers, body = self.call_ssc(req)
  245         self.assertEqual(status, '201 Created')
  246         self.assertTrue(('X-Copied-From', 'c1/o1') in headers)
  247         self.assertEqual(len(self.authorized), 2)
  248         self.assertEqual('GET', self.authorized[0].method)
  249         self.assertEqual('/v1/a/c1/o1', self.authorized[0].path)
  250         self.assertEqual('PUT', self.authorized[1].method)
  251         self.assertEqual('/v1/a/c2/o2', self.authorized[1].path)
  252 
  253     def test_basic_put_with_x_copy_from_across_container_and_account(self):
  254         self.app.register('GET', '/v1/a1/c1/o1', swob.HTTPOk, {}, 'passed')
  255         self.app.register('PUT', '/v1/a2/c2/o2', swob.HTTPCreated, {},
  256                           'passed')
  257         req = Request.blank('/v1/a2/c2/o2', environ={'REQUEST_METHOD': 'PUT'},
  258                             headers={'Content-Length': '0',
  259                                      'X-Copy-From': 'c1/o1',
  260                                      'X-Copy-From-Account': 'a1'})
  261         status, headers, body = self.call_ssc(req)
  262         self.assertEqual(status, '201 Created')
  263         self.assertTrue(('X-Copied-From', 'c1/o1') in headers)
  264         self.assertTrue(('X-Copied-From-Account', 'a1') in headers)
  265         self.assertEqual(len(self.authorized), 2)
  266         self.assertEqual('GET', self.authorized[0].method)
  267         self.assertEqual('/v1/a1/c1/o1', self.authorized[0].path)
  268         self.assertEqual('PUT', self.authorized[1].method)
  269         self.assertEqual('/v1/a2/c2/o2', self.authorized[1].path)
  270 
  271     def test_copy_non_zero_content_length(self):
  272         req = Request.blank('/v1/a/c2/o2', environ={'REQUEST_METHOD': 'PUT'},
  273                             headers={'Content-Length': '10',
  274                                      'X-Copy-From': 'c1/o1'})
  275         status, headers, body = self.call_ssc(req)
  276         self.assertEqual(status, '400 Bad Request')
  277 
  278     def test_copy_non_zero_content_length_with_account(self):
  279         req = Request.blank('/v1/a2/c2/o2', environ={'REQUEST_METHOD': 'PUT'},
  280                             headers={'Content-Length': '10',
  281                                      'X-Copy-From': 'c1/o1',
  282                                      'X-Copy-From-Account': 'a1'})
  283         status, headers, body = self.call_ssc(req)
  284         self.assertEqual(status, '400 Bad Request')
  285 
  286     def test_copy_with_slashes_in_x_copy_from(self):
  287         self.app.register('GET', '/v1/a/c/o/o2', swob.HTTPOk, {}, 'passed')
  288         self.app.register('PUT', '/v1/a/c/o', swob.HTTPCreated, {})
  289         req = Request.blank('/v1/a/c/o', environ={'REQUEST_METHOD': 'PUT'},
  290                             headers={'Content-Length': '0',
  291                                      'X-Copy-From': 'c/o/o2'})
  292         status, headers, body = self.call_ssc(req)
  293         self.assertEqual(status, '201 Created')
  294         self.assertTrue(('X-Copied-From', 'c/o/o2') in headers)
  295         self.assertEqual(len(self.authorized), 2)
  296         self.assertEqual('GET', self.authorized[0].method)
  297         self.assertEqual('/v1/a/c/o/o2', self.authorized[0].path)
  298         self.assertEqual('PUT', self.authorized[1].method)
  299         self.assertEqual('/v1/a/c/o', self.authorized[1].path)
  300 
  301     def test_copy_with_slashes_in_x_copy_from_and_account(self):
  302         self.app.register('GET', '/v1/a1/c1/o/o1', swob.HTTPOk, {}, 'passed')
  303         self.app.register('PUT', '/v1/a2/c2/o2', swob.HTTPCreated, {})
  304         req = Request.blank('/v1/a2/c2/o2', environ={'REQUEST_METHOD': 'PUT'},
  305                             headers={'Content-Length': '0',
  306                                      'X-Copy-From': 'c1/o/o1',
  307                                      'X-Copy-From-Account': 'a1'})
  308         status, headers, body = self.call_ssc(req)
  309         self.assertEqual(status, '201 Created')
  310         self.assertTrue(('X-Copied-From', 'c1/o/o1') in headers)
  311         self.assertTrue(('X-Copied-From-Account', 'a1') in headers)
  312         self.assertEqual(len(self.authorized), 2)
  313         self.assertEqual('GET', self.authorized[0].method)
  314         self.assertEqual('/v1/a1/c1/o/o1', self.authorized[0].path)
  315         self.assertEqual('PUT', self.authorized[1].method)
  316         self.assertEqual('/v1/a2/c2/o2', self.authorized[1].path)
  317 
  318     def test_copy_with_spaces_in_x_copy_from(self):
  319         self.app.register('GET', '/v1/a/c/o o2', swob.HTTPOk, {}, 'passed')
  320         self.app.register('PUT', '/v1/a/c/o', swob.HTTPCreated, {})
  321         # space in soure path
  322         req = Request.blank('/v1/a/c/o', environ={'REQUEST_METHOD': 'PUT'},
  323                             headers={'Content-Length': '0',
  324                                      'X-Copy-From': 'c/o%20o2'})
  325         status, headers, body = self.call_ssc(req)
  326         self.assertEqual(status, '201 Created')
  327         calls = self.app.calls_with_headers
  328         method, path, req_headers = calls[0]
  329         self.assertEqual('GET', method)
  330         self.assertEqual('/v1/a/c/o o2', path)
  331         self.assertTrue(('X-Copied-From', 'c/o%20o2') in headers)
  332         self.assertEqual(len(self.authorized), 2)
  333         self.assertEqual('GET', self.authorized[0].method)
  334         self.assertEqual('/v1/a/c/o%20o2', self.authorized[0].path)
  335         self.assertEqual('PUT', self.authorized[1].method)
  336         self.assertEqual('/v1/a/c/o', self.authorized[1].path)
  337 
  338     def test_copy_with_unicode(self):
  339         self.app.register('GET', '/v1/a/c/\xF0\x9F\x8C\xB4', swob.HTTPOk,
  340                           {}, 'passed')
  341         self.app.register('PUT', '/v1/a/c/\xE2\x98\x83', swob.HTTPCreated, {})
  342         # Just for fun, let's have a mix of properly encoded and not
  343         req = Request.blank('/v1/a/c/%F0\x9F%8C%B4',
  344                             environ={'REQUEST_METHOD': 'COPY'},
  345                             headers={'Content-Length': '0',
  346                                      'Destination': 'c/%E2\x98%83'})
  347         status, headers, body = self.call_ssc(req)
  348         self.assertEqual(status, '201 Created')
  349         calls = self.app.calls_with_headers
  350         method, path, req_headers = calls[0]
  351         self.assertEqual('GET', method)
  352         self.assertEqual('/v1/a/c/\xF0\x9F\x8C\xB4', path)
  353         self.assertIn(('X-Copied-From', 'c/%F0%9F%8C%B4'), headers)
  354 
  355         self.assertEqual(len(self.authorized), 2)
  356         self.assertEqual('GET', self.authorized[0].method)
  357         self.assertEqual('/v1/a/c/%F0%9F%8C%B4', self.authorized[0].path)
  358         self.assertEqual('PUT', self.authorized[1].method)
  359         self.assertEqual('/v1/a/c/%E2%98%83', self.authorized[1].path)
  360 
  361     def test_copy_with_spaces_in_x_copy_from_and_account(self):
  362         self.app.register('GET', '/v1/a/c/o o2', swob.HTTPOk, {}, b'passed')
  363         self.app.register('PUT', '/v1/a1/c1/o', swob.HTTPCreated, {})
  364         # space in soure path
  365         req = Request.blank('/v1/a1/c1/o', environ={'REQUEST_METHOD': 'PUT'},
  366                             headers={'Content-Length': '0',
  367                                      'X-Copy-From': 'c/o%20o2',
  368                                      'X-Copy-From-Account': 'a'})
  369         status, headers, body = self.call_ssc(req)
  370         self.assertEqual(status, '201 Created')
  371         calls = self.app.calls_with_headers
  372         method, path, req_headers = calls[0]
  373         self.assertEqual('GET', method)
  374         self.assertEqual('/v1/a/c/o o2', path)
  375         self.assertTrue(('X-Copied-From', 'c/o%20o2') in headers)
  376         self.assertTrue(('X-Copied-From-Account', 'a') in headers)
  377         self.assertEqual(len(self.authorized), 2)
  378         self.assertEqual('GET', self.authorized[0].method)
  379         self.assertEqual('/v1/a/c/o%20o2', self.authorized[0].path)
  380         self.assertEqual('PUT', self.authorized[1].method)
  381         self.assertEqual('/v1/a1/c1/o', self.authorized[1].path)
  382 
  383     def test_copy_with_leading_slash_in_x_copy_from(self):
  384         self.app.register('GET', '/v1/a/c/o', swob.HTTPOk, {}, 'passed')
  385         self.app.register('PUT', '/v1/a/c/o', swob.HTTPCreated, {})
  386         # repeat tests with leading /
  387         req = Request.blank('/v1/a/c/o', environ={'REQUEST_METHOD': 'PUT'},
  388                             headers={'Content-Length': '0',
  389                                      'X-Copy-From': '/c/o'})
  390         status, headers, body = self.call_ssc(req)
  391         self.assertEqual(status, '201 Created')
  392         calls = self.app.calls_with_headers
  393         method, path, req_headers = calls[0]
  394         self.assertEqual('GET', method)
  395         self.assertEqual('/v1/a/c/o', path)
  396         self.assertTrue(('X-Copied-From', 'c/o') in headers)
  397         self.assertEqual(len(self.authorized), 2)
  398         self.assertEqual('GET', self.authorized[0].method)
  399         self.assertEqual('/v1/a/c/o', self.authorized[0].path)
  400         self.assertEqual('PUT', self.authorized[1].method)
  401         self.assertEqual('/v1/a/c/o', self.authorized[1].path)
  402 
  403     def test_copy_with_leading_slash_in_x_copy_from_and_account(self):
  404         # repeat tests with leading /
  405         self.app.register('GET', '/v1/a/c/o', swob.HTTPOk, {}, 'passed')
  406         self.app.register('PUT', '/v1/a1/c1/o', swob.HTTPCreated, {})
  407         req = Request.blank('/v1/a1/c1/o', environ={'REQUEST_METHOD': 'PUT'},
  408                             headers={'Content-Length': '0',
  409                                      'X-Copy-From': '/c/o',
  410                                      'X-Copy-From-Account': 'a'})
  411         status, headers, body = self.call_ssc(req)
  412         self.assertEqual(status, '201 Created')
  413         calls = self.app.calls_with_headers
  414         method, path, req_headers = calls[0]
  415         self.assertEqual('GET', method)
  416         self.assertEqual('/v1/a/c/o', path)
  417         self.assertTrue(('X-Copied-From', 'c/o') in headers)
  418         self.assertTrue(('X-Copied-From-Account', 'a') in headers)
  419         self.assertEqual(len(self.authorized), 2)
  420         self.assertEqual('GET', self.authorized[0].method)
  421         self.assertEqual('/v1/a/c/o', self.authorized[0].path)
  422         self.assertEqual('PUT', self.authorized[1].method)
  423         self.assertEqual('/v1/a1/c1/o', self.authorized[1].path)
  424 
  425     def test_copy_with_leading_slash_and_slashes_in_x_copy_from(self):
  426         self.app.register('GET', '/v1/a/c/o/o2', swob.HTTPOk, {}, 'passed')
  427         self.app.register('PUT', '/v1/a/c/o', swob.HTTPCreated, {})
  428         req = Request.blank('/v1/a/c/o', environ={'REQUEST_METHOD': 'PUT'},
  429                             headers={'Content-Length': '0',
  430                                      'X-Copy-From': '/c/o/o2'})
  431         status, headers, body = self.call_ssc(req)
  432         self.assertEqual(status, '201 Created')
  433         calls = self.app.calls_with_headers
  434         method, path, req_headers = calls[0]
  435         self.assertEqual('GET', method)
  436         self.assertEqual('/v1/a/c/o/o2', path)
  437         self.assertTrue(('X-Copied-From', 'c/o/o2') in headers)
  438         self.assertEqual(len(self.authorized), 2)
  439         self.assertEqual('GET', self.authorized[0].method)
  440         self.assertEqual('/v1/a/c/o/o2', self.authorized[0].path)
  441         self.assertEqual('PUT', self.authorized[1].method)
  442         self.assertEqual('/v1/a/c/o', self.authorized[1].path)
  443 
  444     def test_copy_with_leading_slash_and_slashes_in_x_copy_from_acct(self):
  445         self.app.register('GET', '/v1/a/c/o/o2', swob.HTTPOk, {}, 'passed')
  446         self.app.register('PUT', '/v1/a1/c1/o', swob.HTTPCreated, {})
  447         req = Request.blank('/v1/a1/c1/o', environ={'REQUEST_METHOD': 'PUT'},
  448                             headers={'Content-Length': '0',
  449                                      'X-Copy-From': '/c/o/o2',
  450                                      'X-Copy-From-Account': 'a'})
  451         status, headers, body = self.call_ssc(req)
  452         self.assertEqual(status, '201 Created')
  453         calls = self.app.calls_with_headers
  454         method, path, req_headers = calls[0]
  455         self.assertEqual('GET', method)
  456         self.assertEqual('/v1/a/c/o/o2', path)
  457         self.assertTrue(('X-Copied-From', 'c/o/o2') in headers)
  458         self.assertTrue(('X-Copied-From-Account', 'a') in headers)
  459         self.assertEqual(len(self.authorized), 2)
  460         self.assertEqual('GET', self.authorized[0].method)
  461         self.assertEqual('/v1/a/c/o/o2', self.authorized[0].path)
  462         self.assertEqual('PUT', self.authorized[1].method)
  463         self.assertEqual('/v1/a1/c1/o', self.authorized[1].path)
  464 
  465     def test_copy_with_no_object_in_x_copy_from(self):
  466         req = Request.blank('/v1/a/c/o', environ={'REQUEST_METHOD': 'PUT'},
  467                             headers={'Content-Length': '0',
  468                                      'X-Copy-From': '/c'})
  469         status, headers, body = self.call_ssc(req)
  470         self.assertEqual(status, '412 Precondition Failed')
  471 
  472     def test_copy_with_no_object_in_x_copy_from_and_account(self):
  473         req = Request.blank('/v1/a/c/o', environ={'REQUEST_METHOD': 'PUT'},
  474                             headers={'Content-Length': '0',
  475                                      'X-Copy-From': '/c',
  476                                      'X-Copy-From-Account': 'a'})
  477         status, headers, body = self.call_ssc(req)
  478         self.assertEqual(status, '412 Precondition Failed')
  479 
  480     def test_copy_with_bad_x_copy_from_account(self):
  481         req = Request.blank('/v1/a/c/o',
  482                             environ={'REQUEST_METHOD': 'PUT'},
  483                             headers={'Content-Length': '0',
  484                                      'X-Copy-From': '/c/o',
  485                                      'X-Copy-From-Account': '/i/am/bad'})
  486         status, headers, body = self.call_ssc(req)
  487         self.assertEqual(status, '412 Precondition Failed')
  488 
  489     def test_copy_server_error_reading_source(self):
  490         self.app.register('GET', '/v1/a/c/o', swob.HTTPServiceUnavailable, {})
  491         req = Request.blank('/v1/a/c/o', environ={'REQUEST_METHOD': 'PUT'},
  492                             headers={'Content-Length': '0',
  493                                      'X-Copy-From': '/c/o'})
  494         status, headers, body = self.call_ssc(req)
  495         self.assertEqual(status, '503 Service Unavailable')
  496 
  497     def test_copy_server_error_reading_source_and_account(self):
  498         self.app.register('GET', '/v1/a/c/o', swob.HTTPServiceUnavailable, {})
  499         req = Request.blank('/v1/a1/c1/o', environ={'REQUEST_METHOD': 'PUT'},
  500                             headers={'Content-Length': '0',
  501                                      'X-Copy-From': '/c/o',
  502                                      'X-Copy-From-Account': 'a'})
  503         status, headers, body = self.call_ssc(req)
  504         self.assertEqual(status, '503 Service Unavailable')
  505         self.assertEqual(len(self.authorized), 1)
  506         self.assertEqual('GET', self.authorized[0].method)
  507         self.assertEqual('/v1/a/c/o', self.authorized[0].path)
  508 
  509     def test_copy_not_found_reading_source(self):
  510         self.app.register('GET', '/v1/a/c/o', swob.HTTPNotFound, {})
  511         req = Request.blank('/v1/a/c/o', environ={'REQUEST_METHOD': 'PUT'},
  512                             headers={'Content-Length': '0',
  513                                      'X-Copy-From': '/c/o'})
  514         status, headers, body = self.call_ssc(req)
  515         self.assertEqual(status, '404 Not Found')
  516         self.assertEqual(len(self.authorized), 1)
  517         self.assertEqual('GET', self.authorized[0].method)
  518         self.assertEqual('/v1/a/c/o', self.authorized[0].path)
  519 
  520     def test_copy_not_found_reading_source_and_account(self):
  521         self.app.register('GET', '/v1/a/c/o', swob.HTTPNotFound, {})
  522         req = Request.blank('/v1/a1/c1/o', environ={'REQUEST_METHOD': 'PUT'},
  523                             headers={'Content-Length': '0',
  524                                      'X-Copy-From': '/c/o',
  525                                      'X-Copy-From-Account': 'a'})
  526         status, headers, body = self.call_ssc(req)
  527         self.assertEqual(status, '404 Not Found')
  528         self.assertEqual(len(self.authorized), 1)
  529         self.assertEqual('GET', self.authorized[0].method)
  530         self.assertEqual('/v1/a/c/o', self.authorized[0].path)
  531 
  532     def test_copy_with_object_metadata(self):
  533         self.app.register('GET', '/v1/a/c/o', swob.HTTPOk, {}, 'passed')
  534         self.app.register('PUT', '/v1/a/c/o', swob.HTTPCreated, {})
  535         req = Request.blank('/v1/a/c/o', environ={'REQUEST_METHOD': 'PUT'},
  536                             headers={'Content-Length': '0',
  537                                      'X-Copy-From': '/c/o',
  538                                      'X-Object-Meta-Ours': 'okay'})
  539         status, headers, body = self.call_ssc(req)
  540         self.assertEqual(status, '201 Created')
  541         calls = self.app.calls_with_headers
  542         method, path, req_headers = calls[1]
  543         self.assertEqual('PUT', method)
  544         self.assertEqual('/v1/a/c/o', path)
  545         self.assertEqual(req_headers['X-Object-Meta-Ours'], 'okay')
  546         self.assertTrue(('X-Object-Meta-Ours', 'okay') in headers)
  547         self.assertEqual(len(self.authorized), 2)
  548         self.assertEqual('GET', self.authorized[0].method)
  549         self.assertEqual('/v1/a/c/o', self.authorized[0].path)
  550         self.assertEqual('PUT', self.authorized[1].method)
  551         self.assertEqual('/v1/a/c/o', self.authorized[1].path)
  552 
  553     def test_copy_with_object_metadata_and_account(self):
  554         self.app.register('GET', '/v1/a1/c/o', swob.HTTPOk, {}, 'passed')
  555         self.app.register('PUT', '/v1/a/c/o', swob.HTTPCreated, {})
  556         req = Request.blank('/v1/a/c/o', environ={'REQUEST_METHOD': 'PUT'},
  557                             headers={'Content-Length': '0',
  558                                      'X-Copy-From': '/c/o',
  559                                      'X-Object-Meta-Ours': 'okay',
  560                                      'X-Copy-From-Account': 'a1'})
  561         status, headers, body = self.call_ssc(req)
  562         self.assertEqual(status, '201 Created')
  563         calls = self.app.calls_with_headers
  564         method, path, req_headers = calls[1]
  565         self.assertEqual('PUT', method)
  566         self.assertEqual('/v1/a/c/o', path)
  567         self.assertEqual(req_headers['X-Object-Meta-Ours'], 'okay')
  568         self.assertTrue(('X-Object-Meta-Ours', 'okay') in headers)
  569         self.assertEqual(len(self.authorized), 2)
  570         self.assertEqual('GET', self.authorized[0].method)
  571         self.assertEqual('/v1/a1/c/o', self.authorized[0].path)
  572         self.assertEqual('PUT', self.authorized[1].method)
  573         self.assertEqual('/v1/a/c/o', self.authorized[1].path)
  574 
  575     def test_copy_source_larger_than_max_file_size(self):
  576         self.app.register('GET', '/v1/a/c/o', swob.HTTPOk, {}, "largebody")
  577         req = Request.blank('/v1/a/c/o', environ={'REQUEST_METHOD': 'PUT'},
  578                             headers={'Content-Length': '0',
  579                                      'X-Copy-From': '/c/o'})
  580         with mock.patch('swift.common.middleware.copy.'
  581                         'MAX_FILE_SIZE', 1):
  582             status, headers, body = self.call_ssc(req)
  583         self.assertEqual(status, '413 Request Entity Too Large')
  584         self.assertEqual(len(self.authorized), 1)
  585         self.assertEqual('GET', self.authorized[0].method)
  586         self.assertEqual('/v1/a/c/o', self.authorized[0].path)
  587 
  588     def test_basic_COPY(self):
  589         self.app.register('GET', '/v1/a/c/o', swob.HTTPOk, {
  590             'etag': 'is sent'}, 'passed')
  591         self.app.register('PUT', '/v1/a/c/o-copy', swob.HTTPCreated, {})
  592         req = Request.blank(
  593             '/v1/a/c/o', method='COPY',
  594             headers={'Content-Length': 0,
  595                      'Destination': 'c/o-copy'})
  596         status, headers, body = self.call_ssc(req)
  597         self.assertEqual(status, '201 Created')
  598         self.assertTrue(('X-Copied-From', 'c/o') in headers)
  599         self.assertEqual(len(self.authorized), 2)
  600         self.assertEqual('GET', self.authorized[0].method)
  601         self.assertEqual('/v1/a/c/o', self.authorized[0].path)
  602         self.assertEqual('PUT', self.authorized[1].method)
  603         self.assertEqual('/v1/a/c/o-copy', self.authorized[1].path)
  604         self.assertEqual(self.app.calls, [
  605             ('GET', '/v1/a/c/o'),
  606             ('PUT', '/v1/a/c/o-copy')])
  607         self.assertIn('etag', self.app.headers[1])
  608         self.assertEqual(self.app.headers[1]['etag'], 'is sent')
  609         # For basic test cases, assert orig_req_method behavior
  610         self.assertEqual(req.environ['swift.orig_req_method'], 'COPY')
  611 
  612     def test_basic_DLO(self):
  613         self.app.register('GET', '/v1/a/c/o', swob.HTTPOk, {
  614             'x-object-manifest': 'some/path',
  615             'etag': 'is not sent'}, 'passed')
  616         self.app.register('PUT', '/v1/a/c/o-copy', swob.HTTPCreated, {})
  617         req = Request.blank(
  618             '/v1/a/c/o', method='COPY',
  619             headers={'Content-Length': 0,
  620                      'Destination': 'c/o-copy'})
  621         status, headers, body = self.call_ssc(req)
  622         self.assertEqual(status, '201 Created')
  623         self.assertTrue(('X-Copied-From', 'c/o') in headers)
  624         self.assertEqual(self.app.calls, [
  625             ('GET', '/v1/a/c/o'),
  626             ('PUT', '/v1/a/c/o-copy')])
  627         self.assertNotIn('x-object-manifest', self.app.headers[1])
  628         self.assertNotIn('etag', self.app.headers[1])
  629 
  630     def test_basic_DLO_manifest(self):
  631         self.app.register('GET', '/v1/a/c/o', swob.HTTPOk, {
  632             'x-object-manifest': 'some/path',
  633             'etag': 'is sent'}, 'passed')
  634         self.app.register('PUT', '/v1/a/c/o-copy', swob.HTTPCreated, {})
  635         req = Request.blank(
  636             '/v1/a/c/o?multipart-manifest=get', method='COPY',
  637             headers={'Content-Length': 0,
  638                      'Destination': 'c/o-copy'})
  639         status, headers, body = self.call_ssc(req)
  640         self.assertEqual(status, '201 Created')
  641         self.assertTrue(('X-Copied-From', 'c/o') in headers)
  642         self.assertEqual(2, len(self.app.calls))
  643         self.assertEqual('GET', self.app.calls[0][0])
  644         get_path, qs = self.app.calls[0][1].split('?')
  645         params = urllib.parse.parse_qs(qs)
  646         self.assertDictEqual(
  647             {'format': ['raw'], 'multipart-manifest': ['get']}, params)
  648         self.assertEqual(get_path, '/v1/a/c/o')
  649         self.assertEqual(self.app.calls[1], ('PUT', '/v1/a/c/o-copy'))
  650         self.assertIn('x-object-manifest', self.app.headers[1])
  651         self.assertEqual(self.app.headers[1]['x-object-manifest'], 'some/path')
  652         self.assertIn('etag', self.app.headers[1])
  653         self.assertEqual(self.app.headers[1]['etag'], 'is sent')
  654 
  655     def test_COPY_source_metadata(self):
  656         source_headers = {
  657             'x-object-sysmeta-test1': 'copy me',
  658             'x-object-meta-test2': 'copy me too',
  659             'x-object-transient-sysmeta-test3': 'ditto',
  660             'x-object-sysmeta-container-update-override-etag': 'etag val',
  661             'x-object-sysmeta-container-update-override-size': 'size val',
  662             'x-object-sysmeta-container-update-override-foo': 'bar',
  663             'x-delete-at': 'delete-at-time'}
  664 
  665         get_resp_headers = source_headers.copy()
  666         get_resp_headers['etag'] = 'source etag'
  667         self.app.register(
  668             'GET', '/v1/a/c/o', swob.HTTPOk,
  669             headers=get_resp_headers, body=b'passed')
  670 
  671         def verify_headers(expected_headers, unexpected_headers,
  672                            actual_headers):
  673             for k, v in actual_headers:
  674                 if k.lower() in expected_headers:
  675                     expected_val = expected_headers.pop(k.lower())
  676                     self.assertEqual(expected_val, v)
  677                 self.assertNotIn(k.lower(), unexpected_headers)
  678             self.assertFalse(expected_headers)
  679 
  680         # use a COPY request
  681         self.app.register('PUT', '/v1/a/c/o-copy0', swob.HTTPCreated, {})
  682         req = Request.blank('/v1/a/c/o', method='COPY',
  683                             headers={'Content-Length': 0,
  684                                      'Destination': 'c/o-copy0'})
  685         status, resp_headers, body = self.call_ssc(req)
  686         self.assertEqual('201 Created', status)
  687         verify_headers(source_headers.copy(), [], resp_headers)
  688         method, path, put_headers = self.app.calls_with_headers[-1]
  689         self.assertEqual('PUT', method)
  690         self.assertEqual('/v1/a/c/o-copy0', path)
  691         verify_headers(source_headers.copy(), [], put_headers.items())
  692         self.assertIn('etag', put_headers)
  693         self.assertEqual(put_headers['etag'], 'source etag')
  694 
  695         req = Request.blank('/v1/a/c/o-copy0', method='GET')
  696         status, resp_headers, body = self.call_ssc(req)
  697         self.assertEqual('200 OK', status)
  698         verify_headers(source_headers.copy(), [], resp_headers)
  699 
  700         # use a COPY request with a Range header
  701         self.app.register('PUT', '/v1/a/c/o-copy1', swob.HTTPCreated, {})
  702         req = Request.blank('/v1/a/c/o', method='COPY',
  703                             headers={'Content-Length': 0,
  704                                      'Destination': 'c/o-copy1',
  705                                      'Range': 'bytes=1-2'})
  706         status, resp_headers, body = self.call_ssc(req)
  707         expected_headers = source_headers.copy()
  708         unexpected_headers = (
  709             'x-object-sysmeta-container-update-override-etag',
  710             'x-object-sysmeta-container-update-override-size',
  711             'x-object-sysmeta-container-update-override-foo')
  712         for h in unexpected_headers:
  713             expected_headers.pop(h)
  714         self.assertEqual('201 Created', status)
  715         verify_headers(expected_headers, unexpected_headers, resp_headers)
  716         method, path, put_headers = self.app.calls_with_headers[-1]
  717         self.assertEqual('PUT', method)
  718         self.assertEqual('/v1/a/c/o-copy1', path)
  719         verify_headers(
  720             expected_headers, unexpected_headers, put_headers.items())
  721         # etag should not be copied with a Range request
  722         self.assertNotIn('etag', put_headers)
  723 
  724         req = Request.blank('/v1/a/c/o-copy1', method='GET')
  725         status, resp_headers, body = self.call_ssc(req)
  726         self.assertEqual('200 OK', status)
  727         verify_headers(expected_headers, unexpected_headers, resp_headers)
  728 
  729         # use a PUT with x-copy-from
  730         self.app.register('PUT', '/v1/a/c/o-copy2', swob.HTTPCreated, {})
  731         req = Request.blank('/v1/a/c/o-copy2', method='PUT',
  732                             headers={'Content-Length': 0,
  733                                      'X-Copy-From': 'c/o'})
  734         status, resp_headers, body = self.call_ssc(req)
  735         self.assertEqual('201 Created', status)
  736         verify_headers(source_headers.copy(), [], resp_headers)
  737         method, path, put_headers = self.app.calls_with_headers[-1]
  738         self.assertEqual('PUT', method)
  739         self.assertEqual('/v1/a/c/o-copy2', path)
  740         verify_headers(source_headers.copy(), [], put_headers.items())
  741         self.assertIn('etag', put_headers)
  742         self.assertEqual(put_headers['etag'], 'source etag')
  743 
  744         req = Request.blank('/v1/a/c/o-copy2', method='GET')
  745         status, resp_headers, body = self.call_ssc(req)
  746         self.assertEqual('200 OK', status)
  747         verify_headers(source_headers.copy(), [], resp_headers)
  748 
  749         # copy to same path as source
  750         self.app.register('PUT', '/v1/a/c/o', swob.HTTPCreated, {})
  751         req = Request.blank('/v1/a/c/o', method='PUT',
  752                             headers={'Content-Length': 0,
  753                                      'X-Copy-From': 'c/o'})
  754         status, resp_headers, body = self.call_ssc(req)
  755         self.assertEqual('201 Created', status)
  756         verify_headers(source_headers.copy(), [], resp_headers)
  757         method, path, put_headers = self.app.calls_with_headers[-1]
  758         self.assertEqual('PUT', method)
  759         self.assertEqual('/v1/a/c/o', path)
  760         verify_headers(source_headers.copy(), [], put_headers.items())
  761         self.assertIn('etag', put_headers)
  762         self.assertEqual(put_headers['etag'], 'source etag')
  763 
  764     def test_COPY_no_destination_header(self):
  765         req = Request.blank(
  766             '/v1/a/c/o', method='COPY', headers={'Content-Length': 0})
  767         status, headers, body = self.call_ssc(req)
  768         self.assertEqual(status, '412 Precondition Failed')
  769         self.assertEqual(len(self.authorized), 0)
  770 
  771     def test_basic_COPY_account(self):
  772         self.app.register('GET', '/v1/a/c/o', swob.HTTPOk, {}, 'passed')
  773         self.app.register('PUT', '/v1/a1/c1/o2', swob.HTTPCreated, {})
  774         req = Request.blank('/v1/a/c/o',
  775                             environ={'REQUEST_METHOD': 'COPY'},
  776                             headers={'Destination': 'c1/o2',
  777                                      'Destination-Account': 'a1'})
  778         status, headers, body = self.call_ssc(req)
  779         self.assertEqual(status, '201 Created')
  780         calls = self.app.calls_with_headers
  781         method, path, req_headers = calls[0]
  782         self.assertEqual('GET', method)
  783         self.assertEqual('/v1/a/c/o', path)
  784         method, path, req_headers = calls[1]
  785         self.assertEqual('PUT', method)
  786         self.assertEqual('/v1/a1/c1/o2', path)
  787         self.assertTrue(('X-Copied-From', 'c/o') in headers)
  788         self.assertTrue(('X-Copied-From-Account', 'a') in headers)
  789         self.assertEqual(len(self.authorized), 2)
  790         self.assertEqual('GET', self.authorized[0].method)
  791         self.assertEqual('/v1/a/c/o', self.authorized[0].path)
  792         self.assertEqual('PUT', self.authorized[1].method)
  793         self.assertEqual('/v1/a1/c1/o2', self.authorized[1].path)
  794 
  795     def test_COPY_across_containers(self):
  796         self.app.register('GET', '/v1/a/c/o', swob.HTTPOk, {}, 'passed')
  797         self.app.register('PUT', '/v1/a/c2/o', swob.HTTPCreated, {})
  798         req = Request.blank('/v1/a/c/o',
  799                             environ={'REQUEST_METHOD': 'COPY'},
  800                             headers={'Destination': 'c2/o'})
  801         status, headers, body = self.call_ssc(req)
  802         self.assertEqual(status, '201 Created')
  803         self.assertTrue(('X-Copied-From', 'c/o') in headers)
  804         self.assertEqual(len(self.authorized), 2)
  805         self.assertEqual('GET', self.authorized[0].method)
  806         self.assertEqual('/v1/a/c/o', self.authorized[0].path)
  807         self.assertEqual('PUT', self.authorized[1].method)
  808         self.assertEqual('/v1/a/c2/o', self.authorized[1].path)
  809 
  810     def test_COPY_source_with_slashes_in_name(self):
  811         self.app.register('GET', '/v1/a/c/o/o2', swob.HTTPOk, {}, 'passed')
  812         self.app.register('PUT', '/v1/a/c/o', swob.HTTPCreated, {})
  813         req = Request.blank('/v1/a/c/o/o2',
  814                             environ={'REQUEST_METHOD': 'COPY'},
  815                             headers={'Destination': 'c/o'})
  816         status, headers, body = self.call_ssc(req)
  817         self.assertEqual(status, '201 Created')
  818         calls = self.app.calls_with_headers
  819         method, path, req_headers = calls[1]
  820         self.assertEqual('PUT', method)
  821         self.assertEqual('/v1/a/c/o', path)
  822         self.assertTrue(('X-Copied-From', 'c/o/o2') in headers)
  823         self.assertEqual(len(self.authorized), 2)
  824         self.assertEqual('GET', self.authorized[0].method)
  825         self.assertEqual('/v1/a/c/o/o2', self.authorized[0].path)
  826         self.assertEqual('PUT', self.authorized[1].method)
  827         self.assertEqual('/v1/a/c/o', self.authorized[1].path)
  828 
  829     def test_COPY_account_source_with_slashes_in_name(self):
  830         self.app.register('GET', '/v1/a/c/o/o2', swob.HTTPOk, {}, 'passed')
  831         self.app.register('PUT', '/v1/a1/c1/o', swob.HTTPCreated, {})
  832         req = Request.blank('/v1/a/c/o/o2',
  833                             environ={'REQUEST_METHOD': 'COPY'},
  834                             headers={'Destination': 'c1/o',
  835                                      'Destination-Account': 'a1'})
  836         status, headers, body = self.call_ssc(req)
  837         self.assertEqual(status, '201 Created')
  838         calls = self.app.calls_with_headers
  839         method, path, req_headers = calls[1]
  840         self.assertEqual('PUT', method)
  841         self.assertEqual('/v1/a1/c1/o', path)
  842         self.assertTrue(('X-Copied-From', 'c/o/o2') in headers)
  843         self.assertTrue(('X-Copied-From-Account', 'a') in headers)
  844         self.assertEqual(len(self.authorized), 2)
  845         self.assertEqual('GET', self.authorized[0].method)
  846         self.assertEqual('/v1/a/c/o/o2', self.authorized[0].path)
  847         self.assertEqual('PUT', self.authorized[1].method)
  848         self.assertEqual('/v1/a1/c1/o', self.authorized[1].path)
  849 
  850     def test_COPY_destination_leading_slash(self):
  851         self.app.register('GET', '/v1/a/c/o', swob.HTTPOk, {}, 'passed')
  852         self.app.register('PUT', '/v1/a/c/o', swob.HTTPCreated, {})
  853         req = Request.blank('/v1/a/c/o',
  854                             environ={'REQUEST_METHOD': 'COPY'},
  855                             headers={'Destination': '/c/o'})
  856         status, headers, body = self.call_ssc(req)
  857         self.assertEqual(status, '201 Created')
  858         self.assertTrue(('X-Copied-From', 'c/o') in headers)
  859         self.assertEqual(len(self.authorized), 2)
  860         self.assertEqual('GET', self.authorized[0].method)
  861         self.assertEqual('/v1/a/c/o', self.authorized[0].path)
  862         self.assertEqual('PUT', self.authorized[1].method)
  863         self.assertEqual('/v1/a/c/o', self.authorized[1].path)
  864 
  865     def test_COPY_account_destination_leading_slash(self):
  866         self.app.register('GET', '/v1/a/c/o', swob.HTTPOk, {}, 'passed')
  867         self.app.register('PUT', '/v1/a1/c1/o', swob.HTTPCreated, {})
  868         req = Request.blank('/v1/a/c/o',
  869                             environ={'REQUEST_METHOD': 'COPY'},
  870                             headers={'Destination': '/c1/o',
  871                                      'Destination-Account': 'a1'})
  872         status, headers, body = self.call_ssc(req)
  873         self.assertEqual(status, '201 Created')
  874         calls = self.app.calls_with_headers
  875         method, path, req_headers = calls[1]
  876         self.assertEqual('PUT', method)
  877         self.assertEqual('/v1/a1/c1/o', path)
  878         self.assertTrue(('X-Copied-From', 'c/o') in headers)
  879         self.assertTrue(('X-Copied-From-Account', 'a') in headers)
  880         self.assertEqual(len(self.authorized), 2)
  881         self.assertEqual('GET', self.authorized[0].method)
  882         self.assertEqual('/v1/a/c/o', self.authorized[0].path)
  883         self.assertEqual('PUT', self.authorized[1].method)
  884         self.assertEqual('/v1/a1/c1/o', self.authorized[1].path)
  885 
  886     def test_COPY_source_with_slashes_destination_leading_slash(self):
  887         self.app.register('GET', '/v1/a/c/o/o2', swob.HTTPOk, {}, 'passed')
  888         self.app.register('PUT', '/v1/a/c/o', swob.HTTPCreated, {})
  889         req = Request.blank('/v1/a/c/o/o2',
  890                             environ={'REQUEST_METHOD': 'COPY'},
  891                             headers={'Destination': '/c/o'})
  892         status, headers, body = self.call_ssc(req)
  893         self.assertEqual(status, '201 Created')
  894         calls = self.app.calls_with_headers
  895         method, path, req_headers = calls[1]
  896         self.assertEqual('PUT', method)
  897         self.assertEqual('/v1/a/c/o', path)
  898         self.assertTrue(('X-Copied-From', 'c/o/o2') in headers)
  899         self.assertEqual(len(self.authorized), 2)
  900         self.assertEqual('GET', self.authorized[0].method)
  901         self.assertEqual('/v1/a/c/o/o2', self.authorized[0].path)
  902         self.assertEqual('PUT', self.authorized[1].method)
  903         self.assertEqual('/v1/a/c/o', self.authorized[1].path)
  904 
  905     def test_COPY_account_source_with_slashes_destination_leading_slash(self):
  906         self.app.register('GET', '/v1/a/c/o/o2', swob.HTTPOk, {}, 'passed')
  907         self.app.register('PUT', '/v1/a1/c1/o', swob.HTTPCreated, {})
  908         req = Request.blank('/v1/a/c/o/o2',
  909                             environ={'REQUEST_METHOD': 'COPY'},
  910                             headers={'Destination': '/c1/o',
  911                                      'Destination-Account': 'a1'})
  912         status, headers, body = self.call_ssc(req)
  913         self.assertEqual(status, '201 Created')
  914         calls = self.app.calls_with_headers
  915         method, path, req_headers = calls[1]
  916         self.assertEqual('PUT', method)
  917         self.assertEqual('/v1/a1/c1/o', path)
  918         self.assertTrue(('X-Copied-From', 'c/o/o2') in headers)
  919         self.assertTrue(('X-Copied-From-Account', 'a') in headers)
  920         self.assertEqual(len(self.authorized), 2)
  921         self.assertEqual('GET', self.authorized[0].method)
  922         self.assertEqual('/v1/a/c/o/o2', self.authorized[0].path)
  923         self.assertEqual('PUT', self.authorized[1].method)
  924         self.assertEqual('/v1/a1/c1/o', self.authorized[1].path)
  925 
  926     def test_COPY_no_object_in_destination(self):
  927         req = Request.blank('/v1/a/c/o',
  928                             environ={'REQUEST_METHOD': 'COPY'},
  929                             headers={'Destination': 'c_o'})
  930         status, headers, body = self.call_ssc(req)
  931 
  932         self.assertEqual(status, '412 Precondition Failed')
  933 
  934     def test_COPY_account_no_object_in_destination(self):
  935         req = Request.blank('/v1/a/c/o',
  936                             environ={'REQUEST_METHOD': 'COPY'},
  937                             headers={'Destination': 'c_o',
  938                                      'Destination-Account': 'a1'})
  939         status, headers, body = self.call_ssc(req)
  940 
  941         self.assertEqual(status, '412 Precondition Failed')
  942 
  943     def test_COPY_account_bad_destination_account(self):
  944         req = Request.blank('/v1/a/c/o',
  945                             environ={'REQUEST_METHOD': 'COPY'},
  946                             headers={'Destination': '/c/o',
  947                                      'Destination-Account': '/i/am/bad'})
  948         status, headers, body = self.call_ssc(req)
  949 
  950         self.assertEqual(status, '412 Precondition Failed')
  951 
  952     def test_COPY_server_error_reading_source(self):
  953         self.app.register('GET', '/v1/a/c/o', swob.HTTPServiceUnavailable, {})
  954         req = Request.blank('/v1/a/c/o',
  955                             environ={'REQUEST_METHOD': 'COPY'},
  956                             headers={'Destination': '/c/o'})
  957         status, headers, body = self.call_ssc(req)
  958         self.assertEqual(status, '503 Service Unavailable')
  959         self.assertEqual(len(self.authorized), 1)
  960         self.assertEqual('GET', self.authorized[0].method)
  961         self.assertEqual('/v1/a/c/o', self.authorized[0].path)
  962 
  963     def test_COPY_account_server_error_reading_source(self):
  964         self.app.register('GET', '/v1/a/c/o', swob.HTTPServiceUnavailable, {})
  965         req = Request.blank('/v1/a/c/o',
  966                             environ={'REQUEST_METHOD': 'COPY'},
  967                             headers={'Destination': '/c1/o',
  968                                      'Destination-Account': 'a1'})
  969         status, headers, body = self.call_ssc(req)
  970         self.assertEqual(status, '503 Service Unavailable')
  971         self.assertEqual(len(self.authorized), 1)
  972         self.assertEqual('GET', self.authorized[0].method)
  973         self.assertEqual('/v1/a/c/o', self.authorized[0].path)
  974 
  975     def test_COPY_not_found_reading_source(self):
  976         self.app.register('GET', '/v1/a/c/o', swob.HTTPNotFound, {})
  977         req = Request.blank('/v1/a/c/o',
  978                             environ={'REQUEST_METHOD': 'COPY'},
  979                             headers={'Destination': '/c/o'})
  980         status, headers, body = self.call_ssc(req)
  981         self.assertEqual(status, '404 Not Found')
  982         self.assertEqual(len(self.authorized), 1)
  983         self.assertEqual('GET', self.authorized[0].method)
  984         self.assertEqual('/v1/a/c/o', self.authorized[0].path)
  985 
  986     def test_COPY_account_not_found_reading_source(self):
  987         self.app.register('GET', '/v1/a/c/o', swob.HTTPNotFound, {})
  988         req = Request.blank('/v1/a/c/o',
  989                             environ={'REQUEST_METHOD': 'COPY'},
  990                             headers={'Destination': '/c1/o',
  991                                      'Destination-Account': 'a1'})
  992         status, headers, body = self.call_ssc(req)
  993         self.assertEqual(status, '404 Not Found')
  994         self.assertEqual(len(self.authorized), 1)
  995         self.assertEqual('GET', self.authorized[0].method)
  996         self.assertEqual('/v1/a/c/o', self.authorized[0].path)
  997 
  998     def test_COPY_with_metadata(self):
  999         self.app.register('GET', '/v1/a/c/o', swob.HTTPOk, {}, "passed")
 1000         self.app.register('PUT', '/v1/a/c/o', swob.HTTPCreated, {})
 1001         req = Request.blank('/v1/a/c/o',
 1002                             environ={'REQUEST_METHOD': 'COPY'},
 1003                             headers={'Destination': '/c/o',
 1004                                      'X-Object-Meta-Ours': 'okay'})
 1005         status, headers, body = self.call_ssc(req)
 1006         self.assertEqual(status, '201 Created')
 1007         calls = self.app.calls_with_headers
 1008         method, path, req_headers = calls[1]
 1009         self.assertEqual('PUT', method)
 1010         self.assertEqual('/v1/a/c/o', path)
 1011         self.assertEqual(req_headers['X-Object-Meta-Ours'], 'okay')
 1012         self.assertTrue(('X-Object-Meta-Ours', 'okay') in headers)
 1013         self.assertEqual(len(self.authorized), 2)
 1014         self.assertEqual('GET', self.authorized[0].method)
 1015         self.assertEqual('/v1/a/c/o', self.authorized[0].path)
 1016         self.assertEqual('PUT', self.authorized[1].method)
 1017         self.assertEqual('/v1/a/c/o', self.authorized[1].path)
 1018 
 1019     def test_COPY_account_with_metadata(self):
 1020         self.app.register('GET', '/v1/a/c/o', swob.HTTPOk, {}, "passed")
 1021         self.app.register('PUT', '/v1/a1/c1/o', swob.HTTPCreated, {})
 1022         req = Request.blank('/v1/a/c/o',
 1023                             environ={'REQUEST_METHOD': 'COPY'},
 1024                             headers={'Destination': '/c1/o',
 1025                                      'X-Object-Meta-Ours': 'okay',
 1026                                      'Destination-Account': 'a1'})
 1027         status, headers, body = self.call_ssc(req)
 1028         self.assertEqual(status, '201 Created')
 1029         calls = self.app.calls_with_headers
 1030         method, path, req_headers = calls[1]
 1031         self.assertEqual('PUT', method)
 1032         self.assertEqual('/v1/a1/c1/o', path)
 1033         self.assertEqual(req_headers['X-Object-Meta-Ours'], 'okay')
 1034         self.assertTrue(('X-Object-Meta-Ours', 'okay') in headers)
 1035         self.assertEqual(len(self.authorized), 2)
 1036         self.assertEqual('GET', self.authorized[0].method)
 1037         self.assertEqual('/v1/a/c/o', self.authorized[0].path)
 1038         self.assertEqual('PUT', self.authorized[1].method)
 1039         self.assertEqual('/v1/a1/c1/o', self.authorized[1].path)
 1040 
 1041     def test_COPY_source_zero_content_length(self):
 1042         self.app.register('GET', '/v1/a/c/o', swob.HTTPOk, {}, None)
 1043         req = Request.blank('/v1/a/c/o',
 1044                             environ={'REQUEST_METHOD': 'COPY'},
 1045                             headers={'Destination': '/c/o'})
 1046         status, headers, body = self.call_ssc(req)
 1047         self.assertEqual(status, '413 Request Entity Too Large')
 1048         self.assertEqual(len(self.authorized), 1)
 1049         self.assertEqual('GET', self.authorized[0].method)
 1050         self.assertEqual('/v1/a/c/o', self.authorized[0].path)
 1051 
 1052     def test_COPY_source_larger_than_max_file_size(self):
 1053         self.app.register('GET', '/v1/a/c/o', swob.HTTPOk, {}, "largebody")
 1054         req = Request.blank('/v1/a/c/o',
 1055                             environ={'REQUEST_METHOD': 'COPY'},
 1056                             headers={'Destination': '/c/o'})
 1057         with mock.patch('swift.common.middleware.copy.'
 1058                         'MAX_FILE_SIZE', 1):
 1059             status, headers, body = self.call_ssc(req)
 1060         self.assertEqual(status, '413 Request Entity Too Large')
 1061         self.assertEqual(len(self.authorized), 1)
 1062         self.assertEqual('GET', self.authorized[0].method)
 1063         self.assertEqual('/v1/a/c/o', self.authorized[0].path)
 1064 
 1065     def test_COPY_account_source_zero_content_length(self):
 1066         self.app.register('GET', '/v1/a/c/o', swob.HTTPOk, {}, None)
 1067         req = Request.blank('/v1/a/c/o',
 1068                             environ={'REQUEST_METHOD': 'COPY'},
 1069                             headers={'Destination': '/c/o',
 1070                                      'Destination-Account': 'a1'})
 1071         status, headers, body = self.call_ssc(req)
 1072         self.assertEqual(status, '413 Request Entity Too Large')
 1073         self.assertEqual(len(self.authorized), 1)
 1074         self.assertEqual('GET', self.authorized[0].method)
 1075         self.assertEqual('/v1/a/c/o', self.authorized[0].path)
 1076 
 1077     def test_COPY_account_source_larger_than_max_file_size(self):
 1078         self.app.register('GET', '/v1/a/c/o', swob.HTTPOk, {}, "largebody")
 1079         req = Request.blank('/v1/a/c/o',
 1080                             environ={'REQUEST_METHOD': 'COPY'},
 1081                             headers={'Destination': '/c1/o',
 1082                                      'Destination-Account': 'a1'})
 1083         with mock.patch('swift.common.middleware.copy.'
 1084                         'MAX_FILE_SIZE', 1):
 1085             status, headers, body = self.call_ssc(req)
 1086         self.assertEqual(status, '413 Request Entity Too Large')
 1087         self.assertEqual(len(self.authorized), 1)
 1088         self.assertEqual('GET', self.authorized[0].method)
 1089         self.assertEqual('/v1/a/c/o', self.authorized[0].path)
 1090 
 1091     def test_COPY_newest(self):
 1092         self.app.register('GET', '/v1/a/c/o', swob.HTTPOk,
 1093                           {'Last-Modified': '123'}, "passed")
 1094         self.app.register('PUT', '/v1/a/c/o', swob.HTTPCreated, {})
 1095         req = Request.blank('/v1/a/c/o',
 1096                             environ={'REQUEST_METHOD': 'COPY'},
 1097                             headers={'Destination': '/c/o'})
 1098         status, headers, body = self.call_ssc(req)
 1099         self.assertEqual(status, '201 Created')
 1100         self.assertTrue(('X-Copied-From-Last-Modified', '123') in headers)
 1101         self.assertEqual(len(self.authorized), 2)
 1102         self.assertEqual('GET', self.authorized[0].method)
 1103         self.assertEqual('/v1/a/c/o', self.authorized[0].path)
 1104         self.assertEqual('PUT', self.authorized[1].method)
 1105         self.assertEqual('/v1/a/c/o', self.authorized[1].path)
 1106 
 1107     def test_COPY_account_newest(self):
 1108         self.app.register('GET', '/v1/a/c/o', swob.HTTPOk,
 1109                           {'Last-Modified': '123'}, "passed")
 1110         self.app.register('PUT', '/v1/a1/c1/o', swob.HTTPCreated, {})
 1111         req = Request.blank('/v1/a/c/o',
 1112                             environ={'REQUEST_METHOD': 'COPY'},
 1113                             headers={'Destination': '/c1/o',
 1114                                      'Destination-Account': 'a1'})
 1115         status, headers, body = self.call_ssc(req)
 1116         self.assertEqual(status, '201 Created')
 1117         self.assertTrue(('X-Copied-From-Last-Modified', '123') in headers)
 1118         self.assertEqual(len(self.authorized), 2)
 1119         self.assertEqual('GET', self.authorized[0].method)
 1120         self.assertEqual('/v1/a/c/o', self.authorized[0].path)
 1121         self.assertEqual('PUT', self.authorized[1].method)
 1122         self.assertEqual('/v1/a1/c1/o', self.authorized[1].path)
 1123 
 1124     def test_COPY_in_OPTIONS_response(self):
 1125         self.app.register('OPTIONS', '/v1/a/c/o', swob.HTTPOk,
 1126                           {'Allow': 'GET, PUT'})
 1127         req = Request.blank('/v1/a/c/o',
 1128                             environ={'REQUEST_METHOD': 'OPTIONS'}, headers={})
 1129         status, headers, body = self.call_ssc(req)
 1130         self.assertEqual(status, '200 OK')
 1131         calls = self.app.calls_with_headers
 1132         method, path, req_headers = calls[0]
 1133         self.assertEqual('OPTIONS', method)
 1134         self.assertEqual('/v1/a/c/o', path)
 1135         self.assertTrue(('Allow', 'GET, PUT, COPY') in headers)
 1136         self.assertEqual(len(self.authorized), 1)
 1137         self.assertEqual('OPTIONS', self.authorized[0].method)
 1138         self.assertEqual('/v1/a/c/o', self.authorized[0].path)
 1139         # For basic test cases, assert orig_req_method behavior
 1140         self.assertNotIn('swift.orig_req_method', req.environ)
 1141 
 1142     def test_COPY_in_OPTIONS_response_CORS(self):
 1143         self.app.register('OPTIONS', '/v1/a/c/o', swob.HTTPOk,
 1144                           {'Allow': 'GET, PUT',
 1145                            'Access-Control-Allow-Methods': 'GET, PUT'})
 1146         req = Request.blank('/v1/a/c/o',
 1147                             environ={'REQUEST_METHOD': 'OPTIONS'}, headers={})
 1148         status, headers, body = self.call_ssc(req)
 1149         self.assertEqual(status, '200 OK')
 1150         calls = self.app.calls_with_headers
 1151         method, path, req_headers = calls[0]
 1152         self.assertEqual('OPTIONS', method)
 1153         self.assertEqual('/v1/a/c/o', path)
 1154         self.assertTrue(('Allow', 'GET, PUT, COPY') in headers)
 1155         self.assertTrue(('Access-Control-Allow-Methods',
 1156                          'GET, PUT, COPY') in headers)
 1157         self.assertEqual(len(self.authorized), 1)
 1158         self.assertEqual('OPTIONS', self.authorized[0].method)
 1159         self.assertEqual('/v1/a/c/o', self.authorized[0].path)
 1160 
 1161     def _test_COPY_source_headers(self, extra_put_headers):
 1162         # helper method to perform a COPY with some metadata headers that
 1163         # should always be sent to the destination
 1164         put_headers = {'Destination': '/c1/o',
 1165                        'X-Object-Meta-Test2': 'added',
 1166                        'X-Object-Sysmeta-Test2': 'added',
 1167                        'X-Object-Transient-Sysmeta-Test2': 'added'}
 1168         put_headers.update(extra_put_headers)
 1169         get_resp_headers = {
 1170             'X-Timestamp': '1234567890.12345',
 1171             'X-Backend-Timestamp': '1234567890.12345',
 1172             'Content-Type': 'text/original',
 1173             'Content-Encoding': 'gzip',
 1174             'Content-Disposition': 'attachment; filename=myfile',
 1175             'X-Object-Meta-Test': 'original',
 1176             'X-Object-Sysmeta-Test': 'original',
 1177             'X-Object-Transient-Sysmeta-Test': 'original',
 1178             'X-Foo': 'Bar'}
 1179         self.app.register(
 1180             'GET', '/v1/a/c/o', swob.HTTPOk, headers=get_resp_headers)
 1181         self.app.register('PUT', '/v1/a/c1/o', swob.HTTPCreated, {})
 1182         req = Request.blank('/v1/a/c/o', method='COPY', headers=put_headers)
 1183         status, headers, body = self.call_ssc(req)
 1184         self.assertEqual(status, '201 Created')
 1185         calls = self.app.calls_with_headers
 1186         self.assertEqual(2, len(calls))
 1187         method, path, req_headers = calls[1]
 1188         self.assertEqual('PUT', method)
 1189         # these headers should always be applied to the destination
 1190         self.assertEqual('added', req_headers.get('X-Object-Meta-Test2'))
 1191         self.assertEqual('added', req_headers.get('X-Object-Sysmeta-Test2'))
 1192         self.assertEqual('added',
 1193                          req_headers.get('X-Object-Transient-Sysmeta-Test2'))
 1194         return req_headers
 1195 
 1196     def test_COPY_source_headers_no_updates(self):
 1197         # copy should preserve existing metadata if not updated
 1198         req_headers = self._test_COPY_source_headers({})
 1199         self.assertEqual('text/original', req_headers.get('Content-Type'))
 1200         self.assertEqual('gzip', req_headers.get('Content-Encoding'))
 1201         self.assertEqual('attachment; filename=myfile',
 1202                          req_headers.get('Content-Disposition'))
 1203         self.assertEqual('original', req_headers.get('X-Object-Meta-Test'))
 1204         self.assertEqual('original', req_headers.get('X-Object-Sysmeta-Test'))
 1205         self.assertEqual('original',
 1206                          req_headers.get('X-Object-Transient-Sysmeta-Test'))
 1207         self.assertEqual('Bar', req_headers.get('X-Foo'))
 1208         self.assertNotIn('X-Timestamp', req_headers)
 1209         self.assertNotIn('X-Backend-Timestamp', req_headers)
 1210 
 1211     def test_COPY_source_headers_with_updates(self):
 1212         # copy should apply any updated values to existing metadata
 1213         put_headers = {
 1214             'Content-Type': 'text/not_original',
 1215             'Content-Encoding': 'not_gzip',
 1216             'Content-Disposition': 'attachment; filename=notmyfile',
 1217             'X-Object-Meta-Test': 'not_original',
 1218             'X-Object-Sysmeta-Test': 'not_original',
 1219             'X-Object-Transient-Sysmeta-Test': 'not_original',
 1220             'X-Foo': 'Not Bar'}
 1221         req_headers = self._test_COPY_source_headers(put_headers)
 1222         self.assertEqual('text/not_original', req_headers.get('Content-Type'))
 1223         self.assertEqual('not_gzip', req_headers.get('Content-Encoding'))
 1224         self.assertEqual('attachment; filename=notmyfile',
 1225                          req_headers.get('Content-Disposition'))
 1226         self.assertEqual('not_original', req_headers.get('X-Object-Meta-Test'))
 1227         self.assertEqual('not_original',
 1228                          req_headers.get('X-Object-Sysmeta-Test'))
 1229         self.assertEqual('not_original',
 1230                          req_headers.get('X-Object-Transient-Sysmeta-Test'))
 1231         self.assertEqual('Not Bar', req_headers.get('X-Foo'))
 1232         self.assertNotIn('X-Timestamp', req_headers)
 1233         self.assertNotIn('X-Backend-Timestamp', req_headers)
 1234 
 1235     def test_COPY_x_fresh_metadata_no_updates(self):
 1236         # existing user metadata should not be copied, sysmeta is copied
 1237         put_headers = {
 1238             'X-Fresh-Metadata': 'true',
 1239             'X-Extra': 'Fresh'}
 1240         req_headers = self._test_COPY_source_headers(put_headers)
 1241         self.assertEqual('text/original', req_headers.get('Content-Type'))
 1242         self.assertEqual('Fresh', req_headers.get('X-Extra'))
 1243         self.assertEqual('original',
 1244                          req_headers.get('X-Object-Sysmeta-Test'))
 1245         self.assertIn('X-Fresh-Metadata', req_headers)
 1246         self.assertNotIn('X-Object-Meta-Test', req_headers)
 1247         self.assertNotIn('X-Object-Transient-Sysmeta-Test', req_headers)
 1248         self.assertNotIn('X-Timestamp', req_headers)
 1249         self.assertNotIn('X-Backend-Timestamp', req_headers)
 1250         self.assertNotIn('Content-Encoding', req_headers)
 1251         self.assertNotIn('Content-Disposition', req_headers)
 1252         self.assertNotIn('X-Foo', req_headers)
 1253 
 1254     def test_COPY_x_fresh_metadata_with_updates(self):
 1255         # existing user metadata should not be copied, new metadata replaces it
 1256         put_headers = {
 1257             'X-Fresh-Metadata': 'true',
 1258             'Content-Type': 'text/not_original',
 1259             'Content-Encoding': 'not_gzip',
 1260             'Content-Disposition': 'attachment; filename=notmyfile',
 1261             'X-Object-Meta-Test': 'not_original',
 1262             'X-Object-Sysmeta-Test': 'not_original',
 1263             'X-Object-Transient-Sysmeta-Test': 'not_original',
 1264             'X-Foo': 'Not Bar',
 1265             'X-Extra': 'Fresh'}
 1266         req_headers = self._test_COPY_source_headers(put_headers)
 1267         self.assertEqual('Fresh', req_headers.get('X-Extra'))
 1268         self.assertEqual('text/not_original', req_headers.get('Content-Type'))
 1269         self.assertEqual('not_gzip', req_headers.get('Content-Encoding'))
 1270         self.assertEqual('attachment; filename=notmyfile',
 1271                          req_headers.get('Content-Disposition'))
 1272         self.assertEqual('not_original', req_headers.get('X-Object-Meta-Test'))
 1273         self.assertEqual('not_original',
 1274                          req_headers.get('X-Object-Sysmeta-Test'))
 1275         self.assertEqual('not_original',
 1276                          req_headers.get('X-Object-Transient-Sysmeta-Test'))
 1277         self.assertEqual('Not Bar', req_headers.get('X-Foo'))
 1278 
 1279     def test_COPY_with_single_range(self):
 1280         # verify that source etag is not copied when copying a range
 1281         self.app.register('GET', '/v1/a/c/o', swob.HTTPOk,
 1282                           {'etag': 'bogus etag'}, "abcdefghijklmnop")
 1283         self.app.register('PUT', '/v1/a/c1/o', swob.HTTPCreated, {})
 1284         req = swob.Request.blank(
 1285             '/v1/a/c/o', method='COPY',
 1286             headers={'Destination': 'c1/o',
 1287                      'Range': 'bytes=5-10'})
 1288 
 1289         status, headers, body = self.call_ssc(req)
 1290 
 1291         self.assertEqual(status, '201 Created')
 1292         calls = self.app.calls_with_headers
 1293         self.assertEqual(2, len(calls))
 1294         method, path, req_headers = calls[1]
 1295         self.assertEqual('PUT', method)
 1296         self.assertEqual('/v1/a/c1/o', path)
 1297         self.assertNotIn('etag', (h.lower() for h in req_headers))
 1298         self.assertEqual('6', req_headers['content-length'])
 1299         req = swob.Request.blank('/v1/a/c1/o', method='GET')
 1300         status, headers, body = self.call_ssc(req)
 1301         self.assertEqual(b'fghijk', body)
 1302 
 1303 
 1304 @patch_policies(with_ec_default=True)
 1305 class TestServerSideCopyMiddlewareWithEC(unittest.TestCase):
 1306     container_info = {
 1307         'status': 200,
 1308         'write_acl': None,
 1309         'read_acl': None,
 1310         'storage_policy': None,
 1311         'sync_key': None,
 1312         'versions': None,
 1313     }
 1314 
 1315     def setUp(self):
 1316         self.logger = debug_logger('proxy-server')
 1317         self.logger.thread_locals = ('txn1', '127.0.0.2')
 1318         self.app = PatchedObjControllerApp(
 1319             None, FakeMemcache(), account_ring=FakeRing(),
 1320             container_ring=FakeRing(), logger=self.logger)
 1321         self.ssc = copy.filter_factory({})(self.app)
 1322         self.ssc.logger = self.app.logger
 1323         self.policy = POLICIES.default
 1324         self.app.container_info = dict(self.container_info)
 1325 
 1326     def test_COPY_with_single_range(self):
 1327         req = swob.Request.blank(
 1328             '/v1/a/c/o', method='COPY',
 1329             headers={'Destination': 'c1/o',
 1330                      'Range': 'bytes=5-10'})
 1331         # turn a real body into fragments
 1332         segment_size = self.policy.ec_segment_size
 1333         real_body = (b'asdf' * segment_size)[:-10]
 1334 
 1335         # split it up into chunks
 1336         chunks = [real_body[x:x + segment_size]
 1337                   for x in range(0, len(real_body), segment_size)]
 1338 
 1339         # we need only first chunk to rebuild 5-10 range
 1340         fragments = self.policy.pyeclib_driver.encode(chunks[0])
 1341         fragment_payloads = []
 1342         fragment_payloads.append(fragments)
 1343 
 1344         node_fragments = list(zip(*fragment_payloads))
 1345         self.assertEqual(len(node_fragments),
 1346                          self.policy.object_ring.replicas)  # sanity
 1347         headers = {'X-Object-Sysmeta-Ec-Content-Length': str(len(real_body))}
 1348         responses = [(200, b''.join(node_fragments[i]), headers)
 1349                      for i in range(POLICIES.default.ec_ndata)]
 1350         responses += [(201, b'', {})] * self.policy.object_ring.replicas
 1351         status_codes, body_iter, headers = zip(*responses)
 1352         expect_headers = {
 1353             'X-Obj-Metadata-Footer': 'yes',
 1354             'X-Obj-Multiphase-Commit': 'yes'
 1355         }
 1356 
 1357         put_hdrs = []
 1358 
 1359         def capture_conn(host, port, dev, part, method, path, *args, **kwargs):
 1360             if method == 'PUT':
 1361                 put_hdrs.append(args[0])
 1362 
 1363         with set_http_connect(*status_codes, body_iter=body_iter,
 1364                               headers=headers, expect_headers=expect_headers,
 1365                               give_connect=capture_conn):
 1366             resp = req.get_response(self.ssc)
 1367 
 1368         self.assertEqual(resp.status_int, 201)
 1369         expected_puts = POLICIES.default.ec_ndata + POLICIES.default.ec_nparity
 1370         self.assertEqual(expected_puts, len(put_hdrs))
 1371         for hdrs in put_hdrs:
 1372             # etag should not be copied from source
 1373             self.assertNotIn('etag', (h.lower() for h in hdrs))
 1374 
 1375     def test_COPY_with_invalid_ranges(self):
 1376         # real body size is segment_size - 10 (just 1 segment)
 1377         segment_size = self.policy.ec_segment_size
 1378         real_body = (b'a' * segment_size)[:-10]
 1379 
 1380         # range is out of real body but in segment size
 1381         self._test_invalid_ranges('COPY', real_body,
 1382                                   segment_size, '%s-' % (segment_size - 10))
 1383         # range is out of both real body and segment size
 1384         self._test_invalid_ranges('COPY', real_body,
 1385                                   segment_size, '%s-' % (segment_size + 10))
 1386 
 1387     def _test_invalid_ranges(self, method, real_body, segment_size, req_range):
 1388         # make a request with range starts from more than real size.
 1389         body_etag = md5(real_body).hexdigest()
 1390         req = swob.Request.blank(
 1391             '/v1/a/c/o', method=method,
 1392             headers={'Destination': 'c1/o',
 1393                      'Range': 'bytes=%s' % (req_range)})
 1394 
 1395         fragments = self.policy.pyeclib_driver.encode(real_body)
 1396         fragment_payloads = [fragments]
 1397 
 1398         node_fragments = list(zip(*fragment_payloads))
 1399         self.assertEqual(len(node_fragments),
 1400                          self.policy.object_ring.replicas)  # sanity
 1401         headers = {'X-Object-Sysmeta-Ec-Content-Length': str(len(real_body)),
 1402                    'X-Object-Sysmeta-Ec-Etag': body_etag}
 1403         start = int(req_range.split('-')[0])
 1404         self.assertTrue(start >= 0)  # sanity
 1405         title, exp = swob.RESPONSE_REASONS[416]
 1406         range_not_satisfiable_body = \
 1407             '<html><h1>%s</h1><p>%s</p></html>' % (title, exp)
 1408         range_not_satisfiable_body = range_not_satisfiable_body.encode('ascii')
 1409         if start >= segment_size:
 1410             responses = [(416, range_not_satisfiable_body, headers)
 1411                          for i in range(POLICIES.default.ec_ndata)]
 1412         else:
 1413             responses = [(200, b''.join(node_fragments[i]), headers)
 1414                          for i in range(POLICIES.default.ec_ndata)]
 1415         status_codes, body_iter, headers = zip(*responses)
 1416         expect_headers = {
 1417             'X-Obj-Metadata-Footer': 'yes',
 1418             'X-Obj-Multiphase-Commit': 'yes'
 1419         }
 1420         # TODO possibly use FakeApp here
 1421         with set_http_connect(*status_codes, body_iter=body_iter,
 1422                               headers=headers, expect_headers=expect_headers):
 1423             resp = req.get_response(self.ssc)
 1424         self.assertEqual(resp.status_int, 416)
 1425         self.assertEqual(resp.content_length, len(range_not_satisfiable_body))
 1426         self.assertEqual(resp.body, range_not_satisfiable_body)
 1427         self.assertEqual(resp.etag, body_etag)
 1428         self.assertEqual(resp.headers['Accept-Ranges'], 'bytes')