passerelle/tests/test_opendatasoft.py

# -*- coding: utf-8 -*-
# passerelle - uniform access to multiple data sources and services
# Copyright (C) 2020  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 mock
import json
import pytest

import utils

from passerelle.apps.opendatasoft.models import OpenDataSoft, Query
from passerelle.utils import import_site

from test_manager import login, admin_user


FAKED_CONTENT_Q_SEARCH = json.dumps({
    "nhits": 76,
    "parameters": {
        "dataset": "referentiel-adresse-test",
        "format": "json",
        "q": "rue de l'aubepine",
        "rows": 3,
        "timezone": "UTC"
    },
    "records": [
        {
            "datasetid": "referentiel-adresse-test",
            "fields": {
                "adresse_complete": "33 RUE DE L'AUBEPINE STRASBOURG",
                "date_exprt": "2019-10-23",
                "geo_point": [
                    48.6060963542,
                    7.76978279836
                ],
                "nom_commun": "Strasbourg",
                "nom_rue": "RUE DE L'AUBEPINE",
                "num_com": 482,
                "numero": "33",
                "source": u"Ville et Eurométropole de Strasbourg"
            },
            "geometry": {
                "coordinates": [
                    7.76978279836,
                    48.6060963542
                ],
                "type": "Point"
            },
            "record_timestamp": "2019-12-02T14:15:08.376000+00:00",
            "recordid": "e00cf6161e52a4c8fe510b2b74d4952036cb3473"
        },
        {
            "datasetid": "referentiel-adresse-test",
            "fields": {
                "adresse_complete": "19 RUE DE L'AUBEPINE LIPSHEIM",
                "date_exprt": "2019-10-23",
                "geo_point": [
                    48.4920620548,
                    7.66177412454
                ],
                "nom_commun": "Lipsheim",
                "nom_rue": "RUE DE L'AUBEPINE",
                "num_com": 268,
                "numero": "19",
                "source": u"Ville et Eurométropole de Strasbourg"
            },
            "geometry": {
                "coordinates": [
                    7.66177412454,
                    48.4920620548
                ],
                "type": "Point"
            },
            "record_timestamp": "2019-12-02T14:15:08.376000+00:00",
            "recordid": "7cafcd5c692773e8b863587b2d38d6be82e023d8"
        },
        {
            "datasetid": "referentiel-adresse-test",
            "fields": {
                "adresse_complete": "29 RUE DE L'AUBEPINE STRASBOURG",
                "date_exprt": "2019-10-23",
                "geo_point": [
                    48.6056497224,
                    7.76988497729
                ],
                "nom_commun": "Strasbourg",
                "nom_rue": "RUE DE L'AUBEPINE",
                "num_com": 482,
                "numero": "29",
                "source": u"Ville et Eurométropole de Strasbourg"
            },
            "geometry": {
                "coordinates": [
                    7.76988497729,
                    48.6056497224
                ],
                "type": "Point"
            },
            "record_timestamp": "2019-12-02T14:15:08.376000+00:00",
            "recordid": "0984a5e1745701f71c91af73ce764e1f7132e0ff"
        }
    ]
})

FAKED_CONTENT_ID_SEARCH = json.dumps({
    "nhits": 1,
    "parameters": {
        "dataset": "referentiel-adresse-test",
        "format": "json",
        "q": "recordid:7cafcd5c692773e8b863587b2d38d6be82e023d8",
        "rows": 1,
        "timezone": "UTC"
    },
    "records": [
        {
            "datasetid": "referentiel-adresse-test",
            "fields": {
                "adresse_complete": "19 RUE DE L'AUBEPINE LIPSHEIM",
                "date_exprt": "2019-10-23",
                "geo_point": [
                    48.4920620548,
                    7.66177412454
                ],
                "nom_commun": "Lipsheim",
                "nom_rue": "RUE DE L'AUBEPINE",
                "num_com": 268,
                "numero": "19",
                u"source": "Ville et Eurométropole de Strasbourg"
            },
            "geometry": {
                "coordinates": [
                    7.66177412454,
                    48.4920620548
                ],
                "type": "Point"
            },
            "record_timestamp": "2019-12-02T14:15:08.376000+00:00",
            "recordid": "7cafcd5c692773e8b863587b2d38d6be82e023d8"
        }
    ]
})


@pytest.fixture
def connector(db):
    return utils.setup_access_rights(OpenDataSoft.objects.create(
        slug='my_connector',
        api_key='my_secret',
    ))


@pytest.fixture
def query(connector):
    return Query.objects.create(
        resource=connector,
        name='Référenciel adresses de test',
        slug='my_query',
        description='Rechercher une adresse',
        dataset='referentiel-adresse-test',
        text_template='{{numero}} {{nom_rue|safe}} {{nom_commun}}',
    )


def test_views(db, admin_user, app, connector):
    app = login(app)
    resp = app.get('/opendatasoft/my_connector/', status=200)
    resp = resp.click('New Query')
    resp.form['name'] = 'my query'
    resp.form['slug'] = 'my-query'
    resp.form['dataset'] = 'my-dataset'
    resp = resp.form.submit()
    resp = resp.follow()
    assert resp.html.find('div', {'id': 'queries'}).ul.li.a.text == 'my query'


def test_export_import(query):
    assert OpenDataSoft.objects.count() == 1
    assert Query.objects.count() == 1
    serialization = {'resources': [query.resource.export_json()]}
    OpenDataSoft.objects.all().delete()
    assert OpenDataSoft.objects.count() == 0
    assert Query.objects.count() == 0
    import_site(serialization)
    assert OpenDataSoft.objects.count() == 1
    assert Query.objects.count() == 1


@mock.patch('passerelle.utils.Request.get')
def test_search_empty_contents(mocked_get, app, connector):
    endpoint = utils.generic_endpoint_url('opendatasoft', 'search', slug=connector.slug)
    assert endpoint == '/opendatasoft/my_connector/search'

    # error returned by opendadasoft server
    json_response = json.dumps({'error': "The query is invalid : Field 00 doesn't exist"})
    mocked_get.return_value = utils.FakedResponse(content=json_response, status_code=200)
    resp = app.get(endpoint, status=200)
    assert resp.json['err']
    assert resp.json['err_desc'] == "The query is invalid : Field 00 doesn't exist"


@mock.patch('passerelle.utils.Request.get')
def test_search_using_q(mocked_get, app, connector):
    endpoint = utils.generic_endpoint_url('opendatasoft', 'search', slug=connector.slug)
    assert endpoint == '/opendatasoft/my_connector/search'
    params = {
        'dataset': 'referentiel-adresse-test',
        'text_template': '{{numero}} {{nom_rue|safe}} {{nom_commun}}',
        'q': "rue de l'aubepine",
        'rows': 3,
    }
    mocked_get.return_value = utils.FakedResponse(content=FAKED_CONTENT_Q_SEARCH, status_code=200)
    resp = app.get(endpoint, params=params, status=200)
    assert not resp.json['err']
    assert len(resp.json['data']) == 3
    # check order is kept
    assert [x['id'] for x in resp.json['data']] == [
            'e00cf6161e52a4c8fe510b2b74d4952036cb3473',
            '7cafcd5c692773e8b863587b2d38d6be82e023d8',
            '0984a5e1745701f71c91af73ce764e1f7132e0ff']
    # check text results
    assert [x['text'] for x in resp.json['data']] == [
            "33 RUE DE L'AUBEPINE Strasbourg",
            "19 RUE DE L'AUBEPINE Lipsheim",
            "29 RUE DE L'AUBEPINE Strasbourg"]
    # check additional attributes
    assert [x['numero'] for x in resp.json['data']] == ['33', '19', '29']


@mock.patch('passerelle.utils.Request.get')
def test_search_using_id(mocked_get, app, connector):
    endpoint = utils.generic_endpoint_url('opendatasoft', 'search', slug=connector.slug)
    assert endpoint == '/opendatasoft/my_connector/search'
    params = {
        'dataset': 'referentiel-adresse-test',
        'text_template': '{{numero}} {{nom_rue|safe}} {{nom_commun}}',
        'id': '7cafcd5c692773e8b863587b2d38d6be82e023d8',
    }
    mocked_get.return_value = utils.FakedResponse(content=FAKED_CONTENT_ID_SEARCH, status_code=200)
    resp = app.get(endpoint, params=params, status=200)
    assert len(resp.json['data']) == 1
    assert resp.json['data'][0]['text'] == "19 RUE DE L'AUBEPINE Lipsheim"


@mock.patch('passerelle.utils.Request.get')
def test_query_q_using_q(mocked_get, app, query):
    endpoint = '/opendatasoft/my_connector/q/my_query/'
    params = {
        'q': "rue de l'aubepine",
        'rows': 3,
    }
    mocked_get.return_value = utils.FakedResponse(content=FAKED_CONTENT_Q_SEARCH, status_code=200)
    resp = app.get(endpoint, params=params, status=200)
    assert not resp.json['err']
    assert len(resp.json['data']) == 3
    # check order is kept
    assert [x['id'] for x in resp.json['data']] == [
            'e00cf6161e52a4c8fe510b2b74d4952036cb3473',
            '7cafcd5c692773e8b863587b2d38d6be82e023d8',
            '0984a5e1745701f71c91af73ce764e1f7132e0ff']
    # check text results
    assert [x['text'] for x in resp.json['data']] == [
            "33 RUE DE L'AUBEPINE Strasbourg",
            "19 RUE DE L'AUBEPINE Lipsheim",
            "29 RUE DE L'AUBEPINE Strasbourg"]
    # check additional attributes
    assert [x['numero'] for x in resp.json['data']] == ['33', '19', '29']


@mock.patch('passerelle.utils.Request.get')
def test_query_q_using_id(mocked_get, app, query):
    endpoint = '/opendatasoft/my_connector/q/my_query/'
    params = {
        'id': '7cafcd5c692773e8b863587b2d38d6be82e023d8',
    }
    mocked_get.return_value = utils.FakedResponse(content=FAKED_CONTENT_ID_SEARCH, status_code=200)
    resp = app.get(endpoint, params=params, status=200)
    assert len(resp.json['data']) == 1
    assert resp.json['data'][0]['text'] == "19 RUE DE L'AUBEPINE Lipsheim"