267 lines
10 KiB
Python
267 lines
10 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import pytest
|
|
import mock
|
|
import requests
|
|
|
|
from django.template.defaultfilters import slugify
|
|
from django.contrib.auth.models import User
|
|
|
|
from combo.data.models import Page
|
|
from combo.apps.newsletters.models import NewslettersCell, SubscriptionsSaveError
|
|
from combo.apps.newsletters.forms import NewslettersManageForm
|
|
from combo.utils import check_query
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
NEWSLETTERS = [{'id': '1',
|
|
'text': 'Democratie locale',
|
|
'transports': [{'id': 'mail', 'text': 'mail'}]},
|
|
{'id': '2',
|
|
'text': 'Rencontres de quartiers',
|
|
'transports': [{'id': 'mail', 'text': 'mail'}]},
|
|
{'id': '3',
|
|
'text': 'Environnement',
|
|
'transports': [{'id': 'mail', 'text': 'mail'},
|
|
{'id': 'sms', 'text': 'sms'},
|
|
{'id': 'rss', 'text': 'rss'}]},
|
|
{'id': '4',
|
|
'text': u'Marchés publics',
|
|
'transports': [{'id': 'mail', 'text': 'mail'},
|
|
{'id': 'rss', 'text': 'rss'}]},
|
|
{'id': '5',
|
|
'text': "Offres d'emploi",
|
|
'transports': [{'id': 'mail', 'text': 'mail'},
|
|
{'id': 'rss', 'text': 'rss'}]},
|
|
{'id': '6',
|
|
'text': 'Infos créche',
|
|
'transports': [{'id': 'sms', 'text': 'sms'},
|
|
{'id': 'rss', 'text': 'rss'}]},
|
|
{'id': '7',
|
|
'text': 'Familles',
|
|
'transports': [{'id': 'mail', 'text': 'mail'},
|
|
{'id': 'sms', 'text': 'sms'}]},
|
|
{'id': '8',
|
|
'text': 'Travaux',
|
|
'transports': [{'id': 'mail', 'text': 'mail'},
|
|
{'id': 'sms', 'text': 'sms'},
|
|
{'id': 'rss', 'text': 'rss'}]}
|
|
]
|
|
|
|
SUBSCRIPTIONS = [
|
|
{'id': '3',
|
|
'text': 'Environnement',
|
|
'transports': [{'id': 'mail', 'text': 'mail'}]},
|
|
{'id': '7',
|
|
'text': 'Familles',
|
|
'transports': [{'id': 'mail', 'text': 'mail'}]},
|
|
{'id': '5',
|
|
'text': "Offres d'emploi",
|
|
'transports': [{'id': 'mail', 'text': 'mail'}]},
|
|
{'id': '6',
|
|
'text': 'Infos créche',
|
|
'transports': [{'id': 'sms', 'text': 'sms'},
|
|
{'id': 'rss', 'text': 'rss'}]},
|
|
]
|
|
|
|
USER_EMAIL = 'foobar@example.com'
|
|
|
|
@pytest.fixture
|
|
def cell():
|
|
page = Page()
|
|
page.save()
|
|
cell = NewslettersCell(title='Newsletter test', url='http://example.org/',
|
|
page=page, order=0)
|
|
cell.save()
|
|
return cell
|
|
|
|
@pytest.fixture
|
|
def user():
|
|
try:
|
|
user = User.objects.get(username='foo')
|
|
except User.DoesNotExist:
|
|
user = User.objects.create(username='foo', email=USER_EMAIL)
|
|
return user
|
|
|
|
|
|
@mock.patch('combo.apps.newsletters.models.requests.get')
|
|
def test_get_newsletters_by_transports(mock_get, cell):
|
|
restrictions = ('mail', 'sms')
|
|
cell.transports_restrictions = ','.join(restrictions)
|
|
expected_newsletters = []
|
|
for n in NEWSLETTERS:
|
|
for t in n['transports']:
|
|
if t['id'] in restrictions:
|
|
expected_newsletters.append(n)
|
|
continue
|
|
mock_json = mock.Mock()
|
|
mock_json.json.return_value = {'err': 0,
|
|
'data': NEWSLETTERS}
|
|
mock_get.return_value = mock_json
|
|
assert cell.get_newsletters() == expected_newsletters
|
|
assert mock_get.call_args[1]['without_user']
|
|
assert 'user' not in mock_get.call_args[1]
|
|
|
|
@mock.patch('combo.apps.newsletters.models.requests.get')
|
|
def test_get_newsletters_by_unrestricted_transports(mock_get, cell):
|
|
cell.transports_restrictions = ''
|
|
expected_newsletters = []
|
|
for n in NEWSLETTERS:
|
|
for t in n['transports']:
|
|
expected_newsletters.append(n)
|
|
mock_json = mock.Mock()
|
|
mock_json.json.return_value = {'err': 0,
|
|
'data': NEWSLETTERS}
|
|
mock_get.return_value = mock_json
|
|
assert cell.get_newsletters() == expected_newsletters
|
|
|
|
@mock.patch('combo.apps.newsletters.models.requests.get')
|
|
def test_get_newsletters_by_resources(mock_get, cell):
|
|
restrictions = ('marches-publics', 'familles', 'democratie-locale')
|
|
cell.transports_restrictions = 'mail'
|
|
cell.resources_restrictions = ','.join(restrictions)
|
|
expected_newsletters = []
|
|
for n in NEWSLETTERS:
|
|
if slugify(n['text']) in restrictions:
|
|
expected_newsletters.append(n)
|
|
mock_json = mock.Mock()
|
|
mock_json.json.return_value = {'err': 0,
|
|
'data': NEWSLETTERS}
|
|
mock_get.return_value = mock_json
|
|
assert cell.get_newsletters() == expected_newsletters
|
|
|
|
|
|
@mock.patch('combo.apps.newsletters.models.requests.get')
|
|
def test_get_subscriptions(mock_get, cell, user):
|
|
restrictions = ('mail', 'sms')
|
|
cell.transports_restrictions = ','.join(restrictions)
|
|
expected_subscriptions = []
|
|
for n in SUBSCRIPTIONS:
|
|
for t in n['transports']:
|
|
if t['id'] in restrictions:
|
|
expected_subscriptions.append(n)
|
|
continue
|
|
mock_json = mock.Mock()
|
|
mock_json.json.return_value = {'err': 0,
|
|
'data': SUBSCRIPTIONS}
|
|
mock_get.return_value = mock_json
|
|
assert cell.get_subscriptions(user) == expected_subscriptions
|
|
assert mock_get.call_args[1]['user'].email == USER_EMAIL
|
|
|
|
@mock.patch('combo.utils.requests_wrapper.RequestsSession.send')
|
|
def test_get_subscriptions_signature_check(mock_send, cell, user):
|
|
restrictions = ('mail', 'sms')
|
|
cell.transports_restrictions = ','.join(restrictions)
|
|
expected_subscriptions = []
|
|
for n in SUBSCRIPTIONS:
|
|
for t in n['transports']:
|
|
if t['id'] in restrictions:
|
|
expected_subscriptions.append(n)
|
|
continue
|
|
mock_json = mock.Mock(status_code=200)
|
|
mock_json.json.return_value = {'err': 0,
|
|
'data': SUBSCRIPTIONS}
|
|
mock_send.return_value = mock_json
|
|
cell.get_subscriptions(user)
|
|
url = mock_send.call_args[0][0].url
|
|
assert check_query(url.split('?', 1)[-1], 'combo')
|
|
|
|
|
|
def mocked_requests_connection_error(*args, **kwargs):
|
|
raise requests.ConnectionError()
|
|
|
|
@mock.patch('combo.apps.newsletters.models.requests.post',
|
|
side_effect=mocked_requests_connection_error)
|
|
def test_failed_set_subscriptions(mock_post, cell, user):
|
|
restrictions = ('sms', 'mail')
|
|
cell.transports_restrictions = ','.join(restrictions)
|
|
subscriptions = [{'id': '1', 'transports': [{'id': 'mail', 'text': 'mail'}]},
|
|
{'id': '7', 'transports': [{'id': 'sms', 'text': 'sms'}]},
|
|
{'id': '8', 'transports': [{'id': 'sms', 'text': 'sms'},
|
|
{'id': 'mail', 'text': 'mail'}]}]
|
|
with pytest.raises(SubscriptionsSaveError):
|
|
cell.set_subscriptions(subscriptions, user, uuid='useruuid')
|
|
|
|
@mock.patch('combo.apps.newsletters.models.requests.post')
|
|
def test_set_subscriptions_with_no_uuid(mocked_post, cell, user):
|
|
restrictions = ('sms', 'mail')
|
|
cell.transports_restrictions = ','.join(restrictions)
|
|
subscriptions = [{'id': '1', 'transports': [{'id': 'mail', 'text': 'mail'}]},
|
|
{'id': '8', 'transports': [{'id': 'sms', 'text': 'sms'},
|
|
{'id': 'mail', 'text': 'mail'}]}]
|
|
mock_json = mock.Mock()
|
|
mock_json.json.return_value = {'err': 0, 'data': True}
|
|
with pytest.raises(SubscriptionsSaveError):
|
|
cell.set_subscriptions(subscriptions, user)
|
|
|
|
@mock.patch('combo.apps.newsletters.models.requests.post')
|
|
def test_set_subscriptions(mock_post, cell, user):
|
|
restrictions = ('sms', 'mail')
|
|
cell.transports_restrictions = ','.join(restrictions)
|
|
subscriptions = [{'id': '1', 'transports': [{'id': 'mail', 'text': 'mail'}]},
|
|
{'id': '7', 'transports': [{'id': 'sms', 'text': 'sms'}]},
|
|
{'id': '8', 'transports': [{'id': 'sms', 'text': 'sms'},
|
|
{'id': 'mail', 'text': 'mail'}]}]
|
|
mock_json = mock.Mock()
|
|
mock_json.json.return_value = {'err': 0,
|
|
'data': True}
|
|
mock_post.return_value = mock_json
|
|
cell.set_subscriptions(subscriptions, user, uuid='useruuid')
|
|
args, kwargs = mock_post.call_args
|
|
assert kwargs['params']['uuid'] == 'useruuid'
|
|
assert kwargs['federation_key'] == 'email'
|
|
assert kwargs['user'].email == USER_EMAIL
|
|
|
|
@mock.patch('combo.apps.newsletters.models.requests.get')
|
|
def test_get_subscriptions_with_name_id_and_mobile(mock_get, cell, user):
|
|
restrictions = ('sms', 'mail')
|
|
cell.transports_restrictions = ','.join(restrictions)
|
|
mock_json = mock.Mock()
|
|
mock_json.json.return_value = {'err': 0,
|
|
'data': NEWSLETTERS}
|
|
mock_get.return_value = mock_json
|
|
|
|
fake_saml_request = mock.Mock()
|
|
fake_saml_request.user = mock.Mock(email=USER_EMAIL)
|
|
fake_saml_request.user.get_name_id.return_value = 'nameid'
|
|
fake_saml_request.session = {'mellon_session': {'mobile': '0607080900'}}
|
|
|
|
form = NewslettersManageForm(instance=cell, request=fake_saml_request)
|
|
args, kwargs = mock_get.call_args
|
|
assert kwargs['params'] == {'uuid': 'nameid', 'mobile': '0607080900'}
|
|
|
|
def mocked_requests_get(*args, **kwargs):
|
|
url = args[0]
|
|
class MockResponse:
|
|
def __init__(self, json_data):
|
|
self.json_data = json_data
|
|
|
|
def ok(self):
|
|
return True
|
|
|
|
def json(self):
|
|
return self.json_data
|
|
|
|
if 'newsletters' in url:
|
|
return MockResponse({'data': NEWSLETTERS})
|
|
else:
|
|
return MockResponse({'data': SUBSCRIPTIONS})
|
|
|
|
@mock.patch('combo.apps.newsletters.models.requests.get',
|
|
side_effect=mocked_requests_get)
|
|
def test_subscriptions_form(mock_get, cell, user):
|
|
restrictions = ('sms', 'mail')
|
|
cell.transports_restrictions = ','.join(restrictions)
|
|
newsletters = [n['id'] for n in cell.get_newsletters()]
|
|
assert mock_get.call_args[1]['without_user']
|
|
subscriptions = [s['id'] for s in cell.get_subscriptions(user)]
|
|
fake_request = mock.Mock(user=mock.Mock(username='username', email=USER_EMAIL),
|
|
session={})
|
|
form = NewslettersManageForm(instance=cell, request=fake_request)
|
|
# test if all newsletters are present
|
|
for f_id, field in form.fields.items():
|
|
assert f_id in newsletters
|
|
# test if initial fields are in the restrictions
|
|
for s in subscriptions:
|
|
assert set(form.fields[s].initial).intersection(set(restrictions))
|