Coverage for tests/test_s3.py: 93%
235 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-23 09:29 +0000
1# This file is part of lsst-resources.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (https://www.lsst.org).
6# See the COPYRIGHT file at the top-level directory of this distribution
7# for details of code ownership.
8#
9# Use of this source code is governed by a 3-clause BSD-style
10# license that can be found in the LICENSE file.
12import datetime
13import os
14import sys
15import time
16import unittest
17from inspect import signature
18from unittest import mock
19from urllib.parse import parse_qs, urlparse
21from lsst.resources import ResourceInfo, ResourcePath
22from lsst.resources.s3 import S3ResourcePath
23from lsst.resources.s3utils import clean_test_environment_for_s3
24from lsst.resources.tests import GenericReadWriteTestCase, GenericTestCase
26# moto mocks S3 by patching botocore in this process, and its stubber reloads
27# the module holding a backend's URL table on every request. Concurrent
28# reloads of the same module race, and the loser raises ImportError, which a
29# bulk operation reports as a URI that could not be reached. Only a
30# free-threaded interpreter issues those requests at the same time, and which
31# request loses varies from run to run, so every test that drives requests in
32# parallel is affected rather than one in particular.
33_MOTO_IS_THREAD_SAFE = getattr(sys, "_is_gil_enabled", lambda: True)()
34skip_if_moto_races = unittest.skipIf(
35 not _MOTO_IS_THREAD_SAFE,
36 "moto reloads modules per request, which races without the GIL",
37)
39try:
40 import boto3
41 import botocore
43 try:
44 from moto import mock_aws # v5
45 except ImportError:
46 from moto import mock_s3 as mock_aws
47except ImportError:
48 boto3 = None
50 def mock_aws(cls):
51 """No-op decorator in case moto mock_aws can not be imported."""
52 return cls
55try:
56 import fsspec
57except ImportError:
58 fsspec = None
61class GenericS3TestCase(GenericTestCase, unittest.TestCase):
62 """Generic tests of S3 URIs."""
64 scheme = "s3"
65 netloc = "my_bucket"
68class S3ReadWriteTestCaseBase(GenericReadWriteTestCase):
69 """Tests of reading and writing S3 URIs."""
71 scheme = "s3"
72 s3_endpoint_url: str | None = None
73 # S3 batches in 1000 files so need more than that.
74 n_mremove_files: int = 1015
76 def setUp(self):
77 self.enterContext(clean_test_environment_for_s3())
79 # Enable S3 mocking of tests.
80 self.enterContext(mock_aws())
82 # MOTO needs to know that we expect Bucket bucketname to exist
83 s3 = boto3.resource("s3", endpoint_url=self.s3_endpoint_url)
84 s3.create_bucket(Bucket=self.bucket)
86 super().setUp()
88 def tearDown(self):
89 s3 = boto3.resource("s3")
90 bucket = s3.Bucket(self.bucket)
91 try:
92 bucket.objects.all().delete()
93 except botocore.exceptions.ClientError as e:
94 if e.response["Error"]["Code"] == "404":
95 # the key was not reachable - pass
96 pass
97 else:
98 raise
100 bucket = s3.Bucket(self.bucket)
101 bucket.delete()
103 S3ResourcePath.use_threads = None
105 super().tearDown()
107 def test_bucket_fail(self):
108 # Deliberately create URI with unknown bucket.
109 uri = ResourcePath("s3://badbucket/something/")
111 with self.assertRaises(ValueError):
112 uri.mkdir()
114 with self.assertRaises(FileNotFoundError):
115 uri.remove()
117 def test_transfer_progress(self):
118 """Test progress bar reporting for upload and download."""
119 remote = self.root_uri.join("test.dat")
120 remote.write(b"42")
121 with ResourcePath.temporary_uri(suffix=".dat") as tmp:
122 # Download from S3.
123 with self.assertLogs("lsst.resources", level="DEBUG") as cm:
124 tmp.transfer_from(remote, transfer="auto")
125 self.assertRegex("".join(cm.output), r"test\.dat.*100\%")
127 # Upload to S3.
128 with self.assertLogs("lsst.resources", level="DEBUG") as cm:
129 remote.transfer_from(tmp, transfer="auto", overwrite=True)
130 self.assertRegex("".join(cm.output), rf"{tmp.basename()}.*100\%")
132 def test_handle(self):
133 remote = self.root_uri.join("test_handle.dat")
134 with remote.open("wb") as handle:
135 self.assertTrue(handle.writable())
136 # write 6 megabytes to make sure partial write work
137 handle.write(6 * 1024 * 1024 * b"a")
138 self.assertEqual(handle.tell(), 6 * 1024 * 1024)
139 handle.flush()
140 self.assertGreaterEqual(len(handle._multiPartUpload), 1)
142 # verify file can't be seeked back
143 with self.assertRaises(OSError):
144 handle.seek(0)
146 # write more bytes
147 handle.write(1024 * b"c")
149 # seek back and overwrite
150 handle.seek(6 * 1024 * 1024)
151 handle.write(1024 * b"b")
153 with remote.open("rb") as handle:
154 self.assertTrue(handle.readable())
155 # read the first 6 megabytes
156 result = handle.read(6 * 1024 * 1024)
157 self.assertEqual(result, 6 * 1024 * 1024 * b"a")
158 self.assertEqual(handle.tell(), 6 * 1024 * 1024)
159 # verify additional read gets the next part
160 result = handle.read(1024)
161 self.assertEqual(result, 1024 * b"b")
162 # see back to the beginning to verify seeking
163 handle.seek(0)
164 result = handle.read(1024)
165 self.assertEqual(result, 1024 * b"a")
167 remote = self.root_uri.join("missing_file.dat")
168 with remote.open("rb") as handle:
169 with self.assertRaises(FileNotFoundError):
170 handle.read()
172 def test_url_signing(self):
173 self._test_url_signing_case("url-signing-test.txt", b"test123")
174 # A zero byte presigned S3 HTTP URL is a weird edge case, because we
175 # emulate HEAD requests using a 1-byte GET.
176 self._test_url_signing_case("url-signing-test-zero-bytes.txt", b"")
177 # Should be the same as a normal case, but check it for paranoia since
178 # it's on the boundary of the read size.
179 self._test_url_signing_case("url-signing-test-one-byte.txt", b"t")
181 def _test_url_signing_case(self, filename: str, test_data: bytes):
182 s3_path = self.root_uri.join(filename)
184 put_url = s3_path.generate_presigned_put_url(expiration_time_seconds=1800)
185 self._check_presigned_url(put_url, 1800)
186 get_url = s3_path.generate_presigned_get_url(expiration_time_seconds=3600)
187 self._check_presigned_url(get_url, 3600)
189 # Check that fragments are retained.
190 s3_path = s3_path.replace(fragment="zip-path=X")
191 put_url = s3_path.generate_presigned_put_url(expiration_time_seconds=1800)
192 self.assertEqual(ResourcePath(put_url).fragment, "zip-path=X")
193 self._check_presigned_url(put_url, 1800)
194 get_url = s3_path.generate_presigned_get_url(expiration_time_seconds=3600)
195 self.assertEqual(ResourcePath(get_url).fragment, "zip-path=X")
196 self._check_presigned_url(get_url, 3600)
198 # Moto monkeypatches the 'requests' library to mock access to presigned
199 # URLs, so we are able to use HttpResourcePath to access the URLs in
200 # this test.
201 ResourcePath(put_url).write(test_data)
202 get_path = ResourcePath(get_url)
203 retrieved = get_path.read()
204 self.assertEqual(retrieved, test_data)
205 self.assertTrue(get_path.exists())
206 self.assertEqual(get_path.size(), len(test_data))
208 # Try again with open().
209 with get_path.open("rb") as fd:
210 self.assertEqual(fd.read(), test_data)
212 def test_nonexistent_presigned_url(self):
213 s3_path = self.root_uri.join("this-is-a-missing-file.txt")
214 get_url = s3_path.generate_presigned_get_url(expiration_time_seconds=3600)
215 get_path = ResourcePath(get_url)
216 # Check the HttpResourcePath implementation for presigned S3 urls.
217 # Nothing has been uploaded to this URL, so it shouldn't exist.
218 self.assertFalse(get_path.exists())
219 with self.assertRaises(FileNotFoundError):
220 get_path.size()
222 def test_get_info(self):
223 now = datetime.datetime.now(tz=datetime.UTC)
224 remote = self.root_uri.join("test-info.dat")
225 remote.write(b"abc")
227 info = remote.get_info()
228 self.assertIsInstance(info, ResourceInfo)
229 self.assertTrue(info.is_file)
230 self.assertEqual(info.size, 3)
231 self.assertIsInstance(info.checksums, dict)
232 self.assertIn("crc32", info.checksums) # Only appears if ChecksumMode=ENABLED
233 self.assertEqual(info.last_modified.tzinfo, datetime.UTC)
234 self.assertGreaterEqual(info.last_modified.timestamp(), now.timestamp() - 1.0)
236 def _check_presigned_url(self, url: str, expiration_time_seconds: int):
237 parsed = urlparse(url)
238 self.assertEqual(parsed.scheme, "https")
240 actual_expiration_timestamp = int(parse_qs(parsed.query)["Expires"][0])
241 current_time = int(time.time())
242 expected_expiration_timestamp = current_time + expiration_time_seconds
243 # Allow some flex in the expiration time in case this test process goes
244 # out to lunch for a while on a busy CI machine
245 self.assertLessEqual(abs(expected_expiration_timestamp - actual_expiration_timestamp), 120)
247 def test_threading_true(self):
248 with mock.patch.dict(os.environ, {"LSST_S3_USE_THREADS": "True"}):
249 S3ResourcePath.use_threads = None
250 test_resource_path = self.root_uri.join("test_file.dat")
251 self.assertTrue(test_resource_path._transfer_config.use_threads)
253 def test_implicit_default_threading(self):
254 S3ResourcePath.use_threads = None
255 boto_default = signature(boto3.s3.transfer.TransferConfig).parameters["use_threads"].default
256 # Newer versions of boto return None as the default.
257 if boto_default is None: 257 ↛ 259line 257 didn't jump to line 259 because the condition on line 257 was always true
258 boto_default = True
259 test_resource_path = self.root_uri.join("test_file.dat")
260 self.assertEqual(test_resource_path._transfer_config.use_threads, boto_default)
262 def test_explicit_default_threading(self):
263 with mock.patch.dict(os.environ, {"LSST_S3_USE_THREADS": "None"}):
264 S3ResourcePath.use_threads = None
265 boto_default = signature(boto3.s3.transfer.TransferConfig).parameters["use_threads"].default
266 # Newer versions of boto return None as the default.
267 if boto_default is None: 267 ↛ 269line 267 didn't jump to line 269 because the condition on line 267 was always true
268 boto_default = True
269 test_resource_path = self.root_uri.join("test_file.dat")
270 self.assertEqual(test_resource_path._transfer_config.use_threads, boto_default)
272 def test_threading_false(self):
273 with mock.patch.dict(os.environ, {"LSST_S3_USE_THREADS": "False"}):
274 S3ResourcePath.use_threads = None
275 test_resource_path = self.root_uri.join("test_file.dat")
276 self.assertFalse(test_resource_path._transfer_config.use_threads)
278 self.test_local()
280 @unittest.skipIf(fsspec is None, "fsspec is not available")
281 def test_fsspec_constructor(self) -> None:
282 """Test that we can obtain an s3fs object."""
283 uri = self.root_uri.join("test_file.dat")
284 fs, path = uri.to_fsspec()
285 self.assertEqual(path, f"{uri._bucket}/{uri.relativeToPathRoot}")
286 self.assertTrue(hasattr(fs, "open"))
288 def test_fsspec(self) -> None:
289 raise unittest.SkipTest("fsspec s3fs incompatible with moto")
291 # These drive many S3 requests in parallel, so under a free-threaded
292 # interpreter they are the tests moto's reload race shows up in.
293 @skip_if_moto_races
294 def test_mexists(self) -> None:
295 super().test_mexists()
297 @skip_if_moto_races
298 def test_mtransfer(self) -> None:
299 super().test_mtransfer()
301 @skip_if_moto_races
302 def test_local_mtransfer(self) -> None:
303 super().test_local_mtransfer()
306@unittest.skipIf(not boto3, "Warning: boto3 AWS SDK not found!")
307class S3ReadWriteTestCase(S3ReadWriteTestCaseBase, unittest.TestCase):
308 """Test S3 with no explicit profile/endpoint specified.
309 (``s3://bucketname/...``).
310 """
312 bucket = "my_2nd_bucket"
313 netloc = bucket
316@unittest.skipIf(not boto3, "Warning: boto3 AWS SDK not found!")
317class S3WithProfileReadWriteTestCase(S3ReadWriteTestCaseBase, unittest.TestCase):
318 """Test S3 URLs with explicit profile specified.
319 (``s3://profile@bucketname/...``).
320 """
322 bucket = "3rd_bucket"
323 netloc = f"myprofile@{bucket}"
324 s3_endpoint_url = "https://endpoint1.test.example"
326 def setUp(self):
327 # Configure custom S3 endpoints that we can target from tests using
328 # non-default profile.
329 self.enterContext(
330 mock.patch.dict(
331 os.environ,
332 {
333 "MOTO_S3_CUSTOM_ENDPOINTS": self.s3_endpoint_url,
334 "LSST_RESOURCES_S3_PROFILE_myprofile": "https://access_key:security_key@endpoint1.test.example",
335 },
336 )
337 )
339 super().setUp()
341 def test_missing_profile(self):
342 with self.assertRaises(botocore.exceptions.ProfileNotFound):
343 ResourcePath("s3://otherprofile@bucket").read()
345 def test_s3_endpoint_url(self):
346 with mock.patch.dict(
347 os.environ,
348 {"S3_ENDPOINT_URL": self.s3_endpoint_url},
349 ):
350 path = ResourcePath(f"s3://{self.bucket}/test-s3-endpoint-url.txt")
351 data = b"123"
352 path.write(data)
353 self.assertEqual(path.read(), data)
354 self.assertIn(
355 "https://endpoint1.test.example",
356 path.generate_presigned_get_url(expiration_time_seconds=3600),
357 )
359 def test_uri_syntax(self):
360 path1 = ResourcePath("s3://profile@bucket/path")
361 self.assertEqual(path1._bucket, "bucket")
362 self.assertEqual(path1._profile, "profile")
363 path2 = ResourcePath("s3://bucket2/path")
364 self.assertEqual(path2._bucket, "bucket2")
365 self.assertIsNone(path2._profile)
367 def test_ceph_uri_syntax(self):
368 # The Ceph S3 'multi-tenant' syntax for buckets can include colons.
369 path1 = ResourcePath("s3://profile@ceph:bucket/path")
370 self.assertEqual(path1._bucket, "ceph:bucket")
371 self.assertEqual(path1._profile, "profile")
372 path2 = ResourcePath("s3://ceph:bucket2/path")
373 self.assertEqual(path2._bucket, "ceph:bucket2")
374 self.assertIsNone(path2._profile)
376 def test_transfer_from_different_endpoints(self):
377 # Create a bucket using a different endpoint (the default endpoint.)
378 boto3.resource("s3").create_bucket(Bucket="source-bucket")
379 source_path = ResourcePath("s3://source-bucket/file.txt")
380 source_path.write(b"123")
381 target_path = ResourcePath(f"s3://{self.netloc}/target.txt")
382 # Transfer from default endpoint to custom endpoint with custom
383 # profile.
384 target_path.transfer_from(source_path)
385 self.assertEqual(target_path.read(), b"123")
388if __name__ == "__main__":
389 unittest.main()