Authenticating to S3 in tests
This commit is contained in:
parent
2055bb62f5
commit
802279d7c4
@ -3,6 +3,7 @@
|
||||
from __future__ import unicode_literals
|
||||
import sure # noqa
|
||||
|
||||
from flask.testing import FlaskClient
|
||||
import moto.server as server
|
||||
|
||||
'''
|
||||
@ -10,18 +11,28 @@ Test the different server responses
|
||||
'''
|
||||
|
||||
|
||||
def test_s3_server_get():
|
||||
backend = server.create_backend_app("s3")
|
||||
test_client = backend.test_client()
|
||||
class AuthenticatedClient(FlaskClient):
|
||||
def open(self, *args, **kwargs):
|
||||
kwargs['headers'] = kwargs.get('headers', {})
|
||||
kwargs['headers']['Authorization'] = "Any authorization header"
|
||||
return super(AuthenticatedClient, self).open(*args, **kwargs)
|
||||
|
||||
|
||||
def authenticated_client():
|
||||
backend = server.create_backend_app("s3")
|
||||
backend.test_client_class = AuthenticatedClient
|
||||
return backend.test_client()
|
||||
|
||||
|
||||
def test_s3_server_get():
|
||||
test_client = authenticated_client()
|
||||
res = test_client.get('/')
|
||||
|
||||
res.data.should.contain(b'ListAllMyBucketsResult')
|
||||
|
||||
|
||||
def test_s3_server_bucket_create():
|
||||
backend = server.create_backend_app("s3")
|
||||
test_client = backend.test_client()
|
||||
test_client = authenticated_client()
|
||||
|
||||
res = test_client.put('/', 'http://foobaz.localhost:5000/')
|
||||
res.status_code.should.equal(200)
|
||||
@ -44,8 +55,7 @@ def test_s3_server_bucket_create():
|
||||
|
||||
|
||||
def test_s3_server_bucket_versioning():
|
||||
backend = server.create_backend_app("s3")
|
||||
test_client = backend.test_client()
|
||||
test_client = authenticated_client()
|
||||
|
||||
# Just enough XML to enable versioning
|
||||
body = '<Status>Enabled</Status>'
|
||||
@ -55,8 +65,7 @@ def test_s3_server_bucket_versioning():
|
||||
|
||||
|
||||
def test_s3_server_post_to_bucket():
|
||||
backend = server.create_backend_app("s3")
|
||||
test_client = backend.test_client()
|
||||
test_client = authenticated_client()
|
||||
|
||||
res = test_client.put('/', 'http://tester.localhost:5000/')
|
||||
res.status_code.should.equal(200)
|
||||
@ -72,8 +81,7 @@ def test_s3_server_post_to_bucket():
|
||||
|
||||
|
||||
def test_s3_server_post_without_content_length():
|
||||
backend = server.create_backend_app("s3")
|
||||
test_client = backend.test_client()
|
||||
test_client = authenticated_client()
|
||||
|
||||
res = test_client.put('/', 'http://tester.localhost:5000/', environ_overrides={'CONTENT_LENGTH': ''})
|
||||
res.status_code.should.equal(411)
|
||||
|
@ -1,6 +1,7 @@
|
||||
from __future__ import unicode_literals
|
||||
import sure # noqa
|
||||
|
||||
from flask.testing import FlaskClient
|
||||
import moto.server as server
|
||||
|
||||
'''
|
||||
@ -8,9 +9,21 @@ Test the different server responses
|
||||
'''
|
||||
|
||||
|
||||
def test_s3_server_get():
|
||||
class AuthenticatedClient(FlaskClient):
|
||||
def open(self, *args, **kwargs):
|
||||
kwargs['headers'] = kwargs.get('headers', {})
|
||||
kwargs['headers']['Authorization'] = "Any authorization header"
|
||||
return super(AuthenticatedClient, self).open(*args, **kwargs)
|
||||
|
||||
|
||||
def authenticated_client():
|
||||
backend = server.create_backend_app("s3bucket_path")
|
||||
test_client = backend.test_client()
|
||||
backend.test_client_class = AuthenticatedClient
|
||||
return backend.test_client()
|
||||
|
||||
|
||||
def test_s3_server_get():
|
||||
test_client = authenticated_client()
|
||||
|
||||
res = test_client.get('/')
|
||||
|
||||
@ -18,8 +31,7 @@ def test_s3_server_get():
|
||||
|
||||
|
||||
def test_s3_server_bucket_create():
|
||||
backend = server.create_backend_app("s3bucket_path")
|
||||
test_client = backend.test_client()
|
||||
test_client = authenticated_client()
|
||||
|
||||
res = test_client.put('/foobar', 'http://localhost:5000')
|
||||
res.status_code.should.equal(200)
|
||||
@ -54,8 +66,7 @@ def test_s3_server_bucket_create():
|
||||
|
||||
|
||||
def test_s3_server_post_to_bucket():
|
||||
backend = server.create_backend_app("s3bucket_path")
|
||||
test_client = backend.test_client()
|
||||
test_client = authenticated_client()
|
||||
|
||||
res = test_client.put('/foobar2', 'http://localhost:5000/')
|
||||
res.status_code.should.equal(200)
|
||||
@ -71,8 +82,7 @@ def test_s3_server_post_to_bucket():
|
||||
|
||||
|
||||
def test_s3_server_put_ipv6():
|
||||
backend = server.create_backend_app("s3bucket_path")
|
||||
test_client = backend.test_client()
|
||||
test_client = authenticated_client()
|
||||
|
||||
res = test_client.put('/foobar2', 'http://[::]:5000/')
|
||||
res.status_code.should.equal(200)
|
||||
@ -88,8 +98,7 @@ def test_s3_server_put_ipv6():
|
||||
|
||||
|
||||
def test_s3_server_put_ipv4():
|
||||
backend = server.create_backend_app("s3bucket_path")
|
||||
test_client = backend.test_client()
|
||||
test_client = authenticated_client()
|
||||
|
||||
res = test_client.put('/foobar2', 'http://127.0.0.1:5000/')
|
||||
res.status_code.should.equal(200)
|
||||
|
Loading…
Reference in New Issue
Block a user