2015-10-28 17:41:14 +01:00
|
|
|
# welco - multichannel request processing
|
2018-05-19 16:38:57 +02:00
|
|
|
# Copyright (C) 2018 Entr'ouvert
|
2015-10-28 17:41:14 +01:00
|
|
|
#
|
|
|
|
# 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 json
|
2017-05-31 23:55:20 +02:00
|
|
|
import re
|
2015-10-28 17:41:14 +01:00
|
|
|
|
|
|
|
import pytest
|
|
|
|
|
|
|
|
from django.core.urlresolvers import reverse
|
2017-05-31 20:37:27 +02:00
|
|
|
from django.test import override_settings
|
2020-01-19 19:11:28 +01:00
|
|
|
from django.utils import six
|
2017-05-31 23:45:52 +02:00
|
|
|
from django.utils.timezone import now, timedelta
|
2015-10-28 17:41:14 +01:00
|
|
|
|
|
|
|
from welco.sources.phone import models
|
|
|
|
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
|
|
|
|
|
|
def test_call_start_stop(client):
|
|
|
|
assert models.PhoneCall.objects.count() == 0
|
|
|
|
payload = {
|
|
|
|
'event': 'start',
|
|
|
|
'caller': '0033699999999',
|
|
|
|
'callee': '102',
|
|
|
|
'data': {
|
|
|
|
'user': 'boby.lapointe',
|
|
|
|
}
|
|
|
|
}
|
|
|
|
response = client.post(reverse('phone-call-event'), json.dumps(payload),
|
|
|
|
content_type='application/json')
|
|
|
|
assert response.status_code == 200
|
|
|
|
assert response['content-type'] == 'application/json'
|
|
|
|
assert json.loads(response.content) == {'err': 0}
|
|
|
|
assert models.PhoneCall.objects.count() == 1
|
|
|
|
assert models.PhoneCall.objects.filter(
|
|
|
|
caller='0033699999999',
|
|
|
|
callee='102',
|
|
|
|
data=json.dumps(payload['data']), stop__isnull=True).count() == 1
|
|
|
|
# new start event
|
|
|
|
response = client.post(reverse('phone-call-event'), json.dumps(payload),
|
|
|
|
content_type='application/json')
|
|
|
|
assert response.status_code == 200
|
|
|
|
assert response['content-type'] == 'application/json'
|
|
|
|
assert json.loads(response.content) == {'err': 0}
|
|
|
|
assert models.PhoneCall.objects.count() == 2
|
|
|
|
assert models.PhoneCall.objects.filter(
|
|
|
|
caller='0033699999999',
|
|
|
|
callee='102',
|
|
|
|
data=json.dumps(payload['data']), stop__isnull=True).count() == 1
|
|
|
|
# first call has been closed
|
|
|
|
assert models.PhoneCall.objects.filter(
|
|
|
|
caller='0033699999999',
|
|
|
|
callee='102',
|
|
|
|
data=json.dumps(payload['data']), stop__isnull=False).count() == 1
|
|
|
|
payload['event'] = 'stop'
|
|
|
|
response = client.post(reverse('phone-call-event'), json.dumps(payload),
|
|
|
|
content_type='application/json')
|
|
|
|
assert response.status_code == 200
|
|
|
|
assert response['content-type'] == 'application/json'
|
|
|
|
assert json.loads(response.content) == {'err': 0}
|
|
|
|
assert models.PhoneCall.objects.count() == 2
|
|
|
|
assert models.PhoneCall.objects.filter(
|
|
|
|
caller='0033699999999',
|
|
|
|
callee='102',
|
|
|
|
data=json.dumps(payload['data']), stop__isnull=False).count() == 2
|
|
|
|
# stop is idempotent
|
|
|
|
response = client.post(reverse('phone-call-event'), json.dumps(payload),
|
|
|
|
content_type='application/json')
|
|
|
|
assert response.status_code == 200
|
|
|
|
assert response['content-type'] == 'application/json'
|
|
|
|
assert json.loads(response.content) == {'err': 0}
|
|
|
|
assert models.PhoneCall.objects.count() == 2
|
|
|
|
assert models.PhoneCall.objects.filter(
|
|
|
|
caller='0033699999999',
|
|
|
|
callee='102',
|
|
|
|
data=json.dumps(payload['data']), stop__isnull=False).count() == 2
|
|
|
|
|
|
|
|
|
2017-05-31 20:37:27 +02:00
|
|
|
def test_one_call_per_callee(user, client):
|
|
|
|
assert models.PhoneCall.objects.count() == 0
|
|
|
|
payload = {'event': 'start', 'caller': '0033699999999', 'callee': '102'}
|
|
|
|
response = client.post(reverse('phone-call-event'), json.dumps(payload),
|
|
|
|
content_type='application/json')
|
|
|
|
assert response.status_code == 200
|
|
|
|
assert models.PhoneCall.objects.filter(callee='102', stop__isnull=True).count() == 1 # active
|
|
|
|
assert models.PhoneCall.objects.filter(callee='102', stop__isnull=False).count() == 0 # inactive
|
|
|
|
|
|
|
|
# new caller, same callee: stops the last call, start a new one
|
|
|
|
payload['caller'] = '00337123456789'
|
|
|
|
response = client.post(reverse('phone-call-event'), json.dumps(payload),
|
|
|
|
content_type='application/json')
|
|
|
|
assert response.status_code == 200
|
|
|
|
assert models.PhoneCall.objects.count() == 2
|
|
|
|
assert models.PhoneCall.objects.filter(
|
|
|
|
caller='00337123456789', callee='102', stop__isnull=True).count() == 1
|
|
|
|
assert models.PhoneCall.objects.filter(
|
|
|
|
caller='0033699999999', callee='102', stop__isnull=False).count() == 1
|
|
|
|
|
|
|
|
with override_settings(PHONE_ONE_CALL_PER_CALLEE=False):
|
|
|
|
# accept multiple call: start a new one, don't stop anything
|
|
|
|
payload['caller'] = '00221774261500'
|
|
|
|
response = client.post(reverse('phone-call-event'), json.dumps(payload),
|
|
|
|
content_type='application/json')
|
|
|
|
assert response.status_code == 200
|
|
|
|
assert models.PhoneCall.objects.count() == 3
|
|
|
|
assert models.PhoneCall.objects.filter(callee='102', stop__isnull=True).count() == 2
|
|
|
|
assert models.PhoneCall.objects.filter(callee='102', stop__isnull=False).count() == 1
|
|
|
|
# same caller: stop his last call, add a new one
|
|
|
|
response = client.post(reverse('phone-call-event'), json.dumps(payload),
|
|
|
|
content_type='application/json')
|
|
|
|
assert response.status_code == 200
|
|
|
|
assert models.PhoneCall.objects.count() == 4
|
|
|
|
assert models.PhoneCall.objects.filter(callee='102', stop__isnull=True).count() == 2
|
|
|
|
assert models.PhoneCall.objects.filter(callee='102', stop__isnull=False).count() == 2
|
|
|
|
|
2015-10-28 17:41:14 +01:00
|
|
|
def test_current_calls(user, client):
|
|
|
|
# create some calls
|
|
|
|
for number in range(0, 10):
|
|
|
|
payload = {
|
|
|
|
'event': 'start',
|
|
|
|
'caller': '00336999999%02d' % number,
|
|
|
|
'callee': '1%02d' % number,
|
|
|
|
'data': {
|
|
|
|
'user': 'boby.lapointe',
|
|
|
|
}
|
|
|
|
}
|
|
|
|
response = client.post(reverse('phone-call-event'), json.dumps(payload),
|
|
|
|
content_type='application/json')
|
|
|
|
assert response.status_code == 200
|
|
|
|
assert response['content-type'] == 'application/json'
|
|
|
|
assert json.loads(response.content) == {'err': 0}
|
|
|
|
|
|
|
|
# register user to some lines
|
|
|
|
# then remove from some
|
|
|
|
for number in range(0, 10):
|
|
|
|
models.PhoneLine.take(callee='1%02d' % number, user=user)
|
|
|
|
for number in range(5, 10):
|
|
|
|
models.PhoneLine.release(callee='1%02d' % number, user=user)
|
|
|
|
client.login(username='toto', password='toto')
|
|
|
|
response = client.get(reverse('phone-current-calls'))
|
|
|
|
assert response.status_code == 200
|
|
|
|
assert response['content-type'] == 'application/json'
|
|
|
|
payload = json.loads(response.content)
|
|
|
|
assert isinstance(payload, dict)
|
|
|
|
assert set(payload.keys()) == set(['err', 'data'])
|
|
|
|
assert payload['err'] == 0
|
|
|
|
data = payload['data']
|
|
|
|
assert set(data.keys()) == set(['calls', 'lines', 'all-lines'])
|
|
|
|
assert isinstance(data['calls'], list)
|
|
|
|
assert isinstance(data['lines'], list)
|
|
|
|
assert isinstance(data['all-lines'], list)
|
|
|
|
assert len(data['calls']) == 5
|
|
|
|
assert len(data['lines']) == 5
|
|
|
|
assert len(data['all-lines']) == 10
|
|
|
|
for call in data['calls']:
|
|
|
|
assert set(call.keys()) <= set(['caller', 'callee', 'start', 'data'])
|
2020-01-19 19:11:28 +01:00
|
|
|
assert isinstance(call['caller'], six.string_types)
|
|
|
|
assert isinstance(call['callee'], six.string_types)
|
|
|
|
assert isinstance(call['start'], six.string_types)
|
2015-10-28 17:41:14 +01:00
|
|
|
if 'data' in call:
|
|
|
|
assert isinstance(call['data'], dict)
|
2020-01-19 19:11:28 +01:00
|
|
|
assert len([call for call in data['lines'] if isinstance(call, six.string_types)]) == 5
|
|
|
|
assert len([call for call in data['all-lines'] if isinstance(call, six.string_types)]) == 10
|
2015-10-28 17:41:14 +01:00
|
|
|
|
|
|
|
# unregister user to all remaining lines
|
|
|
|
for number in range(0, 5):
|
|
|
|
models.PhoneLine.release(callee='1%02d' % number, user=user)
|
|
|
|
response = client.get(reverse('phone-current-calls'))
|
|
|
|
assert response.status_code == 200
|
|
|
|
assert response['content-type'] == 'application/json'
|
|
|
|
payload = json.loads(response.content)
|
|
|
|
assert isinstance(payload, dict)
|
|
|
|
assert set(payload.keys()) == set(['err', 'data'])
|
|
|
|
assert payload['err'] == 0
|
|
|
|
assert set(payload['data'].keys()) == set(['calls', 'lines', 'all-lines'])
|
|
|
|
assert len(payload['data']['calls']) == 0
|
|
|
|
assert len(payload['data']['lines']) == 0
|
|
|
|
assert len(payload['data']['all-lines']) == 10
|
|
|
|
|
|
|
|
|
|
|
|
def test_take_release_line(user, client):
|
|
|
|
client.login(username='toto', password='toto')
|
|
|
|
|
|
|
|
assert models.PhoneLine.objects.count() == 0
|
|
|
|
payload = {
|
|
|
|
'callee': '102',
|
|
|
|
}
|
|
|
|
response = client.post(reverse('phone-take-line'), json.dumps(payload),
|
|
|
|
content_type='application/json')
|
|
|
|
assert response.status_code == 200
|
|
|
|
assert response['content-type'] == 'application/json'
|
|
|
|
assert json.loads(response.content) == {'err': 0}
|
|
|
|
assert models.PhoneLine.objects.count() == 1
|
|
|
|
assert models.PhoneLine.objects.filter(
|
|
|
|
users=user, callee='102').count() == 1
|
|
|
|
response = client.post(reverse('phone-release-line'), json.dumps(payload),
|
|
|
|
content_type='application/json')
|
|
|
|
assert response.status_code == 200
|
|
|
|
assert response['content-type'] == 'application/json'
|
|
|
|
assert json.loads(response.content) == {'err': 0}
|
|
|
|
assert models.PhoneLine.objects.count() == 1
|
|
|
|
assert models.PhoneLine.objects.filter(
|
|
|
|
users=user, callee='102').count() == 0
|
2017-05-31 23:45:52 +02:00
|
|
|
|
|
|
|
|
2017-05-31 23:55:20 +02:00
|
|
|
def test_phone_zone(user, client):
|
|
|
|
client.login(username='toto', password='toto')
|
|
|
|
response = client.get(reverse('phone-zone'))
|
|
|
|
assert response.status_code == 200
|
|
|
|
assert 'You do not have a phoneline configured' in response.content
|
|
|
|
|
|
|
|
models.PhoneLine.take(callee='102', user=user)
|
|
|
|
|
|
|
|
response = client.get(reverse('phone-zone'))
|
|
|
|
assert response.status_code == 200
|
|
|
|
assert 'You do not have a phoneline configured' not in response.content
|
|
|
|
assert '<li>102' in response.content
|
|
|
|
assert 'data-callee="102"' in response.content
|
|
|
|
currents = re.search('<div id="source-mainarea" '
|
|
|
|
'data-current-calls="/api/phone/current-calls/">'
|
|
|
|
'(.*?)</div>', response.content, flags=re.DOTALL)
|
|
|
|
assert currents.group(1).strip() == ''
|
|
|
|
|
|
|
|
# create a call
|
|
|
|
payload = {'event': 'start', 'caller': '003369999999', 'callee': '102'}
|
|
|
|
response = client.post(reverse('phone-call-event'), json.dumps(payload),
|
|
|
|
content_type='application/json')
|
|
|
|
assert response.status_code == 200
|
|
|
|
response = client.get(reverse('phone-zone'))
|
|
|
|
assert response.status_code == 200
|
|
|
|
assert '<h1>Current Call: <strong>003369999999</strong></h1>' in response.content
|
|
|
|
|
|
|
|
# simulate a mellon user
|
|
|
|
session = client.session
|
|
|
|
session['mellon_session'] = {'username': ['agent007@ldap']}
|
|
|
|
session.save()
|
|
|
|
response = client.get(reverse('phone-zone'))
|
|
|
|
assert response.status_code == 200
|
|
|
|
assert 'agent007' not in response.content
|
|
|
|
assert 'data-callee="agent007"' not in response.content
|
|
|
|
assert '<li>102' in response.content
|
|
|
|
assert 'data-callee="102"' in response.content
|
|
|
|
|
|
|
|
with override_settings(PHONE_AUTOTAKE_MELLON_USERNAME=True):
|
|
|
|
response = client.get(reverse('phone-zone'))
|
|
|
|
assert response.status_code == 200
|
|
|
|
assert '<h1>Current Call: <strong>003369999999</strong></h1>' in response.content
|
|
|
|
assert 'agent007' in response.content
|
|
|
|
assert 'data-callee="agent007"' in response.content
|
|
|
|
assert '<li>102' in response.content
|
|
|
|
assert 'data-callee="102"' in response.content
|
|
|
|
|
|
|
|
|
2017-05-31 23:45:52 +02:00
|
|
|
def test_call_expiration(user, client):
|
|
|
|
assert models.PhoneCall.objects.count() == 0
|
|
|
|
# create a call
|
|
|
|
payload = {'event': 'start', 'caller': '003369999999', 'callee': '102'}
|
|
|
|
response = client.post(reverse('phone-call-event'), json.dumps(payload),
|
|
|
|
content_type='application/json')
|
|
|
|
assert response.status_code == 200
|
|
|
|
assert models.PhoneCall.objects.filter(stop__isnull=True).count() == 1
|
|
|
|
|
|
|
|
# get list of calls for line 102
|
|
|
|
models.PhoneLine.take(callee='102', user=user)
|
|
|
|
client.login(username='toto', password='toto')
|
|
|
|
response = client.get(reverse('phone-current-calls'))
|
|
|
|
assert response.status_code == 200
|
|
|
|
payload = json.loads(response.content)
|
|
|
|
assert payload['err'] == 0
|
|
|
|
assert len(payload['data']['calls']) == 1
|
|
|
|
|
|
|
|
# start call 10 minutes ago
|
|
|
|
models.PhoneCall.objects.filter(stop__isnull=True).update(
|
|
|
|
start=now()-timedelta(minutes=10))
|
|
|
|
|
|
|
|
# get list of calls without expiration
|
|
|
|
response = client.get(reverse('phone-current-calls'))
|
|
|
|
assert response.status_code == 200
|
|
|
|
payload = json.loads(response.content)
|
|
|
|
assert payload['err'] == 0
|
|
|
|
assert len(payload['data']['calls']) == 1 # still here
|
|
|
|
|
|
|
|
# get list of calls with an expiration of 2 minutes (< 10 minutes)
|
|
|
|
with override_settings(PHONE_MAX_CALL_DURATION=2):
|
|
|
|
response = client.get(reverse('phone-current-calls'))
|
|
|
|
assert response.status_code == 200
|
|
|
|
payload = json.loads(response.content)
|
|
|
|
assert payload['err'] == 0
|
|
|
|
assert len(payload['data']['calls']) == 0 # call is expired
|
|
|
|
|
|
|
|
assert models.PhoneCall.objects.filter(stop__isnull=True).count() == 0 # active calls
|
|
|
|
assert models.PhoneCall.objects.filter(stop__isnull=False).count() == 1 # stopped calls
|