|
5 | 5 | from nose.plugins.attrib import attr
|
6 | 6 |
|
7 | 7 | from fs.test import FSTestCases
|
8 |
| - |
9 | 8 | from fs_s3fs import S3FS
|
10 | 9 |
|
11 | 10 | import boto3
|
12 | 11 |
|
13 | 12 |
|
14 | 13 | class TestS3FS(FSTestCases, unittest.TestCase):
|
15 | 14 | """Test S3FS implementation from dir_path."""
|
16 |
| - bucket_name = 'fsexample' |
17 |
| - s3 = boto3.resource('s3') |
18 |
| - client = boto3.client('s3') |
| 15 | + |
| 16 | + bucket_name = "fsexample" |
| 17 | + s3 = boto3.resource("s3") |
| 18 | + client = boto3.client("s3") |
19 | 19 |
|
20 | 20 | def make_fs(self):
|
21 | 21 | self._delete_bucket_contents()
|
22 | 22 | return S3FS(self.bucket_name)
|
23 | 23 |
|
24 | 24 | def _delete_bucket_contents(self):
|
25 |
| - response = self.client.list_objects( |
26 |
| - Bucket=self.bucket_name |
27 |
| - ) |
| 25 | + response = self.client.list_objects(Bucket=self.bucket_name) |
28 | 26 | contents = response.get("Contents", ())
|
29 | 27 | for obj in contents:
|
30 |
| - self.client.delete_object( |
31 |
| - Bucket=self.bucket_name, |
32 |
| - Key=obj["Key"] |
33 |
| - ) |
| 28 | + self.client.delete_object(Bucket=self.bucket_name, Key=obj["Key"]) |
34 | 29 |
|
35 | 30 |
|
36 |
| -@attr('slow') |
| 31 | +@attr("slow") |
37 | 32 | class TestS3FSSubDir(FSTestCases, unittest.TestCase):
|
38 | 33 | """Test S3FS implementation from dir_path."""
|
39 |
| - bucket_name = 'fsexample' |
40 |
| - s3 = boto3.resource('s3') |
41 |
| - client = boto3.client('s3') |
| 34 | + |
| 35 | + bucket_name = "fsexample" |
| 36 | + s3 = boto3.resource("s3") |
| 37 | + client = boto3.client("s3") |
42 | 38 |
|
43 | 39 | def make_fs(self):
|
44 | 40 | self._delete_bucket_contents()
|
45 |
| - self.s3.Object(self.bucket_name, 'subdirectory').put() |
46 |
| - return S3FS(self.bucket_name, dir_path='subdirectory') |
| 41 | + self.s3.Object(self.bucket_name, "subdirectory").put() |
| 42 | + return S3FS(self.bucket_name, dir_path="subdirectory") |
47 | 43 |
|
48 | 44 | def _delete_bucket_contents(self):
|
49 |
| - response = self.client.list_objects( |
50 |
| - Bucket=self.bucket_name |
51 |
| - ) |
| 45 | + response = self.client.list_objects(Bucket=self.bucket_name) |
52 | 46 | contents = response.get("Contents", ())
|
53 | 47 | for obj in contents:
|
54 |
| - self.client.delete_object( |
55 |
| - Bucket=self.bucket_name, |
56 |
| - Key=obj["Key"] |
57 |
| - ) |
| 48 | + self.client.delete_object(Bucket=self.bucket_name, Key=obj["Key"]) |
58 | 49 |
|
59 | 50 |
|
60 | 51 | class TestS3FSHelpers(unittest.TestCase):
|
61 |
| - |
62 | 52 | def test_path_to_key(self):
|
63 |
| - s3 = S3FS('foo') |
64 |
| - self.assertEqual(s3._path_to_key('foo.bar'), 'foo.bar') |
65 |
| - self.assertEqual(s3._path_to_key('foo/bar'), 'foo/bar') |
| 53 | + s3 = S3FS("foo") |
| 54 | + self.assertEqual(s3._path_to_key("foo.bar"), "foo.bar") |
| 55 | + self.assertEqual(s3._path_to_key("foo/bar"), "foo/bar") |
66 | 56 |
|
67 | 57 | def test_path_to_key_subdir(self):
|
68 |
| - s3 = S3FS('foo', '/dir') |
69 |
| - self.assertEqual(s3._path_to_key('foo.bar'), 'dir/foo.bar') |
70 |
| - self.assertEqual(s3._path_to_key('foo/bar'), 'dir/foo/bar') |
| 58 | + s3 = S3FS("foo", "/dir") |
| 59 | + self.assertEqual(s3._path_to_key("foo.bar"), "dir/foo.bar") |
| 60 | + self.assertEqual(s3._path_to_key("foo/bar"), "dir/foo/bar") |
71 | 61 |
|
72 | 62 | def test_upload_args(self):
|
73 |
| - s3 = S3FS('foo', acl='acl', cache_control='cc') |
74 |
| - self.assertDictEqual(s3._get_upload_args('test.jpg'), |
75 |
| - {'ACL': 'acl', 'CacheControl': 'cc', 'ContentType': 'image/jpeg'}) |
76 |
| - self.assertDictEqual(s3._get_upload_args('test.mp3'), |
77 |
| - {'ACL': 'acl', 'CacheControl': 'cc', 'ContentType': 'audio/mpeg'}) |
78 |
| - self.assertDictEqual(s3._get_upload_args('test.json'), |
79 |
| - {'ACL': 'acl', 'CacheControl': 'cc', 'ContentType': 'application/json'}) |
80 |
| - self.assertDictEqual(s3._get_upload_args('unknown.unknown'), |
81 |
| - {'ACL': 'acl', 'CacheControl': 'cc', 'ContentType': 'binary/octet-stream'}) |
| 63 | + s3 = S3FS("foo", acl="acl", cache_control="cc") |
| 64 | + self.assertDictEqual( |
| 65 | + s3._get_upload_args("test.jpg"), |
| 66 | + {"ACL": "acl", "CacheControl": "cc", "ContentType": "image/jpeg"}, |
| 67 | + ) |
| 68 | + self.assertDictEqual( |
| 69 | + s3._get_upload_args("test.mp3"), |
| 70 | + {"ACL": "acl", "CacheControl": "cc", "ContentType": "audio/mpeg"}, |
| 71 | + ) |
| 72 | + self.assertDictEqual( |
| 73 | + s3._get_upload_args("test.json"), |
| 74 | + {"ACL": "acl", "CacheControl": "cc", "ContentType": "application/json"}, |
| 75 | + ) |
| 76 | + self.assertDictEqual( |
| 77 | + s3._get_upload_args("unknown.unknown"), |
| 78 | + {"ACL": "acl", "CacheControl": "cc", "ContentType": "binary/octet-stream"}, |
| 79 | + ) |
0 commit comments