1 # Copyright 2017 Sandvine
5 # Licensed under the Apache License, Version 2.0 (the "License"); you may
6 # not use this file except in compliance with the License. You may obtain
7 # a copy of the License at
9 # http://www.apache.org/licenses/LICENSE-2.0
11 # Unless required by applicable law or agreed to in writing, software
12 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
13 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
14 # License for the specific language governing permissions and limitations
17 from io
import BytesIO
24 def __init__(self
, url
, user
='admin', password
='admin'):
27 self
._password
= password
28 self
._http
_header
= None
30 def set_http_header(self
, header
):
31 self
._http
_header
= header
33 def _get_curl_cmd(self
, endpoint
):
34 curl_cmd
= pycurl
.Curl()
35 curl_cmd
.setopt(pycurl
.URL
, self
._url
+ endpoint
)
36 curl_cmd
.setopt(pycurl
.SSL_VERIFYPEER
, 0)
37 curl_cmd
.setopt(pycurl
.SSL_VERIFYHOST
, 0)
44 curl_cmd
.setopt(pycurl
.HTTPHEADER
, self
._http
_header
)
47 def get_cmd(self
, endpoint
):
50 curl_cmd
= self
._get
_curl
_cmd
(endpoint
)
51 curl_cmd
.setopt(pycurl
.HTTPGET
, 1)
52 curl_cmd
.setopt(pycurl
.WRITEFUNCTION
, data
.write
)
53 self
._logger
.info("Request METHOD: {} URL: {}".format("GET",self
._url
+ endpoint
))
55 http_code
= curl_cmd
.getinfo(pycurl
.HTTP_CODE
)
56 self
._logger
.info("Response HTTPCODE: {}".format(http_code
))
59 self
._logger
.debug("Response DATA: {}".format(json
.loads(data
.getvalue().decode())))
60 return json
.loads(data
.getvalue().decode())
63 def delete_cmd(self
, endpoint
):
65 curl_cmd
= self
._get
_curl
_cmd
(endpoint
)
66 curl_cmd
.setopt(pycurl
.CUSTOMREQUEST
, "DELETE")
67 curl_cmd
.setopt(pycurl
.WRITEFUNCTION
, data
.write
)
68 self
._logger
.info("Request METHOD: {} URL: {}".format("DELETE",self
._url
+ endpoint
))
70 http_code
= curl_cmd
.getinfo(pycurl
.HTTP_CODE
)
71 self
._logger
.info("Response HTTPCODE: {}".format(http_code
))
74 self
._logger
.debug("Response DATA: {}".format(json
.loads(data
.getvalue().decode())))
75 return json
.loads(data
.getvalue().decode())
78 def post_cmd(self
, endpoint
='', postfields_dict
=None, formfile
=None, ):
80 curl_cmd
= self
._get
_curl
_cmd
(endpoint
)
81 curl_cmd
.setopt(pycurl
.POST
, 1)
82 curl_cmd
.setopt(pycurl
.WRITEFUNCTION
, data
.write
)
84 if postfields_dict
is not None:
85 jsondata
= json
.dumps(postfields_dict
)
86 curl_cmd
.setopt(pycurl
.POSTFIELDS
, jsondata
)
88 if formfile
is not None:
95 self
._logger
.info("Request METHOD: {} URL: {}".format("POST",self
._url
+ endpoint
))
97 http_code
= curl_cmd
.getinfo(pycurl
.HTTP_CODE
)
98 self
._logger
.info("Response HTTPCODE: {}".format(http_code
))
101 self
._logger
.debug("Response DATA: {}".format(json
.loads(data
.getvalue().decode())))
102 return json
.loads(data
.getvalue().decode())