passerelle/tests/test_proxy.py

# passerelle - uniform access to multiple data sources and services
# Copyright (C) 2023 Entr'ouvert
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import pytest
import responses
import responses.matchers

import tests.utils
from passerelle.apps.proxy.models import Resource


@pytest.fixture
def proxy(db):
    return tests.utils.setup_access_rights(
        Resource.objects.create(slug='echo', upstream_base_url='https://example.org/')
    )


@pytest.fixture
def endpoint(proxy):
    return tests.utils.generic_endpoint_url('proxy', 'request', slug=proxy.slug)


@pytest.fixture
def mocked_responses():
    with responses.RequestsMock(assert_all_requests_are_fired=False) as rsp:
        yield rsp


def test_get(mocked_responses, app, proxy, endpoint):
    mocked_responses.get(
        url='https://example.org/foo/bar',
        body='ok',
        match=[
            responses.matchers.query_string_matcher(''),
        ],
    )

    resp = app.get(endpoint + '/foo/bar', status=200)
    assert resp.text == 'ok'


def test_status(mocked_responses, app, proxy, endpoint):
    mocked_responses.get(url='https://example.org/foo/bar', body='bad request', status=400)
    resp = app.get(endpoint + '/foo/bar', status=400)
    assert resp.text == 'bad request'


def test_post(mocked_responses, app, proxy, endpoint):
    mocked_responses.post(
        url='https://example.org/post',
        body='ok',
        match=[
            responses.matchers.json_params_matcher({'foo': 'bar'}),
        ],
    )
    resp = app.post_json(endpoint + '/post', params={'foo': 'bar'}, status=200)
    assert resp.text == 'ok'


def no_header_matcher(header_name):
    def inner(request):
        if header_name in request.headers:
            return False, f'Found header "{header_name} in requests headers.'
        return True, None

    return inner


def test_headers(mocked_responses, app, proxy, endpoint):
    mocked_responses.get(
        url='https://example.org/foo/bar',
        body='ok',
        match=[
            responses.matchers.header_matcher({'User-Agent': 'test-ua'}),
            no_header_matcher('x-foo'),
            no_header_matcher('dontpass'),
        ],
    )
    resp = app.get(endpoint + '/foo/bar', headers={'user-agent': 'test-ua', 'dontpass': 'x'}, status=200)
    assert resp.text == 'ok'


def test_forced_headers(mocked_responses, app, proxy, endpoint):
    proxy.forced_headers = '''
x-foo :  bar  
badentry

# comment: do not use me
    '''
    proxy.save()

    mocked_responses.get(
        url='https://example.org/foo/bar',
        body='ok',
        match=[
            responses.matchers.header_matcher({'x-foo': 'bar'}),
            no_header_matcher('dontpass'),
        ],
    )
    resp = app.get(endpoint + '/foo/bar', headers={'user-agent': 'test', 'dontpass': 'x'}, status=200)
    assert resp.text == 'ok'


def test_query_parameters(mocked_responses, app, proxy, endpoint):
    mocked_responses.get(
        url='https://example.org/foo/bar',
        body='ok',
        match=[
            responses.matchers.query_param_matcher({'param1': '1', 'param2': '2'}, strict_match=True),
        ],
    )

    resp = app.get(endpoint + '/foo/bar?param1=1&param2=2', status=200)
    assert resp.text == 'ok'


def test_publik_signature_is_removed(mocked_responses, app, proxy, endpoint):
    mocked_responses.get(
        url='https://example.org/foo/bar',
        body='ok',
        match=[
            responses.matchers.query_param_matcher({'param1': '1', 'param2': '2'}, strict_match=True),
        ],
    )
    resp = app.get(
        endpoint + '/foo/bar?param1=1&param2=2&orig=coucou&algo=foo&nonce=bar&timestamp=xxx&signature=okok',
        status=200,
    )
    assert resp.text == 'ok'


def test_auth(mocked_responses, app, proxy, endpoint):
    proxy.basic_auth_username = 'test-login'
    proxy.basic_auth_password = 'test-pass'
    proxy.save()

    mocked_responses.get(
        url='https://example.org/foo/bar',
        body='ok',
        match=[responses.matchers.header_matcher({'Authorization': 'Basic dGVzdC1sb2dpbjp0ZXN0LXBhc3M='})],
    )
    resp = app.get(endpoint + '/foo/bar', status=200)
    assert resp.text == 'ok'


def test_timeout(mocked_responses, app, proxy, endpoint):
    proxy.http_timeout = 5
    proxy.save()
    mocked_responses.get(
        url='https://example.org/foo/bar',
        body='ok',
        match=[responses.matchers.request_kwargs_matcher({'timeout': 5})],
    )
    resp = app.get(endpoint + '/foo/bar', status=200)
    assert resp.text == 'ok'