|
5 | 5 | from pathlib import Path
|
6 | 6 |
|
7 | 7 | import pytest
|
| 8 | +from freezegun import freeze_time |
| 9 | +from yarl import URL |
8 | 10 |
|
9 | 11 | from aiohttp_s3_client import S3Client
|
10 | 12 |
|
@@ -138,3 +140,64 @@ async def async_iterable(iterable: bytes):
|
138 | 140 | # FIXME: uncomment after update fakes3 image
|
139 | 141 | actual = gzip.GzipFile(fileobj=BytesIO(result)).read()
|
140 | 142 | assert actual == data
|
| 143 | + |
| 144 | + |
| 145 | +@pytest.mark.parametrize('method,given_url', [ |
| 146 | + # Simple test |
| 147 | + ('GET', 'test/object'), |
| 148 | +
|
| 149 | + # Check method name is ok in lowercase |
| 150 | + ('get', 'test/object'), |
| 151 | +
|
| 152 | + # Absolute path |
| 153 | + ('GET', '/test/object'), |
| 154 | +
|
| 155 | + # Check URL object with path is given |
| 156 | + ('get', URL('./test/object')), |
| 157 | +
|
| 158 | + # Check URL object with path is given |
| 159 | + ('get', URL('/test/object')), |
| 160 | +]) |
| 161 | +def test_presign_non_absolute_url(s3_client, s3_url, method, given_url): |
| 162 | + presigned = s3_client.presign_url('get', 'test/object') |
| 163 | + assert presigned.is_absolute() |
| 164 | + assert presigned.scheme == s3_url.scheme |
| 165 | + assert presigned.host == s3_url.host |
| 166 | + assert presigned.port == s3_url.port |
| 167 | + assert presigned.path.startswith(s3_url.path) |
| 168 | + assert presigned.path.endswith(str(given_url).lstrip('.')) |
| 169 | + |
| 170 | + |
| 171 | +@pytest.mark.parametrize('method,given_url', [ |
| 172 | + # String url |
| 173 | + ('GET', 'https://absolute-url:123/path'), |
| 174 | +
|
| 175 | + # URL object |
| 176 | + ('GET', URL('https://absolute-url:123/path')), |
| 177 | +]) |
| 178 | +def test_presign_absolute_url(s3_client, method, given_url): |
| 179 | + presigned = s3_client.presign_url(method, given_url) |
| 180 | + assert presigned.is_absolute() |
| 181 | + |
| 182 | + url_object = URL(given_url) |
| 183 | + assert presigned.scheme == url_object.scheme |
| 184 | + assert presigned.host == url_object.host |
| 185 | + assert presigned.port == url_object.port |
| 186 | + assert presigned.path == url_object.path |
| 187 | + |
| 188 | + |
| 189 | +@freeze_time("2024-01-01") |
| 190 | +def test_presign_url(s3_client, s3_url): |
| 191 | + url = s3_client.presign_url('get', URL('./example')) |
| 192 | + assert url.path == (s3_url / 'example').path |
| 193 | + assert url.query == { |
| 194 | + 'X-Amz-Algorithm': 'AWS4-HMAC-SHA256', |
| 195 | + 'X-Amz-Content-Sha256': 'UNSIGNED-PAYLOAD', |
| 196 | + 'X-Amz-Credential': 'user/20240101/us-east-1/s3/aws4_request', |
| 197 | + 'X-Amz-Date': '20240101T000000Z', |
| 198 | + 'X-Amz-Expires': '86400', |
| 199 | + 'X-Amz-SignedHeaders': 'host', |
| 200 | + 'X-Amz-Signature': ( |
| 201 | + '7359f1286edf554b0eab363e3c93ee32855b8d429d975fdb0a5d2cb7ad5c0db0' |
| 202 | + ) |
| 203 | + } |
0 commit comments