entrouvert
/
welco
16
0
Fork 0
Code Issues Pull Requests 2 Releases Activity

trivial: apply black

This commit is contained in:
Frédéric Péters 2021-01-11 20:10:12 +01:00
parent 54aae08cfe
commit ce7f2dd500
44 changed files with 571 additions and 454 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
debian/settings.py vendored
View File

@ -16,15 +16,15 @@
DEBUG = False
TEMPLATE_DEBUG = False
#ADMINS = (
# ADMINS = (
# # ('User 1', 'watchdog@example.net'),
# # ('User 2', 'janitor@example.net'),
#)
# )
# ALLOWED_HOSTS must be correct in production!
# See https://docs.djangoproject.com/en/dev/ref/settings/#allowed-hosts
ALLOWED_HOSTS = [
'*',
'*',
]
# Databases

20
setup.py
View File

@ -25,16 +25,18 @@ class eo_sdist(sdist):
def get_version():
'''Use the VERSION, if absent generates a version with git describe, if not
tag exists, take 0.0- and add the length of the commit log.
'''
"""Use the VERSION, if absent generates a version with git describe, if not
tag exists, take 0.0- and add the length of the commit log.
"""
if os.path.exists('VERSION'):
with open('VERSION', 'r') as v:
return v.read()
if os.path.exists('.git'):
p = subprocess.Popen(
['git', 'describe', '--dirty=.dirty', '--match=v*'],
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
result = p.communicate()[0]
if p.returncode == 0:
result = result.decode('ascii').strip()[1:] # strip spaces/newlines and initial v
@ -45,9 +47,7 @@ def get_version():
version = result
return version
else:
return '0.0.post%s' % len(
subprocess.check_output(
['git', 'rev-list', 'HEAD']).splitlines())
return '0.0.post%s' % len(subprocess.check_output(['git', 'rev-list', 'HEAD']).splitlines())
return '0.0'
@ -65,6 +65,7 @@ class compile_translations(Command):
curdir = os.getcwd()
try:
from django.core.management import call_command
for path, dirs, files in os.walk('welco'):
if 'locale' not in dirs:
continue
@ -108,7 +109,8 @@ setup(
'Programming Language :: Python',
'Programming Language :: Python :: 2',
],
install_requires=['django>=1.11,<2.3',
install_requires=[
'django>=1.11,<2.3',
'gadjo',
'django-ckeditor<4.5.4',
'django-haystack<2.8',
@ -118,7 +120,7 @@ setup(
'whoosh',
'XStatic-Select2',
'python-dateutil',
],
],
zip_safe=False,
cmdclass={
'build': build,

70
tests/test_contacts_manager.py
View File

@ -29,14 +29,9 @@ def test_get_contacts_zone_view(app, db):
resp = app.get('/ajax/contacts', status=200)
assert resp.html.find('button')['data-url'] == '/contacts/add/'
mail = Mail.objects.create(
content=ContentFile('foo', name='bar.txt'),
contact_id='42')
mail = Mail.objects.create(content=ContentFile('foo', name='bar.txt'), contact_id='42')
source_type = ContentType.objects.get_for_model(Mail).pk
resp = app.get(
'/ajax/contacts',
params={'source_type': source_type, 'source_pk': mail.pk},
status=200)
resp = app.get('/ajax/contacts', params={'source_type': source_type, 'source_pk': mail.pk}, status=200)
assert resp.html.find('a').text == '...'
assert resp.html.find('a')['data-page-slug'] == '42'
@ -46,9 +41,8 @@ def test_post_contacts_zone_view(app, db):
assert not mail.contact_id
source_type = ContentType.objects.get_for_model(Mail).pk
resp = app.post(
'/ajax/contacts',
params={'source_type': source_type, 'source_pk': mail.pk, 'user_id': 42},
status=200)
'/ajax/contacts', params={'source_type': source_type, 'source_pk': mail.pk, 'user_id': 42}, status=200
)
assert resp.text == 'ok'
assert Mail.objects.get(id=mail.pk).contact_id == '42'
@ -90,20 +84,24 @@ def test_search_json_view(settings, app, user, mail_group):
def response(url, request):
headers = {'content-type': 'application/json'}
content = {
'data': [{
'user_display_name': 'John Doe',
'user_email': 'john@example.net',
'user_var_phone': '0123456789',
'user_var_mobile': '0612345789',
'user_id': '42',
'user_roles': [{
'name': 'Agent',
'text': 'Agent',
'slug': 'agent',
'id': '8d73434814484aa0b8555ac9c68a9300'
}],
}],
'err': 0
'data': [
{
'user_display_name': 'John Doe',
'user_email': 'john@example.net',
'user_var_phone': '0123456789',
'user_var_mobile': '0612345789',
'user_id': '42',
'user_roles': [
{
'name': 'Agent',
'text': 'Agent',
'slug': 'agent',
'id': '8d73434814484aa0b8555ac9c68a9300',
}
],
}
],
'err': 0,
}
return httmock.response(200, content, headers)
@ -123,6 +121,7 @@ def test_contact_detail_fragment_view(settings, app, db):
}
}
}
@httmock.urlmatch(netloc='wcs.example.net', path='/api/users/42/', method='GET')
def response(url, request):
headers = {'content-type': 'application/json'}
@ -132,12 +131,9 @@ def test_contact_detail_fragment_view(settings, app, db):
'user_var_phone': '0123456789',
'user_var_mobile': '0612345789',
'user_id': '42',
'user_roles': [{
'name': 'Agent',
'text': 'Agent',
'slug': 'agent',
'id': '8d73434814484aa0b8555ac9c68a9300'
}],
'user_roles': [
{'name': 'Agent', 'text': 'Agent', 'slug': 'agent', 'id': '8d73434814484aa0b8555ac9c68a9300'}
],
}
return httmock.response(200, content, headers)
@ -148,15 +144,12 @@ def test_contact_detail_fragment_view(settings, app, db):
assert resp.html.find('li').text == 'Phone: 0123456789'
# unused 'is_pinned_user' context
mail = Mail.objects.create(
content=ContentFile('foo', name='bar.txt'),
contact_id='42')
mail = Mail.objects.create(content=ContentFile('foo', name='bar.txt'), contact_id='42')
source_type = ContentType.objects.get_for_model(Mail).pk
with httmock.HTTMock(response):
resp = app.get(
'/ajax/contacts/42/',
params={'source_type': source_type, 'source_pk': mail.pk},
status=200)
'/ajax/contacts/42/', params={'source_type': source_type, 'source_pk': mail.pk}, status=200
)
assert resp.html.find('h3').text == 'John Doe'
@ -183,7 +176,7 @@ def test_post_contact_add_view(mocked_sleep, settings, app, db):
'orig': 'http://welco.example.net/',
'secret': 'xxx',
}
}
},
}
# normal case
@ -210,7 +203,8 @@ def test_post_contact_add_view(mocked_sleep, settings, app, db):
'first_name': 'John',
'last_name': 'Doe',
},
status=200)
status=200,
)
assert resp.content_type == 'application/json'
assert resp.json['data']['user_id'] == '43'

40
tests/test_mail_manager.py
View File

@ -46,10 +46,7 @@ def test_get_feeder_view(app, user):
def test_post_feeder_view(app, user):
app.set_user(user.username)
resp = app.post(
'/mail/feeder/',
params={'mail': Upload('filename.txt', b'contents')},
status=302)
resp = app.post('/mail/feeder/', params={'mail': Upload('filename.txt', b'contents')}, status=302)
assert resp.location == '/mail/feeder/'
resp = resp.follow()
assert resp.html.find('li', {'class': 'info'}).text == '1 files uploaded successfully.'
@ -63,27 +60,28 @@ def test_qualification_save_view(settings, app, db):
}
}
}
mail = Mail.objects.create(
content=ContentFile('foo', name='bar.txt'),
subject='spam')
mail = Mail.objects.create(content=ContentFile('foo', name='bar.txt'), subject='spam')
assert not mail.contact_id
source_type = ContentType.objects.get_for_model(Mail).pk
resp = app.post(
'/ajax/qualification-mail-save',
params={'source_type': source_type, 'source_pk': mail.pk, 'subject': 'eggs'},
status=302)
assert resp.location == '/ajax/qualification?source_type=%s&source_pk=%s' % (
source_type, mail.pk)
status=302,
)
assert resp.location == '/ajax/qualification?source_type=%s&source_pk=%s' % (source_type, mail.pk)
@httmock.urlmatch(netloc='wcs.example.net', path='/api/formdefs/', method='GET')
def response_get(url, request):
headers = {'content-type': 'application/json'}
content = {
"err": 0,
"data": [{
"title": "Foo",
"slug": "foo",
}]}
"data": [
{
"title": "Foo",
"slug": "foo",
}
],
}
return httmock.response(200, content, headers)
with httmock.HTTMock(response_get):
@ -97,9 +95,7 @@ def test_edit_note_view(app, user):
assert resp.location.startswith('/login/?next=')
app.set_user(user.username)
mail = Mail.objects.create(
content=ContentFile('foo', name='bar.txt'),
note='spam')
mail = Mail.objects.create(content=ContentFile('foo', name='bar.txt'), note='spam')
resp = app.get('/ajax/mail/edit-note/', params={'mail': mail.pk}, status=200)
assert resp.html.find('h2').text == 'Note'
assert resp.html.find('textarea', {'name': 'note'}).text == 'spam'
@ -123,7 +119,7 @@ def test_note_view(app, user):
def test_reject_view(settings, app, user):
settings.MAARCH_FEED= {
settings.MAARCH_FEED = {
'URL': 'http://maarch.example.net',
'ENABLE': True,
'USERNAME': 'xxx',
@ -134,9 +130,7 @@ def test_reject_view(settings, app, user):
assert resp.location.startswith('/login/?next=')
app.set_user(user.username)
mail = Mail.objects.create(
content=ContentFile('foo', name='bar.txt'),
external_id='maarch-42')
mail = Mail.objects.create(content=ContentFile('foo', name='bar.txt'), external_id='maarch-42')
@httmock.urlmatch(netloc='maarch.example.net', path='/rest/res/resource/status', method='PUT')
def response_ok(url, request):
@ -150,9 +144,7 @@ def test_reject_view(settings, app, user):
assert Mail.objects.count() == 0
# errors
mail = Mail.objects.create(
content=ContentFile('foo', name='bar.txt'),
external_id='maarch-42')
mail = Mail.objects.create(content=ContentFile('foo', name='bar.txt'), external_id='maarch-42')
@httmock.urlmatch(netloc='maarch.example.net', path='/rest/res/resource/status', method='PUT')
def response_error1(url, request):

22
tests/test_manager.py
View File

@ -112,15 +112,11 @@ def test_wcs_summary_view(app, mail_group, user):
mail = Mail.objects.create(content=ContentFile('foo', name='bar.txt'))
source_type = ContentType.objects.get_for_model(Mail).pk
resp = app.get(
'/ajax/summary/%s/%s/?callback=spam' % (source_type, mail.pk),
status=302)
resp = app.get('/ajax/summary/%s/%s/?callback=spam' % (source_type, mail.pk), status=302)
assert resp.location.startswith('/login/?next=')
app.set_user(user.username)
resp = app.get(
'/ajax/summary/%s/%s/?callback=spam' % (source_type, mail.pk),
status=200)
resp = app.get('/ajax/summary/%s/%s/?callback=spam' % (source_type, mail.pk), status=200)
assert resp.content_type == 'application/javascript'
assert 'bar' in resp.text
assert resp.text.startswith('spam({')
@ -130,8 +126,8 @@ def test_remove_association_view(app, mail_group, user):
mail = Mail.objects.create(content=ContentFile('foo', name='bar.txt'))
source_type = ContentType.objects.get_for_model(Mail).pk
association = Association.objects.create(
source_type=ContentType.objects.get(id=source_type),
source_pk=mail.pk)
source_type=ContentType.objects.get(id=source_type), source_pk=mail.pk
)
assert Association.objects.filter(id=association.pk).count() == 1
resp = app.get('/ajax/remove-association/%s' % association.pk, status=302)
@ -154,8 +150,8 @@ def test_create_formdata_view(settings, app, mail_group, user):
mail = Mail.objects.create(content=ContentFile('foo', name='bar.txt'))
source_type = ContentType.objects.get_for_model(Mail).pk
association = Association.objects.create(
source_type=ContentType.objects.get(id=source_type),
source_pk=mail.pk)
source_type=ContentType.objects.get(id=source_type), source_pk=mail.pk
)
resp = app.get('/ajax/create-formdata/%s' % association.pk, status=302)
assert resp.location.startswith('/login/?next=')
@ -185,7 +181,8 @@ def test_create_formdata_view(settings, app, mail_group, user):
'data': {
'id': 42,
'backoffice_url': 'http://example.net',
}}
},
}
return httmock.response(200, content, headers)
with httmock.HTTMock(response_get, response_post):
@ -204,8 +201,7 @@ def test_menu_json_view(app, user, mail_group, phone_group, counter_group, kb_gr
app.set_user(user.username)
resp = app.get('/menu.json', status=200)
assert resp.content_type == 'application/json'
assert sorted([x['label'] for x in resp.json]) == [
'Call Center', 'Counter', 'Knowledge Base', 'Mails']
assert sorted([x['label'] for x in resp.json]) == ['Call Center', 'Counter', 'Knowledge Base', 'Mails']
resp = app.get('/menu.json?callback=foo', status=200)
assert resp.content_type == 'application/javascript'

28
tests/test_qualification.py
View File

@ -23,27 +23,25 @@ from welco.forms import QualificationForm
pytestmark = pytest.mark.django_db
KNOWN_SERVICES = {
'wcs': {
'eservices': {
'url': 'http://localhost/',
'title': 'Eservices',
'orig': 'welco'
}
}
}
KNOWN_SERVICES = {'wcs': {'eservices': {'url': 'http://localhost/', 'title': 'Eservices', 'orig': 'welco'}}}
@mock.patch('welco.utils.requests.get')
def test_get_qualification(mocked_get, client):
with override_settings(KNOWN_SERVICES=KNOWN_SERVICES):
forms = mock.Mock()
forms.json.return_value = {'data': [{'category': 'Test',
'authentication_required': False,
'description': '',
'title': 'Test form',
'slug': 'test-form'}],
'err': 0}
forms.json.return_value = {
'data': [
{
'category': 'Test',
'authentication_required': False,
'description': '',
'title': 'Test form',
'slug': 'test-form',
}
],
'err': 0,
}
mocked_get.return_value = forms
user = mock.Mock()

125
tests/test_source_maarch.py
View File

@ -59,11 +59,13 @@ class MaarchMock(BaseMock):
},
'status_code': 200,
}
list_endpoint.path = '^/rest/res/list$'
def update_external_infos(self, url, request):
self.requests.append(('update_external_infos', url, request, json.loads(force_text(request.body))))
return json.dumps({})
update_external_infos.path = '^/rest/res/externalInfos$'
def update_status(self, url, request):
@ -75,10 +77,12 @@ class MaarchMock(BaseMock):
},
'status_code': 200,
}
update_status.path = '^/rest/res/resource/status$'
def post_courrier(self, url, request):
self.requests.append(('post_courrier', url, request, json.loads(force_text(request.body))))
post_courrier.path = '^/rest/res$'
@ -96,37 +100,50 @@ def maarch(settings, mail_group):
class WcsMock(BaseMock):
def api_formdefs(self, url, request):
return json.dumps({
'data': [{
'slug': 'slug1',
'title': 'title1',
}]
})
return json.dumps(
{
'data': [
{
'slug': 'slug1',
'title': 'title1',
}
]
}
)
api_formdefs.path = '^/api/formdefs/$'
def json(self, url, request):
return json.dumps({
'data': [{
'slug': 'slug1',
'title': 'title1',
'category': 'category1',
}]
})
return json.dumps(
{
'data': [
{
'slug': 'slug1',
'title': 'title1',
'category': 'category1',
}
]
}
)
json.path = '^/json$'
def api_formdefs_slug1_schema(self, url, request):
return json.dumps({
})
return json.dumps({})
api_formdefs_slug1_schema.path = '^/api/formdefs/slug-1/schema$'
def api_formdefs_slug1_submit(self, url, request):
return json.dumps({
'err': 0,
'data': {
'id': 1,
'backoffice_url': 'http://wcs.example.net/slug-1/1',
},
})
return json.dumps(
{
'err': 0,
'data': {
'id': 1,
'backoffice_url': 'http://wcs.example.net/slug-1/1',
},
}
)
api_formdefs_slug1_submit.path = '^/api/formdefs/slug-1/submit$'
@ -172,14 +189,16 @@ def test_feed(settings, app, maarch, wcs, user):
# feed mails from maarch
with maarch.ctx_manager:
# list request
maarch.responses.append({
'resources': [
{
'res_id': 1,
'fileBase64Content': force_text(base64.b64encode(PDF_MOCK)),
}
],
})
maarch.responses.append(
{
'resources': [
{
'res_id': 1,
'fileBase64Content': force_text(base64.b64encode(PDF_MOCK)),
}
],
}
)
# update status request
maarch.responses.append({})
# last list request
@ -214,20 +233,26 @@ def test_feed(settings, app, maarch, wcs, user):
maarch.clear()
pk = Mail.objects.get().pk
with wcs.ctx_manager, maarch.ctx_manager:
source_type = str(ContentType.objects.get_for_model(Mail).pk),
source_type = (str(ContentType.objects.get_for_model(Mail).pk),)
source_pk = str(pk)
response = app.get('/ajax/qualification', params={
'source_type': source_type,
'source_pk': source_pk,
})
response = app.get(
'/ajax/qualification',
params={
'source_type': source_type,
'source_pk': source_pk,
},
)
assert len(response.pyquery('a[data-association-pk]')) == 0
response = app.post('/ajax/qualification', params={
'source_type': source_type,
'source_pk': str(pk),
'formdef_reference': 'demarches:slug-1',
})
response = app.post(
'/ajax/qualification',
params={
'source_type': source_type,
'source_pk': str(pk),
'formdef_reference': 'demarches:slug-1',
},
)
# verify qualification was done
assert len(response.pyquery('a[data-association-pk]')) == 1
@ -243,7 +268,7 @@ def test_feed(settings, app, maarch, wcs, user):
'external_link': 'http://wcs.example.net/slug-1/1',
'res_id': 1,
}
]
],
}
# verify we can answer
@ -260,24 +285,24 @@ def test_feed(settings, app, maarch, wcs, user):
assert response.json['err'] == 1
# verify error when maarch feed is not configured
settings.MAARCH_FEED['ENABLE'] = False
response = app.post_json('/api/mail/response/',
params={'mail_id': 'maarch-1', 'content': 'coucou'},
status=200)
response = app.post_json(
'/api/mail/response/', params={'mail_id': 'maarch-1', 'content': 'coucou'}, status=200
)
assert response.json['err'] == 1
assert response.json['err_desc'] == 'maarch is unconfigured'
settings.MAARCH_FEED['ENABLE'] = True
# verify error when mail_id is unknown
response = app.post_json('/api/mail/response/',
params={'mail_id': 'maarch-231', 'content': 'coucou'},
status=404)
response = app.post_json(
'/api/mail/response/', params={'mail_id': 'maarch-231', 'content': 'coucou'}, status=404
)
assert response.json['err'] == 1
# successfull call
maarch.responses.append({})
with maarch.ctx_manager:
response = app.post_json('/api/mail/response/',
params={'mail_id': 'maarch-1', 'content': 'coucou'},
status=200)
response = app.post_json(
'/api/mail/response/', params={'mail_id': 'maarch-1', 'content': 'coucou'}, status=200
)
assert maarch.requests[0][3] == {
'historyMessage': 'coucou',
'resId': [1],

137
tests/test_source_phone.py
View File

@ -38,95 +38,104 @@ def test_call_start_stop(client):
'callee': '102',
'data': {
'user': 'boby.lapointe',
}
},
}
response = client.post(reverse('phone-call-event'), json.dumps(payload),
content_type='application/json')
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 response.json() == {'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
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')
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 response.json() == {'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
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
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')
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 response.json() == {'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
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')
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 response.json() == {'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
assert (
models.PhoneCall.objects.filter(
caller='0033699999999', callee='102', data=json.dumps(payload['data']), stop__isnull=False
).count()
== 2
)
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')
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
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')
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
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')
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')
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
def test_current_calls(user, client):
# create some calls
for number in range(0, 10):
@ -136,10 +145,11 @@ def test_current_calls(user, client):
'callee': '1%02d' % number,
'data': {
'user': 'boby.lapointe',
}
},
}
response = client.post(reverse('phone-call-event'), json.dumps(payload),
content_type='application/json')
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 response.json() == {'err': 0}
@ -199,22 +209,20 @@ def test_take_release_line(user, client):
payload = {
'callee': '102',
}
response = client.post(reverse('phone-take-line'), json.dumps(payload),
content_type='application/json')
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 response.json() == {'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 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 response.json() == {'err': 0}
assert models.PhoneLine.objects.count() == 1
assert models.PhoneLine.objects.filter(
users=user, callee='102').count() == 0
assert models.PhoneLine.objects.filter(users=user, callee='102').count() == 0
def test_phone_zone(user, client):
@ -230,15 +238,16 @@ def test_phone_zone(user, client):
assert 'You do not have a phoneline configured' not in force_text(response.content)
assert '<li>102' in force_text(response.content)
assert 'data-callee="102"' in force_text(response.content)
currents = re.search('<div id="source-mainarea" '
'data-current-calls="/api/phone/current-calls/">'
'(.*?)</div>', force_text(response.content), flags=re.DOTALL)
currents = re.search(
'<div id="source-mainarea" ' 'data-current-calls="/api/phone/current-calls/">' '(.*?)</div>',
force_text(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')
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
@ -269,8 +278,7 @@ 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')
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
@ -284,15 +292,14 @@ def test_call_expiration(user, client):
assert len(payload['data']['calls']) == 1
# start call 10 minutes ago
models.PhoneCall.objects.filter(stop__isnull=True).update(
start=now()-timedelta(minutes=10))
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 = response.json()
assert payload['err'] == 0
assert len(payload['data']['calls']) == 1 # still here
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):
@ -300,7 +307,7 @@ def test_call_expiration(user, client):
assert response.status_code == 200
payload = response.json()
assert payload['err'] == 0
assert len(payload['data']['calls']) == 0 # call is expired
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
assert models.PhoneCall.objects.filter(stop__isnull=False).count() == 1 # stopped calls

1
welco/apps.py
View File

@ -17,6 +17,7 @@
from django.apps import apps
from django.conf.urls import include, url
def register_urls(urlpatterns):
pre_urls = []
post_urls = []

13
welco/contacts/forms.py
View File

@ -24,14 +24,15 @@ DEFAULT_TITLE_CHOICES = (
(pgettext_lazy('title', 'Mr'), pgettext_lazy('title', 'Mr')),
)
class ContactAddForm(forms.Form):
title = forms.CharField(label=_('Title'),
required=False,
widget=forms.Select(choices=DEFAULT_TITLE_CHOICES))
title = forms.CharField(
label=_('Title'), required=False, widget=forms.Select(choices=DEFAULT_TITLE_CHOICES)
)
first_name = forms.CharField(label=_('First Name'), required=False)
last_name = forms.CharField(label=_('Last Name'),
required=True,
widget=forms.TextInput(attrs={'required': 'required'}))
last_name = forms.CharField(
label=_('Last Name'), required=True, widget=forms.TextInput(attrs={'required': 'required'})
)
email = forms.CharField(label=_('Email'), required=False)
address = forms.CharField(label=_('Address'), required=False)
zipcode = forms.CharField(label=_('Zip Code'), required=False)

26
welco/contacts/views.py
View File

@ -33,6 +33,7 @@ from welco.utils import get_wcs_data, sign_url
from .forms import ContactAddForm
class HomeZone(object):
def __init__(self, request):
self.request = request
@ -49,21 +50,20 @@ class ContactsZone(TemplateView):
context = super(ContactsZone, self).get_context_data(**kwargs)
context['source_pk'] = self.request.GET.get('source_pk')
if 'source_pk' in self.request.GET:
source_class = ContentType.objects.get(
id=self.request.GET['source_type']).model_class()
source_class = ContentType.objects.get(id=self.request.GET['source_type']).model_class()
source_object = source_class.objects.get(id=self.request.GET['source_pk'])
context['contact_user_id'] = source_object.contact_id
return context
def post(self, request, *args, **kwargs):
if 'user_id' in request.POST:
source_class = ContentType.objects.get(
id=self.request.POST['source_type']).model_class()
source_class = ContentType.objects.get(id=self.request.POST['source_type']).model_class()
source_object = source_class.objects.get(id=self.request.POST['source_pk'])
source_object.contact_id = request.POST['user_id']
source_object.save()
return HttpResponse('ok')
zone = csrf_exempt(ContactsZone.as_view())
@ -83,8 +83,12 @@ def search_json(request):
raise Exception('error %r' % result)
for user in result.get('data'):
user['title'] = user['user_display_name']
more = [user.get('user_var_address'), user.get('user_var_phone'),
user.get('user_var_mobile'), user.get('user_var_email')]
more = [
user.get('user_var_address'),
user.get('user_var_phone'),
user.get('user_var_mobile'),
user.get('user_var_email'),
]
user['more'] = ' / '.join([x for x in more if x])
if user.get('user_roles'):
user['roles'] = ' / '.join([r['text'] for r in user['user_roles']])
@ -109,13 +113,13 @@ class ContactDetailFragmentView(TemplateView):
context['user_id'] = user_id
if 'source_pk' in self.request.GET:
source_class = ContentType.objects.get(
id=self.request.GET['source_type']).model_class()
source_class = ContentType.objects.get(id=self.request.GET['source_type']).model_class()
source_object = source_class.objects.get(id=self.request.GET['source_pk'])
context['is_pinned_user'] = bool(source_object.contact_id == user_id)
return context
contact_detail_fragment = ContactDetailFragmentView.as_view()
@ -144,9 +148,8 @@ class ContactAdd(FormView):
logger = logging.getLogger(__name__)
logger.info('POST to authentic (%r)', json.dumps(msg))
authentic_response = requests.post(
signed_url,
data=json.dumps(msg),
headers={'Content-type': 'application/json'})
signed_url, data=json.dumps(msg), headers={'Content-type': 'application/json'}
)
logger.info('Got authentic response (%r)', authentic_response.text)
user_uuid = authentic_response.json().get('uuid')
@ -165,4 +168,5 @@ class ContactAdd(FormView):
json.dump(result, response, indent=2)
return response
contact_add = csrf_exempt(ContactAdd.as_view())

1
welco/kb/apps.py
View File

@ -16,6 +16,7 @@
from django.apps import AppConfig
class KbAppConfig(AppConfig):
name = 'welco.kb'

1
welco/kb/forms.py
View File

@ -19,6 +19,7 @@ from django.utils.text import slugify
from .models import Page
class PageForm(forms.ModelForm):
class Meta:
model = Page

8
welco/kb/migrations/0001_initial.py
View File

@ -7,14 +7,16 @@ import ckeditor.fields
class Migration(migrations.Migration):
dependencies = [
]
dependencies = []
operations = [
migrations.CreateModel(
name='Page',
fields=[
('id', models.AutoField(verbose_name='ID', serialize=False, auto_created=True, primary_key=True)),
(
'id',
models.AutoField(verbose_name='ID', serialize=False, auto_created=True, primary_key=True),
),
('title', models.CharField(max_length=200, verbose_name='Title')),
('slug', models.SlugField(verbose_name='Slug')),
('content', ckeditor.fields.RichTextField(verbose_name='Text')),

8
welco/kb/migrations/0003_page_tags.py
View File

@ -16,7 +16,13 @@ class Migration(migrations.Migration):
migrations.AddField(
model_name='page',
name='tags',
field=taggit.managers.TaggableManager(to='taggit.Tag', through='taggit.TaggedItem', blank=True, help_text='A comma-separated list of tags.', verbose_name='Keywords'),
field=taggit.managers.TaggableManager(
to='taggit.Tag',
through='taggit.TaggedItem',
blank=True,
help_text='A comma-separated list of tags.',
verbose_name='Keywords',
),
preserve_default=True,
),
]

3
welco/kb/models.py
View File

@ -26,8 +26,7 @@ class Page(models.Model):
title = models.CharField(_('Title'), max_length=200)
slug = models.SlugField(_('Slug'))
content = RichTextField(_('Text'))
tags = TaggableManager(_('Keywords'), blank=True,
help_text=_('A comma-separated list of tags.'))
tags = TaggableManager(_('Keywords'), blank=True, help_text=_('A comma-separated list of tags.'))
class Meta:
ordering = ['title']

1
welco/kb/search_indexes.py
View File

@ -21,6 +21,7 @@ from haystack import indexes
from .models import Page
class PageIndex(indexes.SearchIndex, indexes.Indexable):
title = indexes.CharField(model_attr='title', boost=3)
text = indexes.CharField(document=True)

19
welco/kb/views.py
View File

@ -25,8 +25,7 @@ from django.db.models import Count
from django.http import HttpResponse, HttpResponseRedirect
from django.template import RequestContext
from django.views.decorators.csrf import csrf_exempt
from django.views.generic import (DetailView, CreateView, UpdateView,
ListView, DeleteView, TemplateView)
from django.views.generic import DetailView, CreateView, UpdateView, ListView, DeleteView, TemplateView
from haystack.forms import SearchForm
from haystack.generic_views import SearchView
@ -36,15 +35,17 @@ from taggit.models import Tag
from .models import Page
from .forms import PageForm
def check_user_perms(user, access=False):
allowed_roles = settings.KB_MANAGE_ROLES[:]
if access:
allowed_roles.extend(settings.KB_ACCESS_ROLES)
if settings.KB_ROLE:
allowed_roles.append(settings.KB_ROLE) # legacy
allowed_roles.append(settings.KB_ROLE) # legacy
user_groups = set([x.name for x in user.groups.all()])
return user_groups.intersection(allowed_roles)
def check_request_perms(request, access=False):
if not check_user_perms(request.user, access=access):
raise PermissionDenied()
@ -63,6 +64,7 @@ class PageListView(ListView):
context['can_manage'] = check_user_perms(self.request.user)
return context
page_list = login_required(PageListView.as_view())
@ -74,6 +76,7 @@ class PageAddView(CreateView):
check_request_perms(request)
return super(PageAddView, self).dispatch(request, *args, **kwargs)
page_add = login_required(PageAddView.as_view())
@ -85,6 +88,7 @@ class PageEditView(UpdateView):
check_request_perms(request)
return super(PageEditView, self).dispatch(request, *args, **kwargs)
page_edit = login_required(PageEditView.as_view())
@ -108,6 +112,7 @@ class PageDetailFragmentView(DetailView):
model = Page
template_name = 'kb/page_detail_fragment.html'
page_detail_fragment = PageDetailFragmentView.as_view()
@ -119,6 +124,7 @@ class PageDeleteView(DeleteView):
check_request_perms(request)
return super(PageDeleteView, self).dispatch(request, *args, **kwargs)
page_delete = login_required(PageDeleteView.as_view())
@ -130,6 +136,7 @@ class PageSearchView(SearchView):
check_request_perms(request, access=True)
return super(PageSearchView, self).dispatch(request, *args, **kwargs)
page_search = login_required(PageSearchView.as_view())
@ -140,8 +147,9 @@ class KbZone(TemplateView):
context = super(KbZone, self).get_context_data(**kwargs)
context['source_pk'] = self.request.GET.get('source_pk')
context['form'] = SearchForm()
context['tags'] = Tag.objects.all().annotate(
num_times=Count('taggit_taggeditem_items')).filter(num_times__gt=0)
context['tags'] = (
Tag.objects.all().annotate(num_times=Count('taggit_taggeditem_items')).filter(num_times__gt=0)
)
num_times = context['tags'].values_list('num_times', flat=True)
if not num_times:
num_times = [0]
@ -163,6 +171,7 @@ class KbZone(TemplateView):
tag.font_size = 'x-large'
return context
zone = csrf_exempt(KbZone.as_view())