Skip to content

Plugin

access_key_getter(request)

Set up cred getter.

Source code in saritasa_s3_tools/testing/plugin.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@pytest.fixture(scope="session")
def access_key_getter(
    request: pytest.FixtureRequest,
) -> collections.abc.Callable[
    [],
    botocore.credentials.Credentials,
]:
    """Set up cred getter."""
    if (
        s3_access_key := request.config.getini(
            "s3_access_key",
        )
    ) and (
        s3_secret_key := request.config.getini(
            "s3_secret_key",
        )
    ):
        return lambda: botocore.credentials.Credentials(
            access_key=str(s3_access_key),
            secret_key=str(s3_secret_key),
        )
    raise NotImplementedError(  # pragma: no cover
        "Please set up `access_key_getter` fixture or "
        "set `s3_access_key` and `s3_secret_key` in `.ini` file.",
    )

async_s3_client(boto3_client, s3_bucket)

Set up async s3 client.

Source code in saritasa_s3_tools/testing/plugin.py 
292
293
294
295
296
297
298
299
300
301
@pytest.fixture(scope="session")
def async_s3_client(
    boto3_client: mypy_boto3_s3.S3Client,
    s3_bucket: str,
) -> saritasa_s3_tools.AsyncS3Client:
    """Set up async s3 client."""
    return saritasa_s3_tools.AsyncS3Client(
        boto3_client=boto3_client,
        default_bucket=s3_bucket,
    )

aws_config()

Get aws config.

Source code in saritasa_s3_tools/testing/plugin.py 
113
114
115
116
@pytest.fixture(scope="session")
def aws_config() -> botocore.config.Config | None:
    """Get aws config."""
    return None

aws_session(access_key_getter, s3_region)

Get aws session.

Source code in saritasa_s3_tools/testing/plugin.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
@pytest.fixture(scope="session")
def aws_session(
    access_key_getter: collections.abc.Callable[
        [],
        botocore.credentials.Credentials,
    ],
    s3_region: str,
) -> boto3.session.Session:
    """Get aws session."""
    return saritasa_s3_tools.client.get_boto3_session(
        access_key_getter=access_key_getter,
        region=s3_region,
    )

boto3_client(boto3_resource)

Prepare boto3 client.

Source code in saritasa_s3_tools/testing/plugin.py 
156
157
158
159
160
161
@pytest.fixture(scope="session")
def boto3_client(
    boto3_resource: mypy_boto3_s3.S3ServiceResource,
) -> mypy_boto3_s3.S3Client:
    """Prepare boto3 client."""
    return boto3_resource.meta.client

boto3_resource(request)

Prepare boto3 resource.

Source code in saritasa_s3_tools/testing/plugin.py 
119
120
121
122
123
124
125
126
127
@pytest.fixture(scope="session")
def boto3_resource(
    request: pytest.FixtureRequest,
) -> mypy_boto3_s3.S3ServiceResource:
    """Prepare boto3 resource."""
    try:
        return request.getfixturevalue("boto3_resource_from_django")
    except ImportError:  # pragma: no cover
        return request.getfixturevalue("boto3_resource_from_config")

boto3_resource_from_config(aws_session, aws_config, s3_endpoint_url_getter)

Prepare boto3 resource.

Source code in saritasa_s3_tools/testing/plugin.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
@pytest.fixture(scope="session")
def boto3_resource_from_config(
    aws_session: boto3.session.Session,
    aws_config: botocore.config.Config,
    s3_endpoint_url_getter: collections.abc.Callable[
        [],
        str | None,
    ]
    | None,
) -> mypy_boto3_s3.S3ServiceResource:
    """Prepare boto3 resource."""
    return saritasa_s3_tools.client.get_boto3_s3_resource(
        session=aws_session,
        s3_endpoint_url_getter=s3_endpoint_url_getter,
        config=aws_config,
    )

boto3_resource_from_django()

Get boto3 resource for django storage.

Source code in saritasa_s3_tools/testing/plugin.py 
130
131
132
133
134
135
@pytest.fixture(scope="session")
def boto3_resource_from_django() -> mypy_boto3_s3.S3ServiceResource:
    """Get boto3 resource for django storage."""
    from django.core.files.storage import default_storage

    return default_storage.connection  # type: ignore

django_adjust_s3_bucket(s3_bucket)

Set bucket to a test one.

Source code in saritasa_s3_tools/testing/plugin.py 
326
327
328
329
330
331
332
333
334
335
336
337
338
339
@pytest.fixture(scope="session")
def django_adjust_s3_bucket(s3_bucket: str) -> None:
    """Set bucket to a test one."""
    from django.conf import settings
    from django.core.files import storage
    from storages.backends import s3

    for storage_alias in settings.STORAGES:
        with contextlib.suppress(storage.InvalidStorageError):
            storage_instance = storage.storages[storage_alias]
            if isinstance(storage_instance, s3.S3Storage):
                storage_instance.bucket_name = s3_bucket  # type: ignore
    settings.AWS_STORAGE_BUCKET_NAME = s3_bucket
    return None

django_storage_changer()

Temporary change default storage settings.

Source code in saritasa_s3_tools/testing/plugin.py 
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
@pytest.fixture
def django_storage_changer() -> collections.abc.Iterator[
    collections.abc.Callable[
        [str, typing.Any],
        None,
    ]
]:
    """Temporary change default storage settings."""
    from django.core.files.storage import default_storage

    old_settings: dict[str, typing.Any] = {}

    def _changer(key: str, value: typing.Any) -> None:
        if key not in old_settings and hasattr(default_storage, key):
            old_settings[key] = getattr(default_storage, key)
        setattr(default_storage, key, value)

    yield _changer
    for key, value in old_settings.items():
        setattr(default_storage, key, value)

pytest_addoption(parser)

Set up cmd args.

Source code in saritasa_s3_tools/testing/plugin.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def pytest_addoption(parser: pytest.Parser) -> None:
    """Set up cmd args."""
    parser.addini(
        "s3_access_key",
        "Access key for s3.",
        default="",
    )
    parser.addini(
        "s3_secret_key",
        "Secret key for s3.",
        default="",
    )
    parser.addini(
        "s3_endpoint_url",
        "Endpoint for s3.",
        default="",
    )
    parser.addini(
        "s3_region",
        "Region for s3.",
        default="",
    )
    parser.addini(
        "s3_bucket_name",
        "Bucket for s3.",
        default="saritasa-s3-tools",
    )

s3_bucket(s3_bucket_factory, s3_bucket_name, s3_bucket_policy, s3_bucket_delete_on_teardown)

Create s3 bucket.

Source code in saritasa_s3_tools/testing/plugin.py 
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
@pytest.fixture(scope="session")
def s3_bucket(
    s3_bucket_factory: collections.abc.Callable[
        [str, dict[str, typing.Any], bool],
        contextlib._GeneratorContextManager[
            mypy_boto3_s3.type_defs.CreateBucketOutputTypeDef
        ],
    ],
    s3_bucket_name: str,
    s3_bucket_policy: dict[str, typing.Any],
    s3_bucket_delete_on_teardown: bool,
) -> collections.abc.Generator[str]:
    """Create s3 bucket."""
    with s3_bucket_factory(
        s3_bucket_name,
        s3_bucket_policy,
        s3_bucket_delete_on_teardown,
    ) as _:
        yield s3_bucket_name

s3_bucket_cleaner(boto3_resource)

Get bucket cleaner.

Source code in saritasa_s3_tools/testing/plugin.py 
199
200
201
202
203
204
205
206
207
208
@pytest.fixture(scope="session")
def s3_bucket_cleaner(
    boto3_resource: mypy_boto3_s3.ServiceResource,
) -> collections.abc.Callable[[str], None]:
    """Get bucket cleaner."""

    def _clean(bucket: str) -> None:
        boto3_resource.Bucket(bucket).objects.all().delete()

    return _clean

s3_bucket_delete_on_teardown()

Delete bucket on teardown or not.

Source code in saritasa_s3_tools/testing/plugin.py 
193
194
195
196
@pytest.fixture(scope="session")
def s3_bucket_delete_on_teardown() -> bool:
    """Delete bucket on teardown or not."""
    return False

s3_bucket_factory(boto3_client, s3_region, s3_bucket_cleaner)

Get factory for generating s3 buckets.

Source code in saritasa_s3_tools/testing/plugin.py 
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
@pytest.fixture(scope="session")
def s3_bucket_factory(
    boto3_client: mypy_boto3_s3.S3Client,
    s3_region: str,
    s3_bucket_cleaner: collections.abc.Callable[[str], None],
) -> collections.abc.Callable[
    [
        str,
        dict[str, typing.Any],
        bool,
    ],
    contextlib._GeneratorContextManager[
        mypy_boto3_s3.type_defs.CreateBucketOutputTypeDef
    ],
]:
    """Get factory for generating s3 buckets."""

    @contextlib.contextmanager
    def _create(
        bucket: str,
        policy: dict[str, typing.Any],
        delete_on_teardown: bool,
    ) -> collections.abc.Generator[
        mypy_boto3_s3.type_defs.CreateBucketOutputTypeDef
    ]:
        with contextlib.suppress(botocore.exceptions.ClientError):
            boto3_client.head_bucket(Bucket=bucket)
            s3_bucket_cleaner(bucket)  # pragma: no cover
            boto3_client.delete_bucket(Bucket=bucket)  # pragma: no cover
        response = boto3_client.create_bucket(
            Bucket=bucket,
            CreateBucketConfiguration={
                "LocationConstraint": s3_region,  # type: ignore
            },
        )
        if policy:
            boto3_client.put_bucket_policy(
                Bucket=bucket,
                Policy=json.dumps(policy),
            )
        yield response
        if delete_on_teardown:
            s3_bucket_cleaner(bucket)
            boto3_client.delete_bucket(Bucket=bucket)

    return _create

s3_bucket_name(request)

Get the name of s3 bucket.

Source code in saritasa_s3_tools/testing/plugin.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
@pytest.fixture(scope="session")
def s3_bucket_name(
    request: pytest.FixtureRequest,
) -> str:
    """Get the name of s3 bucket."""
    worker_input = getattr(
        request.config,
        "workerinput",
        {
            "workerid": "",
        },
    )
    return "-".join(
        filter(
            None,
            (
                str(request.config.getini("s3_bucket_name")),
                worker_input["workerid"],
            ),
        ),
    )

s3_bucket_policy()

Get the policy of s3 bucket.

Source code in saritasa_s3_tools/testing/plugin.py 
187
188
189
190
@pytest.fixture(scope="session")
def s3_bucket_policy() -> dict[str, typing.Any]:
    """Get the policy of s3 bucket."""
    return {}

s3_client(boto3_client, s3_bucket)

Set up s3 client.

Source code in saritasa_s3_tools/testing/plugin.py 
280
281
282
283
284
285
286
287
288
289
@pytest.fixture(scope="session")
def s3_client(
    boto3_client: mypy_boto3_s3.S3Client,
    s3_bucket: str,
) -> saritasa_s3_tools.S3Client:
    """Set up s3 client."""
    return saritasa_s3_tools.S3Client(
        boto3_client=boto3_client,
        default_bucket=s3_bucket,
    )

s3_endpoint_url_getter(request)

Set up url getter.

Source code in saritasa_s3_tools/testing/plugin.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
@pytest.fixture(scope="session")
def s3_endpoint_url_getter(
    request: pytest.FixtureRequest,
) -> (
    collections.abc.Callable[
        [],
        str | None,
    ]
    | None
):
    """Set up url getter."""
    if s3_endpoint_url := request.config.getini("s3_endpoint_url"):
        return lambda: str(s3_endpoint_url)
    return None  # pragma: no cover

s3_region(request)

Get s3 region.

Source code in saritasa_s3_tools/testing/plugin.py 
90
91
92
93
94
95
@pytest.fixture(scope="session")
def s3_region(
    request: pytest.FixtureRequest,
) -> str:
    """Get s3 region."""
    return str(request.config.getini("s3_region"))