passerelle/tests/test_signal_arretes.py

# passerelle - uniform access to multiple data sources and services
# Copyright (C) 2022  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
from base64 import standard_b64encode
from unittest.mock import patch

import pytest
from httmock import HTTMock, response, urlmatch

from passerelle.apps.signal_arretes.models import SignalArretes
from tests.utils import generic_endpoint_url, setup_access_rights


@pytest.fixture()
def connector(db):
    return setup_access_rights(SignalArretes.objects.create(slug='test', base_url='http://sa.net'))


@urlmatch(netloc='^sa.net$', path='^/CreationDemandeService.svc/GetCommunes')
def mock_get_communes(url, request):
    return response(
        200, json.dumps({'GetCommunesResult': json.dumps(['Clapotis Les Canards', 'Grosboule Les Bains'])})
    )


@urlmatch(netloc='^sa.net$', path='^/CreationDemandeService.svc/GetVoies')
def mock_get_voies(url, request):
    assert json.loads(request.body) == {'Commune': 'Clapotis Les Canards'}
    return response(200, json.dumps({'GetVoiesResult': json.dumps(['Rue Sacco', 'Rue Vanzetti'])}))


@urlmatch(netloc='^sa.net$', path='^/CreationDemandeService.svc/GetNaturesOccupation')
def mock_get_natures_occupation(url, request):
    return response(200, json.dumps({'GetNaturesOccupationResult': json.dumps(['Déménagement', 'Festival'])}))


@urlmatch(netloc='^sa.net$', path='^/CreationDemandeService.svc/CreationDODP')
def mock_creation_dodp(url, request):
    data = json.loads(request.body)
    contact = data['contact']
    localization = data['localisation']

    assert data['organisationDeclarante'] == 'ACME'
    assert data['qualite'] == 'Entreprise'
    assert contact['civilite'] == 'MADAME'
    assert contact['nom'] == 'John'
    assert contact['prenom'] == 'Doe'
    assert contact['email'] == 'john@doe.net'
    assert localization['nomVoie'] == 'Sesame Street'
    assert localization['commune'] == 'Melun'
    assert localization['natureOccupation'] == 'Base de vie'
    assert localization['dateDebut'] == '2022-06-02'
    assert localization['dateFin'] == '2022-06-03'

    assert 'SIRET' not in data or data['SIRET'] == '00000000000000'
    assert 'numeroDossier' not in data or data['numeroDossier'] == 'reference_dossier'
    assert 'commentaire' not in data or data['commentaire'] == 'Wubba Lubba Dub Dub'
    assert 'adresseLigne1' not in contact or contact['adresseLigne1'] == '6 Sesame street'
    assert 'CP' not in contact or contact['CP'] == '42 42420'
    assert 'ville' not in contact or contact['ville'] == 'Melun'
    assert 'telephone' not in contact or contact['telephone'] == '0636656565'
    assert 'numeroVoie' not in localization or localization['numeroVoie'] == '10'

    return response(200, json.dumps(json.dumps({'D0000_DOT': 'Enregistré'})))


@urlmatch(netloc='^sa.net$', path='^/CreationDemandeService.svc/GetStatutDemande/test_request_id')
def mock_get_statut_demande(url, request):
    return response(200, json.dumps({'GetStatutDemandeResult': json.dumps('Enregistré')}))


DOCUMENT_CONTENT = b'Test file content'


@urlmatch(netloc='^sa.net$', path='^/CreationDemandeService.svc/GetDocumentDemande/.*')
def mock_get_document_demande(url, request):
    if url.path.endswith('unavailable'):
        return response(
            200,
            json.dumps({'GetDocumentDemandeResult': json.dumps("fichier indisponible")}),
        )

    if url.path.endswith('corrupted'):
        content = '$$$$$$$'
    else:
        content = standard_b64encode(DOCUMENT_CONTENT).decode('utf-8')

    return response(
        200,
        json.dumps(
            {
                'GetDocumentDemandeResult': json.dumps(
                    {'contentType': 'text/test-data', 'name': 'test_filename', 'content': content}
                )
            }
        ),
    )


@pytest.fixture
def mock_signal_arretes():
    with HTTMock(
        mock_creation_dodp,
        mock_get_communes,
        mock_get_voies,
        mock_get_natures_occupation,
        mock_get_statut_demande,
        mock_get_document_demande,
    ) as mock:
        yield mock


def test_cities(app, connector, mock_signal_arretes):
    endpoint = generic_endpoint_url('signal-arretes', 'cities', slug=connector.slug)

    result = app.get(endpoint)
    assert result.json['data'] == [
        {'id': 'clapotis-les-canards', 'text': 'Clapotis Les Canards'},
        {'id': 'grosboule-les-bains', 'text': 'Grosboule Les Bains'},
    ]

    result = app.get(endpoint, params={'id': 'grosboule-les-bains'})
    assert result.json['data'] == [
        {'id': 'grosboule-les-bains', 'text': 'Grosboule Les Bains'},
    ]

    result = app.get(endpoint, params={'q': 'CANARD'})
    assert result.json['data'] == [
        {'id': 'clapotis-les-canards', 'text': 'Clapotis Les Canards'},
    ]

    result = app.get(endpoint, params={'q': 'CANARD', 'id': 'grosboule-les-bains'})
    assert result.json['data'] == []


def test_lanes(app, connector, mock_signal_arretes):
    endpoint = generic_endpoint_url('signal-arretes', 'lanes', slug=connector.slug)

    result = app.get(endpoint, params={'city': 'Clapotis Les Canards'})
    assert result.json['data'] == [
        {'id': 'rue-sacco', 'text': 'Rue Sacco'},
        {'id': 'rue-vanzetti', 'text': 'Rue Vanzetti'},
    ]

    result = app.get(endpoint, params={'city': 'Clapotis Les Canards', 'id': 'rue-sacco'})
    assert result.json['data'] == [
        {'id': 'rue-sacco', 'text': 'Rue Sacco'},
    ]

    result = app.get(endpoint, params={'city': 'Clapotis Les Canards', 'q': 'VAN'})
    assert result.json['data'] == [
        {'id': 'rue-vanzetti', 'text': 'Rue Vanzetti'},
    ]

    result = app.get(endpoint, params={'city': 'Clapotis Les Canards', 'q': 'VAN', 'id': 'rue-sacco'})
    assert result.json['data'] == []


def test_occupation_types(app, connector, mock_signal_arretes):
    endpoint = generic_endpoint_url('signal-arretes', 'occupation_types', slug=connector.slug)

    result = app.get(endpoint)
    assert result.json['data'] == [
        {'id': 'demenagement', 'text': 'Déménagement'},
        {'id': 'festival', 'text': 'Festival'},
    ]

    result = app.get(endpoint, params={'id': 'demenagement'})
    assert result.json['data'] == [
        {'id': 'demenagement', 'text': 'Déménagement'},
    ]

    result = app.get(endpoint, params={'q': 'esti'})
    assert result.json['data'] == [
        {'id': 'festival', 'text': 'Festival'},
    ]

    result = app.get(endpoint, params={'q': 'esti', 'id': 'demenagement'})
    assert result.json['data'] == []


REQUIRED_PARAMETERS = {
    'declarant_organisation': 'ACME',
    'declarant_siret': '',
    'declarant_quality': 'Entreprise',
    'file_number': '',
    'declarant_civility': 'MADAME',
    'declarant_name': 'John',
    'declarant_surname': 'Doe',
    'declarant_address': '',
    'declarant_zip': '',
    'declarant_city': '',
    'declarant_email': 'john@doe.net',
    'declarant_phone': '',
    'occupation_lane': 'Sesame Street',
    'occupation_number': '10',
    'occupation_city': 'Melun',
    'occupation_type': 'Base de vie',
    'occupation_start_date': '02/06/2022',
    'occupation_end_date': '03/06/2022',
    'comment': '',
}


def test_create_request(app, connector, mock_signal_arretes):
    endpoint = generic_endpoint_url('signal-arretes', 'create_request', slug=connector.slug)
    result = app.post_json(endpoint, params=REQUIRED_PARAMETERS)
    assert result.json == {'request_id': 'D0000_DOT', 'request_status': 'Enregistré', 'err': 0}

    all_parameters = dict(REQUIRED_PARAMETERS)
    all_parameters.update(
        {
            'declarant_siret': '00000000000000',
            'file_number': 'reference_dossier',
            'declarant_address': '6 Sesame street',
            'declarant_zip': '42 42420',
            'declarant_city': 'Melun',
            'declarant_phone': '0636656565',
            'occupation_lane': 'Sesame Street',
            'comment': 'Wubba Lubba Dub Dub',
        }
    )

    result = app.post_json(endpoint, params=all_parameters)
    assert result.json == {'request_id': 'D0000_DOT', 'request_status': 'Enregistré', 'err': 0}


@patch('passerelle.utils.Request.post')
@pytest.mark.parametrize(
    'response_body,error_message',
    [
        ('Not valid json', 'Expected valid json'),
        ('[]', 'Expected a string'),
        ('"Invalid json"', 'Returned string should be valid json'),
        ('"[]"', 'Expected a dictionary with one element'),
        ('"{}"', 'Expected a dictionary with one element'),
    ],
)
def test_create_request_errors(mocked_post, app, connector, response_body, error_message):
    endpoint = generic_endpoint_url('signal-arretes', 'create_request', slug=connector.slug)
    mocked_post.return_value = response(200, response_body)
    result = app.post_json(endpoint, params=REQUIRED_PARAMETERS)

    assert 'err' in result.json
    assert result.json['err']
    assert error_message in result.json['err_desc']


def test_request_status(app, connector, mock_signal_arretes):
    endpoint = generic_endpoint_url('signal-arretes', 'request_status', slug=connector.slug)
    result = app.get(endpoint, params={'request_id': 'test_request_id'})

    assert not result.json['err']
    assert result.json['request_status'] == 'Enregistré'


def test_request_document(app, connector, mock_signal_arretes):
    endpoint = generic_endpoint_url('signal-arretes', 'request_document', slug=connector.slug)

    result = app.get(endpoint, params={'request_id': 'document'})

    assert result.headers['Content-Type'] == 'text/test-data'
    assert result.headers['Content-Disposition'] == 'attachment; filename="test_filename"'
    assert result.body == DOCUMENT_CONTENT

    result = app.get(endpoint, params={'request_id': 'document_corrupted'})

    assert 'err' in result.json
    assert result.json['err']
    assert 'Corrupted base64 content' in result.json['err_desc']

    result = app.get(endpoint, params={'request_id': 'document_unavailable'})

    assert 'err' in result.json
    assert result.json['err']
    assert 'fichier indisponible' in result.json['err_desc']


@patch('passerelle.utils.Request.get')
@pytest.mark.parametrize(
    'status_code,body,expected_message',
    [
        (
            400,
            'Le message d\'exception est \'Test error message\'. Pour plus d\'informations thanks a lot to use HTML as return of a json api.',
            'An error occured during the request to Signal Arrêtés: Test error message',
        ),
        (500, 'Unmatched message', 'An error occured during the request to Signal Arrêtés'),
        (200, 'Invalid json', 'Expected valid json'),
        (200, '[]', 'Expected a dictionary with a GetCommunesResult key'),
        (200, '{}', 'Expected a dictionary with a GetCommunesResult key'),
        (200, '{"GetCommunesResult": "Invalid json"}', 'Expected valid json string at GetCommunesResult key'),
        (200, '{"GetCommunesResult": "{}"}', 'Expected a list'),
    ],
)
def test_error_handling(mocked_get, app, connector, status_code, body, expected_message):
    endpoint = generic_endpoint_url('signal-arretes', 'cities', slug=connector.slug)
    mocked_get.return_value = response(status_code, body)
    result = app.get(endpoint)

    assert 'err' in result.json
    assert result.json['err']
    assert expected_message in result.json['err_desc']