import base64 import json import pytest import responses from django.contrib.contenttypes.models import ContentType from passerelle.apps.litteralis.models import Litteralis from passerelle.base.models import AccessRight, ApiUser @pytest.fixture() def connector(db): api = ApiUser.objects.create(username='all', keytype='', key='') connector = Litteralis.objects.create( base_url='http://litteralis.invalid/', basic_auth_username='foo', basic_auth_password='bar', slug='slug-litteralis', ) obj_type = ContentType.objects.get_for_model(connector) AccessRight.objects.create( codename='can_access', apiuser=api, resource_type=obj_type, resource_pk=connector.pk ) return connector @pytest.mark.parametrize( 'geom', [ {'type': 'Point', 'coordinates': ['48.866667', '2.333333']}, {'coordinates': [48.866667, 2.333333]}, {'coordinates': ['48,866667', '2,333333']}, ], ) def test_demandes_recues(app, connector, geom): params = { 'fournisseur': 'the-fournisseur', 'idDemande': '12-34', 'dateEnvoi': '2022-10-03T12:43:18.123456+02:00', 'typeModele': 'DA', 'demandeur': { 'raisonSociale': 'the-world-company', 'mail': 'contact@the-world-company.com', 'adresse': { 'rue': 'rue de la république', 'insee': '69034', }, }, 'destinataire': { 'idCollectivite': '1', 'nomCollectivite': 'Malakoff', }, 'geom': geom, 'additionalInformation': { 'typeDemande': 'Stationnement pour travaux', 'dateDebut': '2019-12-04T14:33:13', 'dateFin': '2019-12-09T14:33:13', }, } with responses.RequestsMock() as rsps: rsps.post('http://litteralis.invalid/demandes-recues', status=200, json={'identifier': '1234'}) resp = app.post_json('/litteralis/slug-litteralis/demandes-recues', params=params) json_resp = resp.json assert json_resp['err'] == 0 assert json_resp['data'] == {'identifier': '1234'} assert len(rsps.calls) == 1 req = rsps.calls[0].request assert req.headers['Content-Type'] == 'application/json' json_req = json.loads(req.body) assert json_req['dateEnvoi'] == '2022-10-03T12:43:18+02:00' assert json_req['geom']['coordinates'] == [48.866667, 2.333333] def test_demandes_recues_clean_empty_fields(app, connector): params = { 'fournisseur': 'the-fournisseur', 'idDemande': '12-34', 'dateEnvoi': '2022-10-03T12:43:18.123456+02:00', 'typeModele': 'DA', 'demandeur': { 'raisonSociale': 'the-world-company', 'mail': 'contact@the-world-company.com', 'adresse': { 'rue': 'rue de la république', 'insee': '69034', }, }, 'destinataire': { 'idCollectivite': '1', 'nomCollectivite': 'Malakoff', }, 'geom': {'type': 'Point', 'coordinates': ['48.866667', '2.333333']}, 'additionalInformation': { 'typeDemande': 'Stationnement pour travaux', 'dateDebut': '2019-12-04T14:33:13', 'dateFin': '2019-12-09T14:33:13', 'dateXX': '', 'foo': {'bar': ''}, 'baz': {'who': 'what', 'which': ''}, }, } with responses.RequestsMock() as rsps: rsps.post('http://litteralis.invalid/demandes-recues', status=200, json={'identifier': '1234'}) resp = app.post_json('/litteralis/slug-litteralis/demandes-recues', params=params) json_resp = resp.json assert json_resp['err'] == 0 assert len(rsps.calls) == 1 req = rsps.calls[0].request json_req = json.loads(req.body) assert 'dateXX' not in json_req['additionalInformation'] assert 'foo' not in json_req['additionalInformation'] assert json_req['additionalInformation']['baz'] == {'who': 'what'} @pytest.mark.parametrize( 'geom, err_desc', [ (None, "geom: None is not of type 'object'"), ( {'type': 'Bubble', 'coordinates': ['48.866667', '2.333333']}, "geom/type: 'Bubble' is not one of ['Point']", ), ({'coordinates': None}, "geom/coordinates: None is not of type 'array'"), ({'coordinates': []}, 'geom/coordinates: [] is too short'), ( {'coordinates': ['4', '2', '42']}, "geom/coordinates: ['4', '2', '42'] is too long", ), ({'coordinates': ['plop', '2.333333']}, "geom/coordinates/0: 'plop' is not a number"), ({'coordinates': ['48.866667', '']}, "geom/coordinates/1: '' is not a number"), ], ) def test_demandes_recues_bad_coordinate(app, connector, geom, err_desc): params = { 'fournisseur': 'the-fournisseur', 'idDemande': '12-34', 'dateEnvoi': '2022-10-03T12:43:18.123456+02:00', 'typeModele': 'DA', 'demandeur': { 'raisonSociale': 'the-world-company', 'mail': 'contact@the-world-company.com', 'adresse': { 'rue': 'rue de la république', 'insee': '69034', }, }, 'destinataire': { 'idCollectivite': '1', 'nomCollectivite': 'Malakoff', }, 'geom': geom, 'additionalInformation': { 'typeDemande': 'Stationnement pour travaux', 'dateDebut': '2019-12-04T14:33:13', 'dateFin': '2019-12-09T14:33:13', }, } resp = app.post_json('/litteralis/slug-litteralis/demandes-recues', params=params, status=400) json_resp = resp.json assert json_resp['err'] == 1 assert 'Error' in json_resp['err_class'] assert json_resp['err_desc'] == err_desc def test_upload(app, connector): params = { 'file': { 'filename': 'bla', 'content': base64.b64encode(b'who what').decode(), 'content_type': 'text/plain', }, 'id_demande': '1234', } with responses.RequestsMock() as rsps: rsps.post('http://litteralis.invalid/demandes-recues/1234/upload', status=200, body='') resp = app.post_json('/litteralis/slug-litteralis/upload', params=params) assert len(rsps.calls) == 1 assert rsps.calls[0].request.headers['Content-Type'].startswith('multipart/form-data') json_resp = resp.json assert json_resp['err'] == 0 assert json_resp['data'] == '' def test_annexes(app, connector): params = { 'file': { 'filename': 'bla', 'content': base64.b64encode(b'who what').decode(), 'content_type': 'text/plain', }, 'id_demande': '1234', } with responses.RequestsMock() as rsps: rsps.post('http://litteralis.invalid/demandes-recues/1234/annexes', status=200, body='') resp = app.post_json('/litteralis/slug-litteralis/annexes', params=params) assert len(rsps.calls) == 1 assert rsps.calls[0].request.headers['Content-Type'].startswith('multipart/form-data') json_resp = resp.json assert json_resp['err'] == 0 assert json_resp['data'] == '' def test_annexes_json(app, connector): params = { 'file': { 'filename': 'bla', 'content': base64.b64encode(b'who what').decode(), 'content_type': 'text/plain', }, 'id_demande': '1234', } with responses.RequestsMock() as rsps: data = {'id': 1234, 'nom': 'PJ.pdf', 'mimeType': 'application/pdf', 'taille': 39114} rsps.post('http://litteralis.invalid/demandes-recues/1234/annexes', status=200, json=data) resp = app.post_json('/litteralis/slug-litteralis/annexes', params=params) assert len(rsps.calls) == 1 assert rsps.calls[0].request.headers['Content-Type'].startswith('multipart/form-data') json_resp = resp.json assert json_resp['err'] == 0 assert json_resp['data'] == data def test_demandes_recues_reponses(app, connector): with responses.RequestsMock() as rsps: rsps.get('http://litteralis.invalid/demandes-recues/1234/reponses', status=200, json={'foo': 'bar'}) resp = app.get('/litteralis/slug-litteralis/demandes-recues-reponses?id_demande=1234') assert len(rsps.calls) == 1 json_resp = resp.json assert json_resp['err'] == 0 assert json_resp['data'] == {'foo': 'bar'} def test_demandes_recues_arrete(app, connector): with responses.RequestsMock() as rsps: headers = {'Content-Type': 'text/plain', 'Content-Disposition': 'attachment; filename=foo.txt'} rsps.get( 'http://litteralis.invalid/demandes-recues/1234/arrete', status=200, body='who hwat', headers=headers, ) resp = app.get('/litteralis/slug-litteralis/demandes-recues-arrete?id_demande=1234') assert len(rsps.calls) == 1 assert resp.text == 'who hwat' assert resp.headers['Content-Type'] == 'text/plain' assert resp.headers['Content-Disposition'] == 'attachment; filename=foo.txt' def test_demandes_recues_arrete_error_with_json(app, connector): with responses.RequestsMock() as rsps: rsps.get('http://litteralis.invalid/demandes-recues/1234/arrete', status=403, json={'foo': 'bar'}) resp = app.get('/litteralis/slug-litteralis/demandes-recues-arrete?id_demande=1234') assert len(rsps.calls) == 1 json_resp = resp.json assert json_resp['err'] == 1 assert json_resp['data'] == {'foo': 'bar'} def test_demandes_recues_arrete_error_with_no_json(app, connector): with responses.RequestsMock() as rsps: rsps.get( 'http://litteralis.invalid/demandes-recues/1234/arrete', status=403, body='something went wrong' ) resp = app.get('/litteralis/slug-litteralis/demandes-recues-arrete?id_demande=1234') assert len(rsps.calls) == 1 json_resp = resp.json assert json_resp['err'] == 1 assert json_resp['err_desc'].startswith('403 Client Error')