slo: add more functional tests for copying

Add functional tests for copying from an SLO using PUT requests with
an x-copy-from header.

Add functional tests for copying from an SLO using part-number parameter.

Change-Id: Ic40c4b104b11de9ab09485e82543521b37ea1daa

test/functional/swift_test_client.py

@@ -866,6 +866,10 @@ class File(Base):
 
     def copy(self, dest_cont, dest_file, hdrs=None, parms=None, cfg=None,
              return_resp=False):
+        """
+        Make a copy of this object using a COPY request with a Destination
+        header.
+        """
         if hdrs is None:
             hdrs = {}
         if parms is None:
@@ -891,6 +895,29 @@ class File(Base):
             return self.conn.response
         return True
 
+    def copy_using_x_copy_from(self, dest_cont, dest_file, hdrs=None,
+                               parms=None, cfg=None, return_resp=False):
+        """
+        Make a copy of this object using a PUT request with an X-Copy-From
+        header.
+        """
+        if hdrs is None:
+            hdrs = {}
+        if parms is None:
+            parms = {}
+        if cfg is None:
+            cfg = {}
+        headers = {'X-Copy-From': '/'.join(self.path)}
+        headers.update(hdrs)
+        path = [dest_cont, dest_file]
+        if self.conn.make_request('PUT', path, hdrs=headers,
+                                  cfg=cfg, parms=parms) != 201:
+            raise ResponseError(self.conn.response, 'PUT',
+                                self.conn.make_path(path))
+        if return_resp:
+            return self.conn.response
+        return True
+
     def copy_account(self, dest_account, dest_cont, dest_file,
                      hdrs=None, parms=None, cfg=None):
         if hdrs is None:

test/functional/test_slo.py

@@ -1031,6 +1031,42 @@ class TestSlo(Base):
         copied_contents = copied.read(parms={'multipart-manifest': 'get'})
         self.assertEqual(4 * 1024 * 1024 + 1, len(copied_contents))
 
+    def test_slo_copy_using_x_copy_from(self):
+        # as per test_slo_copy but using a PUT with x-copy-from
+        file_item = self.env.container.file("manifest-abcde")
+        file_item.copy_using_x_copy_from(
+            self.env.container.name, "copied-abcde")
+
+        copied = self.env.container.file("copied-abcde")
+        copied_contents = copied.read(parms={'multipart-manifest': 'get'})
+        self.assertEqual(4 * 1024 * 1024 + 1, len(copied_contents))
+
+    def test_slo_copy_part_number(self):
+        file_item = self.env.container.file("manifest-abcde")
+        file_item.copy(self.env.container.name, "copied-abcde",
+                       parms={'part-number': '1'})
+
+        copied = self.env.container.file("copied-abcde")
+        copied_contents = copied.read(parms={'multipart-manifest': 'get'})
+        # just the first part is copied
+        self.assertEqual(1024 * 1024, len(copied_contents))
+        self.assertEqual(b'a' * 10, copied_contents[:10])
+
+    def test_slo_copy_part_number_using_x_copy_from(self):
+        # as per test_slo_copy_part_number but using a PUT with x-copy-from
+        file_item = self.env.container.file("manifest-abcde")
+        # part-number on the client PUT target is actually applied to the
+        # internal GET source request
+        file_item.copy_using_x_copy_from(
+            self.env.container.name, "copied-abcde",
+            parms={'part-number': '1'})
+
+        copied = self.env.container.file("copied-abcde")
+        copied_contents = copied.read(parms={'multipart-manifest': 'get'})
+        # just the first part is copied
+        self.assertEqual(1024 * 1024, len(copied_contents))
+        self.assertEqual(b'a' * 10, copied_contents[:10])
+
     def test_slo_copy_account(self):
         acct = urllib.parse.unquote(self.env.conn.account_name)
         # same account copy
@@ -1129,6 +1165,38 @@ class TestSlo(Base):
         self.assertEqual(copied_json[0], {
             'data': base64.b64encode(b'APRE' * 8).decode('ascii')})
 
+    def test_slo_copy_the_manifest_using_x_copy_from(self):
+        # as per test_slo_copy_the_manifest but using a PUT with x-copy-from
+        source = self.env.container.file("manifest-abcde")
+        source.initialize(parms={'multipart-manifest': 'get'})
+        source_contents = source.read(parms={'multipart-manifest': 'get'})
+        source_json = json.loads(source_contents)
+        manifest_etag = md5(source_contents, usedforsecurity=False).hexdigest()
+        if tf.cluster_info.get('etag_quoter', {}).get('enable_by_default'):
+            manifest_etag = '"%s"' % manifest_etag
+        self.assertEqual(manifest_etag, source.etag)
+
+        source.initialize()
+        self.assertEqual('application/octet-stream', source.content_type)
+        self.assertNotEqual(manifest_etag, source.etag)
+
+        # multipart-manifest=get on the client PUT target request actually
+        # applies to the internal GET source request
+        self.assertTrue(
+            source.copy_using_x_copy_from(self.env.container.name,
+                                          "copied-abcde-manifest-only",
+                                          parms={'multipart-manifest': 'get'}))
+
+        copied = self.env.container.file("copied-abcde-manifest-only")
+        copied.initialize(parms={'multipart-manifest': 'get'})
+        copied_contents = copied.read(parms={'multipart-manifest': 'get'})
+        try:
+            copied_json = json.loads(copied_contents)
+        except ValueError:
+            self.fail("COPY didn't copy the manifest (invalid json on GET)")
+        self.assertEqual(source_json, copied_json)
+        self.assertEqual(manifest_etag, copied.etag)
+
     def test_slo_copy_the_manifest_updating_metadata(self):
         source = self.env.container.file("manifest-abcde")
         source.content_type = 'application/octet-stream'