from osmclient.common import http
from osmclient.common.exceptions import OsmHttpException, NotFound
from osmclient.common import http
from osmclient.common.exceptions import OsmHttpException, NotFound
def __init__(self, url, user='admin', password='admin', **kwargs):
self._url = url
def __init__(self, url, user='admin', password='admin', **kwargs):
self._url = url
self._default_query_admin = self._complete_default_query_admin()
def _complete_default_query_admin(self):
self._default_query_admin = self._complete_default_query_admin()
def _complete_default_query_admin(self):
curl_cmd.setopt(pycurl.VERBOSE, True)
if not skip_query_admin:
endpoint = self._complete_endpoint(endpoint)
curl_cmd.setopt(pycurl.VERBOSE, True)
if not skip_query_admin:
endpoint = self._complete_endpoint(endpoint)
curl_cmd.setopt(pycurl.URL, self._url + endpoint)
curl_cmd.setopt(pycurl.SSL_VERIFYPEER, 0)
curl_cmd.setopt(pycurl.SSL_VERIFYHOST, 0)
curl_cmd.setopt(pycurl.URL, self._url + endpoint)
curl_cmd.setopt(pycurl.SSL_VERIFYPEER, 0)
curl_cmd.setopt(pycurl.SSL_VERIFYHOST, 0)
curl_cmd = self._get_curl_cmd(endpoint, skip_query_admin)
curl_cmd.setopt(pycurl.CUSTOMREQUEST, "DELETE")
curl_cmd.setopt(pycurl.WRITEFUNCTION, data.write)
curl_cmd = self._get_curl_cmd(endpoint, skip_query_admin)
curl_cmd.setopt(pycurl.CUSTOMREQUEST, "DELETE")
curl_cmd.setopt(pycurl.WRITEFUNCTION, data.write)
curl_cmd.perform()
http_code = curl_cmd.getinfo(pycurl.HTTP_CODE)
self._logger.info("Response HTTPCODE: {}".format(http_code))
curl_cmd.perform()
http_code = curl_cmd.getinfo(pycurl.HTTP_CODE)
self._logger.info("Response HTTPCODE: {}".format(http_code))
jsondata = json.dumps(postfields_dict)
if 'password' in postfields_dict:
postfields_dict_copy = copy.deepcopy(postfields_dict)
jsondata = json.dumps(postfields_dict)
if 'password' in postfields_dict:
postfields_dict_copy = copy.deepcopy(postfields_dict)
formfile[1])))])
elif filename is not None:
with open(filename, 'rb') as stream:
formfile[1])))])
elif filename is not None:
with open(filename, 'rb') as stream:
self._logger.verbose("Request POSTFIELDS: Binary content")
curl_cmd.setopt(pycurl.POSTFIELDS, postdata)
if put_method:
self._logger.verbose("Request POSTFIELDS: Binary content")
curl_cmd.setopt(pycurl.POSTFIELDS, postdata)
if put_method:
curl_cmd.perform()
http_code = curl_cmd.getinfo(pycurl.HTTP_CODE)
self._logger.info("Response HTTPCODE: {}".format(http_code))
curl_cmd.perform()
http_code = curl_cmd.getinfo(pycurl.HTTP_CODE)
self._logger.info("Response HTTPCODE: {}".format(http_code))
skip_query_admin=skip_query_admin)
def patch_cmd(self, endpoint='', postfields_dict=None,
skip_query_admin=skip_query_admin)
def patch_cmd(self, endpoint='', postfields_dict=None,
- formfile=None, filename=None,
- skip_query_admin=False):
+ formfile=None, filename=None,
+ skip_query_admin=False):
self._logger.debug("")
return self.send_cmd(endpoint=endpoint,
postfields_dict=postfields_dict,
self._logger.debug("")
return self.send_cmd(endpoint=endpoint,
postfields_dict=postfields_dict,
curl_cmd = self._get_curl_cmd(endpoint, skip_query_admin)
curl_cmd.setopt(pycurl.HTTPGET, 1)
curl_cmd.setopt(pycurl.WRITEFUNCTION, data.write)
curl_cmd = self._get_curl_cmd(endpoint, skip_query_admin)
curl_cmd.setopt(pycurl.HTTPGET, 1)
curl_cmd.setopt(pycurl.WRITEFUNCTION, data.write)
curl_cmd.perform()
http_code = curl_cmd.getinfo(pycurl.HTTP_CODE)
self._logger.info("Response HTTPCODE: {}".format(http_code))
curl_cmd.perform()
http_code = curl_cmd.getinfo(pycurl.HTTP_CODE)
self._logger.info("Response HTTPCODE: {}".format(http_code))
def set_query_admin(self, **kwargs):
if 'all_projects' in kwargs:
def set_query_admin(self, **kwargs):
if 'all_projects' in kwargs: