Produits Entr'ouvert » Passerelle

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

from __future__ import unicode_literals

from django.db import models, migrations

import passerelle.pbx.models

class Migration(migrations.Migration):

    dependencies = [

        ('base', '0002_auto_20151009_0326'),

    ]

    operations = [

        migrations.CreateModel(

            name='TlmCom',

            fields=[

                ('id', models.AutoField(verbose_name='ID', serialize=False, auto_created=True, primary_key=True)),

                ('title', models.CharField(verbose_name='Title', max_length=50)),

                ('slug', models.SlugField(verbose_name='Identifier', unique=True)),

                ('description', models.TextField(verbose_name='Description')),

                ('welco_url', models.URLField(default=passerelle.pbx.models.get_default_welco_url, max_length=120, verbose_name='welco URL')),

                ('users', models.ManyToManyField(to='base.ApiUser', related_name='_tlmcom_users_+', related_query_name='+', blank=True)),

            ],

            options={

                'verbose_name': 'TLM COM gateway',

                'verbose_name_plural': 'TLM COM gateways',

            },

            bases=(models.Model,),

        ),

    ]

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

from __future__ import unicode_literals

from django.db import models, migrations

class Migration(migrations.Migration):

    dependencies = [

        ('tlmcom', '0001_initial'),

    ]

    operations = [

        migrations.AlterModelOptions(

            name='tlmcom',

            options={'verbose_name': 'TLM COM PBX', 'verbose_name_plural': 'TLM COM PBX'},

        ),

    ]

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

from __future__ import unicode_literals

from django.db import models, migrations

class Migration(migrations.Migration):

    dependencies = [

        ('tlmcom', '0002_auto_20151222_0732'),

    ]

    operations = [

        migrations.AddField(

            model_name='tlmcom',

            name='log_level',

            field=models.CharField(default=b'NOTSET', max_length=10, verbose_name='Log Level', choices=[(b'NOTSET', b'NOTSET'), (b'DEBUG', b'DEBUG'), (b'INFO', b'INFO'), (b'WARNING', b'WARNING'), (b'ERROR', b'ERROR'), (b'CRITICAL', b'CRITICAL')]),

            preserve_default=True,

        ),

    ]

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

from __future__ import unicode_literals

from django.db import migrations, models

class Migration(migrations.Migration):

    dependencies = [

        ('tlmcom', '0003_tlmcom_log_level'),

    ]

    operations = [

        migrations.AlterField(

            model_name='tlmcom',

            name='slug',

            field=models.SlugField(verbose_name='Identifier', unique=True),

        ),

    ]

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

# Generated by Django 1.11.12 on 2018-11-18 14:07

from __future__ import unicode_literals

from django.db import migrations

class Migration(migrations.Migration):

    dependencies = [

        ('tlmcom', '0004_auto_20170920_0951'),

    ]

    operations = [

        migrations.RemoveField(

            model_name='tlmcom',

            name='log_level',

        ),

    ]

from passerelle.pbx.models import PBXMixin

from django.utils.translation import ugettext_lazy as _

class TlmCom(PBXMixin):

    class Meta:

        verbose_name = _('TLM COM PBX')

        verbose_name_plural = _('TLM COM PBX')

{% extends "passerelle/manage/service_view.html" %}

{% load i18n passerelle %}

{% block description %}

{% endblock %}

{% block endpoints %}

  <ul>

    <li>{% trans 'Call started' %} <a href="{% url 'tlmcom-call-start' slug=object.slug %}?caller=%2B3392323232&amp;callee=102"

            >{{ site_base_uri }}{% url 'tlmcom-call-start' slug=object.slug %}?caller=+3392323232&amp;callee=102</a></li>

    <li>{% trans 'Call stopped:' %} <a href="{% url 'tlmcom-call-stop' slug=object.slug %}?caller=%2B3392323232&amp;callee=102"

            >{{ site_base_uri }}{% url 'tlmcom-call-stop' slug=object.slug %}?caller=+3392323232&amp;callee=102</a></li>

  </ul>

{% endblock %}

{% block security %}

  <p>

    {% trans 'Access is limited to the following API users:' %}

  </p>

  {% access_rights_table resource=object permission='can_access' %}

{% endblock %}

# passerelle - uniform access to multiple data sources and services

# Copyright (C) 2015  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 django.conf.urls import url

from .views import TlmComDetailView, CallStart, CallStop

urlpatterns = [

        url(r'^(?P<slug>[\w,-]+)/$',

            TlmComDetailView.as_view(), name='tlmcom-view'),

        url(r'^(?P<slug>[\w,-]+)/call-start/$',

            CallStart.as_view(), name='tlmcom-call-start'),

        url(r'^(?P<slug>[\w,-]+)/call-stop/$',

            CallStop.as_view(), name='tlmcom-call-stop'),

]

# passerelle - uniform access to multiple data sources and services

# Copyright (C) 2015  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 django.views.generic import DetailView

from django.views.generic.base import View

from django.views.generic.detail import SingleObjectMixin

from django.http import HttpResponse, HttpResponseBadRequest

from .models import TlmCom

class TlmComDetailView(DetailView):

    model = TlmCom

    template_name = 'passerelle/contrib/tlmcom/detail.html'

class CallStart(SingleObjectMixin, View):

    event = 'start'

    model = TlmCom

    def get(self, request, *args, **kwargs):

        try:

            caller = request.GET['caller']

            callee = request.GET['callee']

        except KeyError:

            return HttpResponseBadRequest('caller and callee are mandatory')

        username = request.GET.get('id')

        if username:

            # "id" is a tlmcom dirty hack: the real username is just the part after the dot

            username = username.split('.', 1).pop()

        data = dict(request.GET.items())

        for key in ['caller', 'callee', 'id']:

            data.pop(key, None)

        self.get_object().notify_welco(self.event, caller, username or callee, data)

        return HttpResponse('ok')

class CallStop(CallStart):

    event = 'stop'

import json

import logging

from django.db import models

from django.conf import settings

from django.utils.encoding import force_text

from django.utils.six import string_types

from django.utils.translation import ugettext_lazy as _

import requests

from passerelle.base.models import BaseResource

def get_default_welco_url():

    '''Get default welco URL from a setting than can be set in a Publik

       setting.

    '''

    return getattr(settings, 'PASSERELLE_DEFAULT_WELCO_URL', '')

class PBXMixin(BaseResource):

    welco_url = models.URLField(

        verbose_name=_('welco URL'),

        default=get_default_welco_url,

        max_length=120)

    category = _('Telephony')

    def notify_welco(self, event, caller, callee, data=None):

        assert event in ('start', 'stop')

        assert isinstance(caller, string_types)

        assert isinstance(callee, string_types)

        assert not data or isinstance(data, dict)

        caller = force_text(caller)

        callee = force_text(callee)

        event = force_text(event)

        payload = {

            'event': event,

            'caller': caller,

            'callee': callee,

        }

        if data:

            payload['data'] = data

        requests.post(self.welco_url, data=json.dumps(payload),

                      headers={'content-type': 'application/json'})

        self.logger.info(u'notified welco of event %s from %s to %s',

                    event, caller, callee)

    class Meta:

        abstract = True

    'passerelle.contrib.tlmcom',

import json

import pytest

import mock

from django.core.urlresolvers import reverse

from passerelle.contrib.tlmcom.models import TlmCom

pytestmark = pytest.mark.django_db

@mock.patch('requests.post')

def test_call_start(requests_post, client):

    URL = 'http://welco.example.net/'

    TlmCom.objects.create(slug='tlmcom', description='tlmcom', welco_url=URL)

    url = reverse('tlmcom-call-start', kwargs={'slug': 'tlmcom'}) \

        + '?caller=0033699999999&callee=102&user=zozo'

    response = client.get(url)

    assert response.status_code == 200

    assert response.content == b'ok'

    assert requests_post.call_count == 1

    assert requests_post.call_args == mock.call(URL, data=mock.ANY, headers={'content-type': 'application/json'})

    payload = json.loads(requests_post.call_args[1]['data'])

    assert isinstance(payload, dict)

    assert set(payload.keys()) == set(['data', 'caller', 'callee', 'event'])

    assert payload['event'] == 'start'

    assert payload['data'] == {'user': 'zozo'}

    assert payload['callee'] == '102'

    assert payload['caller'] == '0033699999999'

@mock.patch('requests.post')

def test_call_stop(requests_post, client):

    URL = 'http://welco.example.net/'

    TlmCom.objects.create(slug='tlmcom', description='tlmcom', welco_url=URL)

    url = reverse('tlmcom-call-stop', kwargs={'slug': 'tlmcom'}) \

        + '?caller=0033699999999&callee=102&user=zozo'

    response = client.get(url)

    assert response.status_code == 200

    assert response.content == b'ok'

    assert requests_post.call_count == 1

    assert requests_post.call_args == mock.call(URL, data=mock.ANY, headers={'content-type': 'application/json'})

    payload = json.loads(requests_post.call_args[1]['data'])

    assert isinstance(payload, dict)

    assert set(payload.keys()) == set(['data', 'caller', 'callee', 'event'])

    assert payload['event'] == 'stop'

    assert payload['data'] == {'user': 'zozo'}

    assert payload['callee'] == '102'

    assert payload['caller'] == '0033699999999'

@mock.patch('requests.post')

def test_call_start_with_id(requests_post, client):

    URL = 'http://welco.example.net/'

    TlmCom.objects.create(slug='tlmcom', description='tlmcom', welco_url=URL)

    url = reverse('tlmcom-call-start', kwargs={'slug': 'tlmcom'}) \

        + '?caller=0033699999999&callee=102&id=.zozo&idTLMCOM=XXXXXX01'

    response = client.get(url)

    assert response.status_code == 200

    assert response.content == b'ok'

    assert requests_post.call_count == 1

    assert requests_post.call_args == mock.call(URL, data=mock.ANY, headers={'content-type': 'application/json'})

    payload = json.loads(requests_post.call_args[1]['data'])

    assert isinstance(payload, dict)

    assert set(payload.keys()) == set(['caller', 'callee', 'event', 'data'])

    assert payload['event'] == 'start'

    assert payload['callee'] == 'zozo'

    assert payload['caller'] == '0033699999999'

    assert payload['data'] == {'idTLMCOM': 'XXXXXX01'}

@mock.patch('requests.post')

def test_call_stop_with_id(requests_post, client):

    URL = 'http://welco.example.net/'

    TlmCom.objects.create(slug='tlmcom', description='tlmcom', welco_url=URL)

    url = reverse('tlmcom-call-stop', kwargs={'slug': 'tlmcom'}) \

        + '?caller=0033699999999&callee=102&id=.zozo&idTLMCOM=XXXXXX01'

    response = client.get(url)

    assert response.status_code == 200

    assert response.content == b'ok'

    assert requests_post.call_count == 1

    assert requests_post.call_args == mock.call(URL, data=mock.ANY, headers={'content-type': 'application/json'})

    payload = json.loads(requests_post.call_args[1]['data'])

    assert isinstance(payload, dict)

    assert set(payload.keys()) == set(['caller', 'callee', 'event', 'data'])

    assert payload['event'] == 'stop'

    assert payload['callee'] == 'zozo'

    assert payload['caller'] == '0033699999999'

    assert payload['data'] == {'idTLMCOM': 'XXXXXX01'}

-