155 lines
6.5 KiB
Python
155 lines
6.5 KiB
Python
from unittest import mock
|
|
from urllib import parse
|
|
|
|
import pytest
|
|
from django.contrib.auth.models import AnonymousUser
|
|
|
|
from chrono.utils.requests_wrapper import NothingInCacheException, requests
|
|
from chrono.utils.signature import check_query
|
|
|
|
|
|
class MockUser:
|
|
email = 'foo@example.net'
|
|
is_authenticated = True
|
|
|
|
def get_name_id(self):
|
|
if self.samlized:
|
|
return 'r2d2'
|
|
return None
|
|
|
|
def __init__(self, samlized=True):
|
|
self.samlized = samlized
|
|
|
|
|
|
def test_nosign():
|
|
with mock.patch('chrono.utils.requests_wrapper.RequestsSession.send') as send:
|
|
requests.get('http://example.org/foo/bar/')
|
|
assert send.call_args[0][0].url == 'http://example.org/foo/bar/'
|
|
|
|
|
|
def test_sign():
|
|
remote_service = {'url': 'http://example.org', 'secret': 'secret', 'orig': 'myself'}
|
|
with mock.patch('chrono.utils.requests_wrapper.RequestsSession.send') as send:
|
|
requests.get('/foo/bar/', remote_service=remote_service)
|
|
url = send.call_args[0][0].url
|
|
assert url.startswith('http://example.org/foo/bar/?')
|
|
querystring = parse.urlparse(url)[4]
|
|
query = parse.parse_qs(querystring, keep_blank_values=True)
|
|
assert query['orig'][0] == 'myself'
|
|
assert query['email'][0] == ''
|
|
assert query['NameID'][0] == ''
|
|
assert check_query(querystring, 'secret') is True
|
|
|
|
requests.get('/foo/bar/', remote_service=remote_service, without_user=True)
|
|
url = send.call_args[0][0].url
|
|
assert url.startswith('http://example.org/foo/bar/?')
|
|
querystring = parse.urlparse(url)[4]
|
|
query = parse.parse_qs(querystring, keep_blank_values=True)
|
|
assert query['orig'][0] == 'myself'
|
|
assert 'email' not in query
|
|
assert 'NameID' not in query
|
|
assert check_query(querystring, 'secret') is True
|
|
|
|
|
|
def test_auto_sign():
|
|
with mock.patch('chrono.utils.requests_wrapper.RequestsSession.send') as send:
|
|
requests.get('http://example.org/foo/bar/', remote_service='auto')
|
|
url = send.call_args[0][0].url
|
|
assert url.startswith('http://example.org/foo/bar/?')
|
|
querystring = parse.urlparse(url)[4]
|
|
query = parse.parse_qs(querystring, keep_blank_values=True)
|
|
assert query['orig'][0] == 'chrono'
|
|
assert check_query(querystring, 'chrono') is True
|
|
|
|
requests.get('http://doesnotexist/foo/bar/', remote_service='auto')
|
|
assert send.call_args[0][0].url == 'http://doesnotexist/foo/bar/'
|
|
|
|
|
|
def test_sign_user():
|
|
remote_service = {'url': 'http://example.org', 'secret': 'secret', 'orig': 'myself'}
|
|
with mock.patch('chrono.utils.requests_wrapper.RequestsSession.send') as send:
|
|
|
|
user = MockUser(samlized=True)
|
|
|
|
requests.get('/foo/bar/', remote_service=remote_service, user=user)
|
|
url = send.call_args[0][0].url
|
|
assert url.startswith('http://example.org/foo/bar/?')
|
|
querystring = parse.urlparse(url)[4]
|
|
query = parse.parse_qs(querystring, keep_blank_values=True)
|
|
assert query['NameID'][0] == 'r2d2'
|
|
assert 'email' not in query
|
|
assert query['orig'][0] == 'myself'
|
|
assert check_query(querystring, 'secret') is True
|
|
|
|
requests.get('/foo/bar/', remote_service=remote_service, user=user, federation_key='email')
|
|
url = send.call_args[0][0].url
|
|
assert url.startswith('http://example.org/foo/bar/?')
|
|
querystring = parse.urlparse(url)[4]
|
|
query = parse.parse_qs(querystring, keep_blank_values=True)
|
|
assert query['email'][0] == 'foo@example.net'
|
|
assert 'NameID' not in query
|
|
assert query['orig'][0] == 'myself'
|
|
assert check_query(querystring, 'secret') is True
|
|
|
|
user = MockUser(samlized=False)
|
|
|
|
requests.get('/foo/bar/', remote_service=remote_service, user=user)
|
|
url = send.call_args[0][0].url
|
|
assert url.startswith('http://example.org/foo/bar/?')
|
|
querystring = parse.urlparse(url)[4]
|
|
query = parse.parse_qs(querystring, keep_blank_values=True)
|
|
assert 'NameID' not in query
|
|
assert query['email'][0] == 'foo@example.net'
|
|
assert query['orig'][0] == 'myself'
|
|
assert check_query(querystring, 'secret') is True
|
|
|
|
|
|
def test_sign_anonymous_user():
|
|
remote_service = {'url': 'http://example.org', 'secret': 'secret', 'orig': 'myself'}
|
|
with mock.patch('chrono.utils.requests_wrapper.RequestsSession.send') as send:
|
|
|
|
user = AnonymousUser()
|
|
|
|
requests.get('/foo/bar/', remote_service=remote_service, user=user)
|
|
url = send.call_args[0][0].url
|
|
assert url.startswith('http://example.org/foo/bar/?')
|
|
querystring = parse.urlparse(url)[4]
|
|
query = parse.parse_qs(querystring, keep_blank_values=True)
|
|
assert query['NameID'][0] == ''
|
|
assert query['email'][0] == ''
|
|
assert query['orig'][0] == 'myself'
|
|
assert check_query(querystring, 'secret') is True
|
|
|
|
|
|
def test_requests_cache():
|
|
with mock.patch('chrono.utils.requests_wrapper.RequestsSession.request') as requests_get:
|
|
requests_get.return_value = mock.Mock(content=b'hello world', status_code=200)
|
|
# default cache, nothing in there
|
|
assert requests.get('http://cache.example.org/').content == b'hello world'
|
|
assert requests_get.call_count == 1
|
|
# now there's something in cache
|
|
assert requests.get('http://cache.example.org/').content == b'hello world'
|
|
assert requests_get.call_count == 1
|
|
# value changed
|
|
requests_get.return_value = mock.Mock(content=b'hello second world', status_code=200)
|
|
assert requests.get('http://cache.example.org/').content == b'hello world'
|
|
assert requests_get.call_count == 1
|
|
# force cache invalidation
|
|
assert (
|
|
requests.get('http://cache.example.org/', invalidate_cache=True).content == b'hello second world'
|
|
)
|
|
assert requests_get.call_count == 2
|
|
# check raise_if_not_cached
|
|
with pytest.raises(NothingInCacheException):
|
|
requests.get('http://cache.example.org/other', raise_if_not_cached=True)
|
|
|
|
# check with unicode url
|
|
assert requests.get('http://cache.example.org/éléphant').content == b'hello second world'
|
|
|
|
|
|
def test_requests_to_legacy_urls():
|
|
with mock.patch('chrono.utils.requests_wrapper.RequestsSession.request') as requests_get:
|
|
requests_get.return_value = mock.Mock(content=b'hello world', status_code=200)
|
|
assert requests.get('http://old.org/').content == b'hello world'
|
|
assert requests_get.call_args.args[1] == 'http://new.org/'
|