1import sure # noqa # pylint: disable=unused-import 2 3from flask.testing import FlaskClient 4import moto.server as server 5 6""" 7Test the different server responses 8""" 9 10 11class AuthenticatedClient(FlaskClient): 12 def open(self, *args, **kwargs): 13 kwargs["headers"] = kwargs.get("headers", {}) 14 kwargs["headers"]["Authorization"] = "Any authorization header" 15 kwargs["content_length"] = 0 # Fixes content-length complaints. 16 return super(AuthenticatedClient, self).open(*args, **kwargs) 17 18 19def authenticated_client(): 20 backend = server.create_backend_app("s3bucket_path") 21 backend.test_client_class = AuthenticatedClient 22 return backend.test_client() 23 24 25def test_s3_server_get(): 26 test_client = authenticated_client() 27 28 res = test_client.get("/") 29 30 res.data.should.contain(b"ListAllMyBucketsResult") 31 32 33def test_s3_server_bucket_create(): 34 test_client = authenticated_client() 35 36 res = test_client.put("/foobar", "http://localhost:5000") 37 res.status_code.should.equal(200) 38 39 res = test_client.get("/") 40 res.data.should.contain(b"<Name>foobar</Name>") 41 42 res = test_client.get("/foobar", "http://localhost:5000") 43 res.status_code.should.equal(200) 44 res.data.should.contain(b"ListBucketResult") 45 46 res = test_client.put("/foobar2/", "http://localhost:5000") 47 res.status_code.should.equal(200) 48 49 res = test_client.get("/") 50 res.data.should.contain(b"<Name>foobar2</Name>") 51 52 res = test_client.get("/foobar2/", "http://localhost:5000") 53 res.status_code.should.equal(200) 54 res.data.should.contain(b"ListBucketResult") 55 56 res = test_client.get("/missing-bucket", "http://localhost:5000") 57 res.status_code.should.equal(404) 58 59 res = test_client.put("/foobar/bar", "http://localhost:5000", data="test value") 60 res.status_code.should.equal(200) 61 62 res = test_client.get("/foobar/bar", "http://localhost:5000") 63 res.status_code.should.equal(200) 64 res.data.should.equal(b"test value") 65 66 67def test_s3_server_post_to_bucket(): 68 test_client = authenticated_client() 69 70 res = test_client.put("/foobar2", "http://localhost:5000/") 71 res.status_code.should.equal(200) 72 73 test_client.post( 74 "/foobar2", 75 "https://localhost:5000/", 76 data={"key": "the-key", "file": "nothing"}, 77 ) 78 79 res = test_client.get("/foobar2/the-key", "http://localhost:5000/") 80 res.status_code.should.equal(200) 81 res.data.should.equal(b"nothing") 82 83 84def test_s3_server_put_ipv6(): 85 test_client = authenticated_client() 86 87 res = test_client.put("/foobar2", "http://[::]:5000/") 88 res.status_code.should.equal(200) 89 90 test_client.post( 91 "/foobar2", "https://[::]:5000/", data={"key": "the-key", "file": "nothing"} 92 ) 93 94 res = test_client.get("/foobar2/the-key", "http://[::]:5000/") 95 res.status_code.should.equal(200) 96 res.data.should.equal(b"nothing") 97 98 99def test_s3_server_put_ipv4(): 100 test_client = authenticated_client() 101 102 res = test_client.put("/foobar2", "http://127.0.0.1:5000/") 103 res.status_code.should.equal(200) 104 105 test_client.post( 106 "/foobar2", 107 "https://127.0.0.1:5000/", 108 data={"key": "the-key", "file": "nothing"}, 109 ) 110 111 res = test_client.get("/foobar2/the-key", "http://127.0.0.1:5000/") 112 res.status_code.should.equal(200) 113 res.data.should.equal(b"nothing") 114