110 lines
3.6 KiB
Python
110 lines
3.6 KiB
Python
"""
|
|
Ensure that the responses module plays nice with our mocks
|
|
"""
|
|
|
|
import boto3
|
|
import requests
|
|
import responses
|
|
from moto import mock_dynamodb, mock_s3, 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 unittest import SkipTest, TestCase
|
|
|
|
|
|
class TestResponsesModule(TestCase):
|
|
def setUp(self) -> None:
|
|
if settings.TEST_SERVER_MODE:
|
|
raise SkipTest("No point in testing responses-decorator in ServerMode")
|
|
|
|
@mock_s3
|
|
@responses.activate
|
|
def test_moto_first(self) -> None:
|
|
"""
|
|
Verify we can activate a user-defined `responses` on top of our Moto mocks
|
|
"""
|
|
self.moto_responses_compatibility()
|
|
|
|
@responses.activate
|
|
@mock_s3
|
|
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")
|
|
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_s3():
|
|
s3 = boto3.client("s3")
|
|
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"}
|
|
|
|
|
|
@mock_dynamodb
|
|
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.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("http://ip.jsontest.com")
|
|
|
|
def tearDown(self) -> None:
|
|
self.r_mock.stop()
|
|
self.r_mock.reset()
|
|
override_responses_real_send(None)
|
|
|
|
def http_requests(self) -> str:
|
|
# Mock this website
|
|
requests.post("https://example.org")
|
|
|
|
# Passthrough this website
|
|
assert requests.get("http://ip.jsontest.com").status_code == 200
|
|
|
|
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)
|
|
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)
|
|
self.assertEqual("OK", self.aws_and_http_requests())
|