505 lines
20 KiB
Python
505 lines
20 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import copy
|
|
import json
|
|
import pytest
|
|
import mock
|
|
from decimal import Decimal
|
|
from requests.exceptions import ConnectionError
|
|
|
|
from django.apps import apps
|
|
from django.test.client import RequestFactory
|
|
from django.test import override_settings
|
|
from django.urls import reverse
|
|
from django.conf import settings
|
|
from django.core.management import call_command
|
|
from django.utils.encoding import force_bytes, force_text
|
|
from django.utils.six.moves.urllib import parse as urlparse
|
|
from django.utils.timezone import timedelta, now
|
|
from django.contrib.auth.models import User
|
|
|
|
from combo.utils import check_query, aes_hex_encrypt
|
|
from combo.data.models import Page
|
|
from combo.apps.lingo.models import (Regie, ActiveItems, ItemsHistory, SelfDeclaredInvoicePayment,
|
|
Transaction, BasketItem, PaymentBackend)
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
INVOICES = [
|
|
{
|
|
'id': 'F201601',
|
|
'display_id': 'F-2016-One',
|
|
'label': 'invoice-one',
|
|
'regie': 'remote',
|
|
'created': '2016-02-02',
|
|
'pay_limit_date': '2999-12-31',
|
|
'total_amount': '123.45',
|
|
'amount': '123.45',
|
|
'has_pdf': True,
|
|
'online_payment': True,
|
|
'paid': False,
|
|
'payment_date': '1970-01-01',
|
|
'no_online_payment_reason': '',
|
|
},
|
|
]
|
|
|
|
@pytest.fixture
|
|
def admin():
|
|
try:
|
|
admin = User.objects.get(username='foo')
|
|
except User.DoesNotExist:
|
|
admin = User.objects.create_user('foo', email=None, password='bar')
|
|
admin.email = 'foo@example.net'
|
|
admin.save()
|
|
return admin
|
|
|
|
@pytest.fixture
|
|
def remote_regie():
|
|
try:
|
|
payment_backend = PaymentBackend.objects.get(slug='test1')
|
|
except PaymentBackend.DoesNotExist:
|
|
payment_backend = PaymentBackend.objects.create(
|
|
label='test1', slug='test1', service='dummy', service_options={'siret': '1234'})
|
|
try:
|
|
regie = Regie.objects.get(slug='remote')
|
|
except Regie.DoesNotExist:
|
|
regie = Regie()
|
|
regie.label = 'Remote'
|
|
regie.slug = 'remote'
|
|
regie.description = 'remote'
|
|
regie.payment_min_amount = Decimal(2.0)
|
|
regie.payment_backend = payment_backend
|
|
regie.webservice_url = 'http://example.org/regie' # is_remote
|
|
regie.save()
|
|
return regie
|
|
|
|
|
|
class MockUser(object):
|
|
email = 'foo@example.net'
|
|
is_authenticated = True
|
|
|
|
def get_name_id(self):
|
|
return 'r2d2'
|
|
|
|
|
|
@mock.patch('combo.utils.requests_wrapper.RequestsSession.send')
|
|
def test_remote_regie_active_invoices_cell(mock_send, remote_regie):
|
|
assert remote_regie.is_remote() == True
|
|
|
|
page = Page(title='xxx', slug='test_basket_cell', template_name='standard')
|
|
page.save()
|
|
cell = ActiveItems(regie='remote', page=page, placeholder='content', order=0)
|
|
context = {'request': RequestFactory().get('/')}
|
|
context['synchronous'] = True # to get fresh content
|
|
|
|
user = MockUser()
|
|
context['user'] = user
|
|
context['request'].user = user
|
|
|
|
assert cell.is_relevant(context) is True
|
|
|
|
# show regie with an invoice
|
|
ws_invoices = {'err': 0, 'data': INVOICES}
|
|
mock_response = mock.Mock(status_code=200, content=json.dumps(ws_invoices))
|
|
mock_response.json.return_value = ws_invoices
|
|
mock_send.return_value = mock_response
|
|
content = cell.render(context)
|
|
assert 'F-2016-One' in content
|
|
assert '123.45' in content
|
|
|
|
assert '?page=%s' % page.pk in content
|
|
# check if regie webservice has been correctly called
|
|
assert mock_send.call_args[0][0].method == 'GET'
|
|
url = mock_send.call_args[0][0].url
|
|
scheme, netloc, path, params, querystring, fragment = urlparse.urlparse(url)
|
|
assert scheme == 'http'
|
|
assert netloc == 'example.org'
|
|
assert path == '/regie/invoices/'
|
|
query = urlparse.parse_qs(querystring, keep_blank_values=True)
|
|
assert query['NameID'][0] == 'r2d2'
|
|
assert query['orig'][0] == 'combo'
|
|
assert check_query(querystring, 'combo') == True
|
|
|
|
# with no invoice
|
|
ws_invoices = {'err': 0, 'data': []}
|
|
mock_response = mock.Mock(status_code=200, content=json.dumps(ws_invoices))
|
|
mock_response.json.return_value = ws_invoices
|
|
mock_send.return_value = mock_response
|
|
content = cell.render(context)
|
|
assert 'No items yet' in content
|
|
|
|
# regie is down
|
|
mock_send.side_effect = ConnectionError
|
|
content = cell.render(context)
|
|
assert 'Regie "Remote" is unavailable, please retry later.' in content
|
|
|
|
|
|
@mock.patch('combo.utils.requests_wrapper.RequestsSession.send')
|
|
def test_remote_regie_past_invoices_cell(mock_send, remote_regie):
|
|
assert remote_regie.is_remote() == True
|
|
|
|
page = Page(title='xxx', slug='test_basket_cell', template_name='standard')
|
|
page.save()
|
|
cell = ItemsHistory(regie='remote', page=page, placeholder='content', order=0)
|
|
context = {'request': RequestFactory().get('/')}
|
|
context['synchronous'] = True # to get fresh content
|
|
|
|
user = MockUser()
|
|
context['user'] = user
|
|
context['request'].user = user
|
|
|
|
assert cell.is_relevant(context) is True
|
|
|
|
# show regie with an invoice
|
|
ws_invoices = {'err': 0, 'data': INVOICES}
|
|
mock_response = mock.Mock(status_code=200, content=json.dumps(ws_invoices))
|
|
mock_response.json.return_value = ws_invoices
|
|
mock_send.return_value = mock_response
|
|
content = cell.render(context)
|
|
assert 'F-2016-One' in content
|
|
assert '123.45' in content
|
|
assert 'class="invoice-payment-limit-date"' in content
|
|
|
|
# invoice without limit date
|
|
invoices = copy.deepcopy(INVOICES)
|
|
invoices[0]['pay_limit_date'] = ''
|
|
ws_invoices = {'err': 0, 'data': invoices}
|
|
mock_response = mock.Mock(status_code=200, content=json.dumps(ws_invoices))
|
|
mock_response.json.return_value = ws_invoices
|
|
mock_send.return_value = mock_response
|
|
content = cell.render(context)
|
|
assert 'class="invoice-payment-limit-date"' not in content
|
|
|
|
# invoice with amount_paid
|
|
invoices = copy.deepcopy(INVOICES)
|
|
invoices[0]['amount'] = '100.00'
|
|
invoices[0]['amount_paid'] = '23.45'
|
|
ws_invoices = {'err': 0, 'data': invoices}
|
|
mock_response = mock.Mock(status_code=200, content=json.dumps(ws_invoices))
|
|
mock_response.json.return_value = ws_invoices
|
|
mock_send.return_value = mock_response
|
|
content = cell.render(context)
|
|
assert '100.00' in content
|
|
assert '23.45' in content
|
|
assert 'class="invoice-amount-paid"' in content
|
|
|
|
# invoice with zero amount_paid
|
|
invoices = copy.deepcopy(INVOICES)
|
|
invoices[0]['amount'] = '123.45'
|
|
invoices[0]['amount_paid'] = '0.00'
|
|
ws_invoices = {'err': 0, 'data': invoices}
|
|
mock_response = mock.Mock(status_code=200, content=json.dumps(ws_invoices))
|
|
mock_response.json.return_value = ws_invoices
|
|
mock_send.return_value = mock_response
|
|
content = cell.render(context)
|
|
assert '123.45' in content
|
|
assert 'class="invoice-amount-paid"' not in content
|
|
|
|
# check if regie webservice has been correctly called
|
|
assert mock_send.call_args[0][0].method == 'GET'
|
|
url = mock_send.call_args[0][0].url
|
|
scheme, netloc, path, params, querystring, fragment = urlparse.urlparse(url)
|
|
assert scheme == 'http'
|
|
assert netloc == 'example.org'
|
|
assert path == '/regie/invoices/history/'
|
|
query = urlparse.parse_qs(querystring, keep_blank_values=True)
|
|
assert query['NameID'][0] == 'r2d2'
|
|
assert query['orig'][0] == 'combo'
|
|
assert check_query(querystring, 'combo') == True
|
|
|
|
# with no invoice
|
|
ws_invoices = {'err': 0, 'data': []}
|
|
mock_response = mock.Mock(status_code=200, content=json.dumps(ws_invoices))
|
|
mock_response.json.return_value = ws_invoices
|
|
mock_send.return_value = mock_response
|
|
content = cell.render(context)
|
|
assert 'No items yet' in content
|
|
|
|
cell.hide_if_empty = True
|
|
cell.save()
|
|
content = cell.render(context)
|
|
assert content.strip() == ''
|
|
|
|
# regie is down
|
|
mock_send.side_effect = ConnectionError
|
|
content = cell.render(context)
|
|
assert 'Regie "Remote" is unavailable, please retry later.' in content
|
|
|
|
|
|
@mock.patch('combo.apps.lingo.models.Regie.pay_invoice')
|
|
@mock.patch('combo.apps.lingo.models.requests.get')
|
|
def test_anonymous_successful_item_payment(mock_get, mock_pay_invoice, app, remote_regie):
|
|
assert remote_regie.is_remote() == True
|
|
encrypt_id = aes_hex_encrypt(settings.SECRET_KEY, force_bytes('F201601'))
|
|
# invoice with amount_paid
|
|
invoices = copy.deepcopy(INVOICES)
|
|
invoices[0]['amount'] = '100.00'
|
|
invoices[0]['amount_paid'] = '23.45'
|
|
mock_json = mock.Mock()
|
|
mock_json.json.return_value = {'err': 0, 'data': invoices[0]}
|
|
mock_get.return_value = mock_json
|
|
mock_pay_invoice.return_value = mock.Mock(status_code=200)
|
|
resp = app.get('/lingo/item/%s/%s/' % (remote_regie.id, encrypt_id))
|
|
assert 'Total amount: <span class="amount">123.45€</span>' in resp.text
|
|
assert 'Amount to pay: <span class="amount">100.00€</span>' in resp.text
|
|
assert 'Amount already paid: <span class="amount">23.45€</span>' in resp.text
|
|
|
|
# invoice without amount_paid
|
|
mock_json = mock.Mock()
|
|
mock_json.json.return_value = {'err': 0, 'data': INVOICES[0]}
|
|
mock_get.return_value = mock_json
|
|
mock_pay_invoice.return_value = mock.Mock(status_code=200)
|
|
resp = app.get('/lingo/item/%s/%s/' % (remote_regie.id, encrypt_id))
|
|
assert 'Total amount: <span class="amount">123.45€</span>' in resp.text
|
|
assert 'Amount to pay: <span class="amount">123.45€</span>' in resp.text
|
|
assert 'Amount already paid>' not in resp.text
|
|
|
|
form = resp.form
|
|
|
|
assert 'email' in form.fields
|
|
assert form['email'].value == ''
|
|
assert 'item_url' in form.fields
|
|
assert form['item_url'].value == '/lingo/item/%s/%s/' % (remote_regie.id, encrypt_id)
|
|
assert 'item' in form.fields
|
|
assert form['item'].value == 'F201601'
|
|
assert 'regie' in form.fields
|
|
assert form['regie'].value == force_text(remote_regie.pk)
|
|
|
|
form['email'] = 'ghost@buster.com'
|
|
resp = form.submit()
|
|
|
|
assert resp.status_code == 302
|
|
location = resp.location
|
|
assert 'dummy-payment' in location
|
|
parsed = urlparse.urlparse(location)
|
|
# get return_url and transaction id from location
|
|
qs = urlparse.parse_qs(parsed.query)
|
|
args = {'transaction_id': qs['transaction_id'][0], 'signed': True,
|
|
'ok': True, 'reason': 'Paid'}
|
|
# make sure return url is the user return URL
|
|
return_url = qs['return_url'][0]
|
|
assert urlparse.urlparse(return_url).path.startswith('/lingo/return-payment-backend')
|
|
# simulate successful return URL
|
|
resp = app.get(qs['return_url'][0], params=args)
|
|
# redirect to payment status
|
|
assert resp.status_code == 302
|
|
assert urlparse.urlparse(resp.url).path.startswith('/lingo/payment-status')
|
|
resp = resp.follow()
|
|
assert urlparse.urlparse(resp.html.find('a', {'id': 'next-url'})['href']).path == '/'
|
|
# simulate successful call to callback URL
|
|
resp = app.get(reverse('lingo-callback', kwargs={'regie_pk': remote_regie.id}), params=args)
|
|
trans = Transaction.objects.all()
|
|
b_item = BasketItem.objects.all()
|
|
|
|
assert trans
|
|
assert b_item
|
|
trans = trans[0]
|
|
b_item = b_item[0]
|
|
assert b_item.subject == 'Invoice #%s' % INVOICES[0]['display_id']
|
|
assert b_item.amount == Decimal(INVOICES[0]['amount'])
|
|
assert b_item in trans.items.all()
|
|
|
|
assert resp.status_code == 200
|
|
|
|
@mock.patch('combo.apps.lingo.models.requests.get')
|
|
def test_anonymous_item_payment_email_error(mock_get, app, remote_regie):
|
|
assert remote_regie.is_remote() == True
|
|
encrypt_id = aes_hex_encrypt(settings.SECRET_KEY, force_bytes('F201601'))
|
|
mock_json = mock.Mock()
|
|
mock_json.json.return_value = {'err': 0, 'data': INVOICES[0]}
|
|
mock_get.return_value = mock_json
|
|
resp = app.get('/lingo/item/%s/%s/' % (remote_regie.id, encrypt_id))
|
|
form = resp.form
|
|
resp = form.submit()
|
|
|
|
assert resp.status_code == 302
|
|
path = urlparse.urlparse(resp.location).path
|
|
assert path == '/lingo/item/%s/%s/' % (remote_regie.id, encrypt_id)
|
|
|
|
@mock.patch('combo.apps.lingo.models.requests.get')
|
|
def test_wrong_crypted_item(mock_get, remote_regie, app):
|
|
assert remote_regie.is_remote() == True
|
|
mock_json = mock.Mock()
|
|
mock_json.json.return_value = {'err': 0, 'data': INVOICES[0]}
|
|
mock_get.return_value = mock_json
|
|
resp = app.get('/lingo/item/%s/%s/' % (remote_regie.id, 'zrzer854sfaear45e6rzerzerzef'), status=404)
|
|
|
|
@mock.patch('combo.apps.lingo.models.requests.get')
|
|
def test_self_declared_invoice(mock_get, app, remote_regie):
|
|
mock_json = mock.Mock()
|
|
mock_json.json.return_value = {'err': 0, 'data': INVOICES[0]}
|
|
mock_get.return_value = mock_json
|
|
|
|
page = Page(title='xxx', slug='test-self-invoice', template_name='standard')
|
|
page.save()
|
|
cell = SelfDeclaredInvoicePayment(regie='remote', page=page, placeholder='content', order=0)
|
|
cell.save()
|
|
|
|
resp = app.get('/test-self-invoice/')
|
|
resp = resp.form.submit().follow()
|
|
assert 'Sorry, no invoice were found with that number and amount.'
|
|
|
|
resp = app.get('/test-self-invoice/')
|
|
resp.form['invoice-number'] = 'F201601'
|
|
resp.form['invoice-amount'] = 'FOOBAR' # wrong format
|
|
resp = resp.form.submit().follow()
|
|
assert 'Sorry, the provided amount is invalid.'
|
|
|
|
resp = app.get('/test-self-invoice/')
|
|
resp.form['invoice-number'] = 'F201602' # invalid number
|
|
resp.form['invoice-amount'] = '123.45'
|
|
resp = resp.form.submit().follow()
|
|
assert 'Sorry, no invoice were found with that number and amount.'
|
|
|
|
resp = app.get('/test-self-invoice/')
|
|
resp.form['invoice-number'] = 'F201601'
|
|
resp.form['invoice-amount'] = '123.46' # invalid amount
|
|
resp = resp.form.submit().follow()
|
|
assert 'Sorry, no invoice were found with that number and amount.'
|
|
|
|
resp = app.get('/test-self-invoice/')
|
|
resp.form['invoice-number'] = 'F201601'
|
|
resp.form['invoice-amount'] = '123.45'
|
|
resp = resp.form.submit()
|
|
path = urlparse.urlparse(resp.location).path
|
|
assert path.startswith('/lingo/item/%s/' % remote_regie.id)
|
|
resp = resp.follow()
|
|
|
|
|
|
@mock.patch('combo.apps.lingo.models.Regie.pay_invoice')
|
|
@mock.patch('combo.apps.lingo.models.requests.get')
|
|
@mock.patch('combo.apps.lingo.models.requests.post')
|
|
def test_remote_item_payment_failure(mock_post, mock_get, mock_pay_invoice, app, remote_regie):
|
|
page = Page(title='xxx', slug='active-remote-invoices-page', template_name='standard')
|
|
page.save()
|
|
assert remote_regie.is_remote()
|
|
encrypt_id = aes_hex_encrypt(settings.SECRET_KEY, force_bytes('F201601'))
|
|
mock_json = mock.Mock()
|
|
mock_json.json.return_value = {'err': 0, 'data': INVOICES[0]}
|
|
mock_get.return_value = mock_json
|
|
mock_pay_invoice.return_value = mock.Mock(status_code=200)
|
|
resp = app.get('/lingo/item/%s/%s/?page=%s' % (remote_regie.id, encrypt_id, page.pk))
|
|
form = resp.form
|
|
|
|
assert 'email' in form.fields
|
|
assert form['email'].value == ''
|
|
assert 'item_url' in form.fields
|
|
assert form['item_url'].value == '/lingo/item/%s/%s/?page=%s' % (remote_regie.id, encrypt_id, page.pk)
|
|
assert 'item' in form.fields
|
|
assert form['item'].value == 'F201601'
|
|
assert 'regie' in form.fields
|
|
assert form['regie'].value == force_text(remote_regie.pk)
|
|
|
|
form['email'] = 'test@example.net'
|
|
resp = form.submit()
|
|
|
|
assert resp.status_code == 302
|
|
location = resp.location
|
|
assert 'dummy-payment' in location
|
|
parsed = urlparse.urlparse(location)
|
|
# get return_url and transaction id from location
|
|
qs = urlparse.parse_qs(parsed.query)
|
|
args = {'transaction_id': qs['transaction_id'][0], 'signed': True,
|
|
'ok': True, 'reason': 'Paid'}
|
|
# make sure return url is the user return URL
|
|
return_url = qs['return_url'][0]
|
|
assert urlparse.urlparse(return_url).path.startswith(
|
|
'/lingo/return-payment-backend/%s/' % remote_regie.payment_backend.id)
|
|
# simulate payment failure
|
|
mock_get.side_effect = ConnectionError('where is my hostname?')
|
|
resp = app.get(qs['return_url'][0], params=args)
|
|
# redirect to payment status
|
|
assert resp.status_code == 302
|
|
assert urlparse.urlparse(resp.url).path.startswith('/lingo/payment-status')
|
|
resp = resp.follow()
|
|
assert urlparse.urlparse(resp.html.find('a', {'id': 'next-url'})['href']).path == \
|
|
'/active-remote-invoices-page/'
|
|
|
|
# simulate successful call to callback URL
|
|
resp = app.get(reverse('lingo-callback', kwargs={'regie_pk': remote_regie.id}), params=args)
|
|
trans = Transaction.objects.all()
|
|
b_item = BasketItem.objects.all()
|
|
|
|
assert trans.count() == 1
|
|
assert not b_item
|
|
assert trans[0].to_be_paid_remote_items
|
|
assert resp.status_code == 200
|
|
|
|
mock_get.side_effect = None
|
|
appconfig = apps.get_app_config('lingo')
|
|
appconfig.update_transactions()
|
|
|
|
assert Transaction.objects.count() == 1
|
|
assert BasketItem.objects.count() == 1
|
|
assert Transaction.objects.all()[0].to_be_paid_remote_items is None
|
|
|
|
appconfig.update_transactions()
|
|
|
|
|
|
@mock.patch('combo.apps.lingo.models.Regie.pay_invoice')
|
|
@mock.patch('combo.apps.lingo.models.requests.get')
|
|
def test_remote_invoice_successfull_payment_redirect(mock_get, mock_pay_invoice, app, remote_regie):
|
|
page = Page(title='xxx', slug='active-remote-invoices-page', template_name='standard')
|
|
page.save()
|
|
assert remote_regie.is_remote()
|
|
encrypt_id = aes_hex_encrypt(settings.SECRET_KEY, force_bytes('F201601'))
|
|
mock_json = mock.Mock()
|
|
mock_json.json.return_value = {'err': 0, 'data': INVOICES[0]}
|
|
mock_get.return_value = mock_json
|
|
mock_pay_invoice.return_value = mock.Mock(status_code=200)
|
|
resp = app.get('/lingo/item/%s/%s/?page=%s' % (remote_regie.id, encrypt_id, page.pk))
|
|
form = resp.form
|
|
assert form['next_url'].value == '/active-remote-invoices-page/'
|
|
form['email'] = 'test@example.net'
|
|
resp = form.submit()
|
|
|
|
assert resp.status_code == 302
|
|
location = resp.location
|
|
assert 'dummy-payment' in location
|
|
parsed = urlparse.urlparse(location)
|
|
# get return_url and transaction id from location
|
|
qs = urlparse.parse_qs(parsed.query)
|
|
args = {'transaction_id': qs['transaction_id'][0], 'signed': True,
|
|
'ok': True, 'reason': 'Paid'}
|
|
resp = app.get(qs['return_url'][0], params=args)
|
|
# redirect to payment status
|
|
assert resp.status_code == 302
|
|
assert urlparse.urlparse(resp.url).path.startswith('/lingo/payment-status')
|
|
resp = resp.follow()
|
|
assert urlparse.urlparse(resp.html.find('a', {'id': 'next-url'})['href']).path == \
|
|
'/active-remote-invoices-page/'
|
|
|
|
|
|
@mock.patch('combo.apps.lingo.models.UserSAMLIdentifier')
|
|
@mock.patch('combo.apps.lingo.models.requests.get')
|
|
def test_send_new_remote_invoices_by_email(mock_get, user_saml, admin, app, remote_regie, mailoutbox):
|
|
mocked_objects = mock.Mock()
|
|
mocked_objects.get.return_value = mock.Mock(user=admin)
|
|
user_saml.objects = mocked_objects
|
|
invoice_now = now()
|
|
creation_date = (invoice_now - timedelta(days=1)).date().isoformat()
|
|
pay_limit_date = (invoice_now + timedelta(days=30)).date().isoformat()
|
|
FAKE_PENDING_INVOICES = {
|
|
'data' : {'foo': {'invoices': [{'id': '01', 'label': '010101', 'paid': False,
|
|
'amount': '37.26', 'total_amount': '37.26', 'online_payment': False,
|
|
'has_pdf': True, 'created': creation_date,
|
|
'pay_limit_date': pay_limit_date}]},
|
|
}
|
|
}
|
|
mock_response = mock.Mock(status_code=200, content=json.dumps(FAKE_PENDING_INVOICES))
|
|
mock_response.json.return_value = FAKE_PENDING_INVOICES
|
|
mock_get.return_value = mock_response
|
|
with override_settings(LANGUAGE_CODE='fr'):
|
|
call_command('notify_new_remote_invoices')
|
|
|
|
assert len(mailoutbox) == 1
|
|
assert mailoutbox[0].recipients() == ['foo@example.net']
|
|
assert mailoutbox[0].from_email == settings.DEFAULT_FROM_EMAIL
|
|
assert mailoutbox[0].subject == u'Nouvelle facture numéro 01 disponible'
|
|
html_message = mailoutbox[0].alternatives[0][0]
|
|
assert 'http://localhost' in mailoutbox[0].body
|
|
assert 'http://localhost' in html_message
|
|
assert mailoutbox[0].attachments[0][0] == '01.pdf'
|
|
assert mailoutbox[0].attachments[0][2] == 'application/pdf'
|