How to use the uplink.clients.io function in uplink

To help you get started, we’ve selected a few uplink examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github prkumar / uplink / tests / unit / test_builder.py View on Github external
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)
github prkumar / uplink / tests / integration / test_retry.py View on Github external
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
github prkumar / uplink / tests / unit / test_clients.py View on Github external
def test_io(self):
        assert isinstance(requests_.RequestsClient.io(), io.BlockingStrategy)
github prkumar / uplink / tests / unit / test_clients.py View on Github external
def test_io(self):
        assert isinstance(aiohttp_.AiohttpClient.io(), io.AsyncioStrategy)
github prkumar / uplink / tests / integration / test_retry.py View on Github external
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"}
github prkumar / uplink / uplink / builder.py View on Github external
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,
        )
github prkumar / uplink / uplink / helpers.py View on Github external
def request_template(self):
        return io.CompositeRequestTemplate(self._request_templates)