146 lines
6.4 KiB
Python
146 lines
6.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import mock
|
|
import pytest
|
|
|
|
from django.contrib.auth.models import AnonymousUser
|
|
from django.utils.six.moves.urllib import parse as urlparse
|
|
|
|
from combo.utils import requests, check_query, NothingInCacheException
|
|
|
|
class MockUser(object):
|
|
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('combo.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('combo.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/?')
|
|
scheme, netloc, path, params, querystring, fragment = urlparse.urlparse(url)
|
|
query = urlparse.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') == 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/?')
|
|
scheme, netloc, path, params, querystring, fragment = urlparse.urlparse(url)
|
|
query = urlparse.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') == True
|
|
|
|
|
|
def test_auto_sign():
|
|
with mock.patch('combo.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/?')
|
|
scheme, netloc, path, params, querystring, fragment = urlparse.urlparse(url)
|
|
query = urlparse.parse_qs(querystring, keep_blank_values=True)
|
|
assert query['orig'][0] == 'combo'
|
|
assert check_query(querystring, 'combo') == 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('combo.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/?')
|
|
scheme, netloc, path, params, querystring, fragment = urlparse.urlparse(url)
|
|
query = urlparse.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') == 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/?')
|
|
scheme, netloc, path, params, querystring, fragment = urlparse.urlparse(url)
|
|
query = urlparse.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') == 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/?')
|
|
scheme, netloc, path, params, querystring, fragment = urlparse.urlparse(url)
|
|
query = urlparse.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') == True
|
|
|
|
|
|
def test_sign_anonymous_user():
|
|
remote_service = {'url': 'http://example.org', 'secret': 'secret', 'orig': 'myself'}
|
|
with mock.patch('combo.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/?')
|
|
scheme, netloc, path, params, querystring, fragment = urlparse.urlparse(url)
|
|
query = urlparse.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') == True
|
|
|
|
def test_requests_cache():
|
|
with mock.patch('combo.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(u'http://cache.example.org/éléphant').content == b'hello second world'
|