diff --git a/test/s3api/__init__.py b/test/s3api/__init__.py
new file mode 100644
index 0000000000..43e3fc5d2b
--- /dev/null
+++ b/test/s3api/__init__.py
@@ -0,0 +1,142 @@
+# Copyright (c) 2019 SwiftStack, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+import os
+import unittest
+
+import boto3
+from six.moves import urllib
+
+from swift.common.utils import config_true_value
+
+from test import get_config
+
+_CONFIG = None
+
+
+# boto's loggign can get pretty noisy; require opt-in to see it all
+if not config_true_value(os.environ.get('BOTO3_DEBUG')):
+    logging.getLogger('boto3').setLevel(logging.INFO)
+    logging.getLogger('botocore').setLevel(logging.INFO)
+
+
+class ConfigError(Exception):
+    '''Error test conf misconfigurations'''
+
+
+def get_opt_or_error(option):
+    global _CONFIG
+    if _CONFIG is None:
+        _CONFIG = get_config('s3api_test')
+
+    value = _CONFIG.get(option)
+    if not value:
+        raise ConfigError('must supply [s3api_test]%s' % option)
+    return value
+
+
+def get_opt(option, default=None):
+    try:
+        return get_opt_or_error(option)
+    except ConfigError:
+        return default
+
+
+def get_s3_client(user=1, signature_version='s3v4', addressing_style='path'):
+    '''
+    Get a boto3 client to talk to an S3 endpoint.
+
+    :param user: user number to use. Should be one of:
+        1 -- primary user
+        2 -- secondary user
+        3 -- unprivileged user
+    :param signature_version: S3 signing method. Should be one of:
+        s3 -- v2 signatures; produces Authorization headers like
+              ``AWS access_key:signature``
+        s3-query -- v2 pre-signed URLs; produces query strings like
+                    ``?AWSAccessKeyId=access_key&Signature=signature``
+        s3v4 -- v4 signatures; produces Authorization headers like
+                ``AWS4-HMAC-SHA256
+                Credential=access_key/date/region/s3/aws4_request,
+                Signature=signature``
+        s3v4-query -- v4 pre-signed URLs; produces query strings like
+                      ``?X-Amz-Algorithm=AWS4-HMAC-SHA256&
+                      X-Amz-Credential=access_key/date/region/s3/aws4_request&
+                      X-Amz-Signature=signature``
+    :param addressing_style: One of:
+        path -- produces URLs like ``http(s)://host.domain/bucket/key``
+        virtual -- produces URLs like ``http(s)://bucket.host.domain/key``
+    '''
+    endpoint = get_opt_or_error('endpoint')
+    scheme = urllib.parse.urlsplit(endpoint).scheme
+    if scheme not in ('http', 'https'):
+        raise ConfigError('unexpected scheme in endpoint: %r; '
+                          'expected http or https' % scheme)
+    region = get_opt('region', 'us-east-1')
+    access_key = get_opt_or_error('access_key%d' % user)
+    secret_key = get_opt_or_error('secret_key%d' % user)
+
+    ca_cert = get_opt('ca_cert')
+    if ca_cert is not None:
+        try:
+            # do a quick check now; it's more expensive to have boto check
+            os.stat(ca_cert)
+        except OSError as e:
+            raise ConfigError(str(e))
+
+    return boto3.client(
+        's3',
+        endpoint_url=endpoint,
+        region_name=region,
+        use_ssl=(scheme == 'https'),
+        verify=ca_cert,
+        config=boto3.session.Config(s3={
+            'signature_version': signature_version,
+            'addressing_style': addressing_style,
+        }),
+        aws_access_key_id=access_key,
+        aws_secret_access_key=secret_key,
+    )
+
+
+class BaseS3TestCase(unittest.TestCase):
+    # Default to v4 signatures (as aws-cli does), but subclasses can override
+    signature_version = 's3v4'
+
+    @classmethod
+    def get_s3_client(cls, user):
+        return get_s3_client(user, cls.signature_version)
+
+    @classmethod
+    def clear_bucket(cls, client, bucket):
+        for key in client.list_objects(Bucket=bucket).get('Contents', []):
+            client.delete_key(Bucket=bucket, Key=key['Name'])
+
+    @classmethod
+    def clear_account(cls, client):
+        for bucket in client.list_buckets()['Buckets']:
+            cls.clear_bucket(client, bucket['Name'])
+            client.delete_bucket(Bucket=bucket['Name'])
+
+    def tearDown(self):
+        client = self.get_s3_client(1)
+        self.clear_account(client)
+        try:
+            client = self.get_s3_client(2)
+        except ConfigError:
+            pass
+        else:
+            self.clear_account(client)
diff --git a/test/s3api/test_service.py b/test/s3api/test_service.py
new file mode 100644
index 0000000000..2d7cd7b2e0
--- /dev/null
+++ b/test/s3api/test_service.py
@@ -0,0 +1,91 @@
+# Copyright (c) 2019 SwiftStack, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import unittest
+import uuid
+
+from test.s3api import BaseS3TestCase, ConfigError
+
+
+class TestGetServiceSigV4(BaseS3TestCase):
+    def test_empty_service(self):
+        def do_test(client):
+            access_key = client._request_signer._credentials.access_key
+            resp = client.list_buckets()
+            self.assertEqual(200, resp['ResponseMetadata']['HTTPStatusCode'])
+            self.assertEqual([], resp['Buckets'])
+            self.assertIn('x-amz-request-id',
+                          resp['ResponseMetadata']['HTTPHeaders'])
+            self.assertIn('DisplayName', resp['Owner'])
+            self.assertEqual(access_key, resp['Owner']['DisplayName'])
+            self.assertIn('ID', resp['Owner'])
+
+        client = self.get_s3_client(1)
+        do_test(client)
+        try:
+            client = self.get_s3_client(3)
+        except ConfigError:
+            pass
+        else:
+            do_test(client)
+
+    def test_service_with_buckets(self):
+        c = self.get_s3_client(1)
+        buckets = [str(uuid.uuid4()) for _ in range(5)]
+        for bucket in buckets:
+            c.create_bucket(Bucket=bucket)
+
+        resp = c.list_buckets()
+        self.assertEqual(200, resp['ResponseMetadata']['HTTPStatusCode'])
+        self.assertEqual(sorted(buckets), [
+            bucket['Name'] for bucket in resp['Buckets']])
+        self.assertTrue(all('CreationDate' in bucket
+                            for bucket in resp['Buckets']))
+        self.assertIn('x-amz-request-id',
+                      resp['ResponseMetadata']['HTTPHeaders'])
+        self.assertIn('DisplayName', resp['Owner'])
+        access_key = c._request_signer._credentials.access_key
+        self.assertEqual(access_key, resp['Owner']['DisplayName'])
+        self.assertIn('ID', resp['Owner'])
+
+        # Second user can only see its own buckets
+        try:
+            c2 = self.get_s3_client(2)
+        except ConfigError as err:
+            raise unittest.SkipTest(str(err))
+        buckets2 = [str(uuid.uuid4()) for _ in range(2)]
+        for bucket in buckets2:
+            c2.create_bucket(Bucket=bucket)
+        self.assertEqual(sorted(buckets2), [
+            bucket['Name'] for bucket in c2.list_buckets()['Buckets']])
+
+        # Unprivileged user can't see anything
+        try:
+            c3 = self.get_s3_client(3)
+        except ConfigError as err:
+            raise unittest.SkipTest(str(err))
+        self.assertEqual([], c3.list_buckets()['Buckets'])
+
+
+class TestGetServiceSigV2(TestGetServiceSigV4):
+    signature_version = 's3'
+
+
+class TestGetServicePresignedV2(TestGetServiceSigV4):
+    signature_version = 's3-query'
+
+
+class TestGetServicePresignedV4(TestGetServiceSigV4):
+    signature_version = 's3v4-query'
diff --git a/test/sample.conf b/test/sample.conf
index d33be75486..162cb0797a 100644
--- a/test/sample.conf
+++ b/test/sample.conf
@@ -1,3 +1,16 @@
+[s3api_test]
+endpoint = http://127.0.0.1:8080
+#ca_cert=/path/to/ca.crt
+region = us-east-1
+# First and second users should be account owners
+access_key1 = test:tester
+secret_key1 = testing
+access_key2 = test:tester2
+secret_key2 = testing2
+# Third user should be unprivileged
+access_key3 = test:tester3
+secret_key3 = testing3
+
 [func_test]
 # Sample config for Swift with tempauth
 auth_host = 127.0.0.1