Produits Entr'ouvert » Corbo

# corbo - Announces Manager

# Copyright (C) 2017 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/>.

from hobo.agent.common.management.commands import hobo_notify

from corbo.models import Subscription

class Command(hobo_notify.Command):

    def process_notification(self, tenant, notification):

        super(Command, self).process_notification(tenant, notification)

        object_type = notification['objects']['@type']

        if object_type != 'user':

            return

        users = notification['objects']['data']

        action = notification['@type']

        if notification.get('full'):

            uuids = [user['uuid'] for user in users]

            Subscription.objects.exclude(uuid__in=uuids).delete()

        for user in users:

            for subscription in Subscription.objects.filter(uuid=user['uuid']):

                if action == 'provision':

                    if subscription.identifier.startswith('mailto:'):

                        if user.get('email'):

                            subscription.identifier = 'mailto:%s' % user['email']

                        else:

                            subscription.delete()

                            continue

                    elif subscription.identifier.startswith('sms:'):

                        if user.get('mobile'):

                            subscription.identifier = 'sms:%s' % user['mobile']

                        else:

                            subscription.delete()

                            continue

                    subscription.save()

                elif action == 'deprovision':

                    subscription.delete()

INSTALLED_APPS = ('corbo.hobo_agent', ) + INSTALLED_APPS

import tempfile

import os

import json

import shutil

from django.utils.text import slugify

from corbo.models import Category, Announce, Broadcast, Subscription

CATEGORIES = ('Alerts', 'News')

SUBSCRIBERS = [{'uuid': 'uuid1', 'email': 'foo@example.net', 'mobile': '0102030405'},

               {'uuid': 'uuid2', 'email': 'bar@example.net', 'mobile': '0607080900'},

               {'uuid': '', 'email': 'john@example.net', 'mobile': '0304050607'}]

@pytest.fixture

def categories():

    categories = []

    for category in CATEGORIES:

        c, created = Category.objects.get_or_create(name=category, slug=slugify(category))

        categories.append(c)

    return categories

@pytest.fixture

def subscriptions(categories):

    subscriptions = []

    for category in categories:

        for subscriber in SUBSCRIBERS:

            uuid = subscriber['uuid']

            subscriptions.append(Subscription.objects.create(category=category, uuid=uuid,

                                    identifier='mailto:%(email)s' % subscriber))

            subscriptions.append(Subscription.objects.create(category=category, uuid=uuid,

                                    identifier='sms:%(mobile)s' % subscriber))

    return subscriptions

@pytest.fixture

def tenant_base(request, settings):

    base = tempfile.mkdtemp('corbo-tenant-base')

    settings.TENANT_BASE = base

    def fin():

        shutil.rmtree(base)

    request.addfinalizer(fin)

    return base

@pytest.fixture(scope='function')

def tenant(transactional_db, request, tenant_base):

    from hobo.multitenant.models import Tenant

    base = tenant_base

    @pytest.mark.django_db

    def make_tenant(name):

        tenant_dir = os.path.join(base, name)

        os.mkdir(tenant_dir)

        with open(os.path.join(tenant_dir, 'hobo.json'), 'w') as fd:

            json.dump({

                'variables': {

                    'hobo_test_variable': True,

                    'other_variable': 'foo',

                },

                'services': [

                    {'slug': 'test',

                     'service-id': 'corbo',

                     'title': 'Test',

                     'this': True,

                     'secret_key': '12345',

                     'base_url': 'http://%s' % name,

                     'saml-sp-metadata-url': 'http://%s/metadata/' % name,

                     'variables': {

                         'other_variable': 'bar',

                     }

                    },

                    {'slug': 'passerelle',

                     'title': 'Webserivces',

                     'service-id': 'passerelle',

                     'secret_key': 'abcdef',

                     'base_url': 'http://passerelle.example.net',

                     'saml-sp-metadata-url': 'http://passerelle.example.net/metadata/' },

                    ]}, fd)

        return Tenant(domain_url=name,

                      schema_name=name.replace('-', '_').replace('.', '_'))

    return make_tenant('corbo.example.net')

# Add corbo hobo agent

INSTALLED_APPS = ('corbo.hobo_agent', ) + INSTALLED_APPS

import pytest

from copy import deepcopy

import json

from corbo.models import Subscription

from corbo.hobo_agent.management.commands.hobo_notify import Command

from django.core.management import call_command

NOTIFICATION = {'@type': 'provision',

                'objects': {'@type': 'user',

                            'data': [

                                {'uuid': 'uuid1', 'email': 'foo1@example.net', 'mobile': '0504030201'},

                                {'uuid': 'uuid2', 'email': 'bar1@example.net', 'mobile': '0009080706'},

                            ]},

                'audience': [],

                'full': False

}

def test_notify_provision(tenant, subscriptions):

    command = Command()

    command.process_notification(tenant, NOTIFICATION)

    assert Subscription.objects.count() == len(subscriptions)

    for user in NOTIFICATION['objects']['data']:

        assert Subscription.objects.filter(uuid=user['uuid'], identifier='mailto:%(email)s' % user).exists()

        assert Subscription.objects.filter(uuid=user['uuid'], identifier='sms:%(mobile)s' % user).exists()

def test_notify_other_provision(tenant, subscriptions):

    role_notification = deepcopy(NOTIFICATION)

    role_notification['objects']['@type'] = 'role'

    command = Command()

    command.process_notification(tenant, role_notification)

    assert Subscription.objects.count() == len(subscriptions)

    for subscription in subscriptions:

        assert Subscription.objects.filter(uuid=subscription.uuid, identifier=subscription.identifier)

def test_notify_deprovision(tenant, subscriptions):

    deprovision_notification = deepcopy(NOTIFICATION)

    deprovision_notification['@type'] = 'deprovision'

    command = Command()

    command.process_notification(tenant, deprovision_notification)

    for user in deprovision_notification['objects']['data']:

        assert not Subscription.objects.filter(uuid=user['uuid']).exists()

def test_notify_full(tenant, subscriptions):

    full_notification = deepcopy(NOTIFICATION)

    full_notification['full'] = True

    command = Command()

    command.process_notification(tenant, full_notification)

    assert not Subscription.objects.filter(uuid__isnull=True).exists()

    assert not Subscription.objects.filter(uuid='').exists()

    assert Subscription.objects.count() == 8

def test_notify_emails_only(tenant, subscriptions):

    email_notification = deepcopy(NOTIFICATION)

    for user in email_notification['objects']['data']:

        user['mobile'] = None

        user['email'] = 'new-%(email)s' % user

    command = Command()

    command.process_notification(tenant, email_notification)

    for user in email_notification['objects']['data']:

        assert not Subscription.objects.filter(uuid=user['uuid'], identifier__startswith='sms:').exists()

def test_notify_mobiles_only(tenant, subscriptions):

    mobile_notification = deepcopy(NOTIFICATION)

    for user in mobile_notification['objects']['data']:

        user['email'] = None

        user['mobile'] = '0%(mobile)s' % user

    command = Command()

    command.process_notification(tenant, mobile_notification)

    for user in mobile_notification['objects']['data']:

        assert not Subscription.objects.filter(uuid=user['uuid'], identifier__startswith='mailto:').exists()

  CORBO_SETTINGS_FILE=tests/settings.py

  http://git.entrouvert.org/hobo.git/snapshot/hobo-master.tar.gz