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_init(self):
RequestsNetworkWrapper()
def test_init_with_session(self):
session = {}
rnw = RequestsNetworkWrapper(session=session)
self.assertIs(session, rnw.session)
def setUp(self):
nw = Mock(RequestsNetworkWrapper())
nw.GET.return_value = "GET"
nw.POST.return_value = "POST"
nw.PUT.return_value = "PUT"
nw.DELETE.return_value = "DELETE"
self.sc = ServiceClient("endpoint/", network_wrapper=nw)
self.scClientDoesNotEndWithSlash = ServiceClient("endpoint", network_wrapper=nw)
def setUp(self):
session = Mock(requests.session())
session.get.return_value = self.mock_response(200)
session.post.return_value = self.mock_response(200)
session.put.return_value = self.mock_response(200)
session.delete.return_value = self.mock_response(204)
self.rnw = RequestsNetworkWrapper(session=session)
def __init__(self, endpoint, network_wrapper=None):
if network_wrapper is None:
network_wrapper = RequestsNetworkWrapper(session=requests.session())
self.network_wrapper = network_wrapper
pattern = compile(".*(:[0-9]+)$")
if not endpoint.endswith("/") and not pattern.match(endpoint):
logger.warning(f"endpoint {endpoint} does not end with '/': appending.")
endpoint = endpoint + "/"
self.endpoint = endpoint
endpoint : str, optional
The server URL.
query_timeout : float, optional
The timeout for query operations.
"""
_check_hostname(endpoint)
self._endpoint = endpoint
session = requests.session()
session.verify = False
requests.packages.urllib3.disable_warnings()
# Setup the communications layer.
network_wrapper = RequestsNetworkWrapper(session)
service_client = ServiceClient(self._endpoint, network_wrapper)
self._service = RESTServiceClient(service_client)
if query_timeout is not None and query_timeout > 0:
self.query_timeout = query_timeout
else:
self.query_timeout = 0.0