welco/tests/test_source_maarch.py

289 lines
9.1 KiB
Python

# 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
from django.contrib.auth.models import User
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):
self.requests.append(('list_endpoint', url, request, json.loads(request.body)))
return {
'content': json.dumps(self.next_response()),
'headers': {
'content-type': 'application/json',
},
}
list_endpoint.path = '^/rest/res/list$'
def update_external_infos(self, url, request):
self.requests.append(('update_external_infos', url, request, json.loads(request.body)))
return json.dumps({})
update_external_infos.path = '^/rest/res/externalInfos$'
def update_status(self, url, request):
self.requests.append(('update_status', url, request, json.loads(request.body)))
return {
'content': json.dumps(self.next_response()),
'headers': {
'content-type': 'application/json',
},
}
update_status.path = '^/rest/res/resource/status$'
def post_courrier(self, url, request):
self.requests.append(('post_courrier', url, request, json.loads(request.body)))
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',
}]
})
api_formdefs.path = '^/api/formdefs/$'
def json(self, url, request):
return json.dumps({
'data': [{
'slug': 'slug1',
'title': 'title1',
'category': 'category1',
}]
})
json.path = '^/json$'
def api_formdefs_slug1_schema(self, url, request):
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',
},
})
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'
PDF_MOCK = '%PDF-1.4 ...'
def test_feed(settings, app, maarch, wcs, user):
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,
'fileBase64Content': base64.b64encode(PDF_MOCK),
}
],
})
# 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': [
{
'external_id': '1',
'external_link': 'http://wcs.example.net/slug-1/1',
'res_id': 1,
}
]
}
# 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
response = app.post_json('/api/mail/response/', params={}, status=403)
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',
}
def test_command_is_noop():
from django.core.management import call_command
call_command('feed_mail_maarch')