saltstack/salt
1
0
Fork 0
mirror of https://github.com/saltstack/salt.git synced 2025-04-17 10:10:20 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

Migrate old unit tests for salt.utils.http to pytest and alter proxy tests to conform to the forced use of requests

Browse source
This commit is contained in:
MKLeb 2023-06-13 17:56:54 -04:00 • committed by Daniel Wozniak
parent 055d1daa23
commit 2857ca7ab9
3 changed files with 275 additions and 315 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
salt/utils/http.py
View file

@ -234,20 +234,17 @@ def query(
proxy_username = None
proxy_password = None
proxy_args = None
http_proxy_url = None
if backend != "requests" and proxy_host and proxy_port:
log.debug("Switching to request backend due to the use of proxies.")
backend = "requests"
scheme = urllib.parse.urlparse(proxy_host).scheme
proxy_url = f"{proxy_host}:{proxy_port}"
if proxy_host and proxy_port:
if proxy_username and proxy_password:
proxy_url = salt.utils.url.add_http_basic_auth(
proxy_url, proxy_username, proxy_password
http_proxy_url = (
f"http://{proxy_username}:{proxy_password}@{proxy_host}:{proxy_port}"
)
if scheme and proxy_url:
proxy_args = (scheme, proxy_url)
else:
log.debug("Failed to set proxy details")
http_proxy_url = f"http://{proxy_host}:{proxy_port}"
match = re.match(
r"https?://((25[0-5]|2[0-4]\d|[01]?\d\d?)\.){3}(25[0-5]|2[0-4]\d|[01]?\d\d?)($|/)",
@ -362,8 +359,8 @@ def query(
log.trace("Request Headers: %s", sess.headers)
sess_cookies = sess.cookies
sess.verify = verify_ssl
if proxy_args:
sess.proxies = {proxy_args[0]: proxy_args[1]}
if http_proxy_url is not None:
sess.proxies = {"HTTP": http_proxy_url}
elif backend == "urllib2":
sess_cookies = None
else:

274
tests/pytests/unit/utils/test_http.py
View file

@ -1,7 +1,11 @@
import socket
from contextlib import closing
import pytest
import requests
from werkzeug.wrappers import Response # pylint: disable=3rd-party-module-not-gated
import salt.utils.http
import salt.utils.http as http
from tests.support.mock import MagicMock, patch
@ -16,10 +20,10 @@ def test_requests_session_verify_ssl_false(ssl_webserver, integration_files_dir)
if verify is True or verify is None:
with pytest.raises(requests.exceptions.SSLError) as excinfo:
session = salt.utils.http.session(**kwargs)
session = http.session(**kwargs)
ret = session.get(ssl_webserver.url("this.txt"))
else:
session = salt.utils.http.session(**kwargs)
session = http.session(**kwargs)
ret = session.get(ssl_webserver.url("this.txt"))
assert ret.status_code == 200
@ -29,7 +33,7 @@ def test_session_ca_bundle_verify_false():
test salt.utils.http.session when using
both ca_bunlde and verify_ssl false
"""
ret = salt.utils.http.session(ca_bundle="/tmp/test_bundle", verify_ssl=False)
ret = http.session(ca_bundle="/tmp/test_bundle", verify_ssl=False)
assert ret is False
@ -38,7 +42,7 @@ def test_session_headers():
test salt.utils.http.session when setting
headers
"""
ret = salt.utils.http.session(headers={"Content-Type": "application/json"})
ret = http.session(headers={"Content-Type": "application/json"})
assert ret.headers["Content-Type"] == "application/json"
@ -49,5 +53,263 @@ def test_session_ca_bundle():
fpath = "/tmp/test_bundle"
patch_os = patch("os.path.exists", MagicMock(return_value=True))
with patch_os:
ret = salt.utils.http.session(ca_bundle=fpath)
ret = http.session(ca_bundle=fpath)
assert ret.verify == fpath
def test_sanitize_url_hide_fields_none():
"""
Tests sanitizing a url when the hide_fields kwarg is None.
"""
mock_url = "https://api.testing.com/?&foo=bar&test=testing"
ret = http.sanitize_url(mock_url, hide_fields=None)
assert ret == mock_url
def test_sanitize_url_no_elements():
"""
Tests sanitizing a url when no elements should be sanitized.
"""
mock_url = "https://api.testing.com/?&foo=bar&test=testing"
ret = http.sanitize_url(mock_url, [""])
assert ret == mock_url
def test_sanitize_url_single_element():
"""
Tests sanitizing a url with only a single element to be sanitized.
"""
mock_url = (
"https://api.testing.com/?&keep_it_secret=abcdefghijklmn"
"&api_action=module.function"
)
mock_ret = (
"https://api.testing.com/?&keep_it_secret=XXXXXXXXXX&"
"api_action=module.function"
)
ret = http.sanitize_url(mock_url, ["keep_it_secret"])
assert ret == mock_ret
def test_sanitize_url_multiple_elements():
"""
Tests sanitizing a url with multiple elements to be sanitized.
"""
mock_url = (
"https://api.testing.com/?rootPass=badpassword%21"
"&skipChecks=True&api_key=abcdefghijklmn"
"&NodeID=12345&api_action=module.function"
)
mock_ret = (
"https://api.testing.com/?rootPass=XXXXXXXXXX"
"&skipChecks=True&api_key=XXXXXXXXXX"
"&NodeID=12345&api_action=module.function"
)
ret = http.sanitize_url(mock_url, ["api_key", "rootPass"])
assert ret == mock_ret
# _sanitize_components tests
def test_sanitize_components_no_elements():
"""
Tests when zero elements need to be sanitized.
"""
mock_component_list = ["foo=bar", "bar=baz", "hello=world"]
mock_ret = "foo=bar&bar=baz&hello=world&"
ret = http._sanitize_url_components(mock_component_list, "api_key")
assert ret == mock_ret
def test_sanitize_components_one_element():
"""
Tests a single component to be sanitized.
"""
mock_component_list = ["foo=bar", "api_key=abcdefghijklmnop"]
mock_ret = "foo=bar&api_key=XXXXXXXXXX&"
ret = http._sanitize_url_components(mock_component_list, "api_key")
assert ret == mock_ret
def test_sanitize_components_multiple_elements():
"""
Tests two componenets to be sanitized.
"""
mock_component_list = ["foo=bar", "foo=baz", "api_key=testing"]
mock_ret = "foo=XXXXXXXXXX&foo=XXXXXXXXXX&api_key=testing&"
ret = http._sanitize_url_components(mock_component_list, "foo")
assert ret == mock_ret
@pytest.mark.slow_test
def test_query_null_response():
"""
This tests that we get a null response when raise_error=False and the
host/port cannot be reached.
"""
host = "127.0.0.1"
# Find unused port
with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:
sock.bind((host, 0))
port = sock.getsockname()[1]
url = "http://{host}:{port}/".format(host=host, port=port)
result = http.query(url, raise_error=False)
assert result == {"body": None}, result
def test_query_error_handling():
ret = http.query("http://127.0.0.1:0")
assert isinstance(ret, dict)
assert isinstance(ret.get("error", None), str)
ret = http.query("http://myfoobardomainthatnotexist")
assert isinstance(ret, dict)
assert isinstance(ret.get("error", None), str)
def test_parse_cookie_header():
header = "; ".join(
[
"foo=bar",
"expires=Mon, 03-Aug-20 14:26:27 GMT",
"path=/",
"domain=.mydomain.tld",
"HttpOnly",
"SameSite=Lax",
"Secure",
]
)
ret = http.parse_cookie_header(header)
cookie = ret.pop(0)
assert cookie.name == "foo", cookie.name
assert cookie.value == "bar", cookie.value
assert cookie.expires == 1596464787, cookie.expires
assert cookie.path == "/", cookie.path
assert cookie.domain == ".mydomain.tld", cookie.domain
assert cookie.secure
# Only one cookie should have been returned, if anything is left in the
# parse_cookie_header return then something went wrong.
assert not ret
@pytest.mark.requires_network
def test_requests_multipart_formdata_post(httpserver):
"""
Test handling of a multipart/form-data POST using the requests backend
"""
match_this = (
"{0}\r\nContent-Disposition: form-data;"
' name="fieldname_here"\r\n\r\nmydatahere\r\n{0}--\r\n'
)
def mirror_post_handler(request):
return Response(request.data)
httpserver.expect_request(
"/multipart_form_data",
).respond_with_handler(mirror_post_handler)
url = httpserver.url_for("/multipart_form_data")
ret = http.query(
url,
method="POST",
data="mydatahere",
formdata=True,
formdata_fieldname="fieldname_here",
backend="requests",
)
body = ret.get("body", "")
boundary = body[: body.find("\r")]
assert body == match_this.format(boundary)
def test_query_proxy(httpserver):
"""
Test http.query with tornado and with proxy opts set
and then test with no_proxy set to ensure we dont
run into issue #55192 again.
"""
data = "mydatahere"
opts = {
"proxy_host": "127.0.0.1",
"proxy_port": 88,
"proxy_username": "salt_test",
"proxy_password": "super_secret",
}
with patch("requests.Session") as mock_session:
mock_session.return_value = MagicMock()
ret = http.query(
"https://fake_url",
method="POST",
data=data,
backend="tornado",
opts=opts,
)
assert mock_session.return_value.proxies == {
"HTTP": "http://salt_test:super_secret@127.0.0.1:88"
}
opts["no_proxy"] = [httpserver.host]
httpserver.expect_request(
"/no_proxy_test",
).respond_with_data(data)
url = httpserver.url_for("/no_proxy_test")
with patch("requests.Session") as mock_session:
mock_session.return_value = MagicMock()
ret = http.query(
url,
method="POST",
data=data,
backend="tornado",
opts=opts,
)
assert not isinstance(mock_session.return_value.proxies, dict)
ret = http.query(url, method="POST", data=data, backend="tornado", opts=opts)
body = ret.get("body", "")
assert body == data
@pytest.mark.parametrize("backend", ["requests", "tornado", "urllib2"])
def test_backends_decode_body_false(httpserver, backend):
"""
test all backends when using
decode_body=False that it returns
bytes and does not try to decode
"""
url = "/test-bytes"
data = b"test-bytes"
httpserver.expect_request(
url,
).respond_with_data(data, content_type="application/octet-stream")
ret = http.query(
httpserver.url_for(url),
backend=backend,
decode_body=False,
)
body = ret.get("body", "")
assert isinstance(body, bytes)
@pytest.mark.parametrize("backend", ["requests", "tornado", "urllib2"])
def test_backends_decode_body_true(httpserver, backend):
"""
test all backends when using decode_body=True that it returns string and decodes it.
"""
url = "/test-decoded-bytes"
data = b"test-decoded-bytes"
httpserver.expect_request(
url,
).respond_with_data(data, content_type="application/octet-stream")
ret = http.query(
httpserver.url_for(url),
backend=backend,
)
body = ret.get("body", "")
assert isinstance(body, str)

299
tests/unit/utils/test_http.py
View file

@ -1,299 +0,0 @@
"""
:codeauthor: Nicole Thomas <nicole@saltstack.com>
"""
import socket
from contextlib import closing
import pytest
from saltfactories.utils.tempfiles import temp_file
import salt.utils.http as http
from tests.support.helpers import MirrorPostHandler, Webserver
from tests.support.mock import MagicMock, patch
from tests.support.runtests import RUNTIME_VARS
from tests.support.unit import TestCase
try:
import salt.ext.tornado.curl_httpclient # pylint: disable=unused-import
HAS_CURL = True
except ImportError:
HAS_CURL = False
class HTTPTestCase(TestCase):
"""
Unit TestCase for the salt.utils.http module.
"""
@classmethod
def setUpClass(cls):
cls.post_webserver = Webserver(handler=MirrorPostHandler)
cls.post_webserver.start()
cls.post_web_root = cls.post_webserver.web_root
@classmethod
def tearDownClass(cls):
cls.post_webserver.stop()
del cls.post_webserver
# sanitize_url tests
def test_sanitize_url_hide_fields_none(self):
"""
Tests sanitizing a url when the hide_fields kwarg is None.
"""
mock_url = "https://api.testing.com/?&foo=bar&test=testing"
ret = http.sanitize_url(mock_url, hide_fields=None)
self.assertEqual(ret, mock_url)
def test_sanitize_url_no_elements(self):
"""
Tests sanitizing a url when no elements should be sanitized.
"""
mock_url = "https://api.testing.com/?&foo=bar&test=testing"
ret = http.sanitize_url(mock_url, [""])
self.assertEqual(ret, mock_url)
def test_sanitize_url_single_element(self):
"""
Tests sanitizing a url with only a single element to be sanitized.
"""
mock_url = (
"https://api.testing.com/?&keep_it_secret=abcdefghijklmn"
"&api_action=module.function"
)
mock_ret = (
"https://api.testing.com/?&keep_it_secret=XXXXXXXXXX&"
"api_action=module.function"
)
ret = http.sanitize_url(mock_url, ["keep_it_secret"])
self.assertEqual(ret, mock_ret)
def test_sanitize_url_multiple_elements(self):
"""
Tests sanitizing a url with multiple elements to be sanitized.
"""
mock_url = (
"https://api.testing.com/?rootPass=badpassword%21"
"&skipChecks=True&api_key=abcdefghijklmn"
"&NodeID=12345&api_action=module.function"
)
mock_ret = (
"https://api.testing.com/?rootPass=XXXXXXXXXX"
"&skipChecks=True&api_key=XXXXXXXXXX"
"&NodeID=12345&api_action=module.function"
)
ret = http.sanitize_url(mock_url, ["api_key", "rootPass"])
self.assertEqual(ret, mock_ret)
# _sanitize_components tests
def test_sanitize_components_no_elements(self):
"""
Tests when zero elements need to be sanitized.
"""
mock_component_list = ["foo=bar", "bar=baz", "hello=world"]
mock_ret = "foo=bar&bar=baz&hello=world&"
ret = http._sanitize_url_components(mock_component_list, "api_key")
self.assertEqual(ret, mock_ret)
def test_sanitize_components_one_element(self):
"""
Tests a single component to be sanitized.
"""
mock_component_list = ["foo=bar", "api_key=abcdefghijklmnop"]
mock_ret = "foo=bar&api_key=XXXXXXXXXX&"
ret = http._sanitize_url_components(mock_component_list, "api_key")
self.assertEqual(ret, mock_ret)
def test_sanitize_components_multiple_elements(self):
"""
Tests two componenets to be sanitized.
"""
mock_component_list = ["foo=bar", "foo=baz", "api_key=testing"]
mock_ret = "foo=XXXXXXXXXX&foo=XXXXXXXXXX&api_key=testing&"
ret = http._sanitize_url_components(mock_component_list, "foo")
self.assertEqual(ret, mock_ret)
@pytest.mark.slow_test
def test_query_null_response(self):
"""
This tests that we get a null response when raise_error=False and the
host/port cannot be reached.
"""
host = "127.0.0.1"
# Find unused port
with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:
sock.bind((host, 0))
port = sock.getsockname()[1]
url = "http://{host}:{port}/".format(host=host, port=port)
result = http.query(url, raise_error=False)
assert result == {"body": None}, result
def test_query_error_handling(self):
ret = http.query("http://127.0.0.1:0")
self.assertTrue(isinstance(ret, dict))
self.assertTrue(isinstance(ret.get("error", None), str))
ret = http.query("http://myfoobardomainthatnotexist")
self.assertTrue(isinstance(ret, dict))
self.assertTrue(isinstance(ret.get("error", None), str))
def test_parse_cookie_header(self):
header = "; ".join(
[
"foo=bar",
"expires=Mon, 03-Aug-20 14:26:27 GMT",
"path=/",
"domain=.mydomain.tld",
"HttpOnly",
"SameSite=Lax",
"Secure",
]
)
ret = http.parse_cookie_header(header)
cookie = ret.pop(0)
assert cookie.name == "foo", cookie.name
assert cookie.value == "bar", cookie.value
assert cookie.expires == 1596464787, cookie.expires
assert cookie.path == "/", cookie.path
assert cookie.domain == ".mydomain.tld", cookie.domain
assert cookie.secure
# Only one cookie should have been returned, if anything is left in the
# parse_cookie_header return then something went wrong.
assert not ret
class HTTPPostTestCase(TestCase):
"""
Unit TestCase for the salt.utils.http module when
using POST method
"""
@classmethod
def setUpClass(cls):
cls.post_webserver = Webserver(handler=MirrorPostHandler)
cls.post_webserver.start()
cls.post_web_root = cls.post_webserver.web_root
@classmethod
def tearDownClass(cls):
cls.post_webserver.stop()
del cls.post_webserver
def test_requests_multipart_formdata_post(self):
"""
Test handling of a multipart/form-data POST using the requests backend
"""
match_this = (
"{0}\r\nContent-Disposition: form-data;"
' name="fieldname_here"\r\n\r\nmydatahere\r\n{0}--\r\n'
)
ret = http.query(
self.post_web_root,
method="POST",
data="mydatahere",
formdata=True,
formdata_fieldname="fieldname_here",
backend="requests",
)
body = ret.get("body", "")
boundary = body[: body.find("\r")]
self.assertEqual(body, match_this.format(boundary))
@pytest.mark.skipif(
HAS_CURL is False,
reason="Missing prerequisites for tornado.curl_httpclient library",
)
def test_query_proxy(self):
"""
Test http.query with tornado and with proxy opts set
and then test with no_proxy set to ensure we dont
run into issue #55192 again.
"""
data = "mydatahere"
opts = {
"proxy_host": "127.0.0.1",
"proxy_port": 88,
"proxy_username": "salt_test",
"proxy_password": "super_secret",
}
mock_curl = MagicMock()
with patch("tornado.httpclient.HTTPClient.fetch", mock_curl):
ret = http.query(
self.post_web_root,
method="POST",
data=data,
backend="tornado",
opts=opts,
)
for opt in opts:
assert opt in mock_curl.call_args_list[0][1].keys()
opts["no_proxy"] = ["127.0.0.1"]
ret = http.query(
self.post_web_root, method="POST", data=data, backend="tornado", opts=opts
)
body = ret.get("body", "")
assert body == data
class HTTPGetTestCase(TestCase):
"""
Unit TestCase for the salt.utils.http module when
using Get method
"""
@classmethod
def setUpClass(cls):
cls.get_webserver = Webserver()
cls.get_webserver.start()
@classmethod
def tearDownClass(cls):
cls.get_webserver.stop()
del cls.get_webserver
def test_backends_decode_body_false(self):
"""
test all backends when using
decode_body=False that it returns
bytes and does not try to decode
"""
for backend in ["tornado", "requests", "urllib2"]:
ret = http.query(
self.get_webserver.url("custom.tar.gz"),
backend=backend,
decode_body=False,
)
body = ret.get("body", "")
assert isinstance(body, bytes)
def test_backends_decode_body_true(self):
"""
test all backends when using
decode_body=True that it returns
string and decodes it.
"""
core_state = """
{}:
file:
- managed
- source: salt://testfile
- makedirs: true
""".format(
RUNTIME_VARS.TMP
)
with temp_file("core.sls", core_state, self.get_webserver.root):
for backend in ["tornado", "requests", "urllib2"]:
ret = http.query(self.get_webserver.url("core.sls"), backend=backend)
body = ret.get("body", "")
assert isinstance(body, str)