ansible/test/units/module_utils/urls/test_Request.py

# -*- 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)