passerelle/tests/test_atal.py

import base64
import os
from datetime import datetime
from unittest import mock

import pytest
from django.contrib.contenttypes.models import ContentType
from django.utils.http import urlencode
from requests import RequestException
from zeep.exceptions import Fault

from passerelle.apps.atal.models import ATALConnector
from passerelle.base.models import AccessRight, ApiUser
from passerelle.utils.soap import SOAPError


def get_file(filename):
    with open(os.path.join(os.path.dirname(__file__), 'data', 'atal', filename), 'rb') as f:
        return f.read()


@pytest.fixture()
def connector(db):
    api = ApiUser.objects.create(username='all', keytype='', key='')
    connector = ATALConnector.objects.create(base_soap_url='http://example.atal.com/', slug='slug-atal')
    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


def mock_atal_soap_call(monkeypatch, return_value=None, side_effect=None):
    kwargs = {}
    if return_value is not None:
        kwargs['return_value'] = return_value
    if side_effect is not None:
        kwargs['side_effect'] = side_effect
    mock_soap_call = mock.Mock(**kwargs)
    monkeypatch.setattr(ATALConnector, '_soap_call', mock_soap_call)
    return mock_soap_call


class SoapElem:
    def __init__(self, **kwargs):
        for attr, value in kwargs.items():
            setattr(self, attr, value)


REFS = [SoapElem(code='code1', libelle='elem1'), SoapElem(code='code2', libelle='elem2')]


def test_get_thematique(app, connector, monkeypatch):
    return_value = '''<?xml version="1.0" encoding="UTF-8" standalone="no"?><thematiques><thematique id="9" label="Entretien - Maintenance"></thematique>
<thematique id="12" label="Logistique"></thematique><thematique id="11" label="Réparation- Dépannage"></thematique>
<thematique id="13" label="Sinistre"></thematique><thematique id="15" label="Sécurité - Contrôle"></thematique>
<thematique id="10" label="Travaux Neufs"></thematique><thematique id="14" label="Vandalisme"></thematique></thematiques>
'''
    mock_atal_soap_call(monkeypatch, return_value=return_value)
    response = app.get('/atal/slug-atal/get-thematique')
    assert response.json['err'] == 0
    assert len(response.json['data']) == 7
    assert {'id': '9', 'text': 'Entretien - Maintenance'} in response.json['data']
    assert {'id': '14', 'text': 'Vandalisme'} in response.json['data']


def test_get_type_activite(app, connector, monkeypatch):
    mock_soap_call = mock_atal_soap_call(monkeypatch, return_value=REFS)
    response = app.get('/atal/slug-atal/get-type-activite')
    assert response.json == {
        'err': 0,
        'data': [{'text': 'elem1', 'id': 'code1'}, {'text': 'elem2', 'id': 'code2'}],
    }
    call_params = mock_soap_call.call_args.kwargs
    assert call_params['wsdl'] == 'VilleAgileService'
    assert call_params['method'] == 'getTypeActivite'


def test_get_type_de_voie(app, connector, monkeypatch):
    mock_soap_call = mock_atal_soap_call(monkeypatch, return_value=REFS)
    response = app.get('/atal/slug-atal/get-type-de-voie')
    assert response.json == {
        'err': 0,
        'data': [{'text': 'elem1', 'id': 'code1'}, {'text': 'elem2', 'id': 'code2'}],
    }
    call_params = mock_soap_call.call_args.kwargs
    assert call_params['wsdl'] == 'VilleAgileService'
    assert call_params['method'] == 'getTypeDeVoie'


def test_get_types_equipement(app, connector, monkeypatch):
    return_value = """<?xml version="1.0" encoding="UTF-8" standalone="no"?>
    <types>
        <type id="2" label="Espaces Verts"></type>
        <type id="4" label="Voirie">
    </type>
</types>
"""
    mock_soap_call = mock_atal_soap_call(monkeypatch, return_value=return_value)
    response = app.get('/atal/slug-atal/get-types-equipement')
    assert response.json == {
        'err': 0,
        'data': [{'text': 'Espaces Verts', 'id': '2'}, {'text': 'Voirie', 'id': '4'}],
    }
    call_params = mock_soap_call.call_args.kwargs
    assert call_params['wsdl'] == 'VilleAgileService'
    assert call_params['method'] == 'getTypesEquipement'


def test_insert_action_comment(app, connector, monkeypatch):
    mock_soap_call = mock_atal_soap_call(monkeypatch, return_value='DIT19050001')
    params = {'numero_demande': 'DIT19050001', 'commentaire': 'aaa'}
    response = app.post_json('/atal/slug-atal/insert-action-comment', params=params)
    assert response.json == {'err': 0, 'data': {'demande_number': 'DIT19050001'}}
    call_params = mock_soap_call.call_args.kwargs
    assert call_params['wsdl'] == 'DemandeService'
    assert call_params['method'] == 'insertActionComment'
    assert call_params['numeroDemande'] == 'DIT19050001'
    assert call_params['commentaire'] == 'aaa'


def test_insert_demande_complet_by_type(app, connector, monkeypatch):
    mock_soap_call = mock_atal_soap_call(monkeypatch, return_value='DIT19050001')
    params = {'type_demande': 'VOIRIE', 'coord_x': 48.866667, 'coord_y': 2.333333}
    response = app.post_json('/atal/slug-atal/insert-demande-complet-by-type', params=params)
    assert response.json == {'err': 0, 'data': {'demande_number': 'DIT19050001'}}
    call_params = mock_soap_call.call_args.kwargs
    assert call_params['wsdl'] == 'DemandeService'
    assert call_params['method'] == 'insertDemandeCompletByType'
    assert call_params['typeDemande'] == 'VOIRIE'
    assert call_params['coordX'] == 48.866667
    assert call_params['coordY'] == 2.333333


def test_upload(app, connector, monkeypatch):
    mock_soap_call = mock_atal_soap_call(monkeypatch, return_value=None)
    base64_str = 'eyJsYXN0X2NoZWNrIjoiMjAxOS0wNC0xMFQxMjowODoyOVoiL' + 'CJweXBpX3ZlcnNpb24iOiIxOS4wLjMifQ=='
    params = {'numero_demande': 'DIT19050001', 'nom_fichier': 'data.json', 'file': {'content': base64_str}}
    response = app.post_json('/atal/slug-atal/upload', params=params)
    assert response.json == {'err': 0}
    call_params = mock_soap_call.call_args.kwargs
    assert call_params['wsdl'] == 'ChargementPiecesJointesService'
    assert call_params['method'] == 'upload'
    assert call_params['numeroDemande'] == 'DIT19050001'
    assert call_params['nomFichier'] == 'data.json'
    assert call_params['donneesFichier'] == base64.b64decode(base64_str)

    # use filename sub param
    params = {'numero_demande': 'DIT19050001', 'file': {'content': base64_str, 'filename': 'nicefile.json'}}
    response = app.post_json('/atal/slug-atal/upload', params=params)
    assert response.json == {'err': 0}
    call_params = mock_soap_call.call_args.kwargs
    assert call_params['wsdl'] == 'ChargementPiecesJointesService'
    assert call_params['method'] == 'upload'
    assert call_params['numeroDemande'] == 'DIT19050001'
    assert call_params['nomFichier'] == 'nicefile.json'
    assert call_params['donneesFichier'] == base64.b64decode(base64_str)

    params = {
        'numero_demande': 'DIT19050001',
        'nom_fichier': 'data.json',
        'file': {'content': 'invalidbase64'},
    }
    response = app.post_json('/atal/slug-atal/upload', params=params)
    assert response.json == {
        'data': None,
        'err': 1,
        'err_class': 'passerelle.utils.jsonresponse.APIError',
        'err_desc': 'Invalid base64 string',
    }

    # empty file
    params = {'numero_demande': 'DIT19050001', 'nom_fichier': 'data.json', 'file': {}}
    response = app.post_json('/atal/slug-atal/upload', params=params, status=400)
    assert response.json == {
        'data': None,
        'err': 1,
        'err_class': 'passerelle.utils.json.JSONValidationError',
        'err_desc': "file: 'content' is a required property",
    }

    # no file
    params = {'numero_demande': 'DIT19050001', 'nom_fichier': 'data.json'}
    response = app.post_json('/atal/slug-atal/upload', params=params, status=400)
    assert response.json == {
        'data': None,
        'err': 1,
        'err_class': 'passerelle.utils.json.JSONValidationError',
        'err_desc': "'file' is a required property",
    }


def test_upload_file_too_big(app, connector, monkeypatch):
    mock_atal_soap_call(monkeypatch, side_effect=Fault('File too big'))
    base64_str = 'eyJsYXN0X2NoZWNrIjoiMjAxOS0wNC0xMFQxMjowODoyOVoiL' + 'CJweXBpX3ZlcnNpb24iOiIxOS4wLjMifQ=='
    params = {'numero_demande': 'DIT19050001', 'nom_fichier': 'data.json', 'file': {'content': base64_str}}
    response = app.post_json('/atal/slug-atal/upload', params=params)
    assert response.json['err'] == 1
    assert response.json['err_desc'] == 'File too big'


def test_retrieve_details_demande(app, connector, monkeypatch):
    import passerelle.utils

    wsdl_response = mock.Mock(
        content=get_file('DemandeService.wsdl'), status_code=200, headers={'Content-Type': 'text/xml'}
    )
    monkeypatch.setattr(passerelle.utils.Request, 'get', mock.Mock(return_value=wsdl_response))

    api_response = mock.Mock(
        content=get_file('details_demande_response.xml') % b'EN ATTENTE',
        status_code=200,
        headers={'Content-Type': 'text/xml'},
    )
    monkeypatch.setattr(passerelle.utils.Request, 'post', mock.Mock(return_value=api_response))

    response = app.get('/atal/slug-atal/retrieve-details-demande/DIT18050001/')
    assert response.json['err'] == 0
    assert response.json['data']['etatDemande']['description'] == 'EN ATTENTE'


def test_retrieve_etat_travaux(app, connector, monkeypatch):
    import passerelle.utils

    wsdl_response = mock.Mock(
        content=get_file('DemandeService.wsdl'), status_code=200, headers={'Content-Type': 'text/xml'}
    )
    monkeypatch.setattr(passerelle.utils.Request, 'get', mock.Mock(return_value=wsdl_response))

    api_response = mock.Mock(
        content=get_file('etat_travaux_response.xml') % b'travaux pas commences',
        status_code=200,
        headers={'Content-Type': 'text/xml'},
    )
    monkeypatch.setattr(passerelle.utils.Request, 'post', mock.Mock(return_value=api_response))

    response = app.get('/atal/slug-atal/retrieve-etat-travaux/DIT18050001/')
    assert response.json['err'] == 0
    assert response.json['data']['libelle'] == 'travaux pas commences'


def test_infos(app, connector, monkeypatch):
    import passerelle.utils

    wsdl_response = mock.Mock(
        content=get_file('DemandeService.wsdl'), status_code=200, headers={'Content-Type': 'text/xml'}
    )
    monkeypatch.setattr(passerelle.utils.Request, 'get', mock.Mock(return_value=wsdl_response))

    api_response = mock.Mock(
        content=get_file('details_demande_response.xml') % b'EN ATTENTE',
        status_code=200,
        headers={'Content-Type': 'text/xml'},
    )
    monkeypatch.setattr(passerelle.utils.Request, 'post', mock.Mock(return_value=api_response))
    response = app.get('/atal/slug-atal/infos/DIT18050001/')

    assert response.json['err'] == 0
    assert response.json['data']['status'] == 'EN ATTENTE'

    api_response1 = mock.Mock(
        content=get_file('details_demande_response.xml') % b'PRISE EN COMPTE',
        status_code=200,
        headers={'Content-Type': 'text/xml'},
    )
    api_response2 = mock.Mock(
        content=get_file('etat_travaux_response.xml') % b'travaux pas commences',
        status_code=200,
        headers={'Content-Type': 'text/xml'},
    )
    monkeypatch.setattr(
        passerelle.utils.Request, 'post', mock.Mock(side_effect=[api_response1, api_response2])
    )

    response = app.get('/atal/slug-atal/infos/DIT18050001/')
    assert response.json['err'] == 0
    assert response.json['data']['status'] == 'travaux pas commences'

    # User comments in response
    api_response1 = mock.Mock(
        content=get_file('details_demande_response_with_comments.xml'),
        status_code=200,
        headers={'Content-Type': 'text/xml'},
    )
    api_response2 = mock.Mock(
        content=get_file('etat_travaux_response.xml') % b'travaux pas commences',
        status_code=200,
        headers={'Content-Type': 'text/xml'},
    )
    monkeypatch.setattr(
        passerelle.utils.Request, 'post', mock.Mock(side_effect=[api_response1, api_response2])
    )

    response = app.get('/atal/slug-atal/infos/DIT18050001/')
    assert response.json['err'] == 0
    data = response.json['data']
    assert data['status'] == 'travaux pas commences'
    assert data['works_comment'] == {'text': 'bonjour atal', 'date': 'Thursday 24 October 2019, 16:51'}
    assert data['works_comments'] == []
    assert data['demand_comment'] is None
    assert data['works_status'] is None

    # full in query string
    monkeypatch.setattr(
        passerelle.utils.Request, 'post', mock.Mock(side_effect=[api_response1, api_response2])
    )
    response = app.get('/atal/slug-atal/infos/DIT18050001/?full=true')
    data = response.json['data']
    assert len(data['works_comments']) == 2
    assert data['works_comments'][0] == {'text': 'OK', 'date': 'Thursday 24 October 2019, 16:48'}
    last_comment = {'text': 'bonjour atal', 'date': 'Thursday 24 October 2019, 16:51'}
    assert data['works_comments'][1] == last_comment
    assert data['works_comment'] == last_comment


def test_new_comments(app, connector, monkeypatch):
    import passerelle.utils

    wsdl_response = mock.Mock(
        content=get_file('DemandeService.wsdl'), status_code=200, headers={'Content-Type': 'text/xml'}
    )
    monkeypatch.setattr(passerelle.utils.Request, 'get', mock.Mock(return_value=wsdl_response))

    api_response = mock.Mock(
        content=get_file('details_demande_response_with_comments.xml'),
        status_code=200,
        headers={'Content-Type': 'text/xml'},
    )
    monkeypatch.setattr(passerelle.utils.Request, 'post', mock.Mock(return_value=api_response))

    all_comments = [
        {'text': 'OK', 'date': 'Thursday 24 October 2019, 16:48', 'date_raw': '2019-10-24T16:48:34+02:00'},
        {
            'text': 'bonjour atal',
            'date': 'Thursday 24 October 2019, 16:51',
            'date_raw': '2019-10-24T16:51:37+02:00',
        },
    ]

    last_datetime = datetime(year=2019, month=10, day=23)
    params = urlencode({'last_datetime': last_datetime})
    response = app.get('/atal/slug-atal/new-comments/DIT18050001/?%s' % params)
    assert response.json['err'] == 0
    assert response.json['data']['new_comments'] == all_comments
    assert response.json['data']['all_comments'] == all_comments
    assert response.json['data']['last_date'] == '2019-10-24T16:51:37+02:00'

    last_datetime = datetime(year=2019, month=10, day=24, hour=16, minute=49)
    params = urlencode({'last_datetime': last_datetime})
    response = app.get('/atal/slug-atal/new-comments/DIT18050001/?%s' % params)
    assert response.json['err'] == 0
    assert response.json['data']['new_comments'] == all_comments[1:]
    assert response.json['data']['all_comments'] == all_comments
    assert response.json['data']['last_date'] == '2019-10-24T16:51:37+02:00'

    last_datetime = datetime(year=2019, month=10, day=24, hour=16, minute=52)
    params = urlencode({'last_datetime': last_datetime})
    response = app.get('/atal/slug-atal/new-comments/DIT18050001/?%s' % params)
    assert response.json['err'] == 0
    assert response.json['data']['new_comments'] == []
    assert response.json['data']['all_comments'] == all_comments
    assert response.json['data']['last_date'] == '2019-10-24T16:51:37+02:00'


def test_check_status(app, connector, monkeypatch):
    import passerelle.utils

    wsdl_response = mock.Mock(
        content=get_file('DemandeService.wsdl'), status_code=200, headers={'Content-Type': 'text/xml'}
    )

    monkeypatch.setattr(passerelle.utils.Request, 'get', mock.Mock(return_value=wsdl_response))
    connector.check_status()

    monkeypatch.setattr(passerelle.utils.Request, 'get', mock.Mock(side_effect=RequestException))
    with pytest.raises(SOAPError):
        connector.check_status()