Produits Entr'ouvert » Combo

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

# combo-plugin-gnm - Combo WebPush App

# 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 os

import base64

from datetime import datetime

from cryptography.hazmat.primitives.asymmetric import ec

from cryptography.hazmat.backends import default_backend

import pytest

from mock import patch

import django

from django.contrib.auth.models import User

from django.core.management import call_command

from django.utils.timezone import make_aware

if django.VERSION >= (1, 11):

    from combo.apps.notifications.models import Notification

    from combo.apps.pwa.models import WebPushRecord

    from push_notifications.models import WebPushDevice

pytestmark = pytest.mark.django_db

@pytest.fixture

def user1():

    user1 = User.objects.create_user('user1', email='user1@example.net', password='user1')

    user1.save()

    return user1

def _get_pubkey_str(priv_key):

    '''From pywebpush tests'''

    return base64.urlsafe_b64encode(

        priv_key.public_key().public_numbers().encode_point()

    ).strip(b'=')

def _gen_subscription_info(recv_key=None,

                           endpoint="https://example.com/"):

    '''From pywebpush tests'''

    if not recv_key:

        recv_key = ec.generate_private_key(ec.SECP256R1, default_backend())

    return {

        "keys": {

            'auth': base64.urlsafe_b64encode(os.urandom(16)).strip(b'='),

            'p256dh': _get_pubkey_str(recv_key),

        }

    }

if django.VERSION >= (1, 11):

    @patch('pywebpush.WebPusher.send')

    def test_command_send_webpush(mock_send, user1):

        mock_send.return_value.status_code = 201

        recv_key = ec.generate_private_key(ec.SECP256R1, default_backend())

        subscription_keys_dict = _gen_subscription_info(recv_key)

        webpush_device = WebPushDevice.objects.create(

            name='test-webpush-device',

            browser='CHROME',

            active=True,

            auth=subscription_keys_dict['keys']['auth'],

            p256dh=subscription_keys_dict['keys']['p256dh'],

            user=user1,

            registration_id='test',

        )

        notification = Notification.notify(user1, u'tèst headér', body=u'test uniçode bodéï')

        call_command('send_webpush', '--verbosity=1')

        assert WebPushRecord.objects.filter(status=WebPushRecord.ERR_STATUS).count() == 0

        assert WebPushRecord.objects.filter(status=WebPushRecord.DEFAULT_STATUS).count() == 0

        ok_push = WebPushRecord.objects.filter(status=WebPushRecord.OK_STATUS)

        assert ok_push.count() == 1

        ok_push = ok_push.first()

        assert ok_push.subscription == webpush_device

        assert ok_push.notification == notification

        assert ok_push.creation_date < make_aware(datetime.now())