Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_call(self, mocker, request_definition, request_builder):
args = ()
kwargs = {}
request_preparer = mocker.Mock(spec=builder.RequestPreparer)
request_preparer.create_request_builder.return_value = request_builder
execution_builder = mocker.Mock(spec=io.RequestExecutionBuilder)
execution_builder.build().start.return_value = object()
factory = builder.CallFactory(
request_preparer, request_definition, lambda: execution_builder
)
assert (
factory(*args, **kwargs)
is execution_builder.build().start.return_value
)
request_definition.define_request.assert_called_with(
request_builder, args, kwargs
)
request_preparer.prepare_request.assert_called_with(
request_builder, execution_builder
)
execution_builder.build().start.assert_called_with(
(request_builder.method, request_builder.url, request_builder.info)
def test_retry_fail_with_twisted(mock_client, mock_response):
from twisted.internet import defer
@defer.inlineCallbacks
def return_response():
yield
defer.returnValue(mock_response)
# Setup
CustomException = type("CustomException", (Exception,), {})
mock_response.with_json({"id": 123, "name": "prkumar"})
mock_client.with_side_effect(
[Exception, CustomException, return_response()]
)
mock_client.with_io(io.TwistedStrategy())
github = GitHub(base_url=BASE_URL, client=mock_client)
# Run
with pytest.raises(CustomException):
yield github.get_user("prkumar")
assert len(mock_client.history) == 2
def test_io(self):
assert isinstance(requests_.RequestsClient.io(), io.BlockingStrategy)
def test_io(self):
assert isinstance(aiohttp_.AiohttpClient.io(), io.AsyncioStrategy)
def test_retry_with_asyncio(mock_client, mock_response):
import asyncio
@asyncio.coroutine
def coroutine():
return mock_response
# Setup
mock_response.with_json({"id": 123, "name": "prkumar"})
mock_client.with_side_effect([Exception, coroutine()])
mock_client.with_io(io.AsyncioStrategy())
github = GitHub(base_url=BASE_URL, client=mock_client)
# Run
awaitable = github.get_user("prkumar")
loop = asyncio.get_event_loop()
response = loop.run_until_complete(asyncio.ensure_future(awaitable))
# Verify
assert len(mock_client.history) == 2
assert response.json() == {"id": 123, "name": "prkumar"}
def io():
return io.BlockingStrategy()
def build(self, definition, consumer=None):
"""
Creates a callable that uses the provided definition to execute
HTTP requests when invoked.
"""
return CallFactory(
RequestPreparer(self, consumer),
definition,
io.RequestExecutionBuilder,
)
def io():
return io.AsyncioStrategy()
def io():
return io.TwistedStrategy()
def request_template(self):
return io.CompositeRequestTemplate(self._request_templates)