"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)