Produits Entr'ouvert » Combo

# -*- coding: utf-8 -*-

#

# lingo - basket and payment system

# Copyright (C) 2018  Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# You should have received a copy of the GNU Affero General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import logging

from django.core.management.base import BaseCommand

from combo.apps.lingo.models import Regie

class Command(BaseCommand):

    def handle(self, *args, **kwargs):

        logger = logging.getLogger(__name__)

        for regie in Regie.objects.exclude(webservice_url=''):

            try:

                regie.notify_new_remote_invoices()

            except Exception, e:

                logger.exception('error while notifying new remote invoices: %s', e)

from django.utils import timezone, dateparse

from django.contrib.auth.models import User

from combo.apps.notifications.models import Notification

    def notify_new_remote_invoices(self):

        if not self.is_remote():

            return

        logger = logging.getLogger(__name__)

        url = self.webservice_url + '/users/with-pending-invoices/'

        response = requests.get(url, remote_service='auto', cache_duration=0,

                                log_errors=False)

        if not response.ok:

            return

        data = response.json()['data']

        if not data:

            return

        for uuid, items in data.iteritems():

            try:

                user = User.objects.get(username=uuid)

            except User.DoesNotExist:

                logger.warning('Invoices available for unknown user: %s', uuid)

                continue

            for invoice in items['invoices']:

                invoice_id = 'invoice-%s-%s' % (self.slug, invoice['id'])

                invoice_creation_date = timezone.make_aware(dateparse.parse_datetime(invoice['created']))

                invoice_pay_limit_date = timezone.make_aware(dateparse.parse_datetime(invoice['pay_limit_date']))

                notification_end_timestamp = invoice_creation_date + timezone.timedelta(days=settings.LINGO_NEW_INVOICES_NOTIFICATION_DELAY)

                if invoice_pay_limit_date < notification_end_timestamp:

                    notification_end_timestamp = invoice_pay_limit_date

                notification_id, created = Notification.notify(user, _('Invoice %s to pay') % invoice['label'], id=invoice_id,

                                                               end_timestamp=notification_end_timestamp)

                if not created:

                    notification = Notification.objects.filter_by_id(invoice_id).filter(user=user).last()

                    if notification.end_timestamp < timezone.now():

                        remind_id = 'remind-%s' % invoice_id

                        Notification.notify(user, _('Reminder: invoice %s to pay') % invoice['label'],

                                            id=remind_id, end_timestamp=notification_end_timestamp)

# default delay for invoice payment notifications in days

LINGO_NEW_INVOICES_NOTIFICATION_DELAY = 10

import mock

from decimal import Decimal

from django.test import override_settings

from combo.apps.lingo.models import Regie

@pytest.fixture

def regie():

    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.service = 'dummy'

        regie.save()

    return regie

@mock.patch('combo.apps.lingo.models.requests.get')

def test_notify_remote_items(mock_get, app, user, user2, regie, caplog):

    datetime_format = '%Y-%m-%dT%H:%M:%S'

    invoice_now = now()

    creation_date = (invoice_now - timedelta(days=1)).strftime(datetime_format)

    pay_limit_date = (invoice_now + timedelta(days=30)).strftime(datetime_format)

    FAKE_PENDING_INVOICES = {

        "data" :  {"admin": {"invoices": [{'id': '01', 'label': '010101',

                        'created': creation_date, 'pay_limit_date': pay_limit_date}]},

                   'admin2': {'invoices': [{'id': '02', 'label': '020202',

                        'created': creation_date, 'pay_limit_date': pay_limit_date}]},

                   'foo': {'invoices': [{'id': 'O3', 'label': '030303',

                        'created': creation_date, 'pay_limite_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

    regie.notify_new_remote_invoices()

    assert mock_get.call_count == 0

    regie.webservice_url = 'http://example.org/regie' # is_remote

    regie.save()

    with override_settings(LINGO_NEW_INVOICES_NOTIFICATION_DELAY=0):

        regie.notify_new_remote_invoices()

        assert Notification.objects.count() == 2

        # create remind notifications

        regie.notify_new_remote_invoices()

        assert Notification.objects.count() == 4

    assert len(caplog.records) == 2

@mock.patch('combo.apps.lingo.models.requests.get')

def test_notify_remote_items_expiring_shortly(mock_get, app, user, user2, regie):

    datetime_format = '%Y-%m-%dT%H:%M:%S'

    invoice_now = now()

    creation_date = (invoice_now - timedelta(1)).strftime(datetime_format)

    pay_limit_date = (invoice_now + timedelta(days=5)).strftime(datetime_format)

    SHORT_PENDING_INVOICES = {

        "data" :  {"admin": {"invoices": [{'id': '03', 'label': '030303',

                        'created': creation_date,

                        'pay_limit_date': pay_limit_date}]},

                   'admin2': {'invoices': [{'id': '04', 'label': '040404',

                        'created': creation_date,

                        'pay_limit_date': pay_limit_date}]},

        }

    }

    mock_response = mock.Mock(status_code=200, content=json.dumps(SHORT_PENDING_INVOICES))

    mock_response.json.return_value = SHORT_PENDING_INVOICES

    mock_get.return_value = mock_response

    regie.webservice_url = 'http://example.org/regie'

    regie.save()

    regie.notify_new_remote_invoices()

    assert Notification.objects.count() == 2

    for notification in Notification.objects.all():

        assert notification.end_timestamp.strftime(datetime_format) == pay_limit_date