201 lines
7.6 KiB
Python
201 lines
7.6 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import pytest
|
|
import mock
|
|
import requests
|
|
|
|
from django.contrib.auth.models import User
|
|
from django.template.defaultfilters import slugify
|
|
|
|
from combo.data.models import Page
|
|
from combo.apps.newsletters.models import NewslettersCell, SubscriptionsSaveError
|
|
from combo.apps.newsletters.forms import NewslettersManageForm
|
|
|
|
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
|
|
|
|
|
|
@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
|
|
|
|
@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):
|
|
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() == expected_subscriptions
|
|
|
|
|
|
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):
|
|
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, email=USER_EMAIL)
|
|
|
|
@mock.patch('combo.apps.newsletters.models.requests.post')
|
|
def test_set_subscriptions(mock_post, cell):
|
|
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, email=USER_EMAIL)
|
|
|
|
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):
|
|
restrictions = ('sms', 'mail')
|
|
cell.transports_restrictions = ','.join(restrictions)
|
|
newsletters = [n['id'] for n in cell.get_newsletters()]
|
|
subscriptions = [s['id'] for s in cell.get_subscriptions()]
|
|
form = NewslettersManageForm(instance=cell,
|
|
user=User(email=USER_EMAIL))
|
|
# test if all newsletters are present
|
|
for f_id, field in form.fields.iteritems():
|
|
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))
|