212 lines
9.2 KiB
Python
212 lines
9.2 KiB
Python
import pytest
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.utils.timezone import now, timedelta
|
|
|
|
import tests.utils
|
|
from passerelle.apps.phonecalls.models import Call, PhoneCalls
|
|
from passerelle.base.models import AccessRight, ApiUser
|
|
|
|
|
|
@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 = tests.utils.generic_endpoint_url('phonecalls', 'call-start', slug=phonecalls.slug)
|
|
assert start_endpoint == '/phonecalls/test/call-start'
|
|
stop_endpoint = tests.utils.generic_endpoint_url('phonecalls', 'call-stop', slug=phonecalls.slug)
|
|
assert stop_endpoint == '/phonecalls/test/call-stop'
|
|
calls_endpoint = tests.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 == {}
|
|
|
|
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
|
|
|
|
|
|
def test_phonecalls_start_redirect(app, phonecalls):
|
|
phonecalls.redirect_url = 'https://portail-agent.publik/'
|
|
phonecalls.save()
|
|
|
|
assert Call.objects.count() == 0
|
|
start_endpoint = tests.utils.generic_endpoint_url('phonecalls', 'call-start', slug=phonecalls.slug)
|
|
resp = app.get(
|
|
start_endpoint,
|
|
status=302,
|
|
params={'apikey': '123', 'callee': '42', 'caller': '0612345678', 'redirect': '1'},
|
|
)
|
|
assert resp.location == 'https://portail-agent.publik/?callee=42'
|
|
assert Call.objects.filter(callee='42', caller='0612345678').count() == 1
|
|
|
|
|
|
def test_phonecalls_start_newtab(app, phonecalls):
|
|
phonecalls.redirect_url = 'https://portail-agent.publik/'
|
|
phonecalls.save()
|
|
|
|
assert Call.objects.count() == 0
|
|
start_endpoint = tests.utils.generic_endpoint_url('phonecalls', 'call-start', slug=phonecalls.slug)
|
|
resp = app.get(
|
|
start_endpoint,
|
|
status=200,
|
|
params={'apikey': '123', 'callee': '42', 'caller': '0612345678', 'newtab': '1'},
|
|
)
|
|
assert resp.content_type == 'text/html'
|
|
assert 'X-Frame-Options' not in resp.headers
|
|
assert 'window.open("https://portail\\u002Dagent.publik/?callee\\u003D42")' in resp.text
|
|
assert Call.objects.filter(callee='42', caller='0612345678').count() == 1
|
|
|
|
|
|
def test_phonecalls_add_callee(app, phonecalls):
|
|
phonecalls.redirect_url = 'https://portail-agent.publik/'
|
|
phonecalls.save()
|
|
|
|
start_endpoint = tests.utils.generic_endpoint_url('phonecalls', 'call-start', slug=phonecalls.slug)
|
|
resp = app.get(
|
|
start_endpoint,
|
|
status=302,
|
|
params={'apikey': '123', 'callee': '42', 'caller': '0612345678', 'redirect': '1'},
|
|
)
|
|
assert resp.location == 'https://portail-agent.publik/?callee=42'
|