212 lines
9.8 KiB
Python
212 lines
9.8 KiB
Python
import pytest
|
|
import utils
|
|
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.utils.timezone import now, timedelta
|
|
|
|
from passerelle.apps.phonecalls.models import PhoneCalls, Call
|
|
from passerelle.base.models import ApiUser, AccessRight
|
|
|
|
|
|
@pytest.fixture
|
|
def phonecalls(db):
|
|
phonecalls = PhoneCalls.objects.create(slug='test')
|
|
apikey = ApiUser.objects.create(username='all', keytype='API', key='123')
|
|
obj_type = ContentType.objects.get_for_model(phonecalls)
|
|
AccessRight.objects.create(codename='can_access', apiuser=apikey,
|
|
resource_type=obj_type,
|
|
resource_pk=phonecalls.pk)
|
|
return phonecalls
|
|
|
|
|
|
def test_phonecalls_start_stop(app, phonecalls):
|
|
start_endpoint = utils.generic_endpoint_url('phonecalls', 'call-start',
|
|
slug=phonecalls.slug)
|
|
assert start_endpoint == '/phonecalls/test/call-start'
|
|
stop_endpoint = utils.generic_endpoint_url('phonecalls', 'call-stop',
|
|
slug=phonecalls.slug)
|
|
assert stop_endpoint == '/phonecalls/test/call-stop'
|
|
calls_endpoint = utils.generic_endpoint_url('phonecalls', 'calls',
|
|
slug=phonecalls.slug)
|
|
assert calls_endpoint == '/phonecalls/test/calls'
|
|
|
|
resp = app.get(start_endpoint, status=403)
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_class'] == 'django.core.exceptions.PermissionDenied'
|
|
resp = app.get(stop_endpoint, status=403)
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_class'] == 'django.core.exceptions.PermissionDenied'
|
|
resp = app.get(calls_endpoint, status=403)
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_class'] == 'django.core.exceptions.PermissionDenied'
|
|
|
|
resp = app.get(start_endpoint, params={'apikey': '123'}, status=400)
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_class'] == 'passerelle.views.WrongParameter'
|
|
assert 'missing parameters' in resp.json['err_desc']
|
|
resp = app.get(stop_endpoint, params={'apikey': '123'}, status=400)
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_class'] == 'passerelle.views.WrongParameter'
|
|
assert 'missing parameters' in resp.json['err_desc']
|
|
|
|
Call.objects.all().delete()
|
|
|
|
resp = app.get(start_endpoint, status=200, params={'apikey': '123',
|
|
'callee': '42',
|
|
'caller': '0612345678'})
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data']['callee'] == '42'
|
|
assert resp.json['data']['caller'] == '0612345678'
|
|
assert 'start' in resp.json['data']
|
|
assert resp.json['data']['end'] is None
|
|
assert resp.json['data']['is_current'] is True
|
|
assert resp.json['data']['details'] == {}
|
|
assert Call.objects.count() == 1
|
|
call = Call.objects.first()
|
|
assert call.callee == '42'
|
|
assert call.caller == '0612345678'
|
|
assert call.end_timestamp is None
|
|
assert call.details == {}
|
|
json_start = resp.json['data']['start']
|
|
start = call.start_timestamp
|
|
|
|
# same call, do nothing
|
|
resp = app.get(start_endpoint, status=200, params={'apikey': '123',
|
|
'callee': '42',
|
|
'caller': '0612345678'})
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data']['callee'] == '42'
|
|
assert resp.json['data']['caller'] == '0612345678'
|
|
assert resp.json['data']['start'] == json_start
|
|
assert resp.json['data']['end'] is None
|
|
assert resp.json['data']['is_current'] is True
|
|
assert Call.objects.count() == 1
|
|
call = Call.objects.first()
|
|
assert call.callee == '42'
|
|
assert call.caller == '0612345678'
|
|
assert call.end_timestamp is None
|
|
assert call.start_timestamp == start
|
|
assert call.details == {}
|
|
resp = app.get(start_endpoint, status=200, params={'apikey': '123',
|
|
'callee': '42',
|
|
'caller': '0612345678',
|
|
'foo': 'bar'}) # add details
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data']['callee'] == '42'
|
|
assert resp.json['data']['caller'] == '0612345678'
|
|
assert resp.json['data']['start'] == json_start
|
|
assert resp.json['data']['end'] is None
|
|
assert resp.json['data']['is_current'] is True
|
|
assert resp.json['data']['details'] == {'foo': 'bar'}
|
|
assert Call.objects.count() == 1
|
|
call = Call.objects.first()
|
|
assert call.callee == '42'
|
|
assert call.caller == '0612345678'
|
|
assert call.end_timestamp is None
|
|
assert call.start_timestamp == start
|
|
assert call.details == {'foo': 'bar'}
|
|
|
|
resp = app.get(calls_endpoint, status=200, params={'apikey': '123'})
|
|
assert resp.json['err'] == 0
|
|
assert len(resp.json['data']['current']) == 1
|
|
assert len(resp.json['data']['past']) == 0
|
|
resp = app.get(calls_endpoint, status=200, params={'apikey': '123', 'callee': '42'})
|
|
assert resp.json['err'] == 0
|
|
assert len(resp.json['data']['current']) == 1
|
|
assert len(resp.json['data']['past']) == 0
|
|
resp = app.get(calls_endpoint, status=200, params={'apikey': '123', 'callee': '43'})
|
|
assert resp.json['err'] == 0
|
|
assert len(resp.json['data']['current']) == 0
|
|
assert len(resp.json['data']['past']) == 0
|
|
|
|
resp = app.get(start_endpoint, status=200, params={'apikey': '123',
|
|
'callee': '43',
|
|
'caller': '0687654321'})
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data']['callee'] == '43'
|
|
assert resp.json['data']['caller'] == '0687654321'
|
|
assert Call.objects.count() == 2
|
|
assert Call.objects.filter(end_timestamp__isnull=True).count() == 2
|
|
|
|
resp = app.get(calls_endpoint, status=200, params={'apikey': '123', 'callee': '42'})
|
|
assert resp.json['err'] == 0
|
|
assert len(resp.json['data']['current']) == 1
|
|
assert len(resp.json['data']['past']) == 0
|
|
resp = app.get(calls_endpoint, status=200, params={'apikey': '123', 'callee': '43'})
|
|
assert resp.json['err'] == 0
|
|
assert len(resp.json['data']['current']) == 1
|
|
assert len(resp.json['data']['past']) == 0
|
|
|
|
resp = app.get(stop_endpoint, status=200, params={'apikey': '123',
|
|
'callee': '43',
|
|
'caller': '0687654321'})
|
|
assert resp.json['err'] == 0
|
|
assert len(resp.json['data']) == 1
|
|
assert resp.json['data'][0]['callee'] == '43'
|
|
assert resp.json['data'][0]['caller'] == '0687654321'
|
|
assert resp.json['data'][0]['start'] is not None
|
|
assert resp.json['data'][0]['end'] is not None
|
|
assert resp.json['data'][0]['is_current'] is False
|
|
assert Call.objects.count() == 2
|
|
assert Call.objects.filter(end_timestamp__isnull=True).count() == 1
|
|
assert Call.objects.filter(end_timestamp__isnull=False).count() == 1
|
|
|
|
# calls by callee
|
|
resp = app.get(calls_endpoint, status=200, params={'apikey': '123', 'callee': '42'})
|
|
assert resp.json['err'] == 0
|
|
assert len(resp.json['data']['current']) == 1
|
|
assert len(resp.json['data']['past']) == 0
|
|
resp = app.get(calls_endpoint, status=200, params={'apikey': '123', 'callee': '43'})
|
|
assert resp.json['err'] == 0
|
|
assert len(resp.json['data']['current']) == 0
|
|
assert len(resp.json['data']['past']) == 1
|
|
resp = app.get(calls_endpoint, status=200, params={'apikey': '123'})
|
|
assert resp.json['err'] == 0
|
|
assert len(resp.json['data']['current']) == 1
|
|
assert len(resp.json['data']['past']) == 1
|
|
resp = app.get(calls_endpoint, status=200, params={'apikey': '123', 'callee': 'foo'})
|
|
assert resp.json['err'] == 0
|
|
assert len(resp.json['data']['current']) == 0
|
|
assert len(resp.json['data']['past']) == 0
|
|
|
|
# calls by caller
|
|
resp = app.get(calls_endpoint, status=200, params={'apikey': '123',
|
|
'caller': '0612345678'})
|
|
assert resp.json['err'] == 0
|
|
assert len(resp.json['data']['current']) == 1
|
|
assert len(resp.json['data']['past']) == 0
|
|
resp = app.get(calls_endpoint, status=200, params={'apikey': '123',
|
|
'caller': '0687654321'})
|
|
assert resp.json['err'] == 0
|
|
assert len(resp.json['data']['current']) == 0
|
|
assert len(resp.json['data']['past']) == 1
|
|
resp = app.get(calls_endpoint, status=200, params={'apikey': '123',
|
|
'caller': 'foo'})
|
|
assert resp.json['err'] == 0
|
|
assert len(resp.json['data']['current']) == 0
|
|
assert len(resp.json['data']['past']) == 0
|
|
|
|
# create a "too long" current call (> 120 minutes == phonecalls.max_call_duration)
|
|
assert Call.objects.count() == 2
|
|
assert Call.objects.filter(end_timestamp__isnull=True).count() == 1
|
|
assert Call.objects.filter(end_timestamp__isnull=False).count() == 1
|
|
current_call = Call.objects.filter(end_timestamp__isnull=True).first()
|
|
current_call.start_timestamp = now() - timedelta(minutes=200)
|
|
current_call.save()
|
|
# close too long calls
|
|
phonecalls.hourly()
|
|
assert Call.objects.count() == 2
|
|
assert Call.objects.filter(end_timestamp__isnull=True).count() == 0
|
|
assert Call.objects.filter(end_timestamp__isnull=False).count() == 2
|
|
|
|
# create a "too old" call (> 60 days == phonecalls.data_retention_period)
|
|
old_call = Call.objects.first()
|
|
old_call.start_timestamp = old_call.start_timestamp - timedelta(days=100)
|
|
old_call.end_timestamp = old_call.end_timestamp - timedelta(days=100)
|
|
old_call.save()
|
|
# remove old calls
|
|
phonecalls.daily()
|
|
assert Call.objects.count() == 1
|
|
assert Call.objects.filter(end_timestamp__isnull=True).count() == 0
|
|
assert Call.objects.filter(end_timestamp__isnull=False).count() == 1
|