# 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 . 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')