2018-05-19 16:38:57 +02:00
|
|
|
# welco - multichannel request processing
|
|
|
|
# Copyright (C) 2018 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/>.
|
|
|
|
|
|
|
|
import json
|
|
|
|
|
|
|
|
import pytest
|
|
|
|
|
2018-11-07 13:49:31 +01:00
|
|
|
from django.contrib.auth.models import User
|
2020-01-19 19:32:30 +01:00
|
|
|
from django.utils.encoding import force_text
|
2018-11-07 13:49:31 +01:00
|
|
|
|
2018-05-19 16:38:57 +02:00
|
|
|
from httmock import urlmatch, HTTMock
|
|
|
|
|
|
|
|
|
|
|
|
class BaseMock(object):
|
|
|
|
def __init__(self, netloc):
|
|
|
|
self.netloc = netloc
|
|
|
|
self.clear()
|
|
|
|
|
|
|
|
def clear(self):
|
|
|
|
self.requests = []
|
|
|
|
self.responses = []
|
|
|
|
|
|
|
|
def next_response(self):
|
|
|
|
response, self.responses = self.responses[0], self.responses[1:]
|
|
|
|
return response
|
|
|
|
|
|
|
|
@property
|
|
|
|
def ctx_manager(self):
|
|
|
|
'''Create an HTTMock context manager for all endpoints of a mocked Maarch instance'''
|
|
|
|
endpoints = []
|
|
|
|
for attribute, value in self.__class__.__dict__.items():
|
|
|
|
if hasattr(value, 'path'):
|
|
|
|
value = getattr(self, attribute)
|
|
|
|
match_decorator = urlmatch(netloc=self.netloc, path=value.path)
|
|
|
|
endpoints.append(match_decorator(value))
|
|
|
|
return HTTMock(*endpoints)
|
|
|
|
|
|
|
|
|
|
|
|
class MaarchMock(BaseMock):
|
|
|
|
def list_endpoint(self, url, request):
|
2020-01-19 19:46:54 +01:00
|
|
|
self.requests.append(('list_endpoint', url, request, json.loads(force_text(request.body))))
|
2018-05-19 16:38:57 +02:00
|
|
|
return {
|
|
|
|
'content': json.dumps(self.next_response()),
|
|
|
|
'headers': {
|
|
|
|
'content-type': 'application/json',
|
|
|
|
},
|
2020-01-19 19:32:30 +01:00
|
|
|
'status_code': 200,
|
2018-05-19 16:38:57 +02:00
|
|
|
}
|
2021-01-11 20:10:12 +01:00
|
|
|
|
2018-05-19 16:38:57 +02:00
|
|
|
list_endpoint.path = '^/rest/res/list$'
|
|
|
|
|
|
|
|
def update_external_infos(self, url, request):
|
2020-01-19 19:46:54 +01:00
|
|
|
self.requests.append(('update_external_infos', url, request, json.loads(force_text(request.body))))
|
2018-05-19 16:38:57 +02:00
|
|
|
return json.dumps({})
|
2021-01-11 20:10:12 +01:00
|
|
|
|
2018-05-19 16:38:57 +02:00
|
|
|
update_external_infos.path = '^/rest/res/externalInfos$'
|
|
|
|
|
|
|
|
def update_status(self, url, request):
|
2020-01-19 19:46:54 +01:00
|
|
|
self.requests.append(('update_status', url, request, json.loads(force_text(request.body))))
|
2018-05-19 16:38:57 +02:00
|
|
|
return {
|
|
|
|
'content': json.dumps(self.next_response()),
|
|
|
|
'headers': {
|
|
|
|
'content-type': 'application/json',
|
|
|
|
},
|
2020-01-19 19:32:30 +01:00
|
|
|
'status_code': 200,
|
2018-05-19 16:38:57 +02:00
|
|
|
}
|
2021-01-11 20:10:12 +01:00
|
|
|
|
2018-05-19 16:38:57 +02:00
|
|
|
update_status.path = '^/rest/res/resource/status$'
|
|
|
|
|
|
|
|
def post_courrier(self, url, request):
|
2020-01-19 19:46:54 +01:00
|
|
|
self.requests.append(('post_courrier', url, request, json.loads(force_text(request.body))))
|
2021-01-11 20:10:12 +01:00
|
|
|
|
2018-05-19 16:38:57 +02:00
|
|
|
post_courrier.path = '^/rest/res$'
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.fixture
|
|
|
|
def maarch(settings, mail_group):
|
|
|
|
# configure maarch server
|
|
|
|
settings.MAARCH_FEED = {
|
|
|
|
'ENABLE': True,
|
|
|
|
'URL': 'http://maarch.example.net/',
|
|
|
|
'USERNAME': 'admin',
|
|
|
|
'PASSWORD': 'admin',
|
|
|
|
}
|
|
|
|
return MaarchMock('maarch.example.net')
|
|
|
|
|
|
|
|
|
|
|
|
class WcsMock(BaseMock):
|
|
|
|
def api_formdefs(self, url, request):
|
|
|
|
return json.dumps(
|
|
|
|
{
|
|
|
|
'data': [
|
|
|
|
{
|
|
|
|
'slug': 'slug1',
|
|
|
|
'title': 'title1',
|
|
|
|
}
|
|
|
|
]
|
|
|
|
}
|
|
|
|
)
|
2021-01-11 20:10:12 +01:00
|
|
|
|
2018-05-19 16:38:57 +02:00
|
|
|
api_formdefs.path = '^/api/formdefs/$'
|
|
|
|
|
|
|
|
def json(self, url, request):
|
|
|
|
return json.dumps(
|
|
|
|
{
|
|
|
|
'data': [
|
|
|
|
{
|
|
|
|
'slug': 'slug1',
|
|
|
|
'title': 'title1',
|
|
|
|
'category': 'category1',
|
|
|
|
}
|
|
|
|
]
|
|
|
|
}
|
|
|
|
)
|
2021-01-11 20:10:12 +01:00
|
|
|
|
2018-05-19 16:38:57 +02:00
|
|
|
json.path = '^/json$'
|
|
|
|
|
|
|
|
def api_formdefs_slug1_schema(self, url, request):
|
|
|
|
return json.dumps({})
|
2021-01-11 20:10:12 +01:00
|
|
|
|
2018-05-19 16:38:57 +02:00
|
|
|
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',
|
|
|
|
},
|
|
|
|
}
|
|
|
|
)
|
2021-01-11 20:10:12 +01:00
|
|
|
|
2018-05-19 16:38:57 +02:00
|
|
|
api_formdefs_slug1_submit.path = '^/api/formdefs/slug-1/submit$'
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.fixture
|
|
|
|
def wcs(settings):
|
|
|
|
settings.KNOWN_SERVICES = {
|
|
|
|
'wcs': {
|
|
|
|
'demarches': {
|
|
|
|
'url': 'http://wcs.example.net/',
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return WcsMock('wcs.example.net')
|
|
|
|
|
|
|
|
|
|
|
|
def test_utils(maarch):
|
|
|
|
from welco.sources.mail.utils import get_maarch
|
|
|
|
|
|
|
|
welco_maarch_obj = get_maarch()
|
|
|
|
assert welco_maarch_obj.url == 'http://maarch.example.net/'
|
|
|
|
assert welco_maarch_obj.username == 'admin'
|
|
|
|
assert welco_maarch_obj.password == 'admin'
|
|
|
|
assert welco_maarch_obj.grc_status == 'GRC'
|
|
|
|
assert welco_maarch_obj.grc_received_status == 'GRC_TRT'
|
|
|
|
assert welco_maarch_obj.grc_send_status == 'GRCSENT'
|
|
|
|
assert welco_maarch_obj.grc_refused_status == 'GRCREFUSED'
|
|
|
|
|
|
|
|
|
2020-01-19 19:32:30 +01:00
|
|
|
PDF_MOCK = b'%PDF-1.4 ...'
|
2018-05-19 16:38:57 +02:00
|
|
|
|
|
|
|
|
2018-11-07 13:49:31 +01:00
|
|
|
def test_feed(settings, app, maarch, wcs, user):
|
2018-05-19 16:38:57 +02:00
|
|
|
import base64
|
|
|
|
from django.core.management import call_command
|
|
|
|
from django.contrib.contenttypes.models import ContentType
|
|
|
|
from welco.sources.mail.models import Mail
|
|
|
|
|
|
|
|
app.set_user(user.username)
|
|
|
|
response = app.get('/').follow()
|
|
|
|
# no mail
|
|
|
|
assert len(response.pyquery('li[data-external-id]')) == 0
|
|
|
|
|
|
|
|
# feed mails from maarch
|
|
|
|
with maarch.ctx_manager:
|
|
|
|
# list request
|
|
|
|
maarch.responses.append(
|
|
|
|
{
|
|
|
|
'resources': [
|
|
|
|
{
|
|
|
|
'res_id': 1,
|
2020-01-19 19:32:30 +01:00
|
|
|
'fileBase64Content': force_text(base64.b64encode(PDF_MOCK)),
|
2018-05-19 16:38:57 +02:00
|
|
|
}
|
|
|
|
],
|
|
|
|
}
|
|
|
|
)
|
|
|
|
# update status request
|
|
|
|
maarch.responses.append({})
|
|
|
|
# last list request
|
|
|
|
maarch.responses.append({'resources': []})
|
|
|
|
call_command('feed_mail_maarch')
|
|
|
|
assert len(maarch.requests) == 3
|
|
|
|
assert maarch.requests[0][3] == {
|
|
|
|
'select': '*',
|
|
|
|
'clause': "status='GRC'",
|
|
|
|
'withFile': True,
|
|
|
|
'orderBy': ['res_id'],
|
|
|
|
'limit': 10,
|
|
|
|
}
|
|
|
|
assert maarch.requests[1][3] == {
|
|
|
|
'resId': [1],
|
|
|
|
'status': 'GRC_TRT',
|
|
|
|
}
|
|
|
|
assert maarch.requests[2][3] == {
|
|
|
|
'select': '*',
|
|
|
|
'clause': "status='GRC'",
|
|
|
|
'withFile': True,
|
|
|
|
'orderBy': ['res_id'],
|
|
|
|
'limit': 10,
|
|
|
|
}
|
|
|
|
response = app.get('/').follow()
|
|
|
|
|
|
|
|
# new mail is visible
|
|
|
|
assert len(response.pyquery('li[data-external-id]')) == 1
|
|
|
|
assert len(response.pyquery('li[data-external-id=maarch-1]')) == 1
|
|
|
|
|
|
|
|
# start qualification
|
|
|
|
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_pk = str(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',
|
|
|
|
},
|
|
|
|
)
|
|
|
|
|
|
|
|
# verify qualification was done
|
|
|
|
assert len(response.pyquery('a[data-association-pk]')) == 1
|
|
|
|
association_pk = response.pyquery('a[data-association-pk]')[0].attrib['data-association-pk']
|
|
|
|
|
|
|
|
response = app.post('/ajax/create-formdata/%s' % association_pk)
|
|
|
|
assert len(maarch.requests) == 1
|
|
|
|
assert maarch.requests[0][3] == {
|
|
|
|
'status': 'GRCSENT',
|
|
|
|
'externalInfos': [
|
|
|
|
{
|
2019-01-23 20:48:13 +01:00
|
|
|
'external_id': '1',
|
2018-05-19 16:38:57 +02:00
|
|
|
'external_link': 'http://wcs.example.net/slug-1/1',
|
|
|
|
'res_id': 1,
|
|
|
|
}
|
|
|
|
],
|
|
|
|
}
|
|
|
|
|
2018-11-07 13:49:31 +01:00
|
|
|
# verify we can answer
|
|
|
|
maarch.clear()
|
|
|
|
app.set_user(None)
|
|
|
|
user = User.objects.create(username='test')
|
|
|
|
user.set_password('test')
|
|
|
|
user.save()
|
|
|
|
# verify authentication error
|
2020-01-19 19:32:30 +01:00
|
|
|
response = app.post_json('/api/mail/response/', params={}, status=(401, 403))
|
2018-11-07 13:49:31 +01:00
|
|
|
app.authorization = ('Basic', ('test', 'test'))
|
|
|
|
# verify serializer error
|
|
|
|
response = app.post_json('/api/mail/response/', params={}, status=400)
|
|
|
|
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
|
|
|
|
)
|
|
|
|
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
|
|
|
|
)
|
|
|
|
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
|
|
|
|
)
|
|
|
|
assert maarch.requests[0][3] == {
|
|
|
|
'historyMessage': 'coucou',
|
|
|
|
'resId': [1],
|
|
|
|
'status': 'GRC_RESPONSE',
|
|
|
|
}
|
|
|
|
|
2018-05-19 16:38:57 +02:00
|
|
|
|
|
|
|
def test_command_is_noop():
|
|
|
|
from django.core.management import call_command
|
|
|
|
|
|
|
|
call_command('feed_mail_maarch')
|