Code Coverage

Cobertura Coverage Report > osmclient.common >

http.py

Trend

File Coverage summary

NameClassesLinesConditionals
http.py
100%
1/1
18%
13/71
100%
0/0

Coverage Breakdown by Class

NameLinesConditionals
http.py
18%
13/71
N/A

Source

osmclient/common/http.py
1 # Copyright 2017 Sandvine
2 #
3 # All Rights Reserved.
4 #
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
8 #
9 #         http://www.apache.org/licenses/LICENSE-2.0
10 #
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
15 #    under the License.
16
17 1 import requests
18 1 import urllib3
19 1 from requests.auth import HTTPBasicAuth
20 1 import json
21 1 import logging
22
23 1 urllib3.disable_warnings()
24
25
26 1 class Http(object):
27 1     def __init__(self, url, user="admin", password="admin"):
28 0         self._url = url
29 0         self._user = user
30 0         self._password = password
31 0         self._http_header = None
32 0         self._logger = logging.getLogger("osmclient")
33
34 1     def set_http_header(self, header):
35 0         self._http_header = header
36
37 1     def _get_requests_cmd(self, endpoint):
38 0         requests_cmd = requests.Request()
39 0         requests_cmd.url = self._url + endpoint
40 0         requests_cmd.auth = HTTPBasicAuth(self._user, self._password)
41 0         if self._http_header:
42 0             requests_cmd.headers = self._http_header
43 0         return requests_cmd
44
45 1     def get_cmd(self, endpoint):
46 0         session_cmd = requests.Session()
47 0         requests_cmd = self._get_requests_cmd(endpoint)
48 0         requests_cmd.method = "GET"
49 0         self._logger.info(
50             "Request METHOD: {} URL: {}".format("GET", self._url + endpoint)
51         )
52 0         requests_cmd = requests_cmd.prepare()
53 0         resp = session_cmd.send(requests_cmd, verify=False)
54 0         http_code = resp.status_code
55 0         self._logger.info("Response HTTPCODE: {}".format(http_code))
56 0         data = resp.content
57 0         session_cmd.close()
58 0         if data:
59 0             self._logger.debug("Response DATA: {}".format(json.loads(data.decode())))
60 0             return json.loads(data.decode())
61 0         return None
62
63 1     def delete_cmd(self, endpoint):
64 0         session_cmd = requests.Session()
65 0         requests_cmd = self._get_requests_cmd(endpoint)
66 0         requests_cmd.method = "DELETE"
67 0         self._logger.info(
68             "Request METHOD: {} URL: {}".format("DELETE", self._url + endpoint)
69         )
70 0         requests_cmd = requests_cmd.prepare()
71 0         resp = session_cmd.send(requests_cmd, verify=False)
72 0         http_code = resp.status_code
73 0         self._logger.info("Response HTTPCODE: {}".format(http_code))
74 0         data = resp.content
75 0         session_cmd.close()
76 0         if data:
77 0             self._logger.debug("Response DATA: {}".format(json.loads(data.decode())))
78 0             return json.loads(data.decode())
79 0         return None
80
81 1     def post_cmd(
82         self,
83         endpoint="",
84         postfields_dict=None,
85         formfile=None,
86     ):
87 0         session_cmd = requests.Session()
88 0         requests_cmd = self._get_requests_cmd(endpoint)
89 0         requests_cmd.method = "POST"
90
91 0         if postfields_dict is not None:
92 0             requests_cmd.json = json.dumps(postfields_dict)
93
94 0         if formfile is not None:
95 0             requests_cmd.files = {formfile[0]: formfile[1]}
96
97 0         self._logger.info(
98             "Request METHOD: {} URL: {}".format("POST", self._url + endpoint)
99         )
100 0         requests_cmd = requests_cmd.prepare()
101 0         resp = session_cmd.send(requests_cmd, verify=False)
102 0         http_code = resp.status_code
103 0         self._logger.info("Response HTTPCODE: {}".format(http_code))
104 0         data = resp.content
105 0         session_cmd.close()
106 0         if data:
107 0             self._logger.debug("Response DATA: {}".format(json.loads(data.decode())))
108 0             return json.loads(data.decode())
109 0         return None