Merge pull request #591 from cbodley/wip-68292

s3: test GetObject with PartNumber and SSE-C encryption

README.rst

@@ -22,7 +22,7 @@ Once you have that file copied and edited, you can run the tests with::
 
 You can specify which directory of tests to run::
 
-	S3TEST_CONF=your.conf tox s3tests_boto3/functional
+	S3TEST_CONF=your.conf tox -- s3tests_boto3/functional
 
 You can specify which file of tests to run::
 
@@ -44,7 +44,7 @@ located in the ``s3test_boto3`` directory.
 
 You can run only the boto3 tests with::
 
-	S3TEST_CONF=your.conf tox -- -m 'not fails_on_aws' s3tests_boto3/functional
+	S3TEST_CONF=your.conf tox -- s3tests_boto3/functional
 
 ========================
  STS compatibility tests

s3tests_boto3/functional/__init__.py

@@ -174,7 +174,7 @@ def configured_storage_classes():
 
     return sc
 
-def setup():
+def configure():
     cfg = configparser.RawConfigParser()
     try:
         path = os.environ['S3TEST_CONF']
@@ -285,17 +285,17 @@ def setup():
     template = cfg.get('fixtures', "iam path prefix", fallback="/s3-tests/")
     config.iam_path_prefix = choose_bucket_prefix(template=template)
 
-    alt_client = get_alt_client()
-    tenant_client = get_tenant_client()
-    nuke_prefixed_buckets(prefix=prefix)
-    nuke_prefixed_buckets(prefix=prefix, client=alt_client)
-    nuke_prefixed_buckets(prefix=prefix, client=tenant_client)
-
     if cfg.has_section("s3 cloud"):
         get_cloud_config(cfg)
     else:
         config.cloud_storage_class = None
 
+def setup():
+    alt_client = get_alt_client()
+    tenant_client = get_tenant_client()
+    nuke_prefixed_buckets(prefix=prefix)
+    nuke_prefixed_buckets(prefix=prefix, client=alt_client)
+    nuke_prefixed_buckets(prefix=prefix, client=tenant_client)
 
 def teardown():
     alt_client = get_alt_client()
@@ -324,11 +324,12 @@ def teardown():
 
 @pytest.fixture(scope="package")
 def configfile():
-    setup()
+    configure()
     return config
 
 @pytest.fixture(autouse=True)
 def setup_teardown(configfile):
+    setup()
     yield
     teardown()
 

s3tests_boto3/functional/test_s3.py

@@ -11517,6 +11517,63 @@ def test_bucket_policy_put_obj_tagging_existing_tag():
     assert status == 403
 
 
+@pytest.mark.bucket_policy
+@pytest.mark.fails_on_dbstore
+def test_bucket_policy_upload_part_copy():
+    bucket_name = _create_objects(keys=['public/foo', 'public/bar', 'private/foo'])
+    client = get_client()
+
+    src_resource = _make_arn_resource("{}/{}".format(bucket_name, "public/*"))
+    policy_document = make_json_policy("s3:GetObject", src_resource)
+
+    client.put_bucket_policy(Bucket=bucket_name, Policy=policy_document)
+
+    alt_client = get_alt_client()
+    bucket_name2 = get_new_bucket(alt_client)
+
+    copy_source = {'Bucket': bucket_name, 'Key': 'public/foo'}
+
+    # Create a multipart upload
+    response = alt_client.create_multipart_upload(Bucket=bucket_name2, Key='new_foo')
+    upload_id = response['UploadId']
+    # Upload a part
+    response = alt_client.upload_part_copy(Bucket=bucket_name2, Key='new_foo', PartNumber=1, UploadId=upload_id, CopySource=copy_source)
+    # Complete the multipart upload
+    response = alt_client.complete_multipart_upload(
+        Bucket=bucket_name2, Key='new_foo', UploadId=upload_id,
+        MultipartUpload={'Parts': [{'PartNumber': 1, 'ETag': response['CopyPartResult']['ETag']}]},
+    )
+
+    response = alt_client.get_object(Bucket=bucket_name2, Key='new_foo')
+    body = _get_body(response)
+    assert body == 'public/foo'
+
+    copy_source = {'Bucket': bucket_name, 'Key': 'public/bar'}
+    # Create a multipart upload
+    response = alt_client.create_multipart_upload(Bucket=bucket_name2, Key='new_foo2')
+    upload_id = response['UploadId']
+    # Upload a part
+    response = alt_client.upload_part_copy(Bucket=bucket_name2, Key='new_foo2', PartNumber=1, UploadId=upload_id, CopySource=copy_source)
+    # Complete the multipart upload
+    response = alt_client.complete_multipart_upload(
+        Bucket=bucket_name2, Key='new_foo2', UploadId=upload_id,
+        MultipartUpload={'Parts': [{'PartNumber': 1, 'ETag': response['CopyPartResult']['ETag']}]},
+    )
+
+    response = alt_client.get_object(Bucket=bucket_name2, Key='new_foo2')
+    body = _get_body(response)
+    assert body == 'public/bar'
+
+    copy_source = {'Bucket': bucket_name, 'Key': 'private/foo'}
+    # Create a multipart upload
+    response = alt_client.create_multipart_upload(Bucket=bucket_name2, Key='new_foo2')
+    upload_id = response['UploadId']
+    # Upload a part
+    check_access_denied(alt_client.upload_part_copy, Bucket=bucket_name2, Key='new_foo2', PartNumber=1, UploadId=upload_id, CopySource=copy_source)
+    # Abort the multipart upload
+    alt_client.abort_multipart_upload(Bucket=bucket_name2, Key='new_foo2', UploadId=upload_id)
+
+
 @pytest.mark.tagging
 @pytest.mark.bucket_policy
 @pytest.mark.fails_on_dbstore