Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
client.call(service, mock_endpoint, session=mock_session, timeout_spec=mock_timeout, logger=mock_logger)
mock_session.send.assert_any_call(
request,
timeout=(mock_timeout.connection_timeout, mock_timeout.read_timeout),
stream=mock_endpoint.streaming,
allow_redirects=True,
)
mock_logger.info.assert_any_call("GET http://host1.biz/foo/")
mock_logger.info.assert_any_call("200 http://host1.biz/foo/")
request.method = "PUT"
client.call(
service, mock_endpoint, method="PUT", session=mock_session, timeout_spec=mock_timeout, logger=mock_logger
)
mock_session.send.assert_any_call(
request,
timeout=(mock_timeout.connection_timeout, mock_timeout.read_timeout),
stream=mock_endpoint.streaming,
allow_redirects=True,
)
mock_get_adapted_session.assert_called_once_with(MockAdapter())
mock_session.send.assert_called_once_with(
request,
timeout=(mock_timeout.connection_timeout, mock_timeout.read_timeout),
stream=mock_endpoint.streaming,
allow_redirects=True,
)
mock_logger.info.assert_any_call("GET http://host1.biz/foo/")
mock_logger.info.assert_any_call("200 http://host1.biz/foo/")
mock_endpoint.default_method = "POST"
request.method = "POST"
client.call(service, mock_endpoint, session=mock_session, timeout_spec=mock_timeout, logger=mock_logger)
mock_session.send.assert_any_call(
request,
timeout=(mock_timeout.connection_timeout, mock_timeout.read_timeout),
stream=mock_endpoint.streaming,
allow_redirects=True,
)
mock_logger.info.assert_any_call("GET http://host1.biz/foo/")
mock_logger.info.assert_any_call("200 http://host1.biz/foo/")
request.method = "PUT"
client.call(
service, mock_endpoint, method="PUT", session=mock_session, timeout_spec=mock_timeout, logger=mock_logger
)
def test_call_with_explicit_encoding(self, mock_response, mock_endpoint):
service = mock.Mock()
service.get_hosts.return_value = ["http://host1.biz"]
service.required_headers = {}
session = mock.Mock()
session.send.return_value = mock_response
client.call(service, mock_endpoint, session=session, logger=mock.Mock(), encoding="FAKE-CODEC")
assert "FAKE-CODEC" == mock_response.encoding
mock_endpoint.streaming = True
request = mock.Mock()
request.url = "http://host1.biz/foo/"
mock_build_request_object.return_value = request
mock_logger = mock.Mock()
mock_response.status_code = 200
mock_response.url = "http://host1.biz/foo/"
mock_session = MockSession()
mock_session.send.return_value = mock_response
mock_get_adapted_session.return_value = mock_session
client.call(service, mock_endpoint, timeout_spec=mock_timeout, logger=mock_logger)
mock_get_adapted_session.assert_called_once_with(MockAdapter())
mock_session.send.assert_called_once_with(
request,
timeout=(mock_timeout.connection_timeout, mock_timeout.read_timeout),
stream=mock_endpoint.streaming,
allow_redirects=True,
)
mock_logger.info.assert_any_call("GET http://host1.biz/foo/")
mock_logger.info.assert_any_call("200 http://host1.biz/foo/")
mock_endpoint.default_method = "POST"
request.method = "POST"
client.call(service, mock_endpoint, session=mock_session, timeout_spec=mock_timeout, logger=mock_logger)
def test_call_with_existing_session(self, mock_get_adapted_session, mock_response, mock_endpoint):
service = mock.Mock()
service.get_hosts.return_value = ["http://host1.biz"]
service.required_headers = {}
mock_logger = mock.Mock()
session = mock.Mock()
session.send.return_value = mock_response
client.call(service, mock_endpoint, session=session, logger=mock_logger)
assert not mock_get_adapted_session.called
assert not session.close.called