1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
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
45
46
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
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
|
import sys
from typing import Any
from unittest import SkipTest
import boto3
import pytest
from moto import mock_aws, settings
from moto.utilities.distutils_version import LooseVersion
boto3_version = sys.modules["botocore"].__version__
@pytest.fixture(scope="function", name="aws_credentials")
def fixture_aws_credentials(monkeypatch: Any) -> None: # type: ignore[misc]
"""Mocked AWS Credentials for moto."""
if settings.TEST_SERVER_MODE:
raise SkipTest("No point in testing this in ServerMode.")
monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing")
monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing")
monkeypatch.setenv("AWS_SECURITY_TOKEN", "testing")
monkeypatch.setenv("AWS_SESSION_TOKEN", "testing")
def test_mock_works_with_client_created_inside(
aws_credentials: Any,
) -> None:
m = mock_aws()
m.start()
client = boto3.client("s3", region_name="us-east-1")
b = client.list_buckets()
assert b["Buckets"] == []
m.stop()
def test_mock_works_with_client_created_outside(
aws_credentials: Any,
) -> None:
# Create the boto3 client first
outside_client = boto3.client("s3", region_name="us-east-1")
# Start the mock afterwards - which does not mock an already created client
m = mock_aws()
m.start()
# So remind us to mock this client
from moto.core import patch_client
patch_client(outside_client)
b = outside_client.list_buckets()
assert b["Buckets"] == []
m.stop()
def test_mock_works_with_resource_created_outside(
aws_credentials: Any,
) -> None:
# Create the boto3 client first
outside_resource = boto3.resource("s3", region_name="us-east-1")
# Start the mock afterwards - which does not mock an already created resource
m = mock_aws()
m.start()
# So remind us to mock this resource
from moto.core import patch_resource
patch_resource(outside_resource)
assert not list(outside_resource.buckets.all())
m.stop()
def test_patch_can_be_called_on_a_mocked_client() -> None:
if LooseVersion(boto3_version) < LooseVersion("1.29.0"):
raise SkipTest("Error handling is different in newer versions")
# start S3 after the mock, ensuring that the client contains our event-handler
m = mock_aws()
m.start()
client = boto3.client("s3", region_name="us-east-1")
from moto.core import patch_client
patch_client(client)
patch_client(client)
# At this point, our event-handler should only be registered once, by `mock_aws`
# `patch_client` should not re-register the event-handler
test_object = b"test_object_text"
client.create_bucket(Bucket="buck")
client.put_object(Bucket="buck", Key="to", Body=test_object)
got_object = client.get_object(Bucket="buck", Key="to")["Body"].read()
assert got_object == test_object
m.stop()
def test_patch_client_does_not_work_for_random_parameters() -> None:
from moto.core import patch_client
with pytest.raises(Exception, match="Argument sth should be of type boto3.client"):
patch_client("sth") # type: ignore[arg-type]
def test_patch_resource_does_not_work_for_random_parameters() -> None:
from moto.core import patch_resource
with pytest.raises(
Exception, match="Argument sth should be of type boto3.resource"
):
patch_resource("sth")
class ImportantBusinessLogic:
def __init__(self) -> None:
self._s3 = boto3.client("s3", region_name="us-east-1")
def do_important_things(self) -> list[str]:
return self._s3.list_buckets()["Buckets"]
def test_mock_works_when_replacing_client(
aws_credentials: Any,
) -> None:
if LooseVersion(boto3_version) < LooseVersion("1.29.0"):
raise SkipTest("Error handling is different in newer versions")
logic = ImportantBusinessLogic()
m = mock_aws()
m.start()
# This will fail, as the S3 client was created before the mock was initialized
try:
logic.do_important_things()
except Exception as e:
assert str(e) == "InvalidAccessKeyId"
client_initialized_after_mock = boto3.client("s3", region_name="us-east-1")
logic._s3 = client_initialized_after_mock
# This will work, as we now use a properly mocked client
assert logic.do_important_things() == []
m.stop()
|
