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