mirror of https://github.com/ansible/ansible.git
You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
448 lines
14 KiB
Python
448 lines
14 KiB
Python
# -*- 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 annotations
|
|
|
|
import datetime
|
|
import os
|
|
import urllib.request
|
|
import http.client
|
|
|
|
from ansible.module_utils.urls import (Request, open_url, cookiejar,
|
|
UnixHTTPHandler, UnixHTTPSConnection)
|
|
from ansible.module_utils.urls import HTTPRedirectHandler
|
|
|
|
import pytest
|
|
from unittest.mock import call
|
|
|
|
|
|
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_Request_fallback(urlopen_mock, install_opener_mock, mocker):
|
|
here = os.path.dirname(__file__)
|
|
pem = os.path.join(here, 'fixtures/client.pem')
|
|
client_key = os.path.join(here, 'fixtures/client.key')
|
|
|
|
cookies = cookiejar.CookieJar()
|
|
request = Request(
|
|
headers={'foo': 'bar'},
|
|
use_proxy=False,
|
|
force=True,
|
|
timeout=100,
|
|
validate_certs=False,
|
|
url_username='user',
|
|
url_password='passwd',
|
|
http_agent='ansible-tests',
|
|
force_basic_auth=True,
|
|
follow_redirects='all',
|
|
client_cert=pem,
|
|
client_key=client_key,
|
|
cookies=cookies,
|
|
unix_socket='/foo/bar/baz.sock',
|
|
ca_path=pem,
|
|
ciphers=['ECDHE-RSA-AES128-SHA256'],
|
|
use_netrc=True,
|
|
)
|
|
fallback_mock = mocker.spy(request, '_fallback')
|
|
|
|
r = request.open('GET', 'https://ansible.com')
|
|
|
|
calls = [
|
|
call(None, False), # use_proxy
|
|
call(None, True), # force
|
|
call(None, 100), # timeout
|
|
call(None, False), # validate_certs
|
|
call(None, 'user'), # url_username
|
|
call(None, 'passwd'), # url_password
|
|
call(None, 'ansible-tests'), # http_agent
|
|
call(None, True), # force_basic_auth
|
|
call(None, 'all'), # follow_redirects
|
|
call(None, pem), # client_cert
|
|
call(None, client_key), # client_key
|
|
call(None, cookies), # cookies
|
|
call(None, '/foo/bar/baz.sock'), # unix_socket
|
|
call(None, pem), # ca_path
|
|
call(None, None), # unredirected_headers
|
|
call(None, True), # auto_decompress
|
|
call(None, ['ECDHE-RSA-AES128-SHA256']), # ciphers
|
|
call(None, True), # use_netrc
|
|
call(None, None), # context
|
|
]
|
|
fallback_mock.assert_has_calls(calls)
|
|
|
|
assert fallback_mock.call_count == 19 # All but headers use fallback
|
|
|
|
args = urlopen_mock.call_args[0]
|
|
assert args[1] is None # data, this is handled in the Request not urlopen
|
|
assert args[2] == 100 # timeout
|
|
|
|
req = args[0]
|
|
assert req.headers == {
|
|
'Authorization': b'Basic dXNlcjpwYXNzd2Q=',
|
|
'Cache-control': 'no-cache',
|
|
'Foo': 'bar',
|
|
'User-agent': 'ansible-tests'
|
|
}
|
|
assert req.data is None
|
|
assert req.get_method() == 'GET'
|
|
|
|
|
|
def test_Request_open(urlopen_mock, install_opener_mock):
|
|
r = Request().open('GET', '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 = (
|
|
HTTPRedirectHandler(),
|
|
)
|
|
|
|
found_handlers = []
|
|
for handler in handlers:
|
|
if handler.__class__.__name__ == 'HTTPRedirectHandler':
|
|
found_handlers.append(handler)
|
|
|
|
assert len(found_handlers) == len(expected_handlers)
|
|
|
|
|
|
def test_Request_open_unix_socket(urlopen_mock, install_opener_mock):
|
|
r = Request().open('GET', 'http://ansible.com/', unix_socket='/foo/bar/baz.sock')
|
|
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, UnixHTTPHandler):
|
|
found_handlers.append(handler)
|
|
|
|
assert len(found_handlers) == 1
|
|
|
|
|
|
def test_Request_open_https_unix_socket(urlopen_mock, install_opener_mock, mocker):
|
|
do_open = mocker.patch.object(urllib.request.HTTPSHandler, 'do_open')
|
|
|
|
r = Request().open('GET', 'https://ansible.com/', unix_socket='/foo/bar/baz.sock')
|
|
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, urllib.request.HTTPSHandler):
|
|
found_handlers.append(handler)
|
|
|
|
assert len(found_handlers) == 1
|
|
|
|
found_handlers[0].https_open(None)
|
|
args = do_open.call_args[0]
|
|
cls = args[0]
|
|
assert isinstance(cls, UnixHTTPSConnection)
|
|
|
|
|
|
def test_Request_open_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 = Request().open('GET', 'ftp://foo@ansible.com/')
|
|
|
|
|
|
def test_Request_open_headers(urlopen_mock, install_opener_mock):
|
|
r = Request().open('GET', 'http://ansible.com/', headers={'Foo': 'bar'})
|
|
args = urlopen_mock.call_args[0]
|
|
req = args[0]
|
|
assert req.headers == {'Foo': 'bar'}
|
|
|
|
|
|
def test_Request_open_username(urlopen_mock, install_opener_mock):
|
|
r = Request().open('GET', '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)}
|
|
|
|
|
|
@pytest.mark.parametrize('url, expected', (
|
|
('user2@ansible.com', ('user2', '')),
|
|
('user2%40@ansible.com', ('user2@', '')),
|
|
('user2%40:%40@ansible.com', ('user2@', '@')),
|
|
))
|
|
def test_Request_open_username_in_url(url, expected, urlopen_mock, install_opener_mock):
|
|
r = Request().open('GET', f'http://{url}/')
|
|
|
|
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', '/'),): expected}
|
|
|
|
|
|
def test_Request_open_username_force_basic(urlopen_mock, install_opener_mock):
|
|
r = Request().open('GET', '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_Request_open_auth_in_netloc(urlopen_mock, install_opener_mock):
|
|
r = Request().open('GET', '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_Request_open_netrc(urlopen_mock, install_opener_mock, monkeypatch):
|
|
here = os.path.dirname(__file__)
|
|
|
|
monkeypatch.setenv('NETRC', os.path.join(here, 'fixtures/netrc'))
|
|
r = Request().open('GET', 'http://ansible.com/')
|
|
args = urlopen_mock.call_args[0]
|
|
req = args[0]
|
|
assert req.headers.get('Authorization') == b'Basic dXNlcjpwYXNzd2Q='
|
|
|
|
r = Request().open('GET', '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 = Request().open('GET', 'http://ansible.com/')
|
|
args = urlopen_mock.call_args[0]
|
|
req = args[0]
|
|
assert 'Authorization' not in req.headers
|
|
|
|
|
|
def test_Request_open_no_proxy(urlopen_mock, install_opener_mock, mocker):
|
|
build_opener_mock = mocker.patch('ansible.module_utils.urls.urllib.request.build_opener')
|
|
|
|
r = Request().open('GET', '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
|
|
|
|
|
|
def test_Request_open_no_validate_certs(urlopen_mock, install_opener_mock, mocker):
|
|
do_open = mocker.patch.object(urllib.request.HTTPSHandler, 'do_open')
|
|
|
|
r = Request().open('GET', '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, urllib.request.HTTPSHandler):
|
|
ssl_handler = handler
|
|
break
|
|
|
|
assert ssl_handler is not None
|
|
|
|
ssl_handler.https_open(None)
|
|
args = do_open.call_args[0]
|
|
cls = args[0]
|
|
assert cls is http.client.HTTPSConnection
|
|
|
|
context = ssl_handler._context
|
|
# Differs by Python version
|
|
# 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_Request_open_client_cert(urlopen_mock, install_opener_mock, mocker):
|
|
load_cert_chain = mocker.patch.object(ssl.SSLContext, 'load_cert_chain')
|
|
|
|
here = os.path.dirname(__file__)
|
|
|
|
client_cert = os.path.join(here, 'fixtures/client.pem')
|
|
client_key = os.path.join(here, 'fixtures/client.key')
|
|
|
|
r = Request().open('GET', '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, urllib.request.HTTPSHandler):
|
|
ssl_handler = handler
|
|
break
|
|
|
|
assert ssl_handler is not None
|
|
|
|
load_cert_chain.assert_called_once_with(client_cert, keyfile=client_key)
|
|
|
|
|
|
def test_Request_open_cookies(urlopen_mock, install_opener_mock):
|
|
r = Request().open('GET', '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_Request_open_invalid_method(urlopen_mock, install_opener_mock):
|
|
r = Request().open('UNKNOWN', 'https://ansible.com/')
|
|
|
|
args = urlopen_mock.call_args[0]
|
|
req = args[0]
|
|
|
|
assert req.data is None
|
|
assert req.get_method() == 'UNKNOWN'
|
|
# assert r.status == 504
|
|
|
|
|
|
def test_Request_open_user_agent(urlopen_mock, install_opener_mock):
|
|
r = Request().open('GET', '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_Request_open_force(urlopen_mock, install_opener_mock):
|
|
r = Request().open('GET', '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_Request_open_last_mod(urlopen_mock, install_opener_mock):
|
|
now = datetime.datetime.now()
|
|
r = Request().open('GET', '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 GMT')
|
|
|
|
|
|
def test_Request_open_headers_not_dict(urlopen_mock, install_opener_mock):
|
|
with pytest.raises(ValueError):
|
|
Request().open('GET', 'https://ansible.com/', headers=['bob'])
|
|
|
|
|
|
def test_Request_init_headers_not_dict(urlopen_mock, install_opener_mock):
|
|
with pytest.raises(ValueError):
|
|
Request(headers=['bob'])
|
|
|
|
|
|
@pytest.mark.parametrize('method,kwargs', [
|
|
('get', {}),
|
|
('options', {}),
|
|
('head', {}),
|
|
('post', {'data': None}),
|
|
('put', {'data': None}),
|
|
('patch', {'data': None}),
|
|
('delete', {}),
|
|
])
|
|
def test_methods(method, kwargs, mocker):
|
|
expected = method.upper()
|
|
open_mock = mocker.patch('ansible.module_utils.urls.Request.open')
|
|
request = Request()
|
|
getattr(request, method)('https://ansible.com')
|
|
open_mock.assert_called_once_with(expected, 'https://ansible.com', **kwargs)
|
|
|
|
|
|
def test_open_url(urlopen_mock, install_opener_mock, mocker):
|
|
req_mock = mocker.patch('ansible.module_utils.urls.Request.open')
|
|
open_url('https://ansible.com/')
|
|
req_mock.assert_called_once_with('GET', 'https://ansible.com/', data=None, headers=None, use_proxy=True,
|
|
force=False, last_mod_time=None, timeout=10, validate_certs=True,
|
|
url_username=None, url_password=None, http_agent=None,
|
|
force_basic_auth=False, follow_redirects='urllib2',
|
|
client_cert=None, client_key=None, cookies=None, use_gssapi=False,
|
|
unix_socket=None, ca_path=None, unredirected_headers=None, decompress=True,
|
|
ciphers=None, use_netrc=True)
|