wbrawner/ansible
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add unit tests for ansible.module_utils.urls (#38059)

Browse source 
* Start of tests for ansible.module_utils.urls

* Start adding file for generic functions throughout urls

* Add tests for maybe_add_ssl_handler

* Remove commented out line

* Improve coverage of maybe_add_ssl_handler, test basic_auth_header

* Start tests for open_url

* pep8 and ignore urlopen in test_url_open.py tests

* Extend auth tests, add test for validate_certs=False

* Finish tests for open_url

* Add tests for fetch_url

* Add fetch_url tests to replace-urlopen ignore

* dummy instead of _

* Add BadStatusLine test

* Reorganize/rename tests

* Add tests for RedirectHandlerFactory

* Add POST test to confirm behavior is to convert to GET

* Update tests to handle recent changes to RedirectHandlerFactory

* Special test, just to confirm that aliasing http_error_308 to http_error_307 does not cause issues with urllib2 type redirects
This commit is contained in:
Matt Martz 2018-04-09 10:17:43 -05:00 committed by GitHub
parent eccccfe77f
commit 6332beef65
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
13 changed files with 936 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
lib/ansible/module_utils/urls.py
View file

@ -941,7 +941,7 @@ def open_url(url, data=None, headers=None, method=None, use_proxy=True,
# user defined headers now, which may override things we've set above
if headers:
if not isinstance(headers, dict):
raise ValueError("headers provided to fetch_url() must be a dict")
raise ValueError("headers provided to open_url() must be a dict")
for header in headers:
request.add_header(header, headers[header])

2
test/sanity/code-smell/replace-urlopen.py
View file

@ -10,6 +10,8 @@ def main():
'test/sanity/code-smell/%s' % os.path.basename(__file__),
'lib/ansible/module_utils/six/__init__.py',
'lib/ansible/module_utils/urls.py',
'test/units/module_utils/urls/test_open_url.py',
'test/units/module_utils/urls/test_fetch_url.py',
])
for path in sys.argv[1:] or sys.stdin.read().splitlines():

0
test/units/module_utils/urls/__init__.py Normal file
View file

28
test/units/module_utils/urls/fixtures/client.key Normal file
View file

@ -0,0 +1,28 @@
-----BEGIN PRIVATE KEY-----
MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQDTyiVxrsSyZ+Qr
iMT6sFYCqQtkLqlIWfbpTg9B6fZc793uoMzLUGq3efiZUhhxI78dQ3gNPgs1sK3W
heFpk1n4IL8ll1MS1uJKk2vYqzZVhjgcvQpeV9gm7bt0ndPzGj5h4fh7proPntSy
eBvMKVoqTT7tEnapRKy3anbwRPgTt7B5jEvJkPazuIc+ooMsYOHWfvj4oVsev0N2
SsP0o6cHcsRujFMhz/JTJ1STQxacaVuyKpXacX7Eu1MJgGt/jU/QKNREcV9LdneO
NgqY9tNv0h+9s7DfHYXm8U3POr+bdcW6Yy4791KGCaUNtiNqT1lvu/4yd4WRkXbF
Fm5hJUUpAgMBAAECggEBAJYOac1MSK0nEvENbJM6ERa9cwa+UM6kf176IbFP9XAP
u6zxXWjIR3RMBSmMkyjGbQhs30hypzqZPfH61aUZ8+rsOMKHnyKAAcFZBlZzqIGc
IXGrNwd1Mf8S/Xg4ww1BkOWFV6s0jCu5G3Z/xyI2Ql4qcOVD6bMwpzclRbQjCand
dvqyCdMD0sRDyeOIK5hBhUY60JnWbMCu6pBU+qPoRukbRieaeDLIN1clwEqIQV78
LLnv4n9fuGozH0JdHHfyXFytCgIJvEspZUja/5R4orADhr3ZB010RLzYvs2ndE3B
4cF9RgxspJZeJ/P+PglViZuzj37pXy+7GAcJLR9ka4kCgYEA/l01XKwkCzMgXHW4
UPgl1+on42BsN7T9r3S5tihOjHf4ZJWkgYzisLVX+Nc1oUI3HQfM9PDJZXMMNm7J
ZRvERcopU26wWqr6CFPblGv8oqXHqcpeta8i3xZKoPASsTW6ssuPCEajiLZbQ1rH
H/HP+OZIVLM/WCPgA2BckTU9JnsCgYEA1SbXllXnlwGqmjitmY1Z07rUxQ3ah/fB
iccbbg3E4onontYXIlI5zQms3u+qBdi0ZuwaDm5Y4BetOq0a3UyxAsugqVFnzTba
1w/sFb3fw9KeQ/il4CXkbq87nzJfDmEyqHGCCYXbijHBxnq99PkqwVpaAhHHEW0m
vWyMUvPRY6sCgYAbtUWR0cKfYbNdvwkT8OQWcBBmSWOgcdvMmBd+y0c7L/pj4pUn
85PiEe8CUVcrOM5OIEJoUC5wGacz6r+PfwXTYGE+EGmvhr5z18aslVLQ2OQ2D7Bf
dDOFP6VjgKNYoHS0802iZid8RfkNDj9wsGOqRlOMvnXhAQ9u7rlGrBj8LwKBgFfo
ph99nH8eE9N5LrfWoUZ+loQS258aInsFYB26lgnsYMEpgO8JxIb4x5BGffPdVUHh
fDmZbxQ1D5/UhvDgUVzayI8sYMg1KHpsOa0Z2zCzK8zSvu68EgNISCm3J5cRpUft
UHlG+K19KfMG6lMfdG+8KMUTuetI/iI/o3wOzLvzAoGAIrOh30rHt8wit7ELARyx
wPkp2ARYXrKfX3NES4c67zSAi+3dCjxRqywqTI0gLicyMlj8zEu9YE9Ix/rl8lRZ
nQ9LZmqv7QHzhLTUCPGgZYnemvBzo7r0eW8Oag52dbcJO6FBszfWrxskm/fX25Rb
WPxih2vdRy814dNPW25rgdw=
-----END PRIVATE KEY-----

81
test/units/module_utils/urls/fixtures/client.pem Normal file
View file

@ -0,0 +1,81 @@
Certificate:
Data:
Version: 3 (0x2)
Serial Number: 4099 (0x1003)
Signature Algorithm: sha256WithRSAEncryption
Issuer: C=US, ST=North Carolina, L=Durham, O=Ansible, CN=ansible.http.tests
Validity
Not Before: Mar 21 18:22:47 2018 GMT
Not After : Mar 18 18:22:47 2028 GMT
Subject: C=US, ST=North Carolina, O=Ansible, CN=client.ansible.http.tests
Subject Public Key Info:
Public Key Algorithm: rsaEncryption
Public-Key: (2048 bit)
Modulus:
00:d3:ca:25:71:ae:c4:b2:67:e4:2b:88:c4:fa:b0:
56:02:a9:0b:64:2e:a9:48:59:f6:e9:4e:0f:41:e9:
f6:5c:ef:dd:ee:a0:cc:cb:50:6a:b7:79:f8:99:52:
18:71:23:bf:1d:43:78:0d:3e:0b:35:b0:ad:d6:85:
e1:69:93:59:f8:20:bf:25:97:53:12:d6:e2:4a:93:
6b:d8:ab:36:55:86:38:1c:bd:0a:5e:57:d8:26:ed:
bb:74:9d:d3:f3:1a:3e:61:e1:f8:7b:a6:ba:0f:9e:
d4:b2:78:1b:cc:29:5a:2a:4d:3e:ed:12:76:a9:44:
ac:b7:6a:76:f0:44:f8:13:b7:b0:79:8c:4b:c9:90:
f6:b3:b8:87:3e:a2:83:2c:60:e1:d6:7e:f8:f8:a1:
5b:1e:bf:43:76:4a:c3:f4:a3:a7:07:72:c4:6e:8c:
53:21:cf:f2:53:27:54:93:43:16:9c:69:5b:b2:2a:
95:da:71:7e:c4:bb:53:09:80:6b:7f:8d:4f:d0:28:
d4:44:71:5f:4b:76:77:8e:36:0a:98:f6:d3:6f:d2:
1f:bd:b3:b0:df:1d:85:e6:f1:4d:cf:3a:bf:9b:75:
c5:ba:63:2e:3b:f7:52:86:09:a5:0d:b6:23:6a:4f:
59:6f:bb:fe:32:77:85:91:91:76:c5:16:6e:61:25:
45:29
Exponent: 65537 (0x10001)
X509v3 extensions:
X509v3 Basic Constraints:
CA:FALSE
Netscape Comment:
OpenSSL Generated Certificate
X509v3 Subject Key Identifier:
AF:F3:E5:2A:EB:CF:C7:7E:A4:D6:49:92:F9:29:EE:6A:1B:68:AB:0F
X509v3 Authority Key Identifier:
keyid:13:2E:30:F0:04:EA:41:5F:B7:08:BD:34:31:D7:11:EA:56:A6:99:F0
Signature Algorithm: sha256WithRSAEncryption
29:62:39:25:79:58:eb:a4:b3:0c:ea:aa:1d:2b:96:7c:6e:10:
ce:16:07:b7:70:7f:16:da:fd:20:e6:a2:d9:b4:88:e0:f9:84:
87:f8:b0:0d:77:8b:ae:27:f5:ee:e6:4f:86:a1:2d:74:07:7c:
c7:5d:c2:bd:e4:70:e7:42:e4:14:ee:b9:b7:63:b8:8c:6d:21:
61:56:0b:96:f6:15:ba:7a:ae:80:98:ac:57:99:79:3d:7a:a9:
d8:26:93:30:17:53:7c:2d:02:4b:64:49:25:65:e7:69:5a:08:
cf:84:94:8e:6a:42:a7:d1:4f:ba:39:4b:7c:11:67:31:f7:1b:
2b:cd:79:c2:28:4d:d9:88:66:d6:7f:56:4c:4b:37:d1:3d:a8:
d9:4a:6b:45:1d:4d:a7:12:9f:29:77:6a:55:c1:b5:1d:0e:a5:
b9:4f:38:16:3c:7d:85:ae:ff:23:34:c7:2c:f6:14:0f:55:ef:
b8:00:89:f1:b2:8a:75:15:41:81:72:d0:43:a6:86:d1:06:e6:
ce:81:7e:5f:33:e6:f4:19:d6:70:00:ba:48:6e:05:fd:4c:3c:
c3:51:1b:bd:43:1a:24:c5:79:ea:7a:f0:85:a5:40:10:85:e9:
23:09:09:80:38:9d:bc:81:5e:59:8c:5a:4d:58:56:b9:71:c2:
78:cd:f3:b0
-----BEGIN CERTIFICATE-----
MIIDuTCCAqGgAwIBAgICEAMwDQYJKoZIhvcNAQELBQAwZjELMAkGA1UEBhMCVVMx
FzAVBgNVBAgMDk5vcnRoIENhcm9saW5hMQ8wDQYDVQQHDAZEdXJoYW0xEDAOBgNV
BAoMB0Fuc2libGUxGzAZBgNVBAMMEmFuc2libGUuaHR0cC50ZXN0czAeFw0xODAz
MjExODIyNDdaFw0yODAzMTgxODIyNDdaMFwxCzAJBgNVBAYTAlVTMRcwFQYDVQQI
DA5Ob3J0aCBDYXJvbGluYTEQMA4GA1UECgwHQW5zaWJsZTEiMCAGA1UEAwwZY2xp
ZW50LmFuc2libGUuaHR0cC50ZXN0czCCASIwDQYJKoZIhvcNAQEBBQADggEPADCC
AQoCggEBANPKJXGuxLJn5CuIxPqwVgKpC2QuqUhZ9ulOD0Hp9lzv3e6gzMtQard5
+JlSGHEjvx1DeA0+CzWwrdaF4WmTWfggvyWXUxLW4kqTa9irNlWGOBy9Cl5X2Cbt
u3Sd0/MaPmHh+Humug+e1LJ4G8wpWipNPu0SdqlErLdqdvBE+BO3sHmMS8mQ9rO4
hz6igyxg4dZ++PihWx6/Q3ZKw/SjpwdyxG6MUyHP8lMnVJNDFpxpW7IqldpxfsS7
UwmAa3+NT9Ao1ERxX0t2d442Cpj202/SH72zsN8dhebxTc86v5t1xbpjLjv3UoYJ
pQ22I2pPWW+7/jJ3hZGRdsUWbmElRSkCAwEAAaN7MHkwCQYDVR0TBAIwADAsBglg
hkgBhvhCAQ0EHxYdT3BlblNTTCBHZW5lcmF0ZWQgQ2VydGlmaWNhdGUwHQYDVR0O
BBYEFK/z5Srrz8d+pNZJkvkp7mobaKsPMB8GA1UdIwQYMBaAFBMuMPAE6kFftwi9
NDHXEepWppnwMA0GCSqGSIb3DQEBCwUAA4IBAQApYjkleVjrpLMM6qodK5Z8bhDO
Fge3cH8W2v0g5qLZtIjg+YSH+LANd4uuJ/Xu5k+GoS10B3zHXcK95HDnQuQU7rm3
Y7iMbSFhVguW9hW6eq6AmKxXmXk9eqnYJpMwF1N8LQJLZEklZedpWgjPhJSOakKn
0U+6OUt8EWcx9xsrzXnCKE3ZiGbWf1ZMSzfRPajZSmtFHU2nEp8pd2pVwbUdDqW5
TzgWPH2Frv8jNMcs9hQPVe+4AInxsop1FUGBctBDpobRBubOgX5fM+b0GdZwALpI
bgX9TDzDURu9QxokxXnqevCFpUAQhekjCQmAOJ28gV5ZjFpNWFa5ccJ4zfOw
-----END CERTIFICATE-----

3
test/units/module_utils/urls/fixtures/client.txt Normal file
View file

@ -0,0 +1,3 @@
client.pem and client.key were retrieved from httptester docker image:
ansible/ansible@sha256:fa5def8c294fc50813af131c0b5737594d852abac9cbe7ba38e17bf1c8476f3f

3
test/units/module_utils/urls/fixtures/netrc Normal file
View file

@ -0,0 +1,3 @@
machine ansible.com
login user
password passwd

138
test/units/module_utils/urls/test_RedirectHandlerFactory.py Normal file
View file

@ -0,0 +1,138 @@
# -*- 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
from ansible.module_utils.urls import RedirectHandlerFactory, urllib_request, urllib_error
from ansible.module_utils.six import StringIO
import pytest
@pytest.fixture
def urllib_req():
req = urllib_request.Request(
'https://ansible.com/'
)
return req
@pytest.fixture
def request_body():
return StringIO('TESTS')
def test_no_redirs(urllib_req, request_body):
handler = RedirectHandlerFactory('none', False)
inst = handler()
with pytest.raises(urllib_error.HTTPError):
inst.redirect_request(urllib_req, request_body, 301, '301 Moved Permanently', {}, 'https://docs.ansible.com/')
def test_urllib2_redir(urllib_req, request_body, mocker):
redir_request_mock = mocker.patch('ansible.module_utils.urls.urllib_request.HTTPRedirectHandler.redirect_request')
handler = RedirectHandlerFactory('urllib2', False)
inst = handler()
inst.redirect_request(urllib_req, request_body, 301, '301 Moved Permanently', {}, 'https://docs.ansible.com/')
redir_request_mock.assert_called_once_with(inst, urllib_req, request_body, 301, '301 Moved Permanently', {}, 'https://docs.ansible.com/')
def test_all_redir(urllib_req, request_body, mocker):
req_mock = mocker.patch('ansible.module_utils.urls.RequestWithMethod')
handler = RedirectHandlerFactory('all', False)
inst = handler()
inst.redirect_request(urllib_req, request_body, 301, '301 Moved Permanently', {}, 'https://docs.ansible.com/')
req_mock.assert_called_once_with('https://docs.ansible.com/', data=None, headers={}, method='GET', origin_req_host='ansible.com', unverifiable=True)
def test_all_redir_post(request_body, mocker):
handler = RedirectHandlerFactory('all', False)
inst = handler()
req = urllib_request.Request(
'https://ansible.com/',
'POST'
)
req_mock = mocker.patch('ansible.module_utils.urls.RequestWithMethod')
inst.redirect_request(req, request_body, 301, '301 Moved Permanently', {}, 'https://docs.ansible.com/')
req_mock.assert_called_once_with('https://docs.ansible.com/', data=None, headers={}, method='GET', origin_req_host='ansible.com', unverifiable=True)
def test_redir_headers_removal(urllib_req, request_body, mocker):
req_mock = mocker.patch('ansible.module_utils.urls.RequestWithMethod')
handler = RedirectHandlerFactory('all', False)
inst = handler()
urllib_req.headers = {
'Content-Type': 'application/json',
'Content-Length': 100,
'Foo': 'bar',
}
inst.redirect_request(urllib_req, request_body, 301, '301 Moved Permanently', {}, 'https://docs.ansible.com/')
req_mock.assert_called_once_with('https://docs.ansible.com/', data=None, headers={'Foo': 'bar'}, method='GET', origin_req_host='ansible.com',
unverifiable=True)
def test_redir_url_spaces(urllib_req, request_body, mocker):
req_mock = mocker.patch('ansible.module_utils.urls.RequestWithMethod')
handler = RedirectHandlerFactory('all', False)
inst = handler()
inst.redirect_request(urllib_req, request_body, 301, '301 Moved Permanently', {}, 'https://docs.ansible.com/foo bar')
req_mock.assert_called_once_with('https://docs.ansible.com/foo%20bar', data=None, headers={}, method='GET', origin_req_host='ansible.com',
unverifiable=True)
def test_redir_safe(urllib_req, request_body, mocker):
req_mock = mocker.patch('ansible.module_utils.urls.RequestWithMethod')
handler = RedirectHandlerFactory('safe', False)
inst = handler()
inst.redirect_request(urllib_req, request_body, 301, '301 Moved Permanently', {}, 'https://docs.ansible.com/')
req_mock.assert_called_once_with('https://docs.ansible.com/', data=None, headers={}, method='GET', origin_req_host='ansible.com', unverifiable=True)
def test_redir_safe_not_safe(request_body):
handler = RedirectHandlerFactory('safe', False)
inst = handler()
req = urllib_request.Request(
'https://ansible.com/',
'POST'
)
with pytest.raises(urllib_error.HTTPError):
inst.redirect_request(req, request_body, 301, '301 Moved Permanently', {}, 'https://docs.ansible.com/')
def test_redir_no_error_on_invalid(urllib_req, request_body):
handler = RedirectHandlerFactory('invalid', False)
inst = handler()
with pytest.raises(urllib_error.HTTPError):
inst.redirect_request(urllib_req, request_body, 301, '301 Moved Permanently', {}, 'https://docs.ansible.com/')
def test_redir_validate_certs(urllib_req, request_body, mocker):
opener_mock = mocker.patch('ansible.module_utils.urls.urllib_request._opener')
handler = RedirectHandlerFactory('all', True)
inst = handler()
inst.redirect_request(urllib_req, request_body, 301, '301 Moved Permanently', {}, 'https://docs.ansible.com/')
assert opener_mock.add_handler.call_count == 1
def test_redir_http_error_308_urllib2(urllib_req, request_body):
handler = RedirectHandlerFactory('urllib2', False)
inst = handler()
with pytest.raises(urllib_error.HTTPError):
inst.redirect_request(urllib_req, request_body, 308, '308 Permanent Redirect', {}, 'https://docs.ansible.com/')

22
test/units/module_utils/urls/test_RequestWithMethod.py Normal file
View file

@ -0,0 +1,22 @@
# -*- 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
from ansible.module_utils.urls import RequestWithMethod
def test_RequestWithMethod():
get = RequestWithMethod('https://ansible.com/', 'GET')
assert get.get_method() == 'GET'
post = RequestWithMethod('https://ansible.com/', 'POST', data='foo', headers={'Bar': 'baz'})
assert post.get_method() == 'POST'
assert post.get_full_url() == 'https://ansible.com/'
assert post.data == 'foo'
assert post.headers == {'Bar': 'baz'}
none = RequestWithMethod('https://ansible.com/', '')
assert none.get_method() == 'GET'

196
test/units/module_utils/urls/test_fetch_url.py Normal file
View file

@ -0,0 +1,196 @@
# -*- 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 socket
from ansible.module_utils.six import StringIO
from ansible.module_utils.six.moves.http_cookiejar import Cookie
from ansible.module_utils.urls import fetch_url, urllib_error, ConnectionError, NoSSLError, httplib
import pytest
from mock import MagicMock
class AnsibleModuleExit(Exception):
def __init__(self, *args, **kwargs):
self.args = args
self.kwargs = kwargs
class ExitJson(AnsibleModuleExit):
pass
class FailJson(AnsibleModuleExit):
pass
@pytest.fixture
def open_url_mock(mocker):
return mocker.patch('ansible.module_utils.urls.open_url')
@pytest.fixture
def fake_ansible_module():
return FakeAnsibleModule()
class FakeAnsibleModule:
def __init__(self):
self.params = {}
self.tmpdir = None
def exit_json(self, *args, **kwargs):
raise ExitJson(*args, **kwargs)
def fail_json(self, *args, **kwargs):
raise FailJson(*args, **kwargs)
def test_fetch_url_no_urlparse(mocker, fake_ansible_module):
mocker.patch('ansible.module_utils.urls.HAS_URLPARSE', new=False)
with pytest.raises(FailJson):
fetch_url(fake_ansible_module, 'http://ansible.com/')
def test_fetch_url(open_url_mock, fake_ansible_module):
r, info = fetch_url(fake_ansible_module, 'http://ansible.com/')
dummy, kwargs = open_url_mock.call_args
open_url_mock.assert_called_once_with('http://ansible.com/', client_cert=None, client_key=None, cookies=kwargs['cookies'], data=None,
follow_redirects='urllib2', force=False, force_basic_auth='', headers=None,
http_agent='ansible-httpget', last_mod_time=None, method=None, timeout=10, url_password='', url_username='',
use_proxy=True, validate_certs=True)
def test_fetch_url_params(open_url_mock, fake_ansible_module):
fake_ansible_module.params = {
'validate_certs': False,
'url_username': 'user',
'url_password': 'passwd',
'http_agent': 'ansible-test',
'force_basic_auth': True,
'follow_redirects': 'all',
'client_cert': 'client.pem',
'client_key': 'client.key',
}
r, info = fetch_url(fake_ansible_module, 'http://ansible.com/')
dummy, kwargs = open_url_mock.call_args
open_url_mock.assert_called_once_with('http://ansible.com/', client_cert='client.pem', client_key='client.key', cookies=kwargs['cookies'], data=None,
follow_redirects='all', force=False, force_basic_auth=True, headers=None,
http_agent='ansible-test', last_mod_time=None, method=None, timeout=10, url_password='passwd', url_username='user',
use_proxy=True, validate_certs=False)
def test_fetch_url_cookies(mocker, fake_ansible_module):
def make_cookies(*args, **kwargs):
cookies = kwargs['cookies']
for name, value in (('Foo', 'bar'), ('Baz', 'qux')):
cookie = Cookie(
version=0,
name=name,
value=value,
port=None,
port_specified=False,
domain="ansible.com",
domain_specified=True,
domain_initial_dot=False,
path="/",
path_specified=True,
secure=False,
expires=None,
discard=False,
comment=None,
comment_url=None,
rest=None
)
cookies.set_cookie(cookie)
return MagicMock()
mocker = mocker.patch('ansible.module_utils.urls.open_url', new=make_cookies)
r, info = fetch_url(fake_ansible_module, 'http://ansible.com/')
assert info['cookies'] == {'Baz': 'qux', 'Foo': 'bar'}
def test_fetch_url_nossl(open_url_mock, fake_ansible_module, mocker):
mocker.patch('ansible.module_utils.urls.get_distribution', return_value='notredhat')
open_url_mock.side_effect = NoSSLError
with pytest.raises(FailJson) as excinfo:
fetch_url(fake_ansible_module, 'http://ansible.com/')
assert 'python-ssl' not in excinfo.value.kwargs['msg']
mocker.patch('ansible.module_utils.urls.get_distribution', return_value='redhat')
open_url_mock.side_effect = NoSSLError
with pytest.raises(FailJson) as excinfo:
fetch_url(fake_ansible_module, 'http://ansible.com/')
assert 'python-ssl' in excinfo.value.kwargs['msg']
def test_fetch_url_connectionerror(open_url_mock, fake_ansible_module):
open_url_mock.side_effect = ConnectionError('TESTS')
with pytest.raises(FailJson) as excinfo:
fetch_url(fake_ansible_module, 'http://ansible.com/')
assert excinfo.value.kwargs['msg'] == 'TESTS'
open_url_mock.side_effect = ValueError('TESTS')
with pytest.raises(FailJson) as excinfo:
fetch_url(fake_ansible_module, 'http://ansible.com/')
assert excinfo.value.kwargs['msg'] == 'TESTS'
def test_fetch_url_httperror(open_url_mock, fake_ansible_module):
open_url_mock.side_effect = urllib_error.HTTPError(
'http://ansible.com/',
500,
'Internal Server Error',
{},
StringIO('TESTS')
)
r, info = fetch_url(fake_ansible_module, 'http://ansible.com/')
assert info == {'msg': 'HTTP Error 500: Internal Server Error', 'body': 'TESTS', 'status': 500, 'url': 'http://ansible.com/'}
def test_fetch_url_urlerror(open_url_mock, fake_ansible_module):
open_url_mock.side_effect = urllib_error.URLError('TESTS')
r, info = fetch_url(fake_ansible_module, 'http://ansible.com/')
assert info == {'msg': 'Request failed: <urlopen error TESTS>', 'status': -1, 'url': 'http://ansible.com/'}
def test_fetch_url_socketerror(open_url_mock, fake_ansible_module):
open_url_mock.side_effect = socket.error('TESTS')
r, info = fetch_url(fake_ansible_module, 'http://ansible.com/')
assert info == {'msg': 'Connection failure: TESTS', 'status': -1, 'url': 'http://ansible.com/'}
def test_fetch_url_exception(open_url_mock, fake_ansible_module):
open_url_mock.side_effect = Exception('TESTS')
r, info = fetch_url(fake_ansible_module, 'http://ansible.com/')
exception = info.pop('exception')
assert info == {'msg': 'An unknown error occurred: TESTS', 'status': -1, 'url': 'http://ansible.com/'}
assert "Exception: TESTS" in exception
def test_fetch_url_badstatusline(open_url_mock, fake_ansible_module):
open_url_mock.side_effect = httplib.BadStatusLine('TESTS')
r, info = fetch_url(fake_ansible_module, 'http://ansible.com/')
assert info == {'msg': 'Connection failure: connection was closed before a valid response was received: TESTS', 'status': -1, 'url': 'http://ansible.com/'}

57
test/units/module_utils/urls/test_generic_urlparse.py Normal file
View file

@ -0,0 +1,57 @@
# -*- 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
from ansible.module_utils.urls import generic_urlparse
from ansible.module_utils.six.moves.urllib.parse import urlparse, urlunparse
def test_generic_urlparse():
url = 'https://ansible.com/blog'
parts = urlparse(url)
generic_parts = generic_urlparse(parts)
assert generic_parts.as_list() == list(parts)
assert urlunparse(generic_parts.as_list()) == url
def test_generic_urlparse_netloc():
url = 'https://ansible.com:443/blog'
parts = urlparse(url)
generic_parts = generic_urlparse(parts)
assert generic_parts.hostname == parts.hostname
assert generic_parts.hostname == 'ansible.com'
assert generic_parts.port == 443
assert urlunparse(generic_parts.as_list()) == url
def test_generic_urlparse_no_netloc():
url = 'https://user:passwd@ansible.com:443/blog'
parts = list(urlparse(url))
generic_parts = generic_urlparse(parts)
assert generic_parts.hostname == 'ansible.com'
assert generic_parts.port == 443
assert generic_parts.username == 'user'
assert generic_parts.password == 'passwd'
assert urlunparse(generic_parts.as_list()) == url
def test_generic_urlparse_no_netloc_no_auth():
url = 'https://ansible.com:443/blog'
parts = list(urlparse(url))
generic_parts = generic_urlparse(parts)
assert generic_parts.username is None
assert generic_parts.password is None
def test_generic_urlparse_no_netloc_no_host():
url = '/blog'
parts = list(urlparse(url))
generic_parts = generic_urlparse(parts)
assert generic_parts.username is None
assert generic_parts.password is None
assert generic_parts.port is None
assert generic_parts.hostname == ''

314
test/units/module_utils/urls/test_open_url.py Normal file
View file

@ -0,0 +1,314 @@
# -*- 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
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'])

91
test/units/module_utils/urls/test_urls.py Normal file
View file

@ -0,0 +1,91 @@
# -*- 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
from ansible.module_utils import urls
from ansible.module_utils._text import to_native
import pytest
def test_build_ssl_validation_error(mocker):
mocker.patch.object(urls, 'HAS_SSLCONTEXT', new=False)
mocker.patch.object(urls, 'HAS_URLLIB3_PYOPENSSLCONTEXT', new=False)
mocker.patch.object(urls, 'HAS_URLLIB3_SSL_WRAP_SOCKET', new=False)
with pytest.raises(urls.SSLValidationError) as excinfo:
urls.build_ssl_validation_error('hostname', 'port', 'paths', exc=None)
assert 'python >= 2.7.9' in to_native(excinfo.value)
assert 'the python executable used' in to_native(excinfo.value)
assert 'urllib3' in to_native(excinfo.value)
assert 'python >= 2.6' in to_native(excinfo.value)
assert 'validate_certs=False' in to_native(excinfo.value)
mocker.patch.object(urls, 'HAS_SSLCONTEXT', new=True)
with pytest.raises(urls.SSLValidationError) as excinfo:
urls.build_ssl_validation_error('hostname', 'port', 'paths', exc=None)
assert 'validate_certs=False' in to_native(excinfo.value)
mocker.patch.object(urls, 'HAS_SSLCONTEXT', new=False)
mocker.patch.object(urls, 'HAS_URLLIB3_PYOPENSSLCONTEXT', new=True)
mocker.patch.object(urls, 'HAS_URLLIB3_SSL_WRAP_SOCKET', new=True)
mocker.patch.object(urls, 'HAS_SSLCONTEXT', new=True)
with pytest.raises(urls.SSLValidationError) as excinfo:
urls.build_ssl_validation_error('hostname', 'port', 'paths', exc=None)
assert 'urllib3' not in to_native(excinfo.value)
with pytest.raises(urls.SSLValidationError) as excinfo:
urls.build_ssl_validation_error('hostname', 'port', 'paths', exc='BOOM')
assert 'BOOM' in to_native(excinfo.value)
def test_maybe_add_ssl_handler(mocker):
mocker.patch.object(urls, 'HAS_SSL', new=False)
with pytest.raises(urls.NoSSLError):
urls.maybe_add_ssl_handler('https://ansible.com/', True)
mocker.patch.object(urls, 'HAS_SSL', new=True)
url = 'https://user:passwd@ansible.com/'
handler = urls.maybe_add_ssl_handler(url, True)
assert handler.hostname == 'ansible.com'
assert handler.port == 443
url = 'https://ansible.com:4433/'
handler = urls.maybe_add_ssl_handler(url, True)
assert handler.hostname == 'ansible.com'
assert handler.port == 4433
url = 'https://user:passwd@ansible.com:4433/'
handler = urls.maybe_add_ssl_handler(url, True)
assert handler.hostname == 'ansible.com'
assert handler.port == 4433
url = 'https://ansible.com/'
handler = urls.maybe_add_ssl_handler(url, True)
assert handler.hostname == 'ansible.com'
assert handler.port == 443
url = 'http://ansible.com/'
handler = urls.maybe_add_ssl_handler(url, True)
assert handler is None
def test_basic_auth_header():
header = urls.basic_auth_header('user', 'passwd')
assert header == b'Basic dXNlcjpwYXNzd2Q='
def test_ParseResultDottedDict():
url = 'https://ansible.com/blog'
parts = urls.urlparse(url)
dotted_parts = urls.ParseResultDottedDict(parts._asdict())
assert parts[0] == dotted_parts.scheme
assert dotted_parts.as_list() == list(parts)