Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
curl = pycurl.Curl()
curl.setopt = MagicMock()
curl.perform = MagicMock()
curl.getinfo = MagicMock(return_value=200)
curl.close = MagicMock()
pycurl.Curl = MagicMock(return_value=curl)
method = 'post'
url = 'http://localhost:4502/.cqactions.html'
params = {'foo1': 'bar1', 'foo2': ['bar2a', 'bar2b']}
handlers = {200: self._handler_dummy}
result = pyaem2.bagofrequests.request(method, url, params, handlers)
curl.setopt.assert_any_call(pycurl.POST, 1)
curl.setopt.assert_any_call(pycurl.POSTFIELDS, 'foo1=bar1&foo2=bar2a&foo2=bar2b')
curl.setopt.assert_any_call(pycurl.URL, 'http://localhost:4502/.cqactions.html')
curl.setopt.assert_any_call(pycurl.FOLLOWLOCATION, 1)
curl.setopt.assert_any_call(pycurl.FRESH_CONNECT, 1)
# 6 calls including the one with pycurl.WRITEFUNCTION
self.assertEqual(curl.setopt.call_count, 6)
curl.perform.assert_called_once_with()
curl.getinfo.assert_called_once_with(pycurl.HTTP_CODE)
curl.close.assert_called_once_with()
self.assertEqual(result.is_success(), True)
self.assertEqual(result.message, 'some dummy message')
self.assertEqual(result.response['request']['method'], 'post')
self.assertEqual(result.response['request']['url'], 'http://localhost:4502/.cqactions.html')
self.assertEqual(result.response['request']['params'], params)
self.server_reply = StringIO()
self.c = pycurl.Curl()
self.url = url
self.data = data
self.c.setopt(pycurl.URL, self.url)
self.c.setopt(pycurl.HEADERFUNCTION, self.server_reply.write)
if file is None:
self.fp = StringIO()
self.c.setopt(pycurl.WRITEFUNCTION, self.fp.write)
else:
self.fp = file
self.c.setopt(pycurl.WRITEDATA, self.fp)
if self.data != None:
self.c.setopt(pycurl.POST, 1)
self.c.setopt(pycurl.POSTFIELDS, urllib.urlencode(self.data))
def post(self, url="", params=None):
"""Ship a POST request to a specified CGI, capture the response."""
self.set_option(pycurl.POST, 1)
if params:
self.set_option(pycurl.POSTFIELDS, urllib_parse.urlencode(params))
return self.__request(url)
if self.conn:
self.conn.close()
self.headers = ['Accept: application/json', 'Expect:', 'Connection: close']
self.conn = pycurl.Curl()
self.conn.setopt(pycurl.SSL_VERIFYHOST, False)
self.conn.setopt(pycurl.SSL_VERIFYPEER, False)
self.conn.setopt(pycurl.USERPWD, "%s:%s" % (self.api_user, self.api_password))
self.conn.setopt(pycurl.URL, self.api_url)
self.conn.setopt(pycurl.VERBOSE, 0)
self.conn.setopt(pycurl.WRITEFUNCTION, self.on_receive)
self.conn.setopt(pycurl.NOSIGNAL, 1)
self.conn.setopt(pycurl.NOPROGRESS, 1)
self.conn.setopt(pycurl.HTTPHEADER, self.headers)
if self.method == 'post':
self.conn.setopt(pycurl.POST, 1)
self.conn.setopt(pycurl.POSTFIELDS, urllib.urlencode(POST_PARAMS))
elif self.method == 'get':
self.conn.setopt(pycurl.POST, 0)
assert isinstance(method, str)
assert isinstance(url, str)
assert isinstance(post_data, str)
assert compat.all(isinstance(i, str) for i in headers)
# Buffer for response
resp_buffer = StringIO()
# Configure client for request
curl.setopt(pycurl.VERBOSE, False)
curl.setopt(pycurl.NOSIGNAL, True)
curl.setopt(pycurl.USERAGENT, http.HTTP_GANETI_VERSION)
curl.setopt(pycurl.PROXY, "")
curl.setopt(pycurl.CUSTOMREQUEST, str(method))
curl.setopt(pycurl.URL, url)
curl.setopt(pycurl.POSTFIELDS, post_data)
curl.setopt(pycurl.HTTPHEADER, headers)
if req.read_timeout is None:
curl.setopt(pycurl.TIMEOUT, 0)
else:
curl.setopt(pycurl.TIMEOUT, int(req.read_timeout))
# Disable SSL session ID caching (pycurl >= 7.16.0)
if hasattr(pycurl, "SSL_SESSIONID_CACHE"):
curl.setopt(pycurl.SSL_SESSIONID_CACHE, False)
curl.setopt(pycurl.WRITEFUNCTION, resp_buffer.write)
# Pass cURL object to external config function
if req.curl_config_fn:
req.curl_config_fn(curl)
# let's set the HTTP request method
if method == 'GET':
self.curl.setopt(pycurl.HTTPGET, 1)
elif method == 'POST':
self.curl.setopt(pycurl.POST, 1)
elif method == 'PUT':
self.curl.setopt(pycurl.UPLOAD, 1)
else:
self.curl.setopt(pycurl.CUSTOMREQUEST, method)
if data:
if method == "PUT":
self.data = StringIO.StringIO(data)
self.curl.setopt(pycurl.READFUNCTION, self.data.read)
self.curl.setopt(pycurl.INFILESIZE, len(self.data.getvalue()))
else:
self.curl.setopt(pycurl.POSTFIELDS, self.data)
self.curl.setopt(pycurl.POSTFIELDSIZE, len(self.data))
if headers:
self.curl.setopt(pycurl.HTTPHEADER, headers)
# self reference required cause CurlMulti will only return
# Curl handles
self.curl.request = self
def post_request(self, url, data, timeout_ms=500):
if self.debug:
print url
buffer = cStringIO.StringIO()
curl = pycurl.Curl()
curl.setopt(pycurl.URL, url)
curl.setopt(pycurl.TIMEOUT_MS, timeout_ms)
curl.setopt(pycurl.WRITEFUNCTION, buffer.write)
curl.setopt(pycurl.NOSIGNAL, 1)
curl.setopt(pycurl.POST, 1)
curl.setopt(pycurl.POSTFIELDS, data)
curl.perform()
result = buffer.getvalue()
buffer.close()
curl.close()
return result
if 'proxy_user' in self.proxy:
c.setopt(pycurl.PROXYUSERPWD, "%(proxy_user)s:%(proxy_pass)s" % self.proxy)
self.buf = StringIO()
c.setopt(pycurl.WRITEFUNCTION, self.buf.write)
#c.setopt(pycurl.READFUNCTION, self.read)
#self.body = StringIO(body)
#c.setopt(pycurl.HEADERFUNCTION, self.header)
if self.cacert:
c.setopt(c.CAINFO, self.cacert)
c.setopt(pycurl.SSL_VERIFYPEER, self.cacert and 1 or 0)
c.setopt(pycurl.SSL_VERIFYHOST, self.cacert and 2 or 0)
c.setopt(pycurl.CONNECTTIMEOUT, self.timeout / 6)
c.setopt(pycurl.TIMEOUT, self.timeout)
if method == 'POST':
c.setopt(pycurl.POST, 1)
c.setopt(pycurl.POSTFIELDS, body)
if headers:
hdrs = ['%s: %s' % (k, v) for k, v in headers.items()]
c.setopt(pycurl.HTTPHEADER, hdrs)
c.perform()
c.close()
return {}, self.buf.getvalue()
def post(self, url='', params=None, json=None):
if json:
body = json_lib.dumps(json)
self.set_option(pycurl.POSTFIELDS, body)
response, response_code, msg = super().post(url, params)
return self._response_decode(response, response_code, msg)
:type method: str
:param url: URL to send HTTP request to
:type url: str
:param params: Request parameters key-value pairs, use array value to represent multi parameters with the same name
:type params: dict
:param handlers: Response handlers key-value pairs, keys are response http code, values are callback methods
:type handlers: dict
:returns: PyAem2Result -- Result of the request containing status, response http code and body, and request info
:raises: PyAem2Exception
"""
curl = pycurl.Curl()
body_io = BytesIO()
if method == 'post':
curl.setopt(pycurl.POST, 1)
curl.setopt(pycurl.POSTFIELDS, urlencode(params, True))
elif method == 'delete':
curl.setopt(pycurl.CUSTOMREQUEST, method)
elif method == 'head':
curl.setopt(pycurl.HEADER, True)
curl.setopt(pycurl.NOBODY, True)
else:
url = '{0}?{1}'.format(url, urlencode(params, True))
curl.setopt(pycurl.URL, url)
curl.setopt(pycurl.FOLLOWLOCATION, 1)
curl.setopt(pycurl.FRESH_CONNECT, 1)
curl.setopt(pycurl.WRITEFUNCTION, body_io.write)
curl.perform()
response = {