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
|
"""
Ensure that the responses module plays nice with our mocks
"""
from http.server import BaseHTTPRequestHandler
from unittest import SkipTest, TestCase
import boto3
import requests
import responses
from moto import mock_aws, settings
from moto.core.models import override_responses_real_send
from moto.core.versions import RESPONSES_VERSION
from moto.utilities.distutils_version import LooseVersion
from .utilities import SimpleServer
class TestResponsesModule(TestCase):
def setUp(self) -> None:
if settings.TEST_SERVER_MODE:
raise SkipTest("No point in testing responses-decorator in ServerMode")
@mock_aws
@responses.activate
def test_moto_first(self) -> None: # type: ignore[misc]
"""
Verify we can activate a user-defined `responses` on top of our Moto mocks
"""
self.moto_responses_compatibility()
@responses.activate
@mock_aws
def test_moto_second(self) -> None:
"""
Verify we can load Moto after activating a `responses`-mock
"""
self.moto_responses_compatibility()
def moto_responses_compatibility(self) -> None:
responses.add(
responses.GET, url="http://127.0.0.1/lkdsfjlkdsa", json={"a": "4"}
)
s3 = boto3.client("s3", region_name="us-east-1")
s3.create_bucket(Bucket="mybucket")
s3.put_object(Bucket="mybucket", Key="name", Body="value")
s3.get_object(Bucket="mybucket", Key="name")["Body"].read()
with requests.get("http://127.0.0.1/lkdsfjlkdsa") as r:
assert r.json() == {"a": "4"}
@responses.activate
def test_moto_as_late_as_possible(self) -> None:
"""
Verify we can load moto after registering a response
"""
responses.add(
responses.GET, url="http://127.0.0.1/lkdsfjlkdsa", json={"a": "4"}
)
with mock_aws():
s3 = boto3.client("s3", region_name="us-east-1")
s3.create_bucket(Bucket="mybucket")
s3.put_object(Bucket="mybucket", Key="name", Body="value")
# This mock exists within Moto
with requests.get("http://127.0.0.1/lkdsfjlkdsa") as r:
assert r.json() == {"a": "4"}
# And outside of Moto
with requests.get("http://127.0.0.1/lkdsfjlkdsa") as r:
assert r.json() == {"a": "4"}
class WebRequestHandler(BaseHTTPRequestHandler):
def do_GET(self) -> None:
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(b"real response")
@mock_aws
class TestResponsesMockWithPassThru(TestCase):
"""
https://github.com/getmoto/moto/issues/6417
"""
def setUp(self) -> None:
if RESPONSES_VERSION < LooseVersion("0.24.0"):
raise SkipTest("Can only test this with responses >= 0.24.0")
self.server = SimpleServer(WebRequestHandler)
self.server.start()
host, port = self.server.get_host_and_port()
self.server_url = f"http://{host}:{port}"
self.r_mock = responses.RequestsMock(assert_all_requests_are_fired=True)
override_responses_real_send(self.r_mock)
self.r_mock.start()
self.r_mock.add_passthru(self.server_url)
def tearDown(self) -> None:
self.r_mock.stop() # type: ignore
self.r_mock.reset() # type: ignore
override_responses_real_send(None)
self.server.stop()
def http_requests(self) -> str:
# Mock this website
requests.post("https://example.org")
# Passthrough this website
assert requests.get(self.server_url).content == b"real response"
return "OK"
def aws_and_http_requests(self) -> str:
ddb = boto3.client("dynamodb", "us-east-1")
assert ddb.list_tables()["TableNames"] == []
self.http_requests()
return "OK"
def test_http_requests(self) -> None:
self.r_mock.add(responses.POST, "https://example.org", status=200) # type: ignore
self.assertEqual("OK", self.http_requests())
def test_aws_and_http_requests(self) -> None:
self.r_mock.add(responses.POST, "https://example.org", status=200) # type: ignore
self.assertEqual("OK", self.aws_and_http_requests())
|
