846 lines
32 KiB
Python
846 lines
32 KiB
Python
import copy
|
|
import json
|
|
import urllib.parse
|
|
from decimal import Decimal
|
|
from unittest import mock
|
|
|
|
import eopayment
|
|
import httmock
|
|
import pytest
|
|
from django.apps import apps
|
|
from django.conf import settings
|
|
from django.contrib.auth.models import User
|
|
from django.core.management import call_command
|
|
from django.test import override_settings
|
|
from django.test.client import RequestFactory
|
|
from django.urls import reverse
|
|
from django.utils.encoding import force_bytes, force_text
|
|
from django.utils.timezone import now, timedelta
|
|
from requests.exceptions import ConnectionError
|
|
from requests.models import Response
|
|
|
|
from combo.apps.lingo.models import (
|
|
ActiveItems,
|
|
BasketItem,
|
|
ItemsHistory,
|
|
PaymentBackend,
|
|
Regie,
|
|
SelfDeclaredInvoicePayment,
|
|
Transaction,
|
|
)
|
|
from combo.data.models import Page
|
|
from combo.utils import aes_hex_encrypt, check_query
|
|
|
|
from .test_manager import login
|
|
|
|
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': None,
|
|
'no_online_payment_reason': '',
|
|
'reference_id': 'order-id-1',
|
|
},
|
|
{
|
|
'id': 'F201602',
|
|
'display_id': 'F-2016-Two',
|
|
'label': 'invoice-two',
|
|
'regie': 'remote',
|
|
'created': '2016-02-02',
|
|
'pay_limit_date': '2999-12-31',
|
|
'total_amount': '543.21',
|
|
'amount': '543.21',
|
|
'has_pdf': True,
|
|
'online_payment': True,
|
|
'paid': False,
|
|
'payment_date': None,
|
|
'no_online_payment_reason': '',
|
|
'reference_id': 'order-id-2',
|
|
},
|
|
]
|
|
|
|
|
|
@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.can_pay_only_one_basket_item = False
|
|
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:
|
|
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() is True
|
|
assert remote_regie.can_pay_only_one_basket_item is False
|
|
|
|
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 'F-2016-Two' in content
|
|
assert '543.21' in content
|
|
|
|
# set the second one as paid
|
|
Transaction.objects.create(
|
|
regie=remote_regie, remote_items=INVOICES[1]['id'], status=eopayment.PAID, end_date=now()
|
|
)
|
|
content = cell.render(context)
|
|
assert 'F-2016-One' in content
|
|
assert '123.45' in content
|
|
assert 'F-2016-Two' not in content
|
|
assert '543.21' not 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, dummy, querystring, dummy = urllib.parse.urlparse(url)
|
|
assert scheme == 'http'
|
|
assert netloc == 'example.org'
|
|
assert path == '/regie/invoices/'
|
|
query = urllib.parse.parse_qs(querystring, keep_blank_values=True)
|
|
assert query['NameID'][0] == 'r2d2'
|
|
assert query['orig'][0] == 'combo'
|
|
assert check_query(querystring, 'combo') is 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
|
|
|
|
|
|
@mock.patch('combo.apps.lingo.models.requests.get')
|
|
def test_remote_regie_active_invoices_cell_failure(mock_get, app, remote_regie):
|
|
assert remote_regie.is_remote() is True
|
|
|
|
page = Page.objects.create(title='xxx', slug='test_basket_cell', template_name='standard')
|
|
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
|
|
|
|
mock_json = mock.Mock()
|
|
mock_json.json.return_value = {'err': 1}
|
|
mock_get.return_value = mock_json
|
|
|
|
content = cell.render(context)
|
|
assert 'Regie "Remote" is unavailable, please retry later.' in content
|
|
|
|
mock_json.json.return_value = {'err': 0}
|
|
content = cell.render(context)
|
|
assert 'Regie "Remote" is unavailable, please retry later.' not in content
|
|
|
|
mock_json.json.return_value = {'err': 0, 'data': None}
|
|
content = cell.render(context)
|
|
assert 'Regie "Remote" is unavailable, please retry later.' not in content
|
|
|
|
mock_json.json.return_value = {'err': 0, 'data': 'foo bar'}
|
|
content = cell.render(context)
|
|
assert 'Regie "Remote" is unavailable, please retry later.' in content
|
|
|
|
mock_get.side_effect = ConnectionError('where is my hostname?')
|
|
content = cell.render(context)
|
|
assert 'Regie "Remote" is unavailable, please retry later.' in content
|
|
|
|
mock_resp = Response()
|
|
mock_resp.status_code = 404
|
|
mock_get.return_value = mock_resp
|
|
content = cell.render(context)
|
|
assert 'No items yet' in content
|
|
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() is 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 'F-2016-Two' in content
|
|
assert '543.21' in content
|
|
assert 'class="invoice-payment-limit-date"' in content
|
|
|
|
# invoice without limit date
|
|
invoices = copy.deepcopy(INVOICES)
|
|
invoices[0]['pay_limit_date'] = ''
|
|
invoices[1]['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, dummy, querystring, dummy = urllib.parse.urlparse(url)
|
|
assert scheme == 'http'
|
|
assert netloc == 'example.org'
|
|
assert path == '/regie/invoices/history/'
|
|
query = urllib.parse.parse_qs(querystring, keep_blank_values=True)
|
|
assert query['NameID'][0] == 'r2d2'
|
|
assert query['orig'][0] == 'combo'
|
|
assert check_query(querystring, 'combo') is 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() == ''
|
|
|
|
|
|
@mock.patch('combo.apps.lingo.models.requests.get')
|
|
def test_remote_regie_past_invoices_cell_failure(mock_get, app, remote_regie):
|
|
assert remote_regie.is_remote() is True
|
|
|
|
page = Page.objects.create(title='xxx', slug='test_basket_cell', template_name='standard')
|
|
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
|
|
|
|
mock_json = mock.Mock()
|
|
mock_json.json.return_value = {'err': 1}
|
|
mock_get.return_value = mock_json
|
|
|
|
content = cell.render(context)
|
|
assert 'Regie "Remote" is unavailable, please retry later.' in content
|
|
|
|
mock_json.json.return_value = {'err': 0}
|
|
content = cell.render(context)
|
|
assert 'Regie "Remote" is unavailable, please retry later.' not in content
|
|
|
|
mock_json.json.return_value = {'err': 0, 'data': None}
|
|
content = cell.render(context)
|
|
assert 'Regie "Remote" is unavailable, please retry later.' not in content
|
|
|
|
mock_json.json.return_value = {'err': 0, 'data': 'foo bar'}
|
|
content = cell.render(context)
|
|
assert 'Regie "Remote" is unavailable, please retry later.' in content
|
|
|
|
mock_get.side_effect = ConnectionError('where is my hostname?')
|
|
content = cell.render(context)
|
|
assert 'Regie "Remote" is unavailable, please retry later.' in content
|
|
|
|
mock_resp = Response()
|
|
mock_resp.status_code = 404
|
|
mock_get.return_value = mock_resp
|
|
content = cell.render(context)
|
|
assert 'No items yet' in content
|
|
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() is 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(status_code=200)
|
|
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(status_code=200)
|
|
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
|
|
assert '"buttons"' in resp
|
|
|
|
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'
|
|
|
|
remote_regie.payment_min_amount = Decimal(200)
|
|
remote_regie.save()
|
|
resp = form.submit()
|
|
assert resp.status_code == 302
|
|
resp = resp.follow()
|
|
assert 'Minimal payment amount is 200' in resp.text
|
|
|
|
remote_regie.payment_min_amount = Decimal(2.0)
|
|
remote_regie.save()
|
|
resp = form.submit()
|
|
|
|
assert resp.status_code == 302
|
|
location = resp.location
|
|
assert 'dummy-payment' in location
|
|
parsed = urllib.parse.urlparse(location)
|
|
# get return_url and transaction id from location
|
|
qs = urllib.parse.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 urllib.parse.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 urllib.parse.urlparse(resp.url).path.startswith('/lingo/payment-status')
|
|
resp = resp.follow()
|
|
assert urllib.parse.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
|
|
|
|
# check invoice cannot be paid a second time
|
|
resp = app.get('/lingo/item/%s/%s/' % (remote_regie.id, encrypt_id))
|
|
assert '"buttons"' not in resp
|
|
|
|
resp = form.submit()
|
|
assert resp.location == '/'
|
|
assert 'Some items are already paid' in app.session['_messages']
|
|
|
|
|
|
@mock.patch('combo.apps.lingo.models.requests.get')
|
|
def test_remote_item_failure(mock_get, app, remote_regie):
|
|
assert remote_regie.is_remote() is True
|
|
|
|
mock_json = mock.Mock()
|
|
mock_json.json.return_value = {'err': 1}
|
|
mock_get.return_value = mock_json
|
|
|
|
encrypt_id = aes_hex_encrypt(settings.SECRET_KEY, force_bytes('F201601'))
|
|
resp = app.get('/lingo/item/%s/%s/' % (remote_regie.id, encrypt_id))
|
|
assert '<h2>Technical error: impossible to retrieve invoices.</h2>' in resp.text
|
|
assert '<form></form>' in resp.text
|
|
|
|
mock_json.json.return_value = {'err': 0}
|
|
resp = app.get('/lingo/item/%s/%s/' % (remote_regie.id, encrypt_id))
|
|
assert '<h2>No item was found.</h2>' in resp.text
|
|
|
|
mock_get.side_effect = ConnectionError('where is my hostname?')
|
|
resp = app.get('/lingo/item/%s/%s/' % (remote_regie.id, encrypt_id))
|
|
assert '<h2>Technical error: impossible to retrieve invoices.</h2>' in resp.text
|
|
|
|
mock_resp = Response()
|
|
mock_resp.status_code = 404
|
|
mock_get.return_value = mock_resp
|
|
resp = app.get('/lingo/item/%s/%s/' % (remote_regie.id, encrypt_id))
|
|
assert '<h2>Technical error: impossible to retrieve invoices.</h2>' in resp.text
|
|
|
|
|
|
@mock.patch('combo.apps.lingo.models.Regie.pay_invoice')
|
|
@mock.patch('combo.apps.lingo.models.requests.get')
|
|
def test_pay_remote_item_failure(mock_get, mock_pay_invoice, app, remote_regie):
|
|
encrypt_id = aes_hex_encrypt(settings.SECRET_KEY, force_bytes('F201601'))
|
|
mock_json = mock.Mock(status_code=200)
|
|
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)
|
|
url = '/lingo/item/%s/%s/' % (remote_regie.id, encrypt_id)
|
|
resp = app.get(url)
|
|
|
|
form = resp.form
|
|
|
|
mock_json = mock.Mock()
|
|
mock_json.json.return_value = {'err': 1}
|
|
mock_get.return_value = mock_json
|
|
resp = form.submit().follow()
|
|
assert 'Technical error: impossible to retrieve invoices.' in app.session['_messages']
|
|
|
|
mock_json.json.return_value = {'err': 0}
|
|
resp = form.submit().follow()
|
|
assert 'No invoice was found.' in app.session['_messages']
|
|
|
|
mock_get.side_effect = ConnectionError('where is my hostname?')
|
|
resp = form.submit().follow()
|
|
assert 'Technical error: impossible to retrieve invoices.' in app.session['_messages']
|
|
|
|
mock_resp = Response()
|
|
mock_resp.status_code = 404
|
|
mock_get.return_value = mock_resp
|
|
resp = form.submit().follow()
|
|
assert 'Technical error: impossible to retrieve invoices.' in app.session['_messages']
|
|
|
|
|
|
@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() is 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 = urllib.parse.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() is True
|
|
mock_json = mock.Mock()
|
|
mock_json.json.return_value = {'err': 0, 'data': INVOICES[0]}
|
|
mock_get.return_value = mock_json
|
|
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, the provided amount is invalid.' in resp
|
|
|
|
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.' in resp
|
|
|
|
mock_json = mock.Mock(status_code=404)
|
|
mock_get.return_value = mock_json
|
|
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.' in resp
|
|
|
|
mock_json = mock.Mock()
|
|
mock_json.json.return_value = {'err': 0, 'data': INVOICES[0]}
|
|
mock_get.return_value = mock_json
|
|
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.' in resp
|
|
|
|
resp = app.get('/test-self-invoice/')
|
|
resp.form['invoice-number'] = 'F201601'
|
|
resp.form['invoice-amount'] = '123.45'
|
|
resp = resp.form.submit()
|
|
path = urllib.parse.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(status_code=200)
|
|
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 = urllib.parse.urlparse(location)
|
|
# get return_url and transaction id from location
|
|
qs = urllib.parse.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 urllib.parse.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 urllib.parse.urlparse(resp.url).path.startswith('/lingo/payment-status')
|
|
resp = resp.follow()
|
|
assert (
|
|
urllib.parse.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()
|
|
|
|
|
|
@pytest.mark.parametrize('can_pay_only_one_basket_item', [False, True])
|
|
@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, can_pay_only_one_basket_item, app, remote_regie
|
|
):
|
|
assert remote_regie.is_remote()
|
|
remote_regie.can_pay_only_one_basket_item = can_pay_only_one_basket_item
|
|
remote_regie.save()
|
|
|
|
page = Page(title='xxx', slug='active-remote-invoices-page', template_name='standard')
|
|
page.save()
|
|
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))
|
|
assert '"paid paid-info"' not in resp
|
|
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 = urllib.parse.urlparse(location)
|
|
# get return_url and transaction id from location
|
|
qs = urllib.parse.parse_qs(parsed.query)
|
|
if can_pay_only_one_basket_item:
|
|
assert qs['orderid'] == ['order-id-1']
|
|
assert qs['subject'] == ['invoice-one']
|
|
else:
|
|
assert 'orderid' not in qs
|
|
assert 'subject' not in qs
|
|
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 urllib.parse.urlparse(resp.url).path.startswith('/lingo/payment-status')
|
|
resp = resp.follow()
|
|
assert (
|
|
urllib.parse.urlparse(resp.html.find('a', {'id': 'next-url'})['href']).path
|
|
== '/active-remote-invoices-page/'
|
|
)
|
|
|
|
# check true payment status is visible, even if the remote regie web-service still report the invoice as unpaid
|
|
resp = app.get('/lingo/item/%s/%s/?page=%s' % (remote_regie.id, encrypt_id, page.pk))
|
|
assert not INVOICES[0]['paid']
|
|
assert '"paid paid-info"' in resp
|
|
|
|
|
|
@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 == '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'
|
|
|
|
|
|
@pytest.fixture
|
|
def remote_invoices_httmock():
|
|
invoices = []
|
|
invoice = {}
|
|
|
|
netloc = 'remote.regie.example.com'
|
|
|
|
@httmock.urlmatch(netloc=netloc, path='^/invoice/')
|
|
def invoice_mock(url, request):
|
|
return json.dumps({'err': 0, 'data': invoice})
|
|
|
|
@httmock.urlmatch(netloc=netloc, path='^/invoices/')
|
|
def invoices_mock(url, request):
|
|
return json.dumps({'err': 0, 'data': invoices})
|
|
|
|
context_manager = httmock.HTTMock(invoices_mock, invoice_mock)
|
|
context_manager.url = 'https://%s/' % netloc
|
|
context_manager.invoices = invoices
|
|
context_manager.invoice = invoice
|
|
with context_manager:
|
|
yield context_manager
|
|
|
|
|
|
class TestPolling:
|
|
@mock.patch('eopayment.payfip_ws.Payment.payment_status')
|
|
@mock.patch('eopayment.payfip_ws.Payment.request', return_value=(1, eopayment.URL, 'https://payfip/'))
|
|
def test_in_active_items_cell(
|
|
self,
|
|
payment_request,
|
|
payment_status,
|
|
app,
|
|
remote_regie,
|
|
settings,
|
|
remote_invoices_httmock,
|
|
synchronous_cells,
|
|
):
|
|
|
|
remote_invoices_httmock.invoices.extend(INVOICES)
|
|
remote_invoices_httmock.invoice.update(INVOICES[0])
|
|
remote_regie.webservice_url = remote_invoices_httmock.url
|
|
remote_regie.save()
|
|
# use payfip
|
|
remote_regie.payment_backend.service = 'payfip_ws'
|
|
remote_regie.payment_backend.save()
|
|
|
|
User.objects.create_user('admin', password='admin', email='foo@example.com')
|
|
page = Page.objects.create(title='xxx', slug='test_basket_cell', template_name='standard')
|
|
ActiveItems.objects.create(regie='remote', page=page, placeholder='content', order=0)
|
|
|
|
login(app)
|
|
|
|
assert Transaction.objects.count() == 0
|
|
|
|
resp = app.get('/test_basket_cell/')
|
|
assert 'F-2016-One' in resp
|
|
|
|
resp = resp.click('pay', index=0)
|
|
pay_resp = resp
|
|
resp = resp.form.submit('Pay')
|
|
|
|
transaction = Transaction.objects.get()
|
|
assert transaction.status == 0
|
|
|
|
payment_status.return_value = eopayment.common.PaymentResponse(
|
|
signed=True,
|
|
result=eopayment.WAITING,
|
|
order_id=transaction.order_id,
|
|
)
|
|
|
|
assert payment_status.call_count == 0
|
|
resp = app.get('/test_basket_cell/')
|
|
assert 'F-2016-One' in resp
|
|
assert payment_status.call_count == 1
|
|
transaction.refresh_from_db()
|
|
assert transaction.status == eopayment.WAITING
|
|
|
|
resp = resp.click('pay', index=0)
|
|
assert 'Waiting for payment' in resp
|
|
assert 'button' not in resp
|
|
|
|
resp = pay_resp.form.submit('Pay').follow()
|
|
assert 'Some items are already paid' in resp
|
|
|
|
payment_status.return_value = eopayment.common.PaymentResponse(
|
|
signed=True,
|
|
result=eopayment.PAID,
|
|
order_id=transaction.order_id,
|
|
)
|
|
|
|
resp = app.get('/test_basket_cell/')
|
|
assert 'F-2016-One' not in resp
|
|
transaction.refresh_from_db()
|
|
assert transaction.status == eopayment.PAID
|