passerelle/tests/test_cityweb.py

# Passerelle - uniform access to data and services
# Copyright (C) 2017  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; exclude 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.deepcopy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import json
import os
import shutil
import stat

import pytest
from django.core.files.storage import default_storage
from lxml import etree
from lxml import objectify as xobject

import tests.utils
from passerelle.apps.cityweb.cityweb import DateType
from passerelle.apps.cityweb.models import CityWeb
from passerelle.utils import SFTP
from passerelle.utils.jsonresponse import APIError


def get_test_base_dir(name):
    return os.path.join(os.path.dirname(__file__), 'data', name)


def get_file_from_test_base_dir(filename):
    path = os.path.join(get_test_base_dir('cityweb'), filename)
    with open(path) as fd:
        return fd.read()


@pytest.fixture
def setup(db):
    return tests.utils.setup_access_rights(CityWeb.objects.create(slug='test'))


PAYLOAD = [
    {'birth': json.loads(get_file_from_test_base_dir('payload_birth.json'))},
    {'mariage': json.loads(get_file_from_test_base_dir('payload_mariage.json'))},
    {'death': json.loads(get_file_from_test_base_dir('payload_death.json'))},
]


@pytest.fixture(params=PAYLOAD)
def payload(request):
    return request.param


def assert_xml_doc(filename, assertions):
    with open(os.path.join(get_test_base_dir('cityweb'), 'cityweb.xsd')) as fd:
        schema = etree.XMLSchema(etree.parse(fd))

    with open(filename) as fd:
        content = fd.read()
    xml_content = etree.fromstring(content)
    assert len(xml_content.nsmap) == 1
    assert xml_content.nsmap['xs'] == "http://tempuri.org/XMLSchema.xsd"
    schema.assertValid(xml_content)
    root = xobject.fromstring(content)
    for epath, value in assertions.items():
        path = xobject.ObjectPath('demandeEtatCivil.%s' % epath)
        assert path.find(root) == value


def test_demand_creation(app, setup, payload):
    url = '/cityweb/test/create/'
    if 'birth' in payload:
        response = app.post_json(url, params=payload['birth'])
        assert response.json['data']['demand_id'] == 'N201610154'
        filename = default_storage.path(os.path.join('cityweb', 'test', 'N201610154.xml'))
        assertions = {
            'identifiant': 'N201610154',
            'demandeur.qualiteDemandeur': 'concerne',
            'demandeur.individu.noms.nomDeFamille': 'Whatever',
            'demandeur.individu.prenoms': 'Kim Chelsea',
            'demandeur.individu.adresse.ligneAdr1': "37 Rue de l'Aigle Blanc",
            'demandeur.individu.adresse.codePostal': 54000,
            'demandeur.individu.adresse.lieu.ville': 'Nancy',
            'demandeur.individu.adresse.lieu.pays': 'France',
            'demandeur.individu.adresse.mail': 'chelsea@whatever.com',
            'natureDocument': 'CPI',
            'nbExemplaire': 1,
            'dateDemande': '2016-10-20T14:41:20Z',
            'evenement.natureEvenement': 'NAI',
            'evenement.dateEvenement.dateDebut': '2012-07-14',
            'evenement.lieuEvenement.ville': 'Nancy',
            'evenement.interesse.pere.noms.nomDeFamille': 'Smith',
            'evenement.interesse.pere.prenoms': 'John Oliver',
            'evenement.interesse.mere.noms.nomDeFamille': 'Smith',
            'evenement.interesse.mere.prenoms': 'Kim',
        }
        assert bool(os.stat(filename).st_mode & stat.S_IRUSR)
        assert bool(os.stat(filename).st_mode & stat.S_IRGRP)
        assert_xml_doc(filename, assertions)

    elif 'mariage' in payload:
        response = app.post_json(url, params=payload['mariage'])
        assert response.json['data']['demand_id'] == 'M201610161'
        filename = default_storage.path(os.path.join('cityweb', 'test', 'M201610161.xml'))
        assertions = {
            'identifiant': 'M201610161',
            'demandeur.qualiteDemandeur': 'concerne',
            'demandeur.individu.noms.nomDeFamille': 'Whatever',
            'demandeur.individu.prenoms': 'Kim Chelsea',
            'demandeur.individu.adresse.ligneAdr1': "37 Rue de l'Aigle Blanc",
            'demandeur.individu.adresse.codePostal': 54000,
            'demandeur.individu.adresse.lieu.ville': 'Nancy',
            'demandeur.individu.adresse.lieu.pays': 'France',
            'demandeur.individu.adresse.mail': 'chelsea@whatever.com',
            'natureDocument': 'CPI',
            'nbExemplaire': 1,
            'dateDemande': '2016-10-20T14:41:20Z',
            'evenement.natureEvenement': 'MAR',
            'evenement.dateEvenement.dateDebut': '2012-07-14',
            'evenement.lieuEvenement.ville': 'Nancy',
            'evenement.interesse.pere.noms.nomDeFamille': 'Smith',
            'evenement.interesse.pere.prenoms': 'John Oliver',
            'evenement.interesse.mere.noms.nomDeFamille': 'Smith',
            'evenement.interesse.mere.prenoms': 'Kim',
            'evenement.conjoint.noms.nomDeFamille': 'Contrao',
            'evenement.conjoint.pere.noms.nomDeFamille': 'Scaramucci',
            'evenement.conjoint.pere.prenoms': 'Antonio',
            'evenement.conjoint.mere.noms.nomDeFamille': 'Scaramucci',
            'evenement.conjoint.mere.prenoms': 'Marguerite',
        }
        assert bool(os.stat(filename).st_mode & stat.S_IRUSR)
        assert bool(os.stat(filename).st_mode & stat.S_IRGRP)
        assert_xml_doc(filename, assertions)
    else:
        response = app.post_json(url, params=payload['death'])
        assert response.json['data']['demand_id'] == 'D201610171'
        filename = default_storage.path(os.path.join('cityweb', 'test', 'D201610171.xml'))
        assertions = {
            'identifiant': 'D201610171',
            'demandeur.qualiteDemandeur': 'concerne',
            'demandeur.individu.noms.nomDeFamille': 'Whatever',
            'demandeur.individu.prenoms': 'Kim Chelsea',
            'demandeur.individu.adresse.ligneAdr1': "37 Rue de l'Aigle Blanc",
            'demandeur.individu.adresse.codePostal': 54000,
            'demandeur.individu.adresse.lieu.ville': 'Nancy',
            'demandeur.individu.adresse.lieu.pays': 'France',
            'demandeur.individu.adresse.mail': 'chelsea@whatever.com',
            'natureDocument': 'EXTSF',
            'nbExemplaire': 1,
            'dateDemande': '2016-10-20T14:41:20Z',
            'evenement.natureEvenement': 'DEC',
            'evenement.dateEvenement.dateDebut': '2012-07-14',
            'evenement.lieuEvenement.ville': 'Nancy',
        }
        assert bool(os.stat(filename).st_mode & stat.S_IRUSR)
        assert bool(os.stat(filename).st_mode & stat.S_IRGRP)
        assert_xml_doc(filename, assertions)


def test_date_type_parsing():
    class BirthDate(DateType):
        tagname = 'date'

    with pytest.raises(APIError, match=r'Invalid date \(toto\) for <date>'):
        BirthDate('toto')

    with pytest.raises(
        APIError, match=r'Invalid date \(2017-02-30\) for <date>: day is out of range for month'
    ):
        BirthDate('2017-02-30')

    assert BirthDate('2017-02-22').value == '2017-02-22'


def test_datasource_titles(app, setup):
    response = app.get('/cityweb/test/titles/')
    data = response.json['data']
    assert len(data) == 3
    for datum in data:
        if datum['id'] == 'M':
            assert datum['text'] == 'Monsieur'
        elif datum['id'] == 'Mme':
            assert datum['text'] == 'Madame'
        else:
            assert datum['id'] == 'Mlle'
            assert datum['text'] == 'Mademoiselle'


def test_datasource_sexes(app, setup):
    response = app.get('/cityweb/test/sexes/')
    data = response.json['data']
    assert len(data) == 3
    for datum in data:
        if datum['id'] == 'M':
            assert datum['text']
        elif datum['id'] == 'F':
            assert datum['text'] == 'Femme'
        else:
            assert datum['id'] == 'NA'
            assert datum['text'] == 'Autre'


def test_datasource_concerned(app, setup):
    response = app.get('/cityweb/test/concerned/')
    data = response.json['data']
    assert len(data) == 2
    for datum in data:
        if datum['id'] == 'reconnu':
            assert datum['text'] == 'Reconnu'
        else:
            assert datum['id'] == 'auteur'
            assert datum['text'] == 'Auteur'


def test_datasource_origins(app, setup):
    response = app.get('/cityweb/test/origins/')
    data = response.json['data']
    assert len(data) == 3
    for datum in data:
        if datum['id'] == 'internet':
            assert datum['text'] == 'Internet'
        elif datum['id'] == 'guichet':
            assert datum['text'] == 'Guichet'
        else:
            assert datum['id'] == 'courrier'
            assert datum['text'] == 'Courrier'


def test_datasource_document_types(app, setup):
    response = app.get('/cityweb/test/document-types/')
    data = response.json['data']
    assert len(data) == 4
    for datum in data:
        if datum['id'] == 'CPI':
            assert datum['text'] == 'Copie intégrale'
        elif datum['id'] == 'EXTAF':
            assert datum['text'] == 'Extrait avec filiation'
        elif datum['id'] == 'EXTSF':
            assert datum['text'] == 'Extrait sans filiation'
        else:
            assert datum['id'] == 'EXTPL'
            assert datum['text'] == 'Extrait plurilingue'

    params = {'exclude': 'EXTAF,EXTSF,EXTPL'}
    response = app.get('/cityweb/test/document-types/', params=params)
    data = response.json['data']
    assert len(data) == 1
    assert data[0]['id'] == 'CPI'
    assert data[0]['text'] == 'Copie intégrale'


def test_datasource_certificate_types(app, setup):
    response = app.get('/cityweb/test/certificate-types/')
    data = response.json['data']
    assert len(data) == 4
    for datum in data:
        if datum['id'] == 'NAI':
            assert datum['text'] == 'Naissance'
        elif datum['id'] == 'MAR':
            assert datum['text'] == 'Mariage'
        elif datum['id'] == 'REC':
            assert datum['text'] == 'Reconnaissance'
        else:
            assert datum['id'] == 'DEC'
            assert datum['text'] == 'Décès'


def teardown_module(module):
    shutil.rmtree(os.path.join(get_test_base_dir('cityweb'), 'test'), ignore_errors=True)


def test_demand_creation_with_remote_sftp(app, setup, payload, sftpserver):
    setup.incoming_sftp = SFTP('sftp://foo:bar@{server.host}:{server.port}/input/'.format(server=sftpserver))
    setup.save()
    for payload_value in payload.values():
        resp = app.post_json('/cityweb/test/create', params=payload_value, status=200)
        expected_filename = '%s.xml' % resp.json['data']['demand_id']
        with sftpserver.serve_content({'input': {expected_filename: 'content'}}):
            setup.jobs()