"Fossies" - the Fresh Open Source Software Archive 
Member "salt-3002.2/tests/unit/utils/test_http.py" (18 Nov 2020, 9135 Bytes) of package /linux/misc/salt-3002.2.tar.gz:
As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style:
standard) with prefixed line numbers.
Alternatively you can here
view or
download the uninterpreted source code file.
See also the latest
Fossies "Diffs" side-by-side code changes report for "test_http.py":
3002.1_vs_3002.2.
1 """
2 :codeauthor: Nicole Thomas <nicole@saltstack.com>
3 """
4
5 import socket
6 from contextlib import closing
7
8 import salt.utils.http as http
9 from tests.support.helpers import MirrorPostHandler, Webserver, slowTest
10 from tests.support.mock import MagicMock, patch
11 from tests.support.unit import TestCase, skipIf
12
13 try:
14 import tornado.curl_httpclient # pylint: disable=unused-import
15
16 HAS_CURL = True
17 except ImportError:
18 HAS_CURL = False
19
20
21 class HTTPTestCase(TestCase):
22 """
23 Unit TestCase for the salt.utils.http module.
24 """
25
26 @classmethod
27 def setUpClass(cls):
28 cls.post_webserver = Webserver(handler=MirrorPostHandler)
29 cls.post_webserver.start()
30 cls.post_web_root = cls.post_webserver.web_root
31
32 @classmethod
33 def tearDownClass(cls):
34 cls.post_webserver.stop()
35 del cls.post_webserver
36
37 # sanitize_url tests
38
39 def test_sanitize_url_hide_fields_none(self):
40 """
41 Tests sanitizing a url when the hide_fields kwarg is None.
42 """
43 mock_url = "https://api.testing.com/?&foo=bar&test=testing"
44 ret = http.sanitize_url(mock_url, hide_fields=None)
45 self.assertEqual(ret, mock_url)
46
47 def test_sanitize_url_no_elements(self):
48 """
49 Tests sanitizing a url when no elements should be sanitized.
50 """
51 mock_url = "https://api.testing.com/?&foo=bar&test=testing"
52 ret = http.sanitize_url(mock_url, [""])
53 self.assertEqual(ret, mock_url)
54
55 def test_sanitize_url_single_element(self):
56 """
57 Tests sanitizing a url with only a single element to be sanitized.
58 """
59 mock_url = (
60 "https://api.testing.com/?&keep_it_secret=abcdefghijklmn"
61 "&api_action=module.function"
62 )
63 mock_ret = (
64 "https://api.testing.com/?&keep_it_secret=XXXXXXXXXX&"
65 "api_action=module.function"
66 )
67 ret = http.sanitize_url(mock_url, ["keep_it_secret"])
68 self.assertEqual(ret, mock_ret)
69
70 def test_sanitize_url_multiple_elements(self):
71 """
72 Tests sanitizing a url with multiple elements to be sanitized.
73 """
74 mock_url = (
75 "https://api.testing.com/?rootPass=badpassword%21"
76 "&skipChecks=True&api_key=abcdefghijklmn"
77 "&NodeID=12345&api_action=module.function"
78 )
79 mock_ret = (
80 "https://api.testing.com/?rootPass=XXXXXXXXXX"
81 "&skipChecks=True&api_key=XXXXXXXXXX"
82 "&NodeID=12345&api_action=module.function"
83 )
84 ret = http.sanitize_url(mock_url, ["api_key", "rootPass"])
85 self.assertEqual(ret, mock_ret)
86
87 # _sanitize_components tests
88
89 def test_sanitize_components_no_elements(self):
90 """
91 Tests when zero elements need to be sanitized.
92 """
93 mock_component_list = ["foo=bar", "bar=baz", "hello=world"]
94 mock_ret = "foo=bar&bar=baz&hello=world&"
95 ret = http._sanitize_url_components(mock_component_list, "api_key")
96 self.assertEqual(ret, mock_ret)
97
98 def test_sanitize_components_one_element(self):
99 """
100 Tests a single component to be sanitized.
101 """
102 mock_component_list = ["foo=bar", "api_key=abcdefghijklmnop"]
103 mock_ret = "foo=bar&api_key=XXXXXXXXXX&"
104 ret = http._sanitize_url_components(mock_component_list, "api_key")
105 self.assertEqual(ret, mock_ret)
106
107 def test_sanitize_components_multiple_elements(self):
108 """
109 Tests two componenets to be sanitized.
110 """
111 mock_component_list = ["foo=bar", "foo=baz", "api_key=testing"]
112 mock_ret = "foo=XXXXXXXXXX&foo=XXXXXXXXXX&api_key=testing&"
113 ret = http._sanitize_url_components(mock_component_list, "foo")
114 self.assertEqual(ret, mock_ret)
115
116 @slowTest
117 def test_query_null_response(self):
118 """
119 This tests that we get a null response when raise_error=False and the
120 host/port cannot be reached.
121 """
122 host = "127.0.0.1"
123
124 # Find unused port
125 with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:
126 sock.bind((host, 0))
127 port = sock.getsockname()[1]
128
129 url = "http://{host}:{port}/".format(host=host, port=port)
130 result = http.query(url, raise_error=False)
131 assert result == {"body": None}, result
132
133 def test_query_error_handling(self):
134 ret = http.query("http://127.0.0.1:0")
135 self.assertTrue(isinstance(ret, dict))
136 self.assertTrue(isinstance(ret.get("error", None), str))
137 ret = http.query("http://myfoobardomainthatnotexist")
138 self.assertTrue(isinstance(ret, dict))
139 self.assertTrue(isinstance(ret.get("error", None), str))
140
141 def test_parse_cookie_header(self):
142 header = "; ".join(
143 [
144 "foo=bar",
145 "expires=Mon, 03-Aug-20 14:26:27 GMT",
146 "path=/",
147 "domain=.mydomain.tld",
148 "HttpOnly",
149 "SameSite=Lax",
150 "Secure",
151 ]
152 )
153 ret = http.parse_cookie_header(header)
154 cookie = ret.pop(0)
155 assert cookie.name == "foo", cookie.name
156 assert cookie.value == "bar", cookie.value
157 assert cookie.expires == 1596464787, cookie.expires
158 assert cookie.path == "/", cookie.path
159 assert cookie.domain == ".mydomain.tld", cookie.domain
160 assert cookie.secure
161 # Only one cookie should have been returned, if anything is left in the
162 # parse_cookie_header return then something went wrong.
163 assert not ret
164
165
166 class HTTPPostTestCase(TestCase):
167 """
168 Unit TestCase for the salt.utils.http module when
169 using POST method
170 """
171
172 @classmethod
173 def setUpClass(cls):
174 cls.post_webserver = Webserver(handler=MirrorPostHandler)
175 cls.post_webserver.start()
176 cls.post_web_root = cls.post_webserver.web_root
177
178 @classmethod
179 def tearDownClass(cls):
180 cls.post_webserver.stop()
181 del cls.post_webserver
182
183 def test_requests_multipart_formdata_post(self):
184 """
185 Test handling of a multipart/form-data POST using the requests backend
186 """
187 match_this = '{0}\r\nContent-Disposition: form-data; name="fieldname_here"\r\n\r\nmydatahere\r\n{0}--\r\n'
188 ret = http.query(
189 self.post_web_root,
190 method="POST",
191 data="mydatahere",
192 formdata=True,
193 formdata_fieldname="fieldname_here",
194 backend="requests",
195 )
196 body = ret.get("body", "")
197 boundary = body[: body.find("\r")]
198 self.assertEqual(body, match_this.format(boundary))
199
200 @skipIf(
201 HAS_CURL is False, "Missing prerequisites for tornado.curl_httpclient library"
202 )
203 def test_query_proxy(self):
204 """
205 Test http.query with tornado and with proxy opts set
206 and then test with no_proxy set to ensure we dont
207 run into issue #55192 again.
208 """
209 data = "mydatahere"
210 opts = {
211 "proxy_host": "127.0.0.1",
212 "proxy_port": 88,
213 "proxy_username": "salt_test",
214 "proxy_password": "super_secret",
215 }
216
217 mock_curl = MagicMock()
218
219 with patch("tornado.httpclient.HTTPClient.fetch", mock_curl):
220 ret = http.query(
221 self.post_web_root,
222 method="POST",
223 data=data,
224 backend="tornado",
225 opts=opts,
226 )
227
228 for opt in opts:
229 assert opt in mock_curl.call_args_list[0][1].keys()
230
231 opts["no_proxy"] = ["127.0.0.1"]
232
233 ret = http.query(
234 self.post_web_root, method="POST", data=data, backend="tornado", opts=opts
235 )
236 body = ret.get("body", "")
237 assert body == data
238
239
240 class HTTPGetTestCase(TestCase):
241 """
242 Unit TestCase for the salt.utils.http module when
243 using Get method
244 """
245
246 @classmethod
247 def setUpClass(cls):
248 cls.get_webserver = Webserver()
249 cls.get_webserver.start()
250
251 @classmethod
252 def tearDownClass(cls):
253 cls.get_webserver.stop()
254 del cls.get_webserver
255
256 def test_backends_decode_body_false(self):
257 """
258 test all backends when using
259 decode_body=False that it returns
260 bytes and does not try to decode
261 """
262 for backend in ["tornado", "requests", "urllib2"]:
263 ret = http.query(
264 self.get_webserver.url("custom.tar.gz"),
265 backend=backend,
266 decode_body=False,
267 )
268 body = ret.get("body", "")
269 assert isinstance(body, bytes)
270
271 def test_backends_decode_body_true(self):
272 """
273 test all backends when using
274 decode_body=True that it returns
275 string and decodes it.
276 """
277 for backend in ["tornado", "requests", "urllib2"]:
278 ret = http.query(self.get_webserver.url("core.sls"), backend=backend)
279 body = ret.get("body", "")
280 assert isinstance(body, str)