mirror of
https://github.com/ansible-collections/community.general.git
synced 2025-05-03 15:51:30 -07:00
Add requests.Session like class (#37622)
* Adds requests.Session like class * py2 syntax fix * Add a few examples to the Request docstrings * Add helper methods and docs * Fix test failures * Switch tests to test Request instead of open_url, add simple open_url test to validate funcitonality * Fix filename in replace-urlopen code smell test
This commit is contained in:
Matt Martz
GitHub
parent
c86fd6e2df
commit
d4930e6692
3 changed files with 440 additions and 172 deletions
|
|
@ -1,315 +0,0 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
# (c) 2018 Matt Martz <matt@sivel.net>
|
||||
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
|
||||
|
||||
from __future__ import absolute_import, division, print_function
|
||||
__metaclass__ = type
|
||||
|
||||
import datetime
|
||||
import os
|
||||
|
||||
from ansible.module_utils.urls import open_url, urllib_request, HAS_SSLCONTEXT, cookiejar, ConnectionError, RequestWithMethod
|
||||
from ansible.module_utils.urls import SSLValidationHandler, HTTPSClientAuthHandler, RedirectHandlerFactory
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
if HAS_SSLCONTEXT:
|
||||
import ssl
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def urlopen_mock(mocker):
|
||||
return mocker.patch('ansible.module_utils.urls.urllib_request.urlopen')
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def install_opener_mock(mocker):
|
||||
return mocker.patch('ansible.module_utils.urls.urllib_request.install_opener')
|
||||
|
||||
|
||||
def test_open_url(urlopen_mock, install_opener_mock):
|
||||
r = open_url('https://ansible.com/')
|
||||
args = urlopen_mock.call_args[0]
|
||||
assert args[1] is None # data, this is handled in the Request not urlopen
|
||||
assert args[2] == 10 # timeout
|
||||
|
||||
req = args[0]
|
||||
assert req.headers == {}
|
||||
assert req.data is None
|
||||
assert req.get_method() == 'GET'
|
||||
|
||||
opener = install_opener_mock.call_args[0][0]
|
||||
handlers = opener.handlers
|
||||
|
||||
expected_handlers = (
|
||||
SSLValidationHandler,
|
||||
RedirectHandlerFactory(), # factory, get handler
|
||||
)
|
||||
|
||||
found_handlers = []
|
||||
for handler in handlers:
|
||||
if isinstance(handler, SSLValidationHandler) or handler.__class__.__name__ == 'RedirectHandler':
|
||||
found_handlers.append(handler)
|
||||
|
||||
assert len(found_handlers) == 2
|
||||
|
||||
|
||||
def test_open_url_http(urlopen_mock, install_opener_mock):
|
||||
r = open_url('http://ansible.com/')
|
||||
args = urlopen_mock.call_args[0]
|
||||
|
||||
opener = install_opener_mock.call_args[0][0]
|
||||
handlers = opener.handlers
|
||||
|
||||
found_handlers = []
|
||||
for handler in handlers:
|
||||
if isinstance(handler, SSLValidationHandler):
|
||||
found_handlers.append(handler)
|
||||
|
||||
assert len(found_handlers) == 0
|
||||
|
||||
|
||||
def test_open_url_ftp(urlopen_mock, install_opener_mock, mocker):
|
||||
mocker.patch('ansible.module_utils.urls.ParseResultDottedDict.as_list', side_effect=AssertionError)
|
||||
|
||||
# Using ftp scheme should prevent the AssertionError side effect to fire
|
||||
r = open_url('ftp://foo@ansible.com/')
|
||||
|
||||
|
||||
def test_open_url_headers(urlopen_mock, install_opener_mock):
|
||||
r = open_url('http://ansible.com/', headers={'Foo': 'bar'})
|
||||
args = urlopen_mock.call_args[0]
|
||||
req = args[0]
|
||||
assert req.headers == {'Foo': 'bar'}
|
||||
|
||||
|
||||
def test_open_url_username(urlopen_mock, install_opener_mock):
|
||||
r = open_url('http://ansible.com/', url_username='user')
|
||||
|
||||
opener = install_opener_mock.call_args[0][0]
|
||||
handlers = opener.handlers
|
||||
|
||||
expected_handlers = (
|
||||
urllib_request.HTTPBasicAuthHandler,
|
||||
urllib_request.HTTPDigestAuthHandler,
|
||||
)
|
||||
|
||||
found_handlers = []
|
||||
for handler in handlers:
|
||||
if isinstance(handler, expected_handlers):
|
||||
found_handlers.append(handler)
|
||||
assert len(found_handlers) == 2
|
||||
assert found_handlers[0].passwd.passwd[None] == {(('ansible.com', '/'),): ('user', None)}
|
||||
|
||||
|
||||
def test_open_url_username_in_url(urlopen_mock, install_opener_mock):
|
||||
r = open_url('http://user2@ansible.com/')
|
||||
|
||||
opener = install_opener_mock.call_args[0][0]
|
||||
handlers = opener.handlers
|
||||
|
||||
expected_handlers = (
|
||||
urllib_request.HTTPBasicAuthHandler,
|
||||
urllib_request.HTTPDigestAuthHandler,
|
||||
)
|
||||
|
||||
found_handlers = []
|
||||
for handler in handlers:
|
||||
if isinstance(handler, expected_handlers):
|
||||
found_handlers.append(handler)
|
||||
assert found_handlers[0].passwd.passwd[None] == {(('ansible.com', '/'),): ('user2', '')}
|
||||
|
||||
|
||||
def test_open_url_username_force_basic(urlopen_mock, install_opener_mock):
|
||||
r = open_url('http://ansible.com/', url_username='user', url_password='passwd', force_basic_auth=True)
|
||||
|
||||
opener = install_opener_mock.call_args[0][0]
|
||||
handlers = opener.handlers
|
||||
|
||||
expected_handlers = (
|
||||
urllib_request.HTTPBasicAuthHandler,
|
||||
urllib_request.HTTPDigestAuthHandler,
|
||||
)
|
||||
|
||||
found_handlers = []
|
||||
for handler in handlers:
|
||||
if isinstance(handler, expected_handlers):
|
||||
found_handlers.append(handler)
|
||||
|
||||
assert len(found_handlers) == 0
|
||||
|
||||
args = urlopen_mock.call_args[0]
|
||||
req = args[0]
|
||||
assert req.headers.get('Authorization') == b'Basic dXNlcjpwYXNzd2Q='
|
||||
|
||||
|
||||
def test_open_url_auth_in_netloc(urlopen_mock, install_opener_mock):
|
||||
r = open_url('http://user:passwd@ansible.com/')
|
||||
args = urlopen_mock.call_args[0]
|
||||
req = args[0]
|
||||
assert req.get_full_url() == 'http://ansible.com/'
|
||||
|
||||
opener = install_opener_mock.call_args[0][0]
|
||||
handlers = opener.handlers
|
||||
|
||||
expected_handlers = (
|
||||
urllib_request.HTTPBasicAuthHandler,
|
||||
urllib_request.HTTPDigestAuthHandler,
|
||||
)
|
||||
|
||||
found_handlers = []
|
||||
for handler in handlers:
|
||||
if isinstance(handler, expected_handlers):
|
||||
found_handlers.append(handler)
|
||||
|
||||
assert len(found_handlers) == 2
|
||||
|
||||
|
||||
def test_open_url_netrc(urlopen_mock, install_opener_mock, monkeypatch):
|
||||
here = os.path.dirname(__file__)
|
||||
|
||||
monkeypatch.setenv('NETRC', os.path.join(here, 'fixtures/netrc'))
|
||||
r = open_url('http://ansible.com/')
|
||||
args = urlopen_mock.call_args[0]
|
||||
req = args[0]
|
||||
assert req.headers.get('Authorization') == b'Basic dXNlcjpwYXNzd2Q='
|
||||
|
||||
r = open_url('http://foo.ansible.com/')
|
||||
args = urlopen_mock.call_args[0]
|
||||
req = args[0]
|
||||
assert 'Authorization' not in req.headers
|
||||
|
||||
monkeypatch.setenv('NETRC', os.path.join(here, 'fixtures/netrc.nonexistant'))
|
||||
r = open_url('http://ansible.com/')
|
||||
args = urlopen_mock.call_args[0]
|
||||
req = args[0]
|
||||
assert 'Authorization' not in req.headers
|
||||
|
||||
|
||||
def test_open_url_no_proxy(urlopen_mock, install_opener_mock, mocker):
|
||||
build_opener_mock = mocker.patch('ansible.module_utils.urls.urllib_request.build_opener')
|
||||
|
||||
r = open_url('http://ansible.com/', use_proxy=False)
|
||||
|
||||
handlers = build_opener_mock.call_args[0]
|
||||
found_handlers = []
|
||||
for handler in handlers:
|
||||
if isinstance(handler, urllib_request.ProxyHandler):
|
||||
found_handlers.append(handler)
|
||||
|
||||
assert len(found_handlers) == 1
|
||||
|
||||
|
||||
@pytest.mark.skipif(not HAS_SSLCONTEXT, reason="requires SSLContext")
|
||||
def test_open_url_no_validate_certs(urlopen_mock, install_opener_mock):
|
||||
r = open_url('https://ansible.com/', validate_certs=False)
|
||||
|
||||
opener = install_opener_mock.call_args[0][0]
|
||||
handlers = opener.handlers
|
||||
|
||||
ssl_handler = None
|
||||
for handler in handlers:
|
||||
if isinstance(handler, HTTPSClientAuthHandler):
|
||||
ssl_handler = handler
|
||||
break
|
||||
|
||||
assert ssl_handler is not None
|
||||
context = ssl_handler._context
|
||||
assert context.protocol == ssl.PROTOCOL_SSLv23
|
||||
if ssl.OP_NO_SSLv2:
|
||||
assert context.options & ssl.OP_NO_SSLv2
|
||||
assert context.options & ssl.OP_NO_SSLv3
|
||||
assert context.verify_mode == ssl.CERT_NONE
|
||||
assert context.check_hostname is False
|
||||
|
||||
|
||||
def test_open_url_client_cert(urlopen_mock, install_opener_mock):
|
||||
here = os.path.dirname(__file__)
|
||||
|
||||
client_cert = os.path.join(here, 'fixtures/client.pem')
|
||||
client_key = os.path.join(here, 'fixtures/client.key')
|
||||
|
||||
r = open_url('https://ansible.com/', client_cert=client_cert, client_key=client_key)
|
||||
|
||||
opener = install_opener_mock.call_args[0][0]
|
||||
handlers = opener.handlers
|
||||
|
||||
ssl_handler = None
|
||||
for handler in handlers:
|
||||
if isinstance(handler, HTTPSClientAuthHandler):
|
||||
ssl_handler = handler
|
||||
break
|
||||
|
||||
assert ssl_handler is not None
|
||||
|
||||
assert ssl_handler.client_cert == client_cert
|
||||
assert ssl_handler.client_key == client_key
|
||||
|
||||
https_connection = ssl_handler._build_https_connection('ansible.com')
|
||||
|
||||
assert https_connection.key_file == client_key
|
||||
assert https_connection.cert_file == client_cert
|
||||
|
||||
|
||||
def test_open_url_cookies(urlopen_mock, install_opener_mock):
|
||||
r = open_url('https://ansible.com/', cookies=cookiejar.CookieJar())
|
||||
|
||||
opener = install_opener_mock.call_args[0][0]
|
||||
handlers = opener.handlers
|
||||
|
||||
cookies_handler = None
|
||||
for handler in handlers:
|
||||
if isinstance(handler, urllib_request.HTTPCookieProcessor):
|
||||
cookies_handler = handler
|
||||
break
|
||||
|
||||
assert cookies_handler is not None
|
||||
|
||||
|
||||
def test_open_url_invalid_method(urlopen_mock, install_opener_mock):
|
||||
with pytest.raises(ConnectionError):
|
||||
r = open_url('https://ansible.com/', method='BOGUS')
|
||||
|
||||
|
||||
def test_open_url_custom_method(urlopen_mock, install_opener_mock):
|
||||
r = open_url('https://ansible.com/', method='DELETE')
|
||||
|
||||
args = urlopen_mock.call_args[0]
|
||||
req = args[0]
|
||||
|
||||
assert isinstance(req, RequestWithMethod)
|
||||
|
||||
|
||||
def test_open_url_user_agent(urlopen_mock, install_opener_mock):
|
||||
r = open_url('https://ansible.com/', http_agent='ansible-tests')
|
||||
|
||||
args = urlopen_mock.call_args[0]
|
||||
req = args[0]
|
||||
|
||||
assert req.headers.get('User-agent') == 'ansible-tests'
|
||||
|
||||
|
||||
def test_open_url_force(urlopen_mock, install_opener_mock):
|
||||
r = open_url('https://ansible.com/', force=True, last_mod_time=datetime.datetime.now())
|
||||
|
||||
args = urlopen_mock.call_args[0]
|
||||
req = args[0]
|
||||
|
||||
assert req.headers.get('Cache-control') == 'no-cache'
|
||||
assert 'If-modified-since' not in req.headers
|
||||
|
||||
|
||||
def test_open_url_last_mod(urlopen_mock, install_opener_mock):
|
||||
now = datetime.datetime.now()
|
||||
r = open_url('https://ansible.com/', last_mod_time=now)
|
||||
|
||||
args = urlopen_mock.call_args[0]
|
||||
req = args[0]
|
||||
|
||||
assert req.headers.get('If-modified-since') == now.strftime('%a, %d %b %Y %H:%M:%S +0000')
|
||||
|
||||
|
||||
def test_open_url_headers_not_dict(urlopen_mock, install_opener_mock):
|
||||
with pytest.raises(ValueError):
|
||||
r = open_url('https://ansible.com/', headers=['bob'])
|
||||
Loading…
Add table
Add a link
Reference in a new issue