Merge pull request #591 from cbodley/wip-68292

s3: test GetObject with PartNumber and SSE-C encryption

.gitignore

@@ -10,5 +10,6 @@
 
 /*.egg-info
 /virtualenv
+/venv
 
 config.yaml

README.rst

@@ -6,14 +6,10 @@ This is a set of unofficial Amazon AWS S3 compatibility
 tests, that can be useful to people implementing software
 that exposes an S3-like API. The tests use the Boto2 and Boto3 libraries.
 
-The tests use the Nose test framework. To get started, ensure you have
-the ``virtualenv`` software installed; e.g. on Debian/Ubuntu::
+The tests use the Tox tool. To get started, ensure you have the ``tox``
+software installed; e.g. on Debian/Ubuntu::
 
-	sudo apt-get install python-virtualenv
-
-and then run::
-
-	./bootstrap
+	sudo apt-get install tox
 
 You will need to create a configuration file with the location of the
 service and two different credentials. A sample configuration file named
@@ -22,29 +18,25 @@ used to run the s3 tests on a Ceph cluster started with vstart.
 
 Once you have that file copied and edited, you can run the tests with::
 
-	S3TEST_CONF=your.conf ./virtualenv/bin/nosetests
+	S3TEST_CONF=your.conf tox
 
 You can specify which directory of tests to run::
 
-	S3TEST_CONF=your.conf ./virtualenv/bin/nosetests s3tests_boto3.functional
+	S3TEST_CONF=your.conf tox -- s3tests_boto3/functional
 
 You can specify which file of tests to run::
 
-	S3TEST_CONF=your.conf ./virtualenv/bin/nosetests s3tests_boto3.functional.test_s3
+	S3TEST_CONF=your.conf tox s3tests_boto3/functional/test_s3.py
 
 You can specify which test to run::
 
-	S3TEST_CONF=your.conf ./virtualenv/bin/nosetests s3tests_boto3.functional.test_s3:test_bucket_list_empty
-
-To gather a list of tests being run, use the flags::
-
-	 -v --collect-only
+	S3TEST_CONF=your.conf tox s3tests_boto3/functional/test_s3.py::test_bucket_list_empty
 
 Some tests have attributes set based on their current reliability and
 things like AWS not enforcing their spec stricly. You can filter tests
 based on their attributes::
 
-	S3TEST_CONF=aws.conf ./virtualenv/bin/nosetests -a '!fails_on_aws'
+	S3TEST_CONF=aws.conf tox -- -m 'not fails_on_aws'
 
 Most of the tests have both Boto3 and Boto2 versions. Tests written in
 Boto2 are in the ``s3tests`` directory. Tests written in Boto3 are
@@ -52,7 +44,7 @@ located in the ``s3test_boto3`` directory.
 
 You can run only the boto3 tests with::
 
-        S3TEST_CONF=your.conf ./virtualenv/bin/nosetests -v -s -A 'not fails_on_rgw' s3tests_boto3.functional
+	S3TEST_CONF=your.conf tox -- s3tests_boto3/functional
 
 ========================
  STS compatibility tests
@@ -60,14 +52,51 @@ You can run only the boto3 tests with::
 
 This section contains some basic tests for the AssumeRole, GetSessionToken and AssumeRoleWithWebIdentity API's. The test file is located under ``s3tests_boto3/functional``.
 
+To run the STS tests, the vstart cluster should be started with the following parameter (in addition to any parameters already used with it)::
+
+        vstart.sh -o rgw_sts_key=abcdefghijklmnop -o rgw_s3_auth_use_sts=true
+
+Note that the ``rgw_sts_key`` can be set to anything that is 128 bits in length.
+After the cluster is up the following command should be executed::
+
+      radosgw-admin caps add --tenant=testx --uid="9876543210abcdef0123456789abcdef0123456789abcdef0123456789abcdef" --caps="roles=*"
+
 You can run only the sts tests (all the three API's) with::
 
-        S3TEST_CONF=your.conf ./virtualenv/bin/nosetests s3tests_boto3.functional.test_sts
+        S3TEST_CONF=your.conf tox s3tests_boto3/functional/test_sts.py
 
 You can filter tests based on the attributes. There is a attribute named ``test_of_sts`` to run AssumeRole and GetSessionToken tests and ``webidentity_test`` to run the AssumeRoleWithWebIdentity tests. If you want to execute only ``test_of_sts`` tests you can apply that filter as below::
 
-        S3TEST_CONF=your.conf ./virtualenv/bin/nosetests -v -s -A 'test_of_sts' s3tests_boto3.functional.test_sts
+        S3TEST_CONF=your.conf tox -- -m test_of_sts s3tests_boto3/functional/test_sts.py
 
 For running ``webidentity_test`` you'll need have Keycloak running.
 
 In order to run any STS test you'll need to add "iam" section to the config file. For further reference on how your config file should look check ``s3tests.conf.SAMPLE``.
+
+========================
+ IAM policy tests
+========================
+
+This is a set of IAM policy tests.
+This section covers tests for user policies such as Put, Get, List, Delete, user policies with s3 actions, conflicting user policies etc
+These tests uses Boto3 libraries. Tests are written in the ``s3test_boto3`` directory.
+
+These iam policy tests uses two users with profile name "iam" and "s3 alt" as mentioned in s3tests.conf.SAMPLE.
+If Ceph cluster is started with vstart, then above two users will get created as part of vstart with same access key, secrete key etc as mentioned in s3tests.conf.SAMPLE.
+Out of those two users, "iam" user is with capabilities --caps=user-policy=* and "s3 alt" user is without capabilities.
+Adding above capabilities to "iam" user is also taken care by vstart (If Ceph cluster is started with vstart).
+
+To run these tests, create configuration file with section "iam" and "s3 alt" refer s3tests.conf.SAMPLE.
+Once you have that configuration file copied and edited, you can run all the tests with::
+
+	S3TEST_CONF=your.conf tox s3tests_boto3/functional/test_iam.py
+
+You can also specify specific test to run::
+
+	S3TEST_CONF=your.conf tox s3tests_boto3/functional/test_iam.py::test_put_user_policy
+
+Some tests have attributes set such as "fails_on_rgw".
+You can filter tests based on their attributes::
+
+	S3TEST_CONF=your.conf tox -- s3tests_boto3/functional/test_iam.py -m 'not fails_on_rgw'
+

bootstrap

@@ -1,76 +0,0 @@
-#!/bin/bash
-set -e
-
-virtualenv="virtualenv"
-declare -a packages
-source /etc/os-release
-
-case "$ID" in
-    debian|ubuntu|devuan)
-        packages=(debianutils python3-pip python3-virtualenv python3-dev libevent-dev libffi-dev libxml2-dev libxslt-dev zlib1g-dev)
-        for package in ${packages[@]}; do
-            if [ "$(dpkg --status -- $package 2>/dev/null|sed -n 's/^Status: //p')" != "install ok installed" ]; then
-                # add a space after old values
-                missing="${missing:+$missing }$package"
-            fi
-        done
-
-        if [ -n "$missing" ]; then
-            echo "$0: missing required DEB packages. Installing via sudo." 1>&2
-            sudo apt-get -y install $missing
-        fi
-        ;;
-    centos|fedora|rhel|ol|virtuozzo)
-
-        packages=(which python3-virtualenv python36-devel libevent-devel libffi-devel libxml2-devel libxslt-devel zlib-devel)
-        for package in ${packages[@]}; do
-            # When the package is python36-devel we change it to python3-devel on Fedora
-            if [[ ${package} == "python36-devel" && -f /etc/fedora-release ]]; then
-                package=python36
-            fi
-            if [ "$(rpm -qa $package 2>/dev/null)" == "" ]; then
-                missing="${missing:+$missing }$package"
-            fi
-        done
-
-        if [ -n "$missing" ]; then
-            echo "$0: Missing required RPM packages: ${missing}." 1>&2
-            sudo yum -y install $missing
-        fi
-        ;;
-    opensuse*|suse|sles)
-
-        packages=(which python3-virtualenv python3-devel libev-devel libffi-devel libxml2-devel libxslt-devel zlib-devel)
-        for package in ${packages[@]}; do
-            if [ "$(rpm -qa $package 2>/dev/null)" == "" ]; then
-                missing="${missing:+$missing }$package"
-            fi
-            if [ -n "$missing" ]; then
-                echo "$0: Missing required RPM packages: ${missing}." 1>&2
-                sudo zypper --non-interactive install --no-recommends $missing
-            fi
-        done
-
-        ;;
-    *)
-        echo "Bootstrap script does not support this distro yet, consider adding the packages"
-        exit 1
-esac
-
-
-# s3-tests only works on python 3.6 not newer versions of python3
-${virtualenv} --python=$(which python3.6) virtualenv
-
-# avoid pip bugs
-./virtualenv/bin/pip3 install --upgrade pip
-
-# latest setuptools supporting python 2.7
-./virtualenv/bin/pip install setuptools==44.1.0
-
-./virtualenv/bin/pip3 install -r requirements.txt
-
-# forbid setuptools from using the network because it'll try to use
-# easy_install, and we really wanted pip; next line will fail if pip
-# requirements.txt does not match setup.py requirements -- sucky but
-# good enough for now
-./virtualenv/bin/python3 setup.py develop

pytest.ini

@@ -0,0 +1,51 @@
+[pytest]
+markers =
+    abac_test
+    appendobject
+    auth_aws2
+    auth_aws4
+    auth_common
+    bucket_policy
+    bucket_encryption
+    checksum
+    cloud_transition
+    encryption
+    fails_on_aws
+    fails_on_dbstore
+    fails_on_dho
+    fails_on_mod_proxy_fcgi
+    fails_on_rgw
+    fails_on_s3
+    fails_with_subdomain
+    group
+    group_policy
+    iam_account
+    iam_cross_account
+    iam_role
+    iam_tenant
+    iam_user
+    lifecycle
+    lifecycle_expiration
+    lifecycle_transition
+    list_objects_v2
+    object_lock
+    role_policy
+    session_policy
+    s3select
+    s3website
+    s3website_routing_rules
+    s3website_redirect_location
+    sns
+    sse_s3
+    storage_class
+    tagging
+    test_of_sts
+    token_claims_trust_policy_test
+    token_principal_tag_role_policy_test
+    token_request_tag_trust_policy_test
+    token_resource_tags_test
+    token_role_tags_test
+    token_tag_keys_test
+    user_policy
+    versioning
+    webidentity_test

requirements.txt

@@ -1,12 +1,15 @@
 PyYAML
-nose >=1.0.0
 boto >=2.6.0
 boto3 >=1.0.0
+# botocore-1.28 broke v2 signatures, see https://tracker.ceph.com/issues/58059
+botocore <1.28.0
 munch >=2.0.0
 # 0.14 switches to libev, that means bootstrap needs to change too
 gevent >=1.0
 isodate >=0.4.4
 requests >=2.23.0
-pytz >=2011k
+pytz
 httplib2
 lxml
+pytest
+tox

s3tests.conf.SAMPLE

@@ -19,6 +19,14 @@ ssl_verify = False
 ## the prefix to 30 characters long, and avoid collisions
 bucket prefix = yournamehere-{random}-
 
+# all the iam account resources (users, roles, etc) created
+# will start with this name prefix
+iam name prefix = s3-tests-
+
+# all the iam account resources (users, roles, etc) created
+# will start with this path prefix
+iam path prefix = /s3-tests/
+
 [s3 main]
 # main display_name set in vstart.sh
 display_name = M. Tester
@@ -62,7 +70,7 @@ access_key = NOPQRSTUVWXYZABCDEFG
 # alt AWS secret key set in vstart.sh
 secret_key = nopqrstuvwxyzabcdefghijklmnabcdefghijklm
 
-[s3 cloud]
+#[s3 cloud]
 ## to run the testcases with "cloud_transition" attribute.
 ## Note: the waiting time may have to tweaked depending on
 ## the I/O latency to the cloud endpoint.
@@ -109,6 +117,9 @@ secret_key = opqrstuvwxyzabcdefghijklmnopqrstuvwxyzab
 # tenant email set in vstart.sh
 email = tenanteduser@example.com
 
+# tenant name
+tenant = testx
+
 #following section needs to be added for all sts-tests
 [iam]
 #used for iam operations in sts-tests
@@ -127,6 +138,20 @@ secret_key = abcdefghijklmnopqrstuvwxyzabcdefghijklmn
 #display_name from vstart.sh
 display_name = youruseridhere
 
+# iam account root user for iam_account tests
+[iam root]
+access_key = AAAAAAAAAAAAAAAAAAaa
+secret_key = aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa
+user_id = RGW11111111111111111
+email = account1@ceph.com
+
+# iam account root user in a different account than [iam root]
+[iam alt root]
+access_key = BBBBBBBBBBBBBBBBBBbb
+secret_key = bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb
+user_id = RGW22222222222222222
+email = account2@ceph.com
+
 #following section needs to be added when you want to run Assume Role With Webidentity test
 [webidentity]
 #used for assume role with web identity test in sts-tests

s3tests/functional/__init__.py

@@ -7,6 +7,7 @@ import itertools
 import os
 import random
 import string
+import pytest
 from http.client import HTTPConnection, HTTPSConnection
 from urllib.parse import urlparse
 
@@ -370,6 +371,15 @@ def teardown():
     # remove our buckets here also, to avoid littering
     nuke_prefixed_buckets(prefix=prefix)
 
+@pytest.fixture(scope="package")
+def configfile():
+    setup()
+    yield config
+
+@pytest.fixture(autouse=True)
+def setup_teardown(configfile):
+    yield
+    teardown()
 
 bucket_counter = itertools.count(1)
 

s3tests/functional/test_s3.py

@@ -7,7 +7,7 @@ import datetime
 import time
 import email.utils
 import isodate
-import nose
+import pytest
 import operator
 import socket
 import ssl
@@ -27,16 +27,14 @@ import re
 from collections import defaultdict
 from urllib.parse import urlparse
 
-from nose.tools import eq_ as eq
-from nose.plugins.attrib import attr
-from nose.plugins.skip import SkipTest
-
 from . import utils
 from .utils import assert_raises
 
 from .policy import Policy, Statement, make_json_policy
 
 from . import (
+    configfile,
+    setup_teardown,
     nuke_prefixed_buckets,
     get_new_bucket,
     get_new_bucket_name,
@@ -53,9 +51,9 @@ from . import (
 
 def check_access_denied(fn, *args, **kwargs):
     e = assert_raises(boto.exception.S3ResponseError, fn, *args, **kwargs)
-    eq(e.status, 403)
-    eq(e.reason, 'Forbidden')
-    eq(e.error_code, 'AccessDenied')
+    assert e.status == 403
+    assert e.reason == 'Forbidden'
+    assert e.error_code == 'AccessDenied'
 
 def check_bad_bucket_name(name):
     """
@@ -63,9 +61,9 @@ def check_bad_bucket_name(name):
     that the request fails because of an invalid bucket name.
     """
     e = assert_raises(boto.exception.S3ResponseError, get_new_bucket, targets.main.default, name)
-    eq(e.status, 400)
-    eq(e.reason.lower(), 'bad request') # some proxies vary the case
-    eq(e.error_code, 'InvalidBucketName')
+    assert e.status == 400
+    assert e.reason.lower() == 'bad request' # some proxies vary the case
+    assert e.error_code == 'InvalidBucketName'
 
 def _create_keys(bucket=None, keys=[]):
     """
@@ -94,20 +92,16 @@ def _get_alt_connection():
 
 
 # Breaks DNS with SubdomainCallingFormat
-@attr('fails_with_subdomain')
-@attr(resource='bucket')
-@attr(method='put')
-@attr(operation='create w/! in name')
-@attr(assertion='fails with subdomain')
+@pytest.mark.fails_with_subdomain
 def test_bucket_create_naming_bad_punctuation():
     # characters other than [a-zA-Z0-9._-]
     check_bad_bucket_name('alpha!soup')
 
 def check_versioning(bucket, status):
     try:
-        eq(bucket.get_versioning_status()['Versioning'], status)
+        assert bucket.get_versioning_status()['Versioning'] == status
     except KeyError:
-        eq(status, None)
+        assert status == None
 
 # amazon is eventual consistent, retry a bit if failed
 def check_configure_versioning_retry(bucket, status, expected_string):
@@ -126,13 +120,10 @@ def check_configure_versioning_retry(bucket, status, expected_string):
 
         time.sleep(1)
 
-    eq(expected_string, read_status)
+    assert expected_string == read_status
 
-@attr(resource='object')
-@attr(method='create')
-@attr(operation='create versioned object, read not exist null version')
-@attr(assertion='read null version behaves correctly')
-@attr('versioning')
+@pytest.mark.versioning
+@pytest.mark.fails_on_dbstore
 def test_versioning_obj_read_not_exist_null():
     bucket = get_new_bucket()
     check_versioning(bucket, None)
@@ -146,15 +137,12 @@ def test_versioning_obj_read_not_exist_null():
     key.set_contents_from_string(content)
 
     key = bucket.get_key(objname, version_id='null')
-    eq(key, None)
+    assert key == None
 
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='append object')
-@attr(assertion='success')
-@attr('fails_on_aws')
-@attr('fails_with_subdomain')
-@attr('appendobject')
+@pytest.mark.fails_on_aws
+@pytest.mark.fails_with_subdomain
+@pytest.mark.appendobject
+@pytest.mark.fails_on_dbstore
 def test_append_object():
     bucket = get_new_bucket()
     key = bucket.new_key('foo')
@@ -166,19 +154,16 @@ def test_append_object():
     res = _make_raw_request(host=s3.main.host, port=s3.main.port, method='PUT', path=path1, body='abc', secure=s3.main.is_secure)
     path2 = path + '&append&position=3'
     res = _make_raw_request(host=s3.main.host, port=s3.main.port, method='PUT', path=path2, body='abc', secure=s3.main.is_secure)
-    eq(res.status, 200)
-    eq(res.reason, 'OK')
+    assert res.status == 200
+    assert res.reason == 'OK'
 
     key = bucket.get_key('foo')
-    eq(key.size, 6) 
+    assert key.size == 6 
 
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='append to normal object')
-@attr(assertion='fails 409')
-@attr('fails_on_aws')
-@attr('fails_with_subdomain')
-@attr('appendobject')
+@pytest.mark.fails_on_aws
+@pytest.mark.fails_with_subdomain
+@pytest.mark.appendobject
+@pytest.mark.fails_on_dbstore
 def test_append_normal_object():
     bucket = get_new_bucket()
     key = bucket.new_key('foo')
@@ -189,16 +174,13 @@ def test_append_normal_object():
     path = o.path + '?' + o.query
     path = path + '&append&position=3'
     res = _make_raw_request(host=s3.main.host, port=s3.main.port, method='PUT', path=path, body='abc', secure=s3.main.is_secure)
-    eq(res.status, 409)
+    assert res.status == 409
 
 
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='append position not right')
-@attr(assertion='fails 409')
-@attr('fails_on_aws')
-@attr('fails_with_subdomain')
-@attr('appendobject')
+@pytest.mark.fails_on_aws
+@pytest.mark.fails_with_subdomain
+@pytest.mark.appendobject
+@pytest.mark.fails_on_dbstore
 def test_append_object_position_wrong():
     bucket = get_new_bucket()
     key = bucket.new_key('foo')
@@ -210,17 +192,13 @@ def test_append_object_position_wrong():
     res = _make_raw_request(host=s3.main.host, port=s3.main.port, method='PUT', path=path1, body='abc', secure=s3.main.is_secure)
     path2 = path + '&append&position=9'
     res = _make_raw_request(host=s3.main.host, port=s3.main.port, method='PUT', path=path2, body='abc', secure=s3.main.is_secure)
-    eq(res.status, 409)
-    eq(int(res.getheader('x-rgw-next-append-position')), 3)
+    assert res.status == 409
+    assert int(res.getheader('x-rgw-next-append-position')) == 3
 
 
 # TODO rgw log_bucket.set_as_logging_target() gives 403 Forbidden
 # http://tracker.newdream.net/issues/984
-@attr(resource='bucket.log')
-@attr(method='put')
-@attr(operation='set/enable/disable logging target')
-@attr(assertion='operations succeed')
-@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
 def test_logging_toggle():
     bucket = get_new_bucket()
     log_bucket = get_new_bucket(targets.main.default, bucket.name + '-log')
@@ -307,13 +285,13 @@ def gen_rand_string(size, chars=string.ascii_uppercase + string.digits):
 
 def verify_object(bucket, k, data=None, storage_class=None):
     if storage_class:
-        eq(k.storage_class, storage_class)
+        assert k.storage_class == storage_class
 
     if data:
         read_data = k.get_contents_as_string()
 
-        equal = data == read_data # avoid spamming log if data not equal
-        eq(equal, True)
+        equal = data == read_data.decode() # avoid spamming log if data not equal
+        assert equal == True
 
 def copy_object_storage_class(src_bucket, src_key, dest_bucket, dest_key, storage_class):
             query_args=None
@@ -329,7 +307,7 @@ def copy_object_storage_class(src_bucket, src_key, dest_bucket, dest_key, storag
 
             res = dest_bucket.connection.make_request('PUT', dest_bucket.name, dest_key.name,
                     query_args=query_args, headers=headers)
-            eq(res.status, 200)
+            assert res.status == 200
 
 def _populate_multipart_key(bucket, kname, size, storage_class=None):
     (upload, data) = _multipart_upload(bucket, kname, size, storage_class=storage_class)
@@ -384,6 +362,9 @@ def configured_storage_classes():
             if item != 'STANDARD':
                 sc.append(item)
 
+    sc = [i for i in sc if i]
+    print("storage classes configured: " + str(sc))
+
     return sc
 
 def lc_transition(days=None, date=None, storage_class=None):
@@ -397,15 +378,13 @@ def lc_transitions(transitions=None):
     return result
 
 
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='test create object with storage class')
-@attr('storage_class')
-@attr('fails_on_aws')
+@pytest.mark.storage_class
+@pytest.mark.fails_on_aws
+@pytest.mark.fails_on_dbstore
 def test_object_storage_class():
     sc = configured_storage_classes()
     if len(sc) < 2:
-        raise SkipTest
+        pytest.skip('requires multiple storage classes')
 
     bucket = get_new_bucket()
 
@@ -415,15 +394,13 @@ def test_object_storage_class():
 
         verify_object(bucket, k, data, storage_class)
 
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='test create multipart object with storage class')
-@attr('storage_class')
-@attr('fails_on_aws')
+@pytest.mark.storage_class
+@pytest.mark.fails_on_aws
+@pytest.mark.fails_on_dbstore
 def test_object_storage_class_multipart():
     sc = configured_storage_classes()
     if len(sc) < 2:
-        raise SkipTest
+        pytest.skip('requires multiple storage classes')
 
     bucket = get_new_bucket()
     size = 11 * 1024 * 1024
@@ -433,13 +410,13 @@ def test_object_storage_class_multipart():
         (upload, data) = _multipart_upload(bucket, key, size, storage_class=storage_class)
         upload.complete_upload()
         key2 = bucket.get_key(key)
-        eq(key2.size, size)
-        eq(key2.storage_class, storage_class)
+        assert key2.size == size
+        assert key2.storage_class == storage_class
 
 def _do_test_object_modify_storage_class(obj_write_func, size):
     sc = configured_storage_classes()
     if len(sc) < 2:
-        raise SkipTest
+        pytest.skip('requires multiple storage classes')
 
     bucket = get_new_bucket()
 
@@ -456,27 +433,23 @@ def _do_test_object_modify_storage_class(obj_write_func, size):
             copy_object_storage_class(bucket, k, bucket, k, new_storage_class)
             verify_object(bucket, k, data, storage_class)
 
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='test changing objects storage class')
-@attr('storage_class')
-@attr('fails_on_aws')
+@pytest.mark.storage_class
+@pytest.mark.fails_on_aws
+@pytest.mark.fails_on_dbstore
 def test_object_modify_storage_class():
     _do_test_object_modify_storage_class(_populate_key, size=9*1024*1024)
 
 
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='test changing objects storage class')
-@attr('storage_class')
-@attr('fails_on_aws')
+@pytest.mark.storage_class
+@pytest.mark.fails_on_aws
+@pytest.mark.fails_on_dbstore
 def test_object_modify_storage_class_multipart():
     _do_test_object_modify_storage_class(_populate_multipart_key, size=11*1024*1024)
 
 def _do_test_object_storage_class_copy(obj_write_func, size):
     sc = configured_storage_classes()
     if len(sc) < 2:
-        raise SkipTest
+        pytest.skip('requires multiple storage classes')
 
     src_bucket = get_new_bucket()
     dest_bucket = get_new_bucket()
@@ -494,19 +467,15 @@ def _do_test_object_storage_class_copy(obj_write_func, size):
         copy_object_storage_class(src_bucket, src_key, dest_bucket, dest_key, new_storage_class)
         verify_object(dest_bucket, dest_key, data, new_storage_class)
 
-@attr(resource='object')
-@attr(method='copy')
-@attr(operation='test copy object to object with different storage class')
-@attr('storage_class')
-@attr('fails_on_aws')
+@pytest.mark.storage_class
+@pytest.mark.fails_on_aws
+@pytest.mark.fails_on_dbstore
 def test_object_storage_class_copy():
     _do_test_object_storage_class_copy(_populate_key, size=9*1024*1024)
 
-@attr(resource='object')
-@attr(method='copy')
-@attr(operation='test changing objects storage class')
-@attr('storage_class')
-@attr('fails_on_aws')
+@pytest.mark.storage_class
+@pytest.mark.fails_on_aws
+@pytest.mark.fails_on_dbstore
 def test_object_storage_class_copy_multipart():
     _do_test_object_storage_class_copy(_populate_multipart_key, size=9*1024*1024)
 
@@ -563,7 +532,7 @@ class FakeFileVerifier(object):
         if self.char == None:
             self.char = data[0]
         self.size += size
-        eq(data.decode(), self.char*size)
+        assert data.decode() == self.char*size
 
 def _verify_atomic_key_data(key, size=-1, char=None):
     """
@@ -572,7 +541,7 @@ def _verify_atomic_key_data(key, size=-1, char=None):
     fp_verify = FakeFileVerifier(char)
     key.get_contents_to_file(fp_verify)
     if size >= 0:
-        eq(fp_verify.size, size)
+        assert fp_verify.size == size
 
 def _test_atomic_dual_conditional_write(file_size):
     """
@@ -601,26 +570,20 @@ def _test_atomic_dual_conditional_write(file_size):
     # key.set_contents_from_file(fp_c, headers={'If-Match': etag_fp_a})
     e = assert_raises(boto.exception.S3ResponseError, key.set_contents_from_file, fp_c,
                       headers={'If-Match': etag_fp_a})
-    eq(e.status, 412)
-    eq(e.reason, 'Precondition Failed')
-    eq(e.error_code, 'PreconditionFailed')
+    assert e.status == 412
+    assert e.reason == 'Precondition Failed'
+    assert e.error_code == 'PreconditionFailed'
 
     # verify the file
     _verify_atomic_key_data(key, file_size, 'B')
 
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='write one or the other')
-@attr(assertion='1MB successful')
-@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
+@pytest.mark.fails_on_dbstore
 def test_atomic_dual_conditional_write_1mb():
     _test_atomic_dual_conditional_write(1024*1024)
 
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='write file in deleted bucket')
-@attr(assertion='fail 404')
-@attr('fails_on_aws')
+@pytest.mark.fails_on_aws
+@pytest.mark.fails_on_dbstore
 def test_atomic_write_bucket_gone():
     bucket = get_new_bucket()
 
@@ -632,9 +595,9 @@ def test_atomic_write_bucket_gone():
     key = bucket.new_key('foo')
     fp_a = FakeWriteFile(1024*1024, 'A', remove_bucket)
     e = assert_raises(boto.exception.S3ResponseError, key.set_contents_from_file, fp_a)
-    eq(e.status, 404)
-    eq(e.reason, 'Not Found')
-    eq(e.error_code, 'NoSuchBucket')
+    assert e.status == 404
+    assert e.reason == 'Not Found'
+    assert e.error_code == 'NoSuchBucket'
 
 def _multipart_upload_enc(bucket, s3_key_name, size, part_size=5*1024*1024,
                           do_list=None, init_headers=None, part_headers=None,
@@ -660,11 +623,8 @@ def _multipart_upload_enc(bucket, s3_key_name, size, part_size=5*1024*1024,
 
 
 
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='multipart upload with bad key for uploading chunks')
-@attr(assertion='successful')
-@attr('encryption')
+@pytest.mark.encryption
+@pytest.mark.fails_on_dbstore
 def test_encryption_sse_c_multipart_invalid_chunks_1():
     bucket = get_new_bucket()
     key = "multipart_enc"
@@ -685,13 +645,10 @@ def test_encryption_sse_c_multipart_invalid_chunks_1():
                       _multipart_upload_enc, bucket, key, objlen,
                       init_headers=init_headers, part_headers=part_headers,
                       metadata={'foo': 'bar'})
-    eq(e.status, 400)
+    assert e.status == 400
 
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='multipart upload with bad md5 for chunks')
-@attr(assertion='successful')
-@attr('encryption')
+@pytest.mark.encryption
+@pytest.mark.fails_on_dbstore
 def test_encryption_sse_c_multipart_invalid_chunks_2():
     bucket = get_new_bucket()
     key = "multipart_enc"
@@ -712,14 +669,11 @@ def test_encryption_sse_c_multipart_invalid_chunks_2():
                       _multipart_upload_enc, bucket, key, objlen,
                       init_headers=init_headers, part_headers=part_headers,
                       metadata={'foo': 'bar'})
-    eq(e.status, 400)
+    assert e.status == 400
 
-@attr(resource='bucket')
-@attr(method='get')
-@attr(operation='Test Bucket Policy for a user belonging to a different tenant')
-@attr(assertion='succeeds')
-@attr('fails_with_subdomain')
-@attr('bucket-policy')
+@pytest.mark.fails_with_subdomain
+@pytest.mark.bucket_policy
+@pytest.mark.fails_on_dbstore
 def test_bucket_policy_different_tenant():
     bucket = get_new_bucket()
     key = bucket.new_key('asdf')
@@ -754,10 +708,8 @@ def test_bucket_policy_different_tenant():
     b = new_conn.get_bucket(bucket_name)
     b.get_all_keys()
 
-@attr(resource='bucket')
-@attr(method='put')
-@attr(operation='Test put condition operator end with ifExists')
-@attr('bucket-policy')
+@pytest.mark.bucket_policy
+@pytest.mark.fails_on_dbstore
 def test_bucket_policy_set_condition_operator_end_with_IfExists():
     bucket = _create_keys(keys=['foo'])
     policy = '''{
@@ -776,73 +728,25 @@ def test_bucket_policy_set_condition_operator_end_with_IfExists():
       }
      ]
     }''' % bucket.name
-    eq(bucket.set_policy(policy), True)
+    assert bucket.set_policy(policy) == True
     res = _make_request('GET', bucket.name, bucket.get_key("foo"),
                         request_headers={'referer': 'http://www.example.com/'})
-    eq(res.status, 200)
+    assert res.status == 200
     res = _make_request('GET', bucket.name, bucket.get_key("foo"),
                         request_headers={'referer': 'http://www.example.com/index.html'})
-    eq(res.status, 200)
+    assert res.status == 200
     res = _make_request('GET', bucket.name, bucket.get_key("foo"))
-    eq(res.status, 200)
+    assert res.status == 200
     res = _make_request('GET', bucket.name, bucket.get_key("foo"),
                         request_headers={'referer': 'http://example.com'})
-    eq(res.status, 403)
+    assert res.status == 403
 
 def _make_arn_resource(path="*"):
     return "arn:aws:s3:::{}".format(path)
 
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='Deny put obj requests without encryption')
-@attr(assertion='success')
-@attr('encryption')
-@attr('bucket-policy')
-def test_bucket_policy_put_obj_enc():
-
-    bucket = get_new_bucket()
-
-    deny_incorrect_algo = {
-        "StringNotEquals": {
-          "s3:x-amz-server-side-encryption": "AES256"
-        }
-    }
-
-    deny_unencrypted_obj = {
-        "Null" : {
-          "s3:x-amz-server-side-encryption": "true"
-        }
-    }
-
-    p = Policy()
-    resource = _make_arn_resource("{}/{}".format(bucket.name, "*"))
-
-    s1 = Statement("s3:PutObject", resource, effect="Deny", condition=deny_incorrect_algo)
-    s2 = Statement("s3:PutObject", resource, effect="Deny", condition=deny_unencrypted_obj)
-    policy_document = p.add_statement(s1).add_statement(s2).to_json()
-
-    bucket.set_policy(policy_document)
-
-    key1_str ='testobj'
-    key1  = bucket.new_key(key1_str)
-    check_access_denied(key1.set_contents_from_string, key1_str)
-
-    sse_client_headers = {
-        'x-amz-server-side-encryption' : 'AES256',
-        'x-amz-server-side-encryption-customer-algorithm': 'AES256',
-        'x-amz-server-side-encryption-customer-key': 'pO3upElrwuEXSoFwCfnZPdSsmt/xWeFa0N9KgDijwVs=',
-        'x-amz-server-side-encryption-customer-key-md5': 'DWygnHRtgiJ77HCm+1rvHw=='
-    }
-
-
-    key1.set_contents_from_string(key1_str, headers=sse_client_headers)
-
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='put obj with RequestObjectTag')
-@attr(assertion='success')
-@attr('tagging')
-@attr('bucket-policy')
+@pytest.mark.tagging
+@pytest.mark.bucket_policy
+@pytest.mark.fails_on_dbstore
 def test_bucket_policy_put_obj_request_obj_tag():
 
     bucket = get_new_bucket()

s3tests/functional/test_s3_website.py

@@ -1,7 +1,6 @@
-
 import sys
-import collections
-import nose
+from collections.abc import Container
+import pytest
 import string
 import random
 from pprint import pprint
@@ -11,14 +10,11 @@ import socket
 
 from urllib.parse import urlparse
 
-from nose.tools import eq_ as eq, ok_ as ok
-from nose.plugins.attrib import attr
-from nose.tools import timed
-from nose.plugins.skip import SkipTest
-
 from .. import common
 
 from . import (
+    configfile,
+    setup_teardown,
     get_new_bucket,
     get_new_bucket_name,
     s3,
@@ -43,37 +39,26 @@ ERRORDOC_TEMPLATE = '<html><h1>ErrorDoc</h1><body>{random}</body></html>'
 
 CAN_WEBSITE = None
 
+@pytest.fixture(autouse=True, scope="module")
 def check_can_test_website():
-    global CAN_WEBSITE
-    # This is a bit expensive, so we cache this
-    if CAN_WEBSITE is None:
-        bucket = get_new_bucket()
-        try:
-            wsconf = bucket.get_website_configuration()
-            CAN_WEBSITE = True
-        except boto.exception.S3ResponseError as e:
-            if e.status == 404 and e.reason == 'Not Found' and e.error_code in ['NoSuchWebsiteConfiguration', 'NoSuchKey']:
-                CAN_WEBSITE = True
-            elif e.status == 405 and e.reason == 'Method Not Allowed' and e.error_code == 'MethodNotAllowed':
-                # rgw_enable_static_website is false
-                CAN_WEBSITE = False
-            elif e.status == 403 and e.reason == 'SignatureDoesNotMatch' and e.error_code == 'Forbidden':
-                # This is older versions that do not support the website code
-                CAN_WEBSITE = False
-            elif e.status == 501 and e.error_code == 'NotImplemented':
-                CAN_WEBSITE = False
-            else:
-                raise RuntimeError("Unknown response in checking if WebsiteConf is supported", e)
-        finally:
-            bucket.delete()
-
-    if CAN_WEBSITE is True:
+    bucket = get_new_bucket()
+    try:
+        wsconf = bucket.get_website_configuration()
         return True
-    elif CAN_WEBSITE is False:
-        raise SkipTest
-    else:
-        raise RuntimeError("Unknown cached response in checking if WebsiteConf is supported")
-
+    except boto.exception.S3ResponseError as e:
+        if e.status == 404 and e.reason == 'Not Found' and e.error_code in ['NoSuchWebsiteConfiguration', 'NoSuchKey']:
+            return True
+        elif e.status == 405 and e.reason == 'Method Not Allowed' and e.error_code == 'MethodNotAllowed':
+            pytest.skip('rgw_enable_static_website is false')
+        elif e.status == 403 and e.reason == 'SignatureDoesNotMatch' and e.error_code == 'Forbidden':
+            # This is older versions that do not support the website code
+            pytest.skip('static website is not implemented')
+        elif e.status == 501 and e.error_code == 'NotImplemented':
+            pytest.skip('static website is not implemented')
+        else:
+            raise RuntimeError("Unknown response in checking if WebsiteConf is supported", e)
+    finally:
+        bucket.delete()
 
 def make_website_config(xml_fragment):
     """
@@ -169,20 +154,20 @@ def _test_website_prep(bucket, xml_template, hardcoded_fields = {}, expect_fail=
     # Cleanup for our validation
     common.assert_xml_equal(config_xmlcmp, config_xmlnew)
     #print("config_xmlcmp\n", config_xmlcmp)
-    #eq (config_xmlnew, config_xmlcmp)
+    #assert config_xmlnew == config_xmlcmp
     f['WebsiteConfiguration'] = config_xmlcmp
     return f
 
 def __website_expected_reponse_status(res, status, reason):
-    if not isinstance(status, collections.Container):
+    if not isinstance(status, Container):
         status = set([status])
-    if not isinstance(reason, collections.Container):
+    if not isinstance(reason, Container):
         reason = set([reason])
 
     if status is not IGNORE_FIELD:
-        ok(res.status in status, 'HTTP code was %s should be %s' % (res.status, status))
+        assert res.status in status, 'HTTP code was %s should be %s' % (res.status, status)
     if reason is not IGNORE_FIELD:
-        ok(res.reason in reason, 'HTTP reason was was %s should be %s' % (res.reason, reason))
+        assert res.reason in reason, 'HTTP reason was was %s should be %s' % (res.reason, reason)
 
 def _website_expected_default_html(**kwargs):
     fields = []
@@ -194,7 +179,7 @@ def _website_expected_default_html(**kwargs):
         v = kwargs[k]
         if isinstance(v, str):
             v = [v]
-        elif not isinstance(v, collections.Container):
+        elif not isinstance(v, Container):
             v = [v]
         for v2 in v:
             s = '<li>%s: %s</li>' % (k,v2)
@@ -212,22 +197,22 @@ def _website_expected_error_response(res, bucket_name, status, reason, code, con
     errorcode = res.getheader('x-amz-error-code', None)
     if errorcode is not None:
         if code is not IGNORE_FIELD:
-            eq(errorcode, code)
+            assert errorcode == code
 
-    if not isinstance(content, collections.Container):
+    if not isinstance(content, Container):
         content = set([content])
     for f in content:
         if f is not IGNORE_FIELD and f is not None:
             f = bytes(f, 'utf-8')
-            ok(f in body, 'HTML should contain "%s"' % (f, ))
+            assert f in body, 'HTML should contain "%s"' % (f, )
 
 def _website_expected_redirect_response(res, status, reason, new_url):
     body = res.read()
     print(body)
     __website_expected_reponse_status(res, status, reason)
     loc = res.getheader('Location', None)
-    eq(loc, new_url, 'Location header should be set "%s" != "%s"' % (loc,new_url,))
-    ok(len(body) == 0, 'Body of a redirect should be empty')
+    assert loc == new_url, 'Location header should be set "%s" != "%s"' % (loc,new_url,)
+    assert len(body) == 0, 'Body of a redirect should be empty'
 
 def _website_request(bucket_name, path, connect_hostname=None, method='GET', timeout=None):
     url = get_website_url(proto='http', bucket=bucket_name, path=path)
@@ -246,26 +231,16 @@ def _website_request(bucket_name, path, connect_hostname=None, method='GET', tim
     return res
 
 # ---------- Non-existant buckets via the website endpoint
-@attr(resource='bucket')
-@attr(method='get')
-@attr(operation='list')
-@attr(assertion='non-existant bucket via website endpoint should give NoSuchBucket, exposing security risk')
-@attr('s3website')
-@attr('fails_on_rgw')
-@nose.with_setup(setup=check_can_test_website, teardown=common.teardown)
+@pytest.mark.s3website
+@pytest.mark.fails_on_rgw
 def test_website_nonexistant_bucket_s3():
     bucket_name = get_new_bucket_name()
     res = _website_request(bucket_name, '')
     _website_expected_error_response(res, bucket_name, 404, 'Not Found', 'NoSuchBucket', content=_website_expected_default_html(Code='NoSuchBucket'))
 
-@attr(resource='bucket')
-@attr(method='get')
-@attr(operation='list')
-#@attr(assertion='non-existant bucket via website endpoint should give Forbidden, keeping bucket identity secure')
-@attr(assertion='non-existant bucket via website endpoint should give NoSuchBucket')
-@attr('s3website')
-@attr('fails_on_s3')
-@nose.with_setup(setup=check_can_test_website, teardown=common.teardown)
+@pytest.mark.s3website
+@pytest.mark.fails_on_s3
+@pytest.mark.fails_on_dbstore
 def test_website_nonexistant_bucket_rgw():
     bucket_name = get_new_bucket_name()
     res = _website_request(bucket_name, '')
@@ -273,13 +248,9 @@ def test_website_nonexistant_bucket_rgw():
     _website_expected_error_response(res, bucket_name, 404, 'Not Found', 'NoSuchBucket', content=_website_expected_default_html(Code='NoSuchBucket'))
 
 #------------- IndexDocument only, successes
-@attr(resource='bucket')
-@attr(method='get')
-@attr(operation='list')
-@attr(assertion='non-empty public buckets via s3website return page for /, where page is public')
-@attr('s3website')
-@nose.with_setup(setup=check_can_test_website, teardown=common.teardown)
-@timed(10)
+@pytest.mark.s3website
+@pytest.mark.fails_on_dbstore
+@pytest.mark.timeout(10)
 def test_website_public_bucket_list_public_index():
     bucket = get_new_bucket()
     f = _test_website_prep(bucket, WEBSITE_CONFIGS_XMLFRAG['IndexDoc'])
@@ -296,17 +267,13 @@ def test_website_public_bucket_list_public_index():
     body = res.read()
     print(body)
     indexstring = bytes(indexstring, 'utf-8')
-    eq(body, indexstring) # default content should match index.html set content
+    assert body == indexstring # default content should match index.html set content
     __website_expected_reponse_status(res, 200, 'OK')
     indexhtml.delete()
     bucket.delete()
 
-@attr(resource='bucket')
-@attr(method='get')
-@attr(operation='list')
-@attr(assertion='non-empty private buckets via s3website return page for /, where page is private')
-@attr('s3website')
-@nose.with_setup(setup=check_can_test_website, teardown=common.teardown)
+@pytest.mark.s3website
+@pytest.mark.fails_on_dbstore
 def test_website_private_bucket_list_public_index():
     bucket = get_new_bucket()
     f = _test_website_prep(bucket, WEBSITE_CONFIGS_XMLFRAG['IndexDoc'])
@@ -325,18 +292,14 @@ def test_website_private_bucket_list_public_index():
     body = res.read()
     print(body)
     indexstring = bytes(indexstring, 'utf-8')
-    eq(body, indexstring, 'default content should match index.html set content')
+    assert body == indexstring, 'default content should match index.html set content'
     indexhtml.delete()
     bucket.delete()
 
 
 # ---------- IndexDocument only, failures
-@attr(resource='bucket')
-@attr(method='get')
-@attr(operation='list')
-@attr(assertion='empty private buckets via s3website return a 403 for /')
-@attr('s3website')
-@nose.with_setup(setup=check_can_test_website, teardown=common.teardown)
+@pytest.mark.s3website
+@pytest.mark.fails_on_dbstore
 def test_website_private_bucket_list_empty():
     bucket = get_new_bucket()
     f = _test_website_prep(bucket, WEBSITE_CONFIGS_XMLFRAG['IndexDoc'])
@@ -347,12 +310,8 @@ def test_website_private_bucket_list_empty():
     _website_expected_error_response(res, bucket.name, 403, 'Forbidden', 'AccessDenied', content=_website_expected_default_html(Code='AccessDenied'))
     bucket.delete()
 
-@attr(resource='bucket')
-@attr(method='get')
-@attr(operation='list')
-@attr(assertion='empty public buckets via s3website return a 404 for /')
-@attr('s3website')
-@nose.with_setup(setup=check_can_test_website, teardown=common.teardown)
+@pytest.mark.s3website
+@pytest.mark.fails_on_dbstore
 def test_website_public_bucket_list_empty():
     bucket = get_new_bucket()
     f = _test_website_prep(bucket, WEBSITE_CONFIGS_XMLFRAG['IndexDoc'])
@@ -362,12 +321,8 @@ def test_website_public_bucket_list_empty():
     _website_expected_error_response(res, bucket.name, 404, 'Not Found', 'NoSuchKey', content=_website_expected_default_html(Code='NoSuchKey'))
     bucket.delete()
 
-@attr(resource='bucket')
-@attr(method='get')
-@attr(operation='list')
-@attr(assertion='non-empty public buckets via s3website return page for /, where page is private')
-@attr('s3website')
-@nose.with_setup(setup=check_can_test_website, teardown=common.teardown)
+@pytest.mark.s3website
+@pytest.mark.fails_on_dbstore
 def test_website_public_bucket_list_private_index():
     bucket = get_new_bucket()
     f = _test_website_prep(bucket, WEBSITE_CONFIGS_XMLFRAG['IndexDoc'])
@@ -387,12 +342,8 @@ def test_website_public_bucket_list_private_index():
     indexhtml.delete()
     bucket.delete()
 
-@attr(resource='bucket')
-@attr(method='get')
-@attr(operation='list')
-@attr(assertion='non-empty private buckets via s3website return page for /, where page is private')
-@attr('s3website')
-@nose.with_setup(setup=check_can_test_website, teardown=common.teardown)
+@pytest.mark.s3website
+@pytest.mark.fails_on_dbstore
 def test_website_private_bucket_list_private_index():
     bucket = get_new_bucket()
     f = _test_website_prep(bucket, WEBSITE_CONFIGS_XMLFRAG['IndexDoc'])
@@ -413,12 +364,8 @@ def test_website_private_bucket_list_private_index():
     bucket.delete()
 
 # ---------- IndexDocument & ErrorDocument, failures due to errordoc assigned but missing
-@attr(resource='bucket')
-@attr(method='get')
-@attr(operation='list')
-@attr(assertion='empty private buckets via s3website return a 403 for /, missing errordoc')
-@attr('s3website')
-@nose.with_setup(setup=check_can_test_website, teardown=common.teardown)
+@pytest.mark.s3website
+@pytest.mark.fails_on_dbstore
 def test_website_private_bucket_list_empty_missingerrordoc():
     bucket = get_new_bucket()
     f = _test_website_prep(bucket, WEBSITE_CONFIGS_XMLFRAG['IndexDocErrorDoc'])
@@ -429,12 +376,8 @@ def test_website_private_bucket_list_empty_missingerrordoc():
 
     bucket.delete()
 
-@attr(resource='bucket')
-@attr(method='get')
-@attr(operation='list')
-@attr(assertion='empty public buckets via s3website return a 404 for /, missing errordoc')
-@attr('s3website')
-@nose.with_setup(setup=check_can_test_website, teardown=common.teardown)
+@pytest.mark.s3website
+@pytest.mark.fails_on_dbstore
 def test_website_public_bucket_list_empty_missingerrordoc():
     bucket = get_new_bucket()
     f = _test_website_prep(bucket, WEBSITE_CONFIGS_XMLFRAG['IndexDocErrorDoc'])
@@ -444,12 +387,8 @@ def test_website_public_bucket_list_empty_missingerrordoc():
     _website_expected_error_response(res, bucket.name, 404, 'Not Found', 'NoSuchKey')
     bucket.delete()
 
-@attr(resource='bucket')
-@attr(method='get')
-@attr(operation='list')
-@attr(assertion='non-empty public buckets via s3website return page for /, where page is private, missing errordoc')
-@attr('s3website')
-@nose.with_setup(setup=check_can_test_website, teardown=common.teardown)
+@pytest.mark.s3website
+@pytest.mark.fails_on_dbstore
 def test_website_public_bucket_list_private_index_missingerrordoc():
     bucket = get_new_bucket()
     f = _test_website_prep(bucket, WEBSITE_CONFIGS_XMLFRAG['IndexDocErrorDoc'])
@@ -468,12 +407,8 @@ def test_website_public_bucket_list_private_index_missingerrordoc():
     indexhtml.delete()
     bucket.delete()
 
-@attr(resource='bucket')
-@attr(method='get')
-@attr(operation='list')
-@attr(assertion='non-empty private buckets via s3website return page for /, where page is private, missing errordoc')
-@attr('s3website')
-@nose.with_setup(setup=check_can_test_website, teardown=common.teardown)
+@pytest.mark.s3website
+@pytest.mark.fails_on_dbstore
 def test_website_private_bucket_list_private_index_missingerrordoc():
     bucket = get_new_bucket()
     f = _test_website_prep(bucket, WEBSITE_CONFIGS_XMLFRAG['IndexDocErrorDoc'])
@@ -493,12 +428,8 @@ def test_website_private_bucket_list_private_index_missingerrordoc():
     bucket.delete()
 
 # ---------- IndexDocument & ErrorDocument, failures due to errordoc assigned but not accessible
-@attr(resource='bucket')
-@attr(method='get')
-@attr(operation='list')
-@attr(assertion='empty private buckets via s3website return a 403 for /, blocked errordoc')
-@attr('s3website')
-@nose.with_setup(setup=check_can_test_website, teardown=common.teardown)
+@pytest.mark.s3website
+@pytest.mark.fails_on_dbstore
 def test_website_private_bucket_list_empty_blockederrordoc():
     bucket = get_new_bucket()
     f = _test_website_prep(bucket, WEBSITE_CONFIGS_XMLFRAG['IndexDocErrorDoc'])
@@ -516,17 +447,13 @@ def test_website_private_bucket_list_empty_blockederrordoc():
     print(body)
     _website_expected_error_response(res, bucket.name, 403, 'Forbidden', 'AccessDenied', content=_website_expected_default_html(Code='AccessDenied'), body=body)
     errorstring = bytes(errorstring, 'utf-8')
-    ok(errorstring not in body, 'error content should NOT match error.html set content')
+    assert errorstring not in body, 'error content should NOT match error.html set content'
 
     errorhtml.delete()
     bucket.delete()
 
-@attr(resource='bucket')
-@attr(method='get')
-@attr(operation='list')
-@attr(assertion='check if there is an invalid payload after serving error doc')
-@attr('s3website')
-@nose.with_setup(setup=check_can_test_website, teardown=common.teardown)
+@pytest.mark.s3website
+@pytest.mark.fails_on_dbstore
 def test_website_public_bucket_list_pubilc_errordoc():
     bucket = get_new_bucket()
     f = _test_website_prep(bucket, WEBSITE_CONFIGS_XMLFRAG['IndexDocErrorDoc'])
@@ -567,17 +494,13 @@ def test_website_public_bucket_list_pubilc_errordoc():
     except socket.timeout:
         print('no invalid payload')
 
-    ok(resp_len == 0, 'invalid payload')
+    assert resp_len == 0, 'invalid payload'
 
     errorhtml.delete()
     bucket.delete()
 
-@attr(resource='bucket')
-@attr(method='get')
-@attr(operation='list')
-@attr(assertion='empty public buckets via s3website return a 404 for /, blocked errordoc')
-@attr('s3website')
-@nose.with_setup(setup=check_can_test_website, teardown=common.teardown)
+@pytest.mark.s3website
+@pytest.mark.fails_on_dbstore
 def test_website_public_bucket_list_empty_blockederrordoc():
     bucket = get_new_bucket()
     f = _test_website_prep(bucket, WEBSITE_CONFIGS_XMLFRAG['IndexDocErrorDoc'])
@@ -594,17 +517,13 @@ def test_website_public_bucket_list_empty_blockederrordoc():
     print(body)
     _website_expected_error_response(res, bucket.name, 404, 'Not Found', 'NoSuchKey', content=_website_expected_default_html(Code='NoSuchKey'), body=body)
     errorstring = bytes(errorstring, 'utf-8')
-    ok(errorstring not in body, 'error content should match error.html set content')
+    assert errorstring not in body, 'error content should match error.html set content'
 
     errorhtml.delete()
     bucket.delete()
 
-@attr(resource='bucket')
-@attr(method='get')
-@attr(operation='list')
-@attr(assertion='non-empty public buckets via s3website return page for /, where page is private, blocked errordoc')
-@attr('s3website')
-@nose.with_setup(setup=check_can_test_website, teardown=common.teardown)
+@pytest.mark.s3website
+@pytest.mark.fails_on_dbstore
 def test_website_public_bucket_list_private_index_blockederrordoc():
     bucket = get_new_bucket()
     f = _test_website_prep(bucket, WEBSITE_CONFIGS_XMLFRAG['IndexDocErrorDoc'])
@@ -626,18 +545,14 @@ def test_website_public_bucket_list_private_index_blockederrordoc():
     print(body)
     _website_expected_error_response(res, bucket.name, 403, 'Forbidden', 'AccessDenied', content=_website_expected_default_html(Code='AccessDenied'), body=body)
     errorstring = bytes(errorstring, 'utf-8')
-    ok(errorstring not in body, 'error content should match error.html set content')
+    assert errorstring not in body, 'error content should match error.html set content'
 
     indexhtml.delete()
     errorhtml.delete()
     bucket.delete()
 
-@attr(resource='bucket')
-@attr(method='get')
-@attr(operation='list')
-@attr(assertion='non-empty private buckets via s3website return page for /, where page is private, blocked errordoc')
-@attr('s3website')
-@nose.with_setup(setup=check_can_test_website, teardown=common.teardown)
+@pytest.mark.s3website
+@pytest.mark.fails_on_dbstore
 def test_website_private_bucket_list_private_index_blockederrordoc():
     bucket = get_new_bucket()
     f = _test_website_prep(bucket, WEBSITE_CONFIGS_XMLFRAG['IndexDocErrorDoc'])
@@ -659,19 +574,15 @@ def test_website_private_bucket_list_private_index_blockederrordoc():
     print(body)
     _website_expected_error_response(res, bucket.name, 403, 'Forbidden', 'AccessDenied', content=_website_expected_default_html(Code='AccessDenied'), body=body)
     errorstring = bytes(errorstring, 'utf-8')
-    ok(errorstring not in body, 'error content should match error.html set content')
+    assert errorstring not in body, 'error content should match error.html set content'
 
     indexhtml.delete()
     errorhtml.delete()
     bucket.delete()
 
 # ---------- IndexDocument & ErrorDocument, failures with errordoc available
-@attr(resource='bucket')
-@attr(method='get')
-@attr(operation='list')
-@attr(assertion='empty private buckets via s3website return a 403 for /, good errordoc')
-@attr('s3website')
-@nose.with_setup(setup=check_can_test_website, teardown=common.teardown)
+@pytest.mark.s3website
+@pytest.mark.fails_on_dbstore
 def test_website_private_bucket_list_empty_gooderrordoc():
     bucket = get_new_bucket()
     f = _test_website_prep(bucket, WEBSITE_CONFIGS_XMLFRAG['IndexDocErrorDoc'])
@@ -689,12 +600,8 @@ def test_website_private_bucket_list_empty_gooderrordoc():
     errorhtml.delete()
     bucket.delete()
 
-@attr(resource='bucket')
-@attr(method='get')
-@attr(operation='list')
-@attr(assertion='empty public buckets via s3website return a 404 for /, good errordoc')
-@attr('s3website')
-@nose.with_setup(setup=check_can_test_website, teardown=common.teardown)
+@pytest.mark.s3website
+@pytest.mark.fails_on_dbstore
 def test_website_public_bucket_list_empty_gooderrordoc():
     bucket = get_new_bucket()
     f = _test_website_prep(bucket, WEBSITE_CONFIGS_XMLFRAG['IndexDocErrorDoc'])
@@ -713,12 +620,8 @@ def test_website_public_bucket_list_empty_gooderrordoc():
     errorhtml.delete()
     bucket.delete()
 
-@attr(resource='bucket')
-@attr(method='get')
-@attr(operation='list')
-@attr(assertion='non-empty public buckets via s3website return page for /, where page is private')
-@attr('s3website')
-@nose.with_setup(setup=check_can_test_website, teardown=common.teardown)
+@pytest.mark.s3website
+@pytest.mark.fails_on_dbstore
 def test_website_public_bucket_list_private_index_gooderrordoc():
     bucket = get_new_bucket()
     f = _test_website_prep(bucket, WEBSITE_CONFIGS_XMLFRAG['IndexDocErrorDoc'])
@@ -742,12 +645,8 @@ def test_website_public_bucket_list_private_index_gooderrordoc():
     errorhtml.delete()
     bucket.delete()
 
-@attr(resource='bucket')
-@attr(method='get')
-@attr(operation='list')
-@attr(assertion='non-empty private buckets via s3website return page for /, where page is private')
-@attr('s3website')
-@nose.with_setup(setup=check_can_test_website, teardown=common.teardown)
+@pytest.mark.s3website
+@pytest.mark.fails_on_dbstore
 def test_website_private_bucket_list_private_index_gooderrordoc():
     bucket = get_new_bucket()
     f = _test_website_prep(bucket, WEBSITE_CONFIGS_XMLFRAG['IndexDocErrorDoc'])
@@ -772,12 +671,8 @@ def test_website_private_bucket_list_private_index_gooderrordoc():
     bucket.delete()
 
 # ------ RedirectAll tests
-@attr(resource='bucket')
-@attr(method='get')
-@attr(operation='list')
-@attr(assertion='RedirectAllRequestsTo without protocol should TODO')
-@attr('s3website')
-@nose.with_setup(setup=check_can_test_website, teardown=common.teardown)
+@pytest.mark.s3website
+@pytest.mark.fails_on_dbstore
 def test_website_bucket_private_redirectall_base():
     bucket = get_new_bucket()
     f = _test_website_prep(bucket, WEBSITE_CONFIGS_XMLFRAG['RedirectAll'])
@@ -789,12 +684,8 @@ def test_website_bucket_private_redirectall_base():
 
     bucket.delete()
 
-@attr(resource='bucket')
-@attr(method='get')
-@attr(operation='list')
-@attr(assertion='RedirectAllRequestsTo without protocol should TODO')
-@attr('s3website')
-@nose.with_setup(setup=check_can_test_website, teardown=common.teardown)
+@pytest.mark.s3website
+@pytest.mark.fails_on_dbstore
 def test_website_bucket_private_redirectall_path():
     bucket = get_new_bucket()
     f = _test_website_prep(bucket, WEBSITE_CONFIGS_XMLFRAG['RedirectAll'])
@@ -808,12 +699,8 @@ def test_website_bucket_private_redirectall_path():
 
     bucket.delete()
 
-@attr(resource='bucket')
-@attr(method='get')
-@attr(operation='list')
-@attr(assertion='RedirectAllRequestsTo without protocol should TODO')
-@attr('s3website')
-@nose.with_setup(setup=check_can_test_website, teardown=common.teardown)
+@pytest.mark.s3website
+@pytest.mark.fails_on_dbstore
 def test_website_bucket_private_redirectall_path_upgrade():
     bucket = get_new_bucket()
     x = string.Template(WEBSITE_CONFIGS_XMLFRAG['RedirectAll+Protocol']).safe_substitute(RedirectAllRequestsTo_Protocol='https')
@@ -829,13 +716,9 @@ def test_website_bucket_private_redirectall_path_upgrade():
     bucket.delete()
 
 # ------ x-amz redirect tests
-@attr(resource='bucket')
-@attr(method='get')
-@attr(operation='list')
-@attr(assertion='x-amz-website-redirect-location should not fire without websiteconf')
-@attr('s3website')
-@attr('x-amz-website-redirect-location')
-@nose.with_setup(setup=check_can_test_website, teardown=common.teardown)
+@pytest.mark.s3website
+@pytest.mark.s3website_redirect_location
+@pytest.mark.fails_on_dbstore
 def test_website_xredirect_nonwebsite():
     bucket = get_new_bucket()
     #f = _test_website_prep(bucket, WEBSITE_CONFIGS_XMLFRAG['RedirectAll'])
@@ -847,7 +730,7 @@ def test_website_xredirect_nonwebsite():
     headers = {'x-amz-website-redirect-location': redirect_dest}
     k.set_contents_from_string(content, headers=headers, policy='public-read')
     redirect = k.get_redirect()
-    eq(k.get_redirect(), redirect_dest)
+    assert k.get_redirect() == redirect_dest
 
     res = _website_request(bucket.name, '/page')
     body = res.read()
@@ -861,13 +744,9 @@ def test_website_xredirect_nonwebsite():
     k.delete()
     bucket.delete()
 
-@attr(resource='bucket')
-@attr(method='get')
-@attr(operation='list')
-@attr(assertion='x-amz-website-redirect-location should fire websiteconf, relative path, public key')
-@attr('s3website')
-@attr('x-amz-website-redirect-location')
-@nose.with_setup(setup=check_can_test_website, teardown=common.teardown)
+@pytest.mark.s3website
+@pytest.mark.s3website_redirect_location
+@pytest.mark.fails_on_dbstore
 def test_website_xredirect_public_relative():
     bucket = get_new_bucket()
     f = _test_website_prep(bucket, WEBSITE_CONFIGS_XMLFRAG['IndexDoc'])
@@ -879,7 +758,7 @@ def test_website_xredirect_public_relative():
     headers = {'x-amz-website-redirect-location': redirect_dest}
     k.set_contents_from_string(content, headers=headers, policy='public-read')
     redirect = k.get_redirect()
-    eq(k.get_redirect(), redirect_dest)
+    assert k.get_redirect() == redirect_dest
 
     res = _website_request(bucket.name, '/page')
     #new_url =  get_website_url(bucket_name=bucket.name, path=redirect_dest)
@@ -888,13 +767,9 @@ def test_website_xredirect_public_relative():
     k.delete()
     bucket.delete()
 
-@attr(resource='bucket')
-@attr(method='get')
-@attr(operation='list')
-@attr(assertion='x-amz-website-redirect-location should fire websiteconf, absolute, public key')
-@attr('s3website')
-@attr('x-amz-website-redirect-location')
-@nose.with_setup(setup=check_can_test_website, teardown=common.teardown)
+@pytest.mark.s3website
+@pytest.mark.s3website_redirect_location
+@pytest.mark.fails_on_dbstore
 def test_website_xredirect_public_abs():
     bucket = get_new_bucket()
     f = _test_website_prep(bucket, WEBSITE_CONFIGS_XMLFRAG['IndexDoc'])
@@ -906,7 +781,7 @@ def test_website_xredirect_public_abs():
     headers = {'x-amz-website-redirect-location': redirect_dest}
     k.set_contents_from_string(content, headers=headers, policy='public-read')
     redirect = k.get_redirect()
-    eq(k.get_redirect(), redirect_dest)
+    assert k.get_redirect() == redirect_dest
 
     res = _website_request(bucket.name, '/page')
     new_url =  get_website_url(proto='http', hostname='example.com', path='/foo')
@@ -915,13 +790,9 @@ def test_website_xredirect_public_abs():
     k.delete()
     bucket.delete()
 
-@attr(resource='bucket')
-@attr(method='get')
-@attr(operation='list')
-@attr(assertion='x-amz-website-redirect-location should fire websiteconf, relative path, private key')
-@attr('s3website')
-@attr('x-amz-website-redirect-location')
-@nose.with_setup(setup=check_can_test_website, teardown=common.teardown)
+@pytest.mark.s3website
+@pytest.mark.s3website_redirect_location
+@pytest.mark.fails_on_dbstore
 def test_website_xredirect_private_relative():
     bucket = get_new_bucket()
     f = _test_website_prep(bucket, WEBSITE_CONFIGS_XMLFRAG['IndexDoc'])
@@ -933,7 +804,7 @@ def test_website_xredirect_private_relative():
     headers = {'x-amz-website-redirect-location': redirect_dest}
     k.set_contents_from_string(content, headers=headers, policy='private')
     redirect = k.get_redirect()
-    eq(k.get_redirect(), redirect_dest)
+    assert k.get_redirect() == redirect_dest
 
     res = _website_request(bucket.name, '/page')
     # We get a 403 because the page is private
@@ -942,13 +813,9 @@ def test_website_xredirect_private_relative():
     k.delete()
     bucket.delete()
 
-@attr(resource='bucket')
-@attr(method='get')
-@attr(operation='list')
-@attr(assertion='x-amz-website-redirect-location should fire websiteconf, absolute, private key')
-@attr('s3website')
-@attr('x-amz-website-redirect-location')
-@nose.with_setup(setup=check_can_test_website, teardown=common.teardown)
+@pytest.mark.s3website
+@pytest.mark.s3website_redirect_location
+@pytest.mark.fails_on_dbstore
 def test_website_xredirect_private_abs():
     bucket = get_new_bucket()
     f = _test_website_prep(bucket, WEBSITE_CONFIGS_XMLFRAG['IndexDoc'])
@@ -960,7 +827,7 @@ def test_website_xredirect_private_abs():
     headers = {'x-amz-website-redirect-location': redirect_dest}
     k.set_contents_from_string(content, headers=headers, policy='private')
     redirect = k.get_redirect()
-    eq(k.get_redirect(), redirect_dest)
+    assert k.get_redirect() == redirect_dest
 
     res = _website_request(bucket.name, '/page')
     new_url =  get_website_url(proto='http', hostname='example.com', path='/foo')
@@ -1173,8 +1040,6 @@ def routing_teardown(**kwargs):
     print('Deleting', str(o))
     o.delete()
 
-@common.with_setup_kwargs(setup=routing_setup, teardown=routing_teardown)
-#@timed(10)
 def routing_check(*args, **kwargs):
     bucket = kwargs['bucket']
     args=args[0]
@@ -1200,8 +1065,8 @@ def routing_check(*args, **kwargs):
     if args['code'] >= 200 and args['code'] < 300:
         #body = res.read()
         #print(body)
-        #eq(body, args['content'], 'default content should match index.html set content')
-        ok(int(res.getheader('Content-Length', -1)) > 0)
+        #assert body == args['content'], 'default content should match index.html set content'
+        assert int(res.getheader('Content-Length', -1)) > 0
     elif args['code'] >= 300 and args['code'] < 400:
         _website_expected_redirect_response(res, args['code'], IGNORE_FIELD, new_url)
     elif args['code'] >= 400:
@@ -1209,9 +1074,9 @@ def routing_check(*args, **kwargs):
     else:
         assert(False)
 
-@attr('s3website_RoutingRules')
-@attr('s3website')
-@nose.with_setup(setup=check_can_test_website, teardown=common.teardown)
+@pytest.mark.s3website_routing_rules
+@pytest.mark.s3website
+@pytest.mark.fails_on_dbstore
 def test_routing_generator():
     for t in ROUTING_RULES_TESTS:
         if 'xml' in t and 'RoutingRules' in t['xml'] and len(t['xml']['RoutingRules']) > 0:

s3tests/functional/test_utils.py

@@ -1,11 +1,9 @@
-from nose.tools import eq_ as eq
-
 from . import utils
 
 def test_generate():
     FIVE_MB = 5 * 1024 * 1024
-    eq(len(''.join(utils.generate_random(0))), 0)
-    eq(len(''.join(utils.generate_random(1))), 1)
-    eq(len(''.join(utils.generate_random(FIVE_MB - 1))), FIVE_MB - 1)
-    eq(len(''.join(utils.generate_random(FIVE_MB))), FIVE_MB)
-    eq(len(''.join(utils.generate_random(FIVE_MB + 1))), FIVE_MB + 1)
+    assert len(''.join(utils.generate_random(0))) == 0
+    assert len(''.join(utils.generate_random(1))) == 1
+    assert len(''.join(utils.generate_random(FIVE_MB - 1))) == FIVE_MB - 1
+    assert len(''.join(utils.generate_random(FIVE_MB))) == FIVE_MB
+    assert len(''.join(utils.generate_random(FIVE_MB + 1))) == FIVE_MB + 1

s3tests/functional/utils.py

@@ -3,8 +3,6 @@ import requests
 import string
 import time
 
-from nose.tools import eq_ as eq
-
 def assert_raises(excClass, callableObj, *args, **kwargs):
     """
     Like unittest.TestCase.assertRaises, but returns the exception.
@@ -48,7 +46,7 @@ def region_sync_meta(targets, region):
         conf = r.conf
         if conf.sync_agent_addr:
             ret = requests.post('http://{addr}:{port}/metadata/incremental'.format(addr = conf.sync_agent_addr, port = conf.sync_agent_port))
-            eq(ret.status_code, 200)
+            assert ret.status_code == 200
         if conf.sync_meta_wait:
             time.sleep(conf.sync_meta_wait)
 

s3tests_boto3/functional/__init__.py

@@ -1,3 +1,4 @@
+import pytest
 import boto3
 from botocore import UNSIGNED
 from botocore.client import Config
@@ -81,18 +82,13 @@ def get_objects_list(bucket, client=None, prefix=None):
 # generator function that returns object listings in batches, where each
 # batch is a list of dicts compatible with delete_objects()
 def list_versions(client, bucket, batch_size):
-    key_marker = ''
-    version_marker = ''
+    kwargs = {'Bucket': bucket, 'MaxKeys': batch_size}
     truncated = True
     while truncated:
-        listing = client.list_object_versions(
-                Bucket=bucket,
-                KeyMarker=key_marker,
-                VersionIdMarker=version_marker,
-                MaxKeys=batch_size)
+        listing = client.list_object_versions(**kwargs)
 
-        key_marker = listing.get('NextKeyMarker')
-        version_marker = listing.get('NextVersionIdMarker')
+        kwargs['KeyMarker'] = listing.get('NextKeyMarker')
+        kwargs['VersionIdMarker'] = listing.get('NextVersionIdMarker')
         truncated = listing['IsTruncated']
 
         objs = listing.get('Versions', []) + listing.get('DeleteMarkers', [])
@@ -178,7 +174,7 @@ def configured_storage_classes():
 
     return sc
 
-def setup():
+def configure():
     cfg = configparser.RawConfigParser()
     try:
         path = os.environ['S3TEST_CONF']
@@ -263,25 +259,43 @@ def setup():
     config.tenant_display_name = cfg.get('s3 tenant',"display_name")
     config.tenant_user_id = cfg.get('s3 tenant',"user_id")
     config.tenant_email = cfg.get('s3 tenant',"email")
+    config.tenant_name = cfg.get('s3 tenant',"tenant")
+
+    config.iam_access_key = cfg.get('iam',"access_key")
+    config.iam_secret_key = cfg.get('iam',"secret_key")
+    config.iam_display_name = cfg.get('iam',"display_name")
+    config.iam_user_id = cfg.get('iam',"user_id")
+    config.iam_email = cfg.get('iam',"email")
+
+    config.iam_root_access_key = cfg.get('iam root',"access_key")
+    config.iam_root_secret_key = cfg.get('iam root',"secret_key")
+    config.iam_root_user_id = cfg.get('iam root',"user_id")
+    config.iam_root_email = cfg.get('iam root',"email")
+
+    config.iam_alt_root_access_key = cfg.get('iam alt root',"access_key")
+    config.iam_alt_root_secret_key = cfg.get('iam alt root',"secret_key")
+    config.iam_alt_root_user_id = cfg.get('iam alt root',"user_id")
+    config.iam_alt_root_email = cfg.get('iam alt root',"email")
 
     # vars from the fixtures section
-    try:
-        template = cfg.get('fixtures', "bucket prefix")
-    except (configparser.NoOptionError):
-        template = 'test-{random}-'
+    template = cfg.get('fixtures', "bucket prefix", fallback='test-{random}-')
     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)
+    template = cfg.get('fixtures', "iam name prefix", fallback="s3-tests-")
+    config.iam_name_prefix = choose_bucket_prefix(template=template)
+    template = cfg.get('fixtures', "iam path prefix", fallback="/s3-tests/")
+    config.iam_path_prefix = choose_bucket_prefix(template=template)
 
     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()
@@ -308,6 +322,17 @@ def teardown():
     except:
         pass
 
+@pytest.fixture(scope="package")
+def configfile():
+    configure()
+    return config
+
+@pytest.fixture(autouse=True)
+def setup_teardown(configfile):
+    setup()
+    yield
+    teardown()
+
 def check_webidentity():
     cfg = configparser.RawConfigParser()
     try:
@@ -389,52 +414,65 @@ def get_v2_client():
                         config=Config(signature_version='s3'))
     return client
 
-def get_sts_client(client_config=None):
-    if client_config == None:
-        client_config = Config(signature_version='s3v4')
+def get_sts_client(**kwargs):
+    kwargs.setdefault('aws_access_key_id', config.alt_access_key)
+    kwargs.setdefault('aws_secret_access_key', config.alt_secret_key)
+    kwargs.setdefault('config', Config(signature_version='s3v4'))
 
     client = boto3.client(service_name='sts',
-                        aws_access_key_id=config.alt_access_key,
-                        aws_secret_access_key=config.alt_secret_key,
-                        endpoint_url=config.default_endpoint,
-                        region_name='',
-                        use_ssl=config.default_is_secure,
-                        verify=config.default_ssl_verify,
-                        config=client_config)
+                          endpoint_url=config.default_endpoint,
+                          region_name='',
+                          use_ssl=config.default_is_secure,
+                          verify=config.default_ssl_verify,
+                          **kwargs)
     return client
 
-def get_iam_client(client_config=None):
-    cfg = configparser.RawConfigParser()
-    try:
-        path = os.environ['S3TEST_CONF']
-    except KeyError:
-        raise RuntimeError(
-            'To run tests, point environment '
-            + 'variable S3TEST_CONF to a config file.',
-            )
-    cfg.read(path)
-    if not cfg.has_section("iam"):
-        raise RuntimeError('Your config file is missing the "iam" section!')
+def get_iam_client(**kwargs):
+    kwargs.setdefault('aws_access_key_id', config.iam_access_key)
+    kwargs.setdefault('aws_secret_access_key', config.iam_secret_key)
 
-    config.iam_access_key = cfg.get('iam',"access_key")
-    config.iam_secret_key = cfg.get('iam',"secret_key")
-    config.iam_display_name = cfg.get('iam',"display_name")
-    config.iam_user_id = cfg.get('iam',"user_id")
-    config.iam_email = cfg.get('iam',"email")    
-
-    if client_config == None:
-        client_config = Config(signature_version='s3v4')
-    
     client = boto3.client(service_name='iam',
-                        aws_access_key_id=config.iam_access_key,
-                        aws_secret_access_key=config.iam_secret_key,
                         endpoint_url=config.default_endpoint,
                         region_name='',
                         use_ssl=config.default_is_secure,
                         verify=config.default_ssl_verify,
-                        config=client_config)
+                        **kwargs)
     return client
 
+def get_iam_s3client(**kwargs):
+    kwargs.setdefault('aws_access_key_id', config.iam_access_key)
+    kwargs.setdefault('aws_secret_access_key', config.iam_secret_key)
+    kwargs.setdefault('config', Config(signature_version='s3v4'))
+
+    client = boto3.client(service_name='s3',
+                          endpoint_url=config.default_endpoint,
+                          use_ssl=config.default_is_secure,
+                          verify=config.default_ssl_verify,
+                          **kwargs)
+    return client
+
+def get_iam_root_client(**kwargs):
+    kwargs.setdefault('service_name', 'iam')
+    kwargs.setdefault('aws_access_key_id', config.iam_root_access_key)
+    kwargs.setdefault('aws_secret_access_key', config.iam_root_secret_key)
+
+    return boto3.client(endpoint_url=config.default_endpoint,
+                        region_name='',
+                        use_ssl=config.default_is_secure,
+                        verify=config.default_ssl_verify,
+                        **kwargs)
+
+def get_iam_alt_root_client(**kwargs):
+    kwargs.setdefault('service_name', 'iam')
+    kwargs.setdefault('aws_access_key_id', config.iam_alt_root_access_key)
+    kwargs.setdefault('aws_secret_access_key', config.iam_alt_root_secret_key)
+
+    return boto3.client(endpoint_url=config.default_endpoint,
+                        region_name='',
+                        use_ssl=config.default_is_secure,
+                        verify=config.default_ssl_verify,
+                        **kwargs)
+
 def get_alt_client(client_config=None):
     if client_config == None:
         client_config = Config(signature_version='s3v4')
@@ -473,6 +511,17 @@ def get_tenant_client(client_config=None):
                         config=client_config)
     return client
 
+def get_v2_tenant_client():
+    client_config = Config(signature_version='s3')
+    client = boto3.client(service_name='s3',
+                          aws_access_key_id=config.tenant_access_key,
+                          aws_secret_access_key=config.tenant_secret_key,
+                          endpoint_url=config.default_endpoint,
+                          use_ssl=config.default_is_secure,
+                          verify=config.default_ssl_verify,
+                          config=client_config)
+    return client
+
 def get_tenant_iam_client():
 
     client = boto3.client(service_name='iam',
@@ -484,6 +533,17 @@ def get_tenant_iam_client():
                           use_ssl=config.default_is_secure)
     return client
 
+def get_alt_iam_client():
+
+    client = boto3.client(service_name='iam',
+                          region_name='',
+                          aws_access_key_id=config.alt_access_key,
+                          aws_secret_access_key=config.alt_secret_key,
+                          endpoint_url=config.default_endpoint,
+                          verify=config.default_ssl_verify,
+                          use_ssl=config.default_is_secure)
+    return client
+
 def get_unauthenticated_client():
     client = boto3.client(service_name='s3',
                         aws_access_key_id='',
@@ -646,6 +706,9 @@ def get_tenant_aws_secret_key():
 def get_tenant_display_name():
     return config.tenant_display_name
 
+def get_tenant_name():
+    return config.tenant_name
+
 def get_tenant_user_id():
     return config.tenant_user_id
 
@@ -670,12 +733,33 @@ def get_token():
 def get_realm_name():
     return config.webidentity_realm
 
+def get_iam_name_prefix():
+    return config.iam_name_prefix
+
+def make_iam_name(name):
+    return config.iam_name_prefix + name
+
+def get_iam_path_prefix():
+    return config.iam_path_prefix
+
 def get_iam_access_key():
     return config.iam_access_key
 
 def get_iam_secret_key():
     return config.iam_secret_key
 
+def get_iam_root_user_id():
+    return config.iam_root_user_id
+
+def get_iam_root_email():
+    return config.iam_root_email
+
+def get_iam_alt_root_user_id():
+    return config.iam_alt_root_user_id
+
+def get_iam_alt_root_email():
+    return config.iam_alt_root_email
+
 def get_user_token():
     return config.webidentity_user_token
 

s3tests_boto3/functional/iam.py

@@ -0,0 +1,199 @@
+from botocore.exceptions import ClientError
+import pytest
+
+from . import (
+    configfile,
+    get_iam_root_client,
+    get_iam_root_user_id,
+    get_iam_root_email,
+    get_iam_alt_root_client,
+    get_iam_alt_root_user_id,
+    get_iam_alt_root_email,
+    get_iam_path_prefix,
+)
+
+def nuke_user_keys(client, name):
+    p = client.get_paginator('list_access_keys')
+    for response in p.paginate(UserName=name):
+        for key in response['AccessKeyMetadata']:
+            try:
+                client.delete_access_key(UserName=name, AccessKeyId=key['AccessKeyId'])
+            except:
+                pass
+
+def nuke_user_policies(client, name):
+    p = client.get_paginator('list_user_policies')
+    for response in p.paginate(UserName=name):
+        for policy in response['PolicyNames']:
+            try:
+                client.delete_user_policy(UserName=name, PolicyName=policy)
+            except:
+                pass
+
+def nuke_attached_user_policies(client, name):
+    p = client.get_paginator('list_attached_user_policies')
+    for response in p.paginate(UserName=name):
+        for policy in response['AttachedPolicies']:
+            try:
+                client.detach_user_policy(UserName=name, PolicyArn=policy['PolicyArn'])
+            except:
+                pass
+
+def nuke_user(client, name):
+    # delete access keys, user policies, etc
+    try:
+        nuke_user_keys(client, name)
+    except:
+        pass
+    try:
+        nuke_user_policies(client, name)
+    except:
+        pass
+    try:
+        nuke_attached_user_policies(client, name)
+    except:
+        pass
+    client.delete_user(UserName=name)
+
+def nuke_users(client, **kwargs):
+    p = client.get_paginator('list_users')
+    for response in p.paginate(**kwargs):
+        for user in response['Users']:
+            try:
+                nuke_user(client, user['UserName'])
+            except:
+                pass
+
+def nuke_group_policies(client, name):
+    p = client.get_paginator('list_group_policies')
+    for response in p.paginate(GroupName=name):
+        for policy in response['PolicyNames']:
+            try:
+                client.delete_group_policy(GroupName=name, PolicyName=policy)
+            except:
+                pass
+
+def nuke_attached_group_policies(client, name):
+    p = client.get_paginator('list_attached_group_policies')
+    for response in p.paginate(GroupName=name):
+        for policy in response['AttachedPolicies']:
+            try:
+                client.detach_group_policy(GroupName=name, PolicyArn=policy['PolicyArn'])
+            except:
+                pass
+
+def nuke_group_users(client, name):
+    p = client.get_paginator('get_group')
+    for response in p.paginate(GroupName=name):
+        for user in response['Users']:
+            try:
+                client.remove_user_from_group(GroupName=name, UserName=user['UserName'])
+            except:
+                pass
+
+def nuke_group(client, name):
+    # delete group policies and remove all users
+    try:
+        nuke_group_policies(client, name)
+    except:
+        pass
+    try:
+        nuke_attached_group_policies(client, name)
+    except:
+        pass
+    try:
+        nuke_group_users(client, name)
+    except:
+        pass
+    client.delete_group(GroupName=name)
+
+def nuke_groups(client, **kwargs):
+    p = client.get_paginator('list_groups')
+    for response in p.paginate(**kwargs):
+        for user in response['Groups']:
+            try:
+                nuke_group(client, user['GroupName'])
+            except:
+                pass
+
+def nuke_role_policies(client, name):
+    p = client.get_paginator('list_role_policies')
+    for response in p.paginate(RoleName=name):
+        for policy in response['PolicyNames']:
+            try:
+                client.delete_role_policy(RoleName=name, PolicyName=policy)
+            except:
+                pass
+
+def nuke_attached_role_policies(client, name):
+    p = client.get_paginator('list_attached_role_policies')
+    for response in p.paginate(RoleName=name):
+        for policy in response['AttachedPolicies']:
+            try:
+                client.detach_role_policy(RoleName=name, PolicyArn=policy['PolicyArn'])
+            except:
+                pass
+
+def nuke_role(client, name):
+    # delete role policies, etc
+    try:
+        nuke_role_policies(client, name)
+    except:
+        pass
+    try:
+        nuke_attached_role_policies(client, name)
+    except:
+        pass
+    client.delete_role(RoleName=name)
+
+def nuke_roles(client, **kwargs):
+    p = client.get_paginator('list_roles')
+    for response in p.paginate(**kwargs):
+        for role in response['Roles']:
+            try:
+                nuke_role(client, role['RoleName'])
+            except:
+                pass
+
+def nuke_oidc_providers(client, prefix):
+    result = client.list_open_id_connect_providers()
+    for provider in result['OpenIDConnectProviderList']:
+        arn = provider['Arn']
+        if f':oidc-provider{prefix}' in arn:
+            try:
+                client.delete_open_id_connect_provider(OpenIDConnectProviderArn=arn)
+            except:
+                pass
+
+
+# fixture for iam account root user
+@pytest.fixture
+def iam_root(configfile):
+    client = get_iam_root_client()
+    try:
+        arn = client.get_user()['User']['Arn']
+        if not arn.endswith(':root'):
+            pytest.skip('[iam root] user does not have :root arn')
+    except ClientError as e:
+        pytest.skip('[iam root] user does not belong to an account')
+
+    yield client
+    nuke_users(client, PathPrefix=get_iam_path_prefix())
+    nuke_groups(client, PathPrefix=get_iam_path_prefix())
+    nuke_roles(client, PathPrefix=get_iam_path_prefix())
+    nuke_oidc_providers(client, get_iam_path_prefix())
+
+# fixture for iam alt account root user
+@pytest.fixture
+def iam_alt_root(configfile):
+    client = get_iam_alt_root_client()
+    try:
+        arn = client.get_user()['User']['Arn']
+        if not arn.endswith(':root'):
+            pytest.skip('[iam alt root] user does not have :root arn')
+    except ClientError as e:
+        pytest.skip('[iam alt root] user does not belong to an account')
+
+    yield client
+    nuke_users(client, PathPrefix=get_iam_path_prefix())
+    nuke_roles(client, PathPrefix=get_iam_path_prefix())

s3tests_boto3/functional/policy.py

@@ -37,10 +37,10 @@ class Policy(object):
 
         return json.dumps(policy_dict)
 
-def make_json_policy(action, resource, principal={"AWS": "*"}, conditions=None):
+def make_json_policy(action, resource, principal={"AWS": "*"}, effect="Allow", conditions=None):
     """
     Helper function to make single statement policies
     """
-    s = Statement(action, resource, principal, condition=conditions)
+    s = Statement(action, resource, principal, effect=effect, condition=conditions)
     p = Policy()
     return p.add_statement(s).to_json()

s3tests_boto3/functional/test_headers.py

@@ -1,7 +1,5 @@
 import boto3
-from nose.tools import eq_ as eq
-from nose.plugins.attrib import attr
-import nose
+import pytest
 from botocore.exceptions import ClientError
 from email.utils import formatdate
 
@@ -10,6 +8,8 @@ from .utils import _get_status_and_error_code
 from .utils import _get_status
 
 from . import (
+    configfile,
+    setup_teardown,
     get_client,
     get_v2_client,
     get_new_bucket,
@@ -149,178 +149,97 @@ def _remove_header_create_bad_bucket(remove, client=None):
 
     return e
 
-def tag(*tags):
-    def wrap(func):
-        for tag in tags:
-            setattr(func, tag, True)
-        return func
-    return wrap
-
 #
 # common tests
 #
 
-@tag('auth_common')
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='create w/invalid MD5')
-@attr(assertion='fails 400')
+@pytest.mark.auth_common
 def test_object_create_bad_md5_invalid_short():
     e = _add_header_create_bad_object({'Content-MD5':'YWJyYWNhZGFicmE='})
     status, error_code = _get_status_and_error_code(e.response)
-    eq(status, 400)
-    eq(error_code, 'InvalidDigest')
+    assert status == 400
+    assert error_code == 'InvalidDigest'
 
-@tag('auth_common')
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='create w/mismatched MD5')
-@attr(assertion='fails 400')
+@pytest.mark.auth_common
 def test_object_create_bad_md5_bad():
     e = _add_header_create_bad_object({'Content-MD5':'rL0Y20xC+Fzt72VPzMSk2A=='})
     status, error_code = _get_status_and_error_code(e.response)
-    eq(status, 400)
-    eq(error_code, 'BadDigest')
+    assert status == 400
+    assert error_code == 'BadDigest'
 
-@tag('auth_common')
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='create w/empty MD5')
-@attr(assertion='fails 400')
+@pytest.mark.auth_common
 def test_object_create_bad_md5_empty():
     e = _add_header_create_bad_object({'Content-MD5':''})
     status, error_code = _get_status_and_error_code(e.response)
-    eq(status, 400)
-    eq(error_code, 'InvalidDigest')
+    assert status == 400
+    assert error_code == 'InvalidDigest'
 
-@tag('auth_common')
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='create w/no MD5 header')
-@attr(assertion='succeeds')
+@pytest.mark.auth_common
 def test_object_create_bad_md5_none():
     bucket_name, key_name = _remove_header_create_object('Content-MD5')
     client = get_client()
     client.put_object(Bucket=bucket_name, Key=key_name, Body='bar')
 
-@tag('auth_common')
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='create w/Expect 200')
-@attr(assertion='garbage, but S3 succeeds!')
+@pytest.mark.auth_common
 def test_object_create_bad_expect_mismatch():
     bucket_name, key_name = _add_header_create_object({'Expect': 200})
     client = get_client()
     client.put_object(Bucket=bucket_name, Key=key_name, Body='bar')
 
-@tag('auth_common')
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='create w/empty expect')
-@attr(assertion='succeeds ... should it?')
+@pytest.mark.auth_common
 def test_object_create_bad_expect_empty():
     bucket_name, key_name = _add_header_create_object({'Expect': ''})
     client = get_client()
     client.put_object(Bucket=bucket_name, Key=key_name, Body='bar')
 
-@tag('auth_common')
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='create w/no expect')
-@attr(assertion='succeeds')
+@pytest.mark.auth_common
 def test_object_create_bad_expect_none():
     bucket_name, key_name = _remove_header_create_object('Expect')
     client = get_client()
     client.put_object(Bucket=bucket_name, Key=key_name, Body='bar')
 
-@tag('auth_common')
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='create w/empty content length')
-@attr(assertion='fails 400')
+@pytest.mark.auth_common
 # TODO: remove 'fails_on_rgw' and once we have learned how to remove the content-length header
-@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
 def test_object_create_bad_contentlength_empty():
     e = _add_header_create_bad_object({'Content-Length':''})
     status, error_code = _get_status_and_error_code(e.response)
-    eq(status, 400)
+    assert status == 400
 
-@tag('auth_common')
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='create w/negative content length')
-@attr(assertion='fails 400')
-@attr('fails_on_mod_proxy_fcgi')
+@pytest.mark.auth_common
+@pytest.mark.fails_on_mod_proxy_fcgi
 def test_object_create_bad_contentlength_negative():
     client = get_client()
     bucket_name = get_new_bucket()
     key_name = 'foo'
     e = assert_raises(ClientError, client.put_object, Bucket=bucket_name, Key=key_name, ContentLength=-1)
     status = _get_status(e.response)
-    eq(status, 400)
+    assert status == 400
 
-@tag('auth_common')
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='create w/no content length')
-@attr(assertion='fails 411')
+@pytest.mark.auth_common
 # TODO: remove 'fails_on_rgw' and once we have learned how to remove the content-length header
-@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
 def test_object_create_bad_contentlength_none():
     remove = 'Content-Length'
     e = _remove_header_create_bad_object('Content-Length')
     status, error_code = _get_status_and_error_code(e.response)
-    eq(status, 411)
-    eq(error_code, 'MissingContentLength')
+    assert status == 411
+    assert error_code == 'MissingContentLength'
 
-@tag('auth_common')
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='create w/content length too long')
-@attr(assertion='fails 400')
-# TODO: remove 'fails_on_rgw' and once we have learned how to remove the content-length header
-@attr('fails_on_rgw')
-def test_object_create_bad_contentlength_mismatch_above():
-    content = 'bar'
-    length = len(content) + 1
-
-    client = get_client()
-    bucket_name = get_new_bucket()
-    key_name = 'foo'
-    headers = {'Content-Length': str(length)}
-    add_headers = (lambda **kwargs: kwargs['params']['headers'].update(headers))
-    client.meta.events.register('before-sign.s3.PutObject', add_headers)
-
-    e = assert_raises(ClientError, client.put_object, Bucket=bucket_name, Key=key_name, Body=content)
-    status, error_code = _get_status_and_error_code(e.response)
-    eq(status, 400)
-
-@tag('auth_common')
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='create w/content type text/plain')
-@attr(assertion='succeeds')
+@pytest.mark.auth_common
 def test_object_create_bad_contenttype_invalid():
     bucket_name, key_name = _add_header_create_object({'Content-Type': 'text/plain'})
     client = get_client()
     client.put_object(Bucket=bucket_name, Key=key_name, Body='bar')
 
-@tag('auth_common')
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='create w/empty content type')
-@attr(assertion='succeeds')
+@pytest.mark.auth_common
 def test_object_create_bad_contenttype_empty():
     client = get_client()
     key_name = 'foo'
     bucket_name = get_new_bucket()
     client.put_object(Bucket=bucket_name, Key=key_name, Body='bar', ContentType='')
 
-@tag('auth_common')
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='create w/no content type')
-@attr(assertion='succeeds')
+@pytest.mark.auth_common
 def test_object_create_bad_contenttype_none():
     bucket_name = get_new_bucket()
     key_name = 'foo'
@@ -329,38 +248,26 @@ def test_object_create_bad_contenttype_none():
     client.put_object(Bucket=bucket_name, Key=key_name, Body='bar')
 
 
-@tag('auth_common')
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='create w/empty authorization')
-@attr(assertion='fails 403')
+@pytest.mark.auth_common
 # TODO: remove 'fails_on_rgw' and once we have learned how to remove the authorization header
-@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
 def test_object_create_bad_authorization_empty():
     e = _add_header_create_bad_object({'Authorization': ''})
     status, error_code = _get_status_and_error_code(e.response)
-    eq(status, 403)
+    assert status == 403
 
-@tag('auth_common')
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='create w/date and x-amz-date')
-@attr(assertion='succeeds')
+@pytest.mark.auth_common
 # TODO: remove 'fails_on_rgw' and once we have learned how to pass both the 'Date' and 'X-Amz-Date' header during signing and not 'X-Amz-Date' before
-@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
 def test_object_create_date_and_amz_date():
     date = formatdate(usegmt=True)
     bucket_name, key_name = _add_header_create_object({'Date': date, 'X-Amz-Date': date})
     client = get_client()
     client.put_object(Bucket=bucket_name, Key=key_name, Body='bar')
 
-@tag('auth_common')
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='create w/x-amz-date and no date')
-@attr(assertion='succeeds')
+@pytest.mark.auth_common
 # TODO: remove 'fails_on_rgw' and once we have learned how to pass both the 'Date' and 'X-Amz-Date' header during signing and not 'X-Amz-Date' before
-@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
 def test_object_create_amz_date_and_no_date():
     date = formatdate(usegmt=True)
     bucket_name, key_name = _add_header_create_object({'Date': '', 'X-Amz-Date': date})
@@ -368,36 +275,24 @@ def test_object_create_amz_date_and_no_date():
     client.put_object(Bucket=bucket_name, Key=key_name, Body='bar')
 
 # the teardown is really messed up here. check it out
-@tag('auth_common')
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='create w/no authorization')
-@attr(assertion='fails 403')
+@pytest.mark.auth_common
 # TODO: remove 'fails_on_rgw' and once we have learned how to remove the authorization header
-@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
 def test_object_create_bad_authorization_none():
     e = _remove_header_create_bad_object('Authorization')
     status, error_code = _get_status_and_error_code(e.response)
-    eq(status, 403)
+    assert status == 403
 
-@tag('auth_common')
-@attr(resource='bucket')
-@attr(method='put')
-@attr(operation='create w/no content length')
-@attr(assertion='succeeds')
+@pytest.mark.auth_common
 # TODO: remove 'fails_on_rgw' and once we have learned how to remove the content-length header
-@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
 def test_bucket_create_contentlength_none():
     remove = 'Content-Length'
     _remove_header_create_bucket(remove)
 
-@tag('auth_common')
-@attr(resource='bucket')
-@attr(method='acls')
-@attr(operation='set w/no content length')
-@attr(assertion='succeeds')
+@pytest.mark.auth_common
 # TODO: remove 'fails_on_rgw' and once we have learned how to remove the content-length header
-@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
 def test_object_acl_create_contentlength_none():
     bucket_name = get_new_bucket()
     client = get_client()
@@ -411,11 +306,7 @@ def test_object_acl_create_contentlength_none():
     client.meta.events.register('before-call.s3.PutObjectAcl', remove_header)
     client.put_object_acl(Bucket=bucket_name, Key='foo', ACL='public-read')
 
-@tag('auth_common')
-@attr(resource='bucket')
-@attr(method='acls')
-@attr(operation='set w/invalid permission')
-@attr(assertion='fails 400')
+@pytest.mark.auth_common
 def test_bucket_put_bad_canned_acl():
     bucket_name = get_new_bucket()
     client = get_client()
@@ -426,13 +317,9 @@ def test_bucket_put_bad_canned_acl():
 
     e = assert_raises(ClientError, client.put_bucket_acl, Bucket=bucket_name, ACL='public-read')
     status = _get_status(e.response)
-    eq(status, 400)
+    assert status == 400
 
-@tag('auth_common')
-@attr(resource='bucket')
-@attr(method='put')
-@attr(operation='create w/expect 200')
-@attr(assertion='garbage, but S3 succeeds!')
+@pytest.mark.auth_common
 def test_bucket_create_bad_expect_mismatch():
     bucket_name = get_new_bucket_name()
     client = get_client()
@@ -442,99 +329,67 @@ def test_bucket_create_bad_expect_mismatch():
     client.meta.events.register('before-call.s3.CreateBucket', add_headers)
     client.create_bucket(Bucket=bucket_name)
 
-@tag('auth_common')
-@attr(resource='bucket')
-@attr(method='put')
-@attr(operation='create w/expect empty')
-@attr(assertion='garbage, but S3 succeeds!')
+@pytest.mark.auth_common
 def test_bucket_create_bad_expect_empty():
     headers = {'Expect': ''}
     _add_header_create_bucket(headers)
 
-@tag('auth_common')
-@attr(resource='bucket')
-@attr(method='put')
-@attr(operation='create w/empty content length')
-@attr(assertion='fails 400')
+@pytest.mark.auth_common
 # TODO: The request isn't even making it to the RGW past the frontend
 # This test had 'fails_on_rgw' before the move to boto3
-@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
 def test_bucket_create_bad_contentlength_empty():
     headers = {'Content-Length': ''}
     e = _add_header_create_bad_bucket(headers)
     status, error_code = _get_status_and_error_code(e.response)
-    eq(status, 400)
+    assert status == 400
 
-@tag('auth_common')
-@attr(resource='bucket')
-@attr(method='put')
-@attr(operation='create w/negative content length')
-@attr(assertion='fails 400')
-@attr('fails_on_mod_proxy_fcgi')
+@pytest.mark.auth_common
+@pytest.mark.fails_on_mod_proxy_fcgi
 def test_bucket_create_bad_contentlength_negative():
     headers = {'Content-Length': '-1'}
     e = _add_header_create_bad_bucket(headers)
     status = _get_status(e.response)
-    eq(status, 400)
+    assert status == 400
 
-@tag('auth_common')
-@attr(resource='bucket')
-@attr(method='put')
-@attr(operation='create w/no content length')
-@attr(assertion='succeeds')
+@pytest.mark.auth_common
 # TODO: remove 'fails_on_rgw' and once we have learned how to remove the content-length header
-@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
 def test_bucket_create_bad_contentlength_none():
     remove = 'Content-Length'
     _remove_header_create_bucket(remove)
 
-@tag('auth_common')
-@attr(resource='bucket')
-@attr(method='put')
-@attr(operation='create w/empty authorization')
-@attr(assertion='fails 403')
+@pytest.mark.auth_common
 # TODO: remove 'fails_on_rgw' and once we have learned how to manipulate the authorization header
-@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
 def test_bucket_create_bad_authorization_empty():
     headers = {'Authorization': ''}
     e = _add_header_create_bad_bucket(headers)
     status, error_code = _get_status_and_error_code(e.response)
-    eq(status, 403)
-    eq(error_code, 'AccessDenied')
+    assert status == 403
+    assert error_code == 'AccessDenied'
 
-@tag('auth_common')
-@attr(resource='bucket')
-@attr(method='put')
-@attr(operation='create w/no authorization')
-@attr(assertion='fails 403')
+@pytest.mark.auth_common
 # TODO: remove 'fails_on_rgw' and once we have learned how to manipulate the authorization header
-@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
 def test_bucket_create_bad_authorization_none():
     e = _remove_header_create_bad_bucket('Authorization')
     status, error_code = _get_status_and_error_code(e.response)
-    eq(status, 403)
-    eq(error_code, 'AccessDenied')
+    assert status == 403
+    assert error_code == 'AccessDenied'
 
-@tag('auth_aws2')
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='create w/invalid MD5')
-@attr(assertion='fails 400')
+@pytest.mark.auth_aws2
 def test_object_create_bad_md5_invalid_garbage_aws2():
     v2_client = get_v2_client()
     headers = {'Content-MD5': 'AWS HAHAHA'}
     e = _add_header_create_bad_object(headers, v2_client)
     status, error_code = _get_status_and_error_code(e.response)
-    eq(status, 400)
-    eq(error_code, 'InvalidDigest')
+    assert status == 400
+    assert error_code == 'InvalidDigest'
 
-@tag('auth_aws2')
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='create w/content length too short')
-@attr(assertion='fails 400')
+@pytest.mark.auth_aws2
 # TODO: remove 'fails_on_rgw' and once we have learned how to manipulate the Content-Length header
-@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
 def test_object_create_bad_contentlength_mismatch_below_aws2():
     v2_client = get_v2_client()
     content = 'bar'
@@ -542,252 +397,176 @@ def test_object_create_bad_contentlength_mismatch_below_aws2():
     headers = {'Content-Length': str(length)}
     e = _add_header_create_bad_object(headers, v2_client)
     status, error_code = _get_status_and_error_code(e.response)
-    eq(status, 400)
-    eq(error_code, 'BadDigest')
+    assert status == 400
+    assert error_code == 'BadDigest'
 
-@tag('auth_aws2')
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='create w/incorrect authorization')
-@attr(assertion='fails 403')
+@pytest.mark.auth_aws2
 # TODO: remove 'fails_on_rgw' and once we have learned how to manipulate the authorization header
-@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
 def test_object_create_bad_authorization_incorrect_aws2():
     v2_client = get_v2_client()
     headers = {'Authorization': 'AWS AKIAIGR7ZNNBHC5BKSUB:FWeDfwojDSdS2Ztmpfeubhd9isU='}
     e = _add_header_create_bad_object(headers, v2_client)
     status, error_code = _get_status_and_error_code(e.response)
-    eq(status, 403)
-    eq(error_code, 'InvalidDigest')
+    assert status == 403
+    assert error_code == 'InvalidDigest'
 
-@tag('auth_aws2')
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='create w/invalid authorization')
-@attr(assertion='fails 400')
+@pytest.mark.auth_aws2
 # TODO: remove 'fails_on_rgw' and once we have learned how to manipulate the authorization header
-@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
 def test_object_create_bad_authorization_invalid_aws2():
     v2_client = get_v2_client()
     headers = {'Authorization': 'AWS HAHAHA'}
     e = _add_header_create_bad_object(headers, v2_client)
     status, error_code = _get_status_and_error_code(e.response)
-    eq(status, 400)
-    eq(error_code, 'InvalidArgument')
+    assert status == 400
+    assert error_code == 'InvalidArgument'
 
-@tag('auth_aws2')
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='create w/empty user agent')
-@attr(assertion='succeeds')
+@pytest.mark.auth_aws2
 def test_object_create_bad_ua_empty_aws2():
     v2_client = get_v2_client()
     headers = {'User-Agent': ''}
     bucket_name, key_name = _add_header_create_object(headers, v2_client)
     v2_client.put_object(Bucket=bucket_name, Key=key_name, Body='bar')
 
-@tag('auth_aws2')
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='create w/no user agent')
-@attr(assertion='succeeds')
+@pytest.mark.auth_aws2
 def test_object_create_bad_ua_none_aws2():
     v2_client = get_v2_client()
     remove = 'User-Agent'
     bucket_name, key_name = _remove_header_create_object(remove, v2_client)
     v2_client.put_object(Bucket=bucket_name, Key=key_name, Body='bar')
 
-@tag('auth_aws2')
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='create w/invalid date')
-@attr(assertion='fails 403')
+@pytest.mark.auth_aws2
 def test_object_create_bad_date_invalid_aws2():
     v2_client = get_v2_client()
     headers = {'x-amz-date': 'Bad Date'}
     e = _add_header_create_bad_object(headers, v2_client)
     status, error_code = _get_status_and_error_code(e.response)
-    eq(status, 403)
-    eq(error_code, 'AccessDenied')
+    assert status == 403
+    assert error_code == 'AccessDenied'
 
-@tag('auth_aws2')
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='create w/empty date')
-@attr(assertion='fails 403')
+@pytest.mark.auth_aws2
 def test_object_create_bad_date_empty_aws2():
     v2_client = get_v2_client()
     headers = {'x-amz-date': ''}
     e = _add_header_create_bad_object(headers, v2_client)
     status, error_code = _get_status_and_error_code(e.response)
-    eq(status, 403)
-    eq(error_code, 'AccessDenied')
+    assert status == 403
+    assert error_code == 'AccessDenied'
 
-@tag('auth_aws2')
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='create w/no date')
-@attr(assertion='fails 403')
+@pytest.mark.auth_aws2
 # TODO: remove 'fails_on_rgw' and once we have learned how to remove the date header
-@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
 def test_object_create_bad_date_none_aws2():
     v2_client = get_v2_client()
     remove = 'x-amz-date'
     e = _remove_header_create_bad_object(remove, v2_client)
     status, error_code = _get_status_and_error_code(e.response)
-    eq(status, 403)
-    eq(error_code, 'AccessDenied')
+    assert status == 403
+    assert error_code == 'AccessDenied'
 
-@tag('auth_aws2')
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='create w/date in past')
-@attr(assertion='fails 403')
+@pytest.mark.auth_aws2
 def test_object_create_bad_date_before_today_aws2():
     v2_client = get_v2_client()
     headers = {'x-amz-date': 'Tue, 07 Jul 2010 21:53:04 GMT'}
     e = _add_header_create_bad_object(headers, v2_client)
     status, error_code = _get_status_and_error_code(e.response)
-    eq(status, 403)
-    eq(error_code, 'RequestTimeTooSkewed')
+    assert status == 403
+    assert error_code == 'RequestTimeTooSkewed'
 
-@tag('auth_aws2')
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='create w/date before epoch')
-@attr(assertion='fails 403')
+@pytest.mark.auth_aws2
 def test_object_create_bad_date_before_epoch_aws2():
     v2_client = get_v2_client()
     headers = {'x-amz-date': 'Tue, 07 Jul 1950 21:53:04 GMT'}
     e = _add_header_create_bad_object(headers, v2_client)
     status, error_code = _get_status_and_error_code(e.response)
-    eq(status, 403)
-    eq(error_code, 'AccessDenied')
+    assert status == 403
+    assert error_code == 'AccessDenied'
 
-@tag('auth_aws2')
-@attr(resource='object')
-@attr(method='put')
-@attr(operation='create w/date after 9999')
-@attr(assertion='fails 403')
+@pytest.mark.auth_aws2
 def test_object_create_bad_date_after_end_aws2():
     v2_client = get_v2_client()
     headers = {'x-amz-date': 'Tue, 07 Jul 9999 21:53:04 GMT'}
     e = _add_header_create_bad_object(headers, v2_client)
     status, error_code = _get_status_and_error_code(e.response)
-    eq(status, 403)
-    eq(error_code, 'RequestTimeTooSkewed')
+    assert status == 403
+    assert error_code == 'RequestTimeTooSkewed'
 
-@tag('auth_aws2')
-@attr(resource='bucket')
-@attr(method='put')
-@attr(operation='create w/invalid authorization')
-@attr(assertion='fails 400')
+@pytest.mark.auth_aws2
 # TODO: remove 'fails_on_rgw' and once we have learned how to remove the date header
-@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
 def test_bucket_create_bad_authorization_invalid_aws2():
     v2_client = get_v2_client()
     headers = {'Authorization': 'AWS HAHAHA'}
     e = _add_header_create_bad_bucket(headers, v2_client)
     status, error_code = _get_status_and_error_code(e.response)
-    eq(status, 400)
-    eq(error_code, 'InvalidArgument')
+    assert status == 400
+    assert error_code == 'InvalidArgument'
 
-@tag('auth_aws2')
-@attr(resource='bucket')
-@attr(method='put')
-@attr(operation='create w/empty user agent')
-@attr(assertion='succeeds')
+@pytest.mark.auth_aws2
 def test_bucket_create_bad_ua_empty_aws2():
     v2_client = get_v2_client()
     headers = {'User-Agent': ''}
     _add_header_create_bucket(headers, v2_client)
 
-@tag('auth_aws2')
-@attr(resource='bucket')
-@attr(method='put')
-@attr(operation='create w/no user agent')
-@attr(assertion='succeeds')
+@pytest.mark.auth_aws2
 def test_bucket_create_bad_ua_none_aws2():
     v2_client = get_v2_client()
     remove = 'User-Agent'
     _remove_header_create_bucket(remove, v2_client)
 
-@tag('auth_aws2')
-@attr(resource='bucket')
-@attr(method='put')
-@attr(operation='create w/invalid date')
-@attr(assertion='fails 403')
+@pytest.mark.auth_aws2
 def test_bucket_create_bad_date_invalid_aws2():
     v2_client = get_v2_client()
     headers = {'x-amz-date': 'Bad Date'}
     e = _add_header_create_bad_bucket(headers, v2_client)
     status, error_code = _get_status_and_error_code(e.response)
-    eq(status, 403)
-    eq(error_code, 'AccessDenied')
+    assert status == 403
+    assert error_code == 'AccessDenied'
 
-@tag('auth_aws2')
-@attr(resource='bucket')
-@attr(method='put')
-@attr(operation='create w/empty date')
-@attr(assertion='fails 403')
+@pytest.mark.auth_aws2
 def test_bucket_create_bad_date_empty_aws2():
     v2_client = get_v2_client()
     headers = {'x-amz-date': ''}
     e = _add_header_create_bad_bucket(headers, v2_client)
     status, error_code = _get_status_and_error_code(e.response)
-    eq(status, 403)
-    eq(error_code, 'AccessDenied')
+    assert status == 403
+    assert error_code == 'AccessDenied'
 
-@tag('auth_aws2')
-@attr(resource='bucket')
-@attr(method='put')
-@attr(operation='create w/no date')
-@attr(assertion='fails 403')
+@pytest.mark.auth_aws2
 # TODO: remove 'fails_on_rgw' and once we have learned how to remove the date header
-@attr('fails_on_rgw')
+@pytest.mark.fails_on_rgw
 def test_bucket_create_bad_date_none_aws2():
     v2_client = get_v2_client()
     remove = 'x-amz-date'
     e = _remove_header_create_bad_bucket(remove, v2_client)
     status, error_code = _get_status_and_error_code(e.response)
-    eq(status, 403)
-    eq(error_code, 'AccessDenied')
+    assert status == 403
+    assert error_code == 'AccessDenied'
 
-@tag('auth_aws2')
-@attr(resource='bucket')
-@attr(method='put')
-@attr(operation='create w/date in past')
-@attr(assertion='fails 403')
+@pytest.mark.auth_aws2
 def test_bucket_create_bad_date_before_today_aws2():
     v2_client = get_v2_client()
     headers = {'x-amz-date': 'Tue, 07 Jul 2010 21:53:04 GMT'}
     e = _add_header_create_bad_bucket(headers, v2_client)
     status, error_code = _get_status_and_error_code(e.response)
-    eq(status, 403)
-    eq(error_code, 'RequestTimeTooSkewed')
+    assert status == 403
+    assert error_code == 'RequestTimeTooSkewed'
 
-@tag('auth_aws2')
-@attr(resource='bucket')
-@attr(method='put')
-@attr(operation='create w/date in future')
-@attr(assertion='fails 403')
+@pytest.mark.auth_aws2
 def test_bucket_create_bad_date_after_today_aws2():
     v2_client = get_v2_client()
     headers = {'x-amz-date': 'Tue, 07 Jul 2030 21:53:04 GMT'}
     e = _add_header_create_bad_bucket(headers, v2_client)
     status, error_code = _get_status_and_error_code(e.response)
-    eq(status, 403)
-    eq(error_code, 'RequestTimeTooSkewed')
+    assert status == 403
+    assert error_code == 'RequestTimeTooSkewed'
 
-@tag('auth_aws2')
-@attr(resource='bucket')
-@attr(method='put')
-@attr(operation='create w/date before epoch')
-@attr(assertion='fails 403')
+@pytest.mark.auth_aws2
 def test_bucket_create_bad_date_before_epoch_aws2():
     v2_client = get_v2_client()
     headers = {'x-amz-date': 'Tue, 07 Jul 1950 21:53:04 GMT'}
     e = _add_header_create_bad_bucket(headers, v2_client)
     status, error_code = _get_status_and_error_code(e.response)
-    eq(status, 403)
-    eq(error_code, 'AccessDenied')
+    assert status == 403
+    assert error_code == 'AccessDenied'

s3tests_boto3/functional/test_sns.py

@@ -0,0 +1,159 @@
+import json
+import pytest
+from botocore.exceptions import ClientError
+from . import (
+    configfile,
+    get_iam_root_client,
+    get_iam_alt_root_client,
+    get_new_bucket_name,
+    get_prefix,
+    nuke_prefixed_buckets,
+)
+from .iam import iam_root, iam_alt_root
+from .utils import assert_raises, _get_status_and_error_code
+
+def get_new_topic_name():
+    return get_new_bucket_name()
+
+def nuke_topics(client, prefix):
+    p = client.get_paginator('list_topics')
+    for response in p.paginate():
+        for topic in response['Topics']:
+            arn = topic['TopicArn']
+            if prefix not in arn:
+                pass
+            try:
+                client.delete_topic(TopicArn=arn)
+            except:
+                pass
+
+@pytest.fixture
+def sns(iam_root):
+    client = get_iam_root_client(service_name='sns')
+    yield client
+    nuke_topics(client, get_prefix())
+
+@pytest.fixture
+def sns_alt(iam_alt_root):
+    client = get_iam_alt_root_client(service_name='sns')
+    yield client
+    nuke_topics(client, get_prefix())
+
+@pytest.fixture
+def s3(iam_root):
+    client = get_iam_root_client(service_name='s3')
+    yield client
+    nuke_prefixed_buckets(get_prefix(), client)
+
+@pytest.fixture
+def s3_alt(iam_alt_root):
+    client = get_iam_alt_root_client(service_name='s3')
+    yield client
+    nuke_prefixed_buckets(get_prefix(), client)
+
+
+@pytest.mark.iam_account
+@pytest.mark.sns
+def test_account_topic(sns):
+    name = get_new_topic_name()
+
+    response = sns.create_topic(Name=name)
+    arn = response['TopicArn']
+    assert arn.startswith('arn:aws:sns:')
+    assert arn.endswith(f':{name}')
+
+    response = sns.list_topics()
+    assert arn in [p['TopicArn'] for p in response['Topics']]
+
+    sns.set_topic_attributes(TopicArn=arn, AttributeName='Policy', AttributeValue='')
+
+    response = sns.get_topic_attributes(TopicArn=arn)
+    assert 'Attributes' in response
+
+    sns.delete_topic(TopicArn=arn)
+
+    response = sns.list_topics()
+    assert arn not in [p['TopicArn'] for p in response['Topics']]
+
+    with pytest.raises(sns.exceptions.NotFoundException):
+        sns.get_topic_attributes(TopicArn=arn)
+    sns.delete_topic(TopicArn=arn)
+
+@pytest.mark.iam_account
+@pytest.mark.sns
+def test_cross_account_topic(sns, sns_alt):
+    name = get_new_topic_name()
+    arn = sns.create_topic(Name=name)['TopicArn']
+
+    # not visible to any alt user apis
+    with pytest.raises(sns.exceptions.NotFoundException):
+        sns_alt.get_topic_attributes(TopicArn=arn)
+    with pytest.raises(sns.exceptions.NotFoundException):
+        sns_alt.set_topic_attributes(TopicArn=arn, AttributeName='Policy', AttributeValue='')
+
+    # delete returns success
+    sns_alt.delete_topic(TopicArn=arn)
+
+    response = sns_alt.list_topics()
+    assert arn not in [p['TopicArn'] for p in response['Topics']]
+
+@pytest.mark.iam_account
+@pytest.mark.sns
+def test_account_topic_publish(sns, s3):
+    name = get_new_topic_name()
+
+    response = sns.create_topic(Name=name)
+    topic_arn = response['TopicArn']
+
+    bucket = get_new_bucket_name()
+    s3.create_bucket(Bucket=bucket)
+
+    config = {'TopicConfigurations': [{
+        'Id': 'id',
+        'TopicArn': topic_arn,
+        'Events': [ 's3:ObjectCreated:*' ],
+        }]}
+    s3.put_bucket_notification_configuration(
+            Bucket=bucket, NotificationConfiguration=config)
+
+@pytest.mark.iam_account
+@pytest.mark.iam_cross_account
+@pytest.mark.sns
+def test_cross_account_topic_publish(sns, s3_alt, iam_alt_root):
+    name = get_new_topic_name()
+
+    response = sns.create_topic(Name=name)
+    topic_arn = response['TopicArn']
+
+    bucket = get_new_bucket_name()
+    s3_alt.create_bucket(Bucket=bucket)
+
+    config = {'TopicConfigurations': [{
+        'Id': 'id',
+        'TopicArn': topic_arn,
+        'Events': [ 's3:ObjectCreated:*' ],
+        }]}
+
+    # expect AccessDenies because no resource policy allows cross-account access
+    e = assert_raises(ClientError, s3_alt.put_bucket_notification_configuration,
+                      Bucket=bucket, NotificationConfiguration=config)
+    status, error_code = _get_status_and_error_code(e.response)
+    assert status == 403
+    assert error_code == 'AccessDenied'
+
+    # add topic policy to allow the alt user
+    alt_principal = iam_alt_root.get_user()['User']['Arn']
+    policy = json.dumps({
+        'Version': '2012-10-17',
+        'Statement': [{
+            'Effect': 'Allow',
+            'Principal': {'AWS': alt_principal},
+            'Action': 'sns:Publish',
+            'Resource': topic_arn
+            }]
+        })
+    sns.set_topic_attributes(TopicArn=topic_arn, AttributeName='Policy',
+                             AttributeValue=policy)
+
+    s3_alt.put_bucket_notification_configuration(
+            Bucket=bucket, NotificationConfiguration=config)

s3tests_boto3/functional/test_utils.py

@@ -1,11 +1,9 @@
-from nose.tools import eq_ as eq
-
 from . import utils
 
 def test_generate():
     FIVE_MB = 5 * 1024 * 1024
-    eq(len(''.join(utils.generate_random(0))), 0)
-    eq(len(''.join(utils.generate_random(1))), 1)
-    eq(len(''.join(utils.generate_random(FIVE_MB - 1))), FIVE_MB - 1)
-    eq(len(''.join(utils.generate_random(FIVE_MB))), FIVE_MB)
-    eq(len(''.join(utils.generate_random(FIVE_MB + 1))), FIVE_MB + 1)
+    assert len(''.join(utils.generate_random(0))) == 0
+    assert len(''.join(utils.generate_random(1))) == 1
+    assert len(''.join(utils.generate_random(FIVE_MB - 1))) == FIVE_MB - 1
+    assert len(''.join(utils.generate_random(FIVE_MB))) == FIVE_MB
+    assert len(''.join(utils.generate_random(FIVE_MB + 1))) == FIVE_MB + 1

s3tests_boto3/functional/utils.py

@@ -3,8 +3,6 @@ import requests
 import string
 import time
 
-from nose.tools import eq_ as eq
-
 def assert_raises(excClass, callableObj, *args, **kwargs):
     """
     Like unittest.TestCase.assertRaises, but returns the exception.

tox.ini

@@ -0,0 +1,9 @@
+[tox]
+envlist = py
+
+[testenv]
+deps = -rrequirements.txt
+passenv =
+  S3TEST_CONF
+  S3_USE_SIGV4
+commands = pytest {posargs}