Produits Entr'ouvert » Passerelle

from django.core.files import File

from django.utils import six

from httmock import HTTMock, remember_called, response, urlmatch

if six.PY2:

    collect_ignore = ['wcs/']

import stat

import pytest

from django.utils import six

from passerelle.apps.actesweb.models import ActesWeb

@pytest.fixture

def actesweb(db):

PAYLOAD = [

import mock

import pytest

from django.core.files import File

from django.utils.six import StringIO

from passerelle.apps.base_adresse.models import BaseAdresse

from passerelle.apps.sector.models import SectorResource

@pytest.fixture

def sector(db):

        SectorResource.objects.create(

            slug='ecole',

            title='Secteur scolaire',

@pytest.fixture

def base_adresse(db):

@mock.patch('passerelle.utils.Request.get')

def test_sectorization(mocked_get, app, base_adresse, sector):

    assert base_adresse.sectors.count() == 0

    addresses = app.get(url, params={'q': 'foo'}, status=200).json

    assert len(addresses) == 8

    for address in addresses:

    assert addresses[7]['sectors'] == {}

    assert addresses[7]['sectorization_error'] == 'missing street_id in address'

    for params in (None, {'id': 'ecole'}, {'q': 'scolaire'}):

        sectors = app.get(url, params=params, status=200).json

        assert sectors['err'] == 0

import mock

import pytest

from passerelle.contrib.adict.models import Adict

FAKE_FEATURE_INFO = json.dumps(

@pytest.fixture

def connector(db):

        Adict.objects.create(

            slug='test',

            service_root_url='http://adict.example.net/',

@mock.patch('passerelle.utils.Request.get')

def test_feature_info(mocked_get, app, connector):

    assert endpoint == '/adict/test/feature_info'

    resp = app.get(endpoint, params={'lat': '48.570472877', 'lon': '7.75659804'})

    assert mocked_get.call_args[0][0] == 'http://adict.example.net/api/v1.0/secteurs'

    assert mocked_get.call_args[1]['params']['x'] == '7.75659804'

@mock.patch('passerelle.utils.Request.get')

def test_no_feature_info(mocked_get, app, connector):

    assert endpoint == '/adict/test/feature_info'

    resp = app.get(endpoint, params={'lat': '48.570472877', 'lon': '7.75659804'})

    assert resp.json['err'] == 1

    assert resp.json['err_msg'] == 'not found'

import freezegun

import pytest

from httmock import HTTMock, response

from passerelle.apps.airquality.models import AirQuality

SAMPLE_RESPONSE = {

@freezegun.freeze_time('2020-12-21')

def test_airquality_details(app, airquality):

    assert endpoint == '/airquality/atmo/details'

    with HTTMock(mocked_http):

        resp = app.get(endpoint + '/fr/lyon/', status=200)

def test_airquality_details_unknown_city(app, airquality):

    resp = app.get(endpoint + '/fr/paris/', status=404)

from django.contrib.contenttypes.models import ContentType

from django.test import override_settings

from django.urls import reverse

from passerelle.apps.ovh.models import OVHSMSGateway

from passerelle.base.models import AccessRight, ApiUser, Job

pytestmark = pytest.mark.django_db

import sys

import pytest

from django.contrib.auth.models import User

from django.contrib.contenttypes.models import ContentType

from django.test import override_settings

from django.urls import reverse

from passerelle.apps.oxyd.models import OxydSMSGateway

from passerelle.base import signature

from passerelle.base.models import AccessRight, ApiUser, ResourceLog

import requests

from django.utils import timezone

from httmock import HTTMock, response, urlmatch

from passerelle.apps.api_entreprise.models import APIEntreprise

ETABLISSEMENTS_RESPONSE = {

    "etablissement": {

import requests

from django.urls import reverse

from httmock import HTTMock, response, urlmatch

from passerelle.apps.api_particulier.models import APIParticulier

from passerelle.base.models import ResourceLog

SVAIR_RESPONSE = {

    "declarant1": {

import mock

import pytest

from django.contrib.auth.models import User

from django.contrib.contenttypes.models import ContentType

from django.core.exceptions import ValidationError

from passerelle.apps.arcgis.models import ArcGIS, Query, SqlFormatter, validate_where

from passerelle.base.models import AccessRight, ApiUser

from passerelle.utils import import_site

pytestmark = pytest.mark.django_db

def test_arcgis_mapservice_query(app, arcgis):

    assert endpoint == '/arcgis/test/mapservice-query'

    params = {'folder': 'fold', 'service': 'serv', 'layer': '1'}

    with mock.patch('passerelle.utils.Request.get') as requests_get:

        resp = app.get(endpoint, params=params, status=403)

        assert requests_get.call_count == 0

    # call errors

    with mock.patch('passerelle.utils.Request.get') as requests_get:

        resp = app.get(endpoint, params={}, status=400)

        assert requests_get.call_count == 0

        assert resp.json['err'] == 1

        'meta': {},

    }

    with mock.patch('passerelle.utils.Request.get') as requests_get:

        assert query.q(rf.get('/', data={'adress': "AVENUE D'ANNAM"}), full=True) == {

            "data": [

                {

def test_q_endpoint(arcgis, query, app):

    assert endpoint == '/arcgis/test/q/adresses/'

    with mock.patch('passerelle.utils.Request.get') as requests_get:

        resp = app.get(endpoint, params={}, status=403)

        assert requests_get.call_count == 0

    assert arcgis.base_url == 'https://arcgis.example.net/'

    with mock.patch('passerelle.utils.Request.get') as requests_get:

        resp = app.get('/arcgis/test/tile/layer1/13/4258/2991.png')

        assert requests_get.call_count == 1

    # test layer and folders

    with mock.patch('passerelle.utils.Request.get') as requests_get:

        resp = app.get('/arcgis/test/tile/layer1/foo/bar/13/4258/2991.png')

        assert requests_get.call_count == 1

    arcgis.base_url = 'https://arcgis.example.net'

    arcgis.save()

    with mock.patch('passerelle.utils.Request.get') as requests_get:

        resp = app.get('/arcgis/test/tile/layer1/13/4258/2991.png')

        assert requests_get.call_count == 1

import mock

import pytest

from django.urls import reverse

from passerelle.apps.arpege_ecp.models import ArpegeECP

from passerelle.utils.jsonresponse import APIError

    resource = ArpegeECP.objects.create(

        slug='test', webservice_base_url='http://arpege.net', hawk_auth_id='id', hawk_auth_key='secret'

    )

@mock.patch('passerelle.utils.Request.get')

def test_check_status(mocked_get, connector):

    resp = connector.check_status()

    assert resp['data'] == u'InteropAPI v1 (c) Arpège 2017'

def test_check_status(mocked_get, connector):

    hello_response = json.loads(FAKE_HELLO_RESPONSE)

    del hello_response['Data']

    with pytest.raises(Exception) as error:

        resp = connector.check_status()

    assert str(error.value) == 'Invalid credentials'

def test_get_access_token(connector):

        token = connector.get_access_token('nameid')

    assert token == '0f86353f2d87b8b78aaaacc2ecc763e287ded44f773289a5e336546a251718b3'

        with pytest.raises(APIError) as error:

            token = connector.get_access_token('nameid')

    assert ' 404 ' in str(error.value)

        with pytest.raises(APIError) as error:

            token = connector.get_access_token('nameid')

    assert 'no JSON content' in str(error.value)

        with pytest.raises(APIError) as error:

            token = connector.get_access_token('nameid')

    assert 'no JSON content' in str(error.value)

        response='{"IsSuccess": false, "CodErreur": "Fail", "LibErreur": "Auth FAIL"}', status_code=200

    ):

        with pytest.raises(APIError) as error:

    )

    assert endpoint == '/arpege-ecp/test/api/users/nameid/forms'

    resp = app.get(endpoint)

    result = resp.json

    assert resp.json['data']

        },

    )

        content='{"IsSuccess": false, "CodErreur": "Fail", "LibErreur": "Failed to get demands"}',

        status_code=200,

    )

        },

    )

            resp = app.get(endpoint)

            result = resp.json

            assert result['err'] == 1

        },

    )

    resp = app.get(endpoint)

    result = resp.json

    assert result['err'] == 1

    demands_response = json.loads(FAKE_USER_DEMANDS_RESPONSE)

    demands_response['Data']['results'][0]['data_administratives']['date_depot'] = None

    resp = app.get(endpoint)

    result = resp.json

    assert result['err'] == 1

    token_response = json.loads(FAKE_LOGIN_OIDC_RESPONSE)

    del token_response['Data']['AccessToken']

    resp = app.get(endpoint)

    result = resp.json

    assert result['err'] == 1

    token_response = json.loads(FAKE_LOGIN_OIDC_RESPONSE)

    token_response['Data']['AccessToken'] = None

    resp = app.get(endpoint)

    result = resp.json

    assert result['err'] == 1

import mock

import pytest

from django.urls import reverse

from requests import Request

from requests.exceptions import ConnectionError

from passerelle.apps.astech.models import ASTech

from passerelle.utils.jsonresponse import APIError

@pytest.fixture

def setup(db):

        ASTech.objects.create(

            slug='test',

            base_url='https://astech.example.net/app.php/',

@mock.patch("passerelle.utils.Request.request")

def test_connections(mocked_request, app, setup):

    endpoint = reverse(

        "generic-endpoint",

        kwargs={"connector": "astech", "slug": setup.slug, "endpoint": "connections"},

    assert response.json["data"]["default"] == "TEST"

    # bad response

        content='{"msg": "not found"}', status_code=404, reason="Not Found"

    )

    response = app.get(endpoint)

    assert response.json["err_desc"] == "AS-TECH response: 404 Not Found"

    assert response.json["data"]["error"]["status"] == 404

    assert response.json["data"]["error"]["content"]["msg"] == "not found"

    response = app.get(endpoint)

    assert response.json["err"] == 1

    assert response.json["err_class"].endswith("APIError")

    assert response.json["err_desc"] == "AS-TECH response: 500 Crashhhh"

    assert response.json["data"]["error"]["status"] == 500

    assert response.json["data"]["error"]["content"] == "crash"

    response = app.get(endpoint)

    assert response.json["err"] == 1

    assert response.json["err_class"].endswith("APIError")

@mock.patch("passerelle.utils.Request.request")

def test_authorization(mocked_request, app, setup):

    mocked_request.side_effect = [

    ]

    endpoint = reverse(

        "generic-endpoint",

        kwargs={"connector": "astech", "slug": setup.slug, "endpoint": "services"},

    )

    response = app.get(endpoint)

    assert mocked_request.call_count == 1

    assert mocked_request.call_args[0][0] == 'post'

        kwargs={"connector": "astech", "slug": setup.slug, "endpoint": "companies"},

    )

    response = app.get(endpoint)

    assert mocked_request.call_count == 1

    assert mocked_request.call_args[0][0] == 'post'

    )

    mocked_request.side_effect = [

    ]

    response = app.get(endpoint, status=200)

    assert mocked_request.call_count == 2

    ]

    mocked_request.reset_mock(side_effect=True)

    response = app.get(endpoint + '?company=42', status=200)

    assert mocked_request.call_count == 1

    assert mocked_request.call_args[0][0] == 'post'

    )

    mocked_request.side_effect = [

    ]

    response = app.get(endpoint + '?name=LIBELDEMDEF', status=200)

    assert mocked_request.call_count == 2

    assert response.json['data']['LIBELDEMDEF'] == 'O'

    mocked_request.reset_mock(side_effect=True)

    response = app.get(endpoint + '?name=LIBELDEMDEF&company=00', status=200)

    assert mocked_request.call_count == 1

    assert mocked_request.call_args[0][0] == 'get'

        kwargs={"connector": "astech", "slug": setup.slug, "endpoint": "create-demand"},

    )

    demand = {

        'company': '99',

        'service': '11',

    assert response.json['data']['demand_id'] == '000000000001234'

    # mock invalid AS-TECH response

    response = app.post_json(endpoint, params=demand, status=200)

    assert mocked_request.call_count == 2

    assert response.json['err'] == 1

        "generic-endpoint",

        kwargs={"connector": "astech", "slug": setup.slug, "endpoint": "add-document"},

    )

    document = {

        'demand_id': '000000000001234',

        'title': 'test document',

        "generic-endpoint",

        kwargs={"connector": "astech", "slug": setup.slug, "endpoint": "demand-position"},

    )

    response = app.get(endpoint + '?demand_id=000000000001234', status=200)

    assert mocked_request.call_count == 1

    assert mocked_request.call_args[0][0] == 'get'

    }

    # invalid AS-TECH response

    response = app.get(endpoint + '?demand_id=000000000001234', status=200)

    assert mocked_request.call_count == 2

    assert response.json['err'] == 1

        "generic-endpoint",

        kwargs={"connector": "astech", "slug": setup.slug, "endpoint": "demand-all-positions"},

    )

    response = app.get(endpoint, status=200)

    assert mocked_request.call_count == 1

    assert mocked_request.call_args[0][0] == 'get'

import mock

import pytest

from django.utils.encoding import force_bytes, force_str, force_text

from django.utils.http import urlencode

from requests import Request

from requests.exceptions import ConnectionError

from passerelle.apps.astregs.models import AstreGS, Link

BASE_URL = 'https://test-ws-astre-gs.departement06.fr/axis2/services/'

@pytest.fixture

def connector(db):

        AstreGS,

        title='Test',

        slug='test',

import os

import pytest

from django.utils.http import urlencode

from passerelle.apps.atos_genesys.models import Link, Resource

FAKE_URL = 'https://sirus.fr/'

@pytest.fixture

def genesys(db):

        Resource, title='Test 1', slug='test1', description='Connecteur de test', webservice_base_url=FAKE_URL

    )

    response = open(

        os.path.join(os.path.dirname(__file__), 'data', 'genesys_select_codifications.xml')

    ).read()

        yield mock

def test_ws_categories(app, genesys, mock_codifications_ok):

    response = app.get(url)

    assert response.json['err'] == 0

    assert response.json['data']

def test_ws_codifications(app, genesys, mock_codifications_ok):

    url += '/MOT_APA/'

    response = app.get(url)

    assert response.json['err'] == 0

    cache.clear()

    url += '/MOT_APA/'

    response = app.get(url)

    assert response.json['err'] == 1

def test_ws_link_unknown_appairage(app, genesys):

        response = app.post(

            url

            + '?'

def test_ws_link_created(app, genesys):

    assert Link.objects.count() == 0

        response = app.post(

            url

            + '?'

    assert response.json['new']

    assert Link.objects.filter(name_id='zob', id_per='789', resource=genesys).count() == 1

        response = app.get(

            dossiers_url

            + '?'

            )

        )

    response = app.post(

        url

        + '?'

def test_ws_dossiers(app, genesys):

    link = Link.objects.create(resource=genesys, name_id='zob', id_per='1234')

        response = app.get(

            url

            + '?'

    link2 = Link.objects.create(resource=genesys, name_id='zob', id_per='4567')

        response = app.get(

            url

            + '?'

        assert response.json['data'][1]['id'] == str(link2.id)

        assert response.json['data'][1]['text'] == u'%s - John DOE' % link2.id_per

        response = app.get(

            url

            + '?'

def test_ws_search(app, genesys):

        FAKE_URL + 'WSUsagerPublik/services/PublikService/chercheBeneficiaire', RESPONSE_SEARCH

    ):

            FAKE_URL + 'WSUsagerPublik/services/PublikService/selectUsager', RESPONSE_SELECT_USAGER

        ):

            response = app.get(

        },

    ]

        FAKE_URL + 'WSUsagerPublik/services/PublikService/chercheBeneficiaire', RESPONSE_SEARCH

    ):

            FAKE_URL + 'WSUsagerPublik/services/PublikService/selectUsager', RESPONSE_SELECT_USAGER

        ):

            response = app.get(

            assert response.json['err'] == 1

            assert response.json['err_desc'] == 'not-found'

        FAKE_URL + 'WSUsagerPublik/services/PublikService/chercheBeneficiaire', RESPONSE_SEARCH

    ):

            FAKE_URL + 'WSUsagerPublik/services/PublikService/selectUsager', RESPONSE_SELECT_USAGER

        ):

            response = app.get(

            assert response.json['err'] == 1

            assert response.json['err_desc'] == 'not-found'

        FAKE_URL + 'WSUsagerPublik/services/PublikService/chercheBeneficiaire', RESPONSE_SEARCH

    ):

            FAKE_URL + 'WSUsagerPublik/services/PublikService/selectUsager',

            RESPONSE_SELECT_USAGER_NO_CONTACTS,

        ):

def test_ws_link_by_id_per(app, genesys):

    assert Link.objects.count() == 0

        FAKE_URL + 'WSUsagerPublik/services/PublikService/selectUsager', RESPONSE_SELECT_USAGER

    ):

        response = app.post(

    assert data['new']

    assert data['link_id'] == link.pk

        FAKE_URL + 'WSUsagerPublik/services/PublikService/chercheBeneficiaire', RESPONSE_SEARCH

    ):

            FAKE_URL + 'WSUsagerPublik/services/PublikService/selectUsager', RESPONSE_SELECT_USAGER

        ):

            response = app.get(

# -*- coding: utf-8 -*-

import pytest

from httmock import HTTMock, all_requests

from passerelle.apps.feeds.models import Feed

FEED_EXAMPLE = u"""<?xml version="1.0" encoding="UTF-8"?>

@pytest.fixture

def connector(db):

@all_requests

import mock

import pytest

from django.core.management import call_command

from django.core.management.base import CommandError

from django.utils.six.moves.urllib.parse import urljoin

from requests.exceptions import ConnectionError, HTTPError

from passerelle.apps.base_adresse.models import (

    AddressCacheModel,

    BaseAdresse,

@pytest.fixture

def base_adresse(db):

@pytest.fixture

def base_adresse_97x(db):

@pytest.fixture

def base_adresse_corsica(db):

@pytest.fixture

def base_adresse_multiple(db):

        BaseAdresse.objects.create(slug='base-adresse', zipcode='73, 73100, 97425,20000 ')

    )

@pytest.fixture

def base_adresse_coordinates(db):

        BaseAdresse.objects.create(slug='base-adresse', latitude=1.2, longitude=2.1)

    )

@mock.patch('passerelle.utils.Request.get')

def test_base_adresse_search(mocked_get, app, base_adresse):

    assert endpoint == '/base-adresse/base-adresse/search'

    resp = app.get(endpoint, params={'q': 'plop'}, status=200)

    data = resp.json[0]

    assert data['lat'] == '47.474633'

def test_base_adresse_search_path(mocked_get, app, base_adresse):

    base_adresse.service_url = 'http://example.net/path/'

    base_adresse.save()

    resp = app.get(endpoint, params={'q': 'plop'}, status=200)

    assert mocked_get.call_args[0][0].startswith('http://example.net/path/search/?')

    data = resp.json[0]

    def raise_for_status():

        raise HTTPError("400 Client Error: Bad Request for url: xxx")

    response.raise_for_status = raise_for_status

    mocked_get.return_value = response

    resp = app.get('/base-adresse/%s/search' % base_adresse.slug, params={'q': 'plop'}, status=200)

        }

    )

    faked_content = json.dumps(content)

    resp = app.get('/base-adresse/%s/reverse?lon=-0.593775&lat=47.474633' % base_adresse.slug)

    data = resp.json

    assert data['address']['road'] == 'Rue Roger Halope'

@mock.patch('passerelle.utils.Request.get')

def test_base_adresse_reverse_path(mocked_get, app, base_adresse):

    base_adresse.service_url = 'http://example.net/path/'

    base_adresse.save()

    app.get('/base-adresse/%s/reverse?lon=-0.593775&lat=47.474633' % base_adresse.slug)

def test_base_adresse_command_update(mocked_get, db, base_adresse):

    filepath = os.path.join(os.path.dirname(__file__), 'data', 'update_streets_test.gz')

    with open(filepath, 'rb') as ban_file:

    call_command('cron', 'daily')

    mocked_get.assert_called_once_with(

        'https://adresse.data.gouv.fr/data/ban/adresses/latest/addok/adresses-addok-73.ndjson.gz'

    base_adresse.update_api_geo_data = lambda: None

    filepath = os.path.join(os.path.dirname(__file__), 'data', 'update_streets_test.gz')

    with open(filepath, 'rb') as ban_file:

    # check the job added at save() downloads streets

    base_adresse.jobs()

    mocked_get.assert_called_once_with(

    base_adresse_97x.update_api_geo_data = lambda: None

    filepath = os.path.join(os.path.dirname(__file__), 'data', 'update_streets_test.gz')

    with open(filepath, 'rb') as ban_file:

    call_command('cron', 'daily')

    mocked_get.assert_called_once_with(

        'https://adresse.data.gouv.fr/data/ban/adresses/latest/addok/adresses-addok-974.ndjson.gz'

    base_adresse_corsica.update_api_geo_data = lambda: None

    filepath = os.path.join(os.path.dirname(__file__), 'data', 'update_streets_test.gz')

    with open(filepath, 'rb') as ban_file:

    call_command('cron', 'daily')

    assert mocked_get.call_count == 2

    mocked_get.assert_any_call(

    base_adresse_multiple.update_api_geo_data = lambda: None

    filepath = os.path.join(os.path.dirname(__file__), 'data', 'update_streets_test.gz')

    with open(filepath, 'rb') as ban_file:

    call_command('cron', 'daily')

    assert mocked_get.call_count == 4

    mocked_get.assert_any_call(

@mock.patch('passerelle.utils.Request.get')

def test_base_adresse_command_update_geo(mocked_get, db, base_adresse, base_adresse_97x):

    return_values = [

        for content in (FAKE_API_GEO_REGIONS, FAKE_API_GEO_DEPARTMENTS, FAKE_API_GEO) * 2

    ]

    mocked_get.side_effect = return_values

@mock.patch('passerelle.utils.Request.get')

def test_base_adresse_command_update_geo_delete(mocked_get, db, base_adresse):

    return_values = [

        for content in (FAKE_API_GEO_REGIONS, FAKE_API_GEO_DEPARTMENTS, FAKE_API_GEO)

    ]

    mocked_get.side_effect = return_values

    new_fake_api_geo = json.dumps([FAKE_API_GEO_LIST[1]])

    return_values = [

        for content in (FAKE_API_GEO_REGIONS, FAKE_API_GEO_DEPARTMENTS, new_fake_api_geo)

    ]

    mocked_get.side_effect = return_values

@mock.patch('passerelle.utils.Request.get')

def test_base_adresse_command_job_update_geo(mocked_get, db, base_adresse):

    return_values = [

        for content in (FAKE_API_GEO_REGIONS, FAKE_API_GEO_DEPARTMENTS, FAKE_API_GEO)

    ]

    mocked_get.side_effect = return_values

@pytest.mark.usefixtures('mock_update_streets')

@mock.patch('passerelle.utils.Request.get')

def test_base_adresse_command_update_geo_invalid(mocked_get, db, base_adresse):

    with pytest.raises(CommandError):

        call_command('cron', 'daily')

    assert mocked_get.call_count == 1

    assert not RegionModel.objects.exists()

    call_command('cron', 'daily')

    assert mocked_get.call_count == 4

    assert not RegionModel.objects.exists()

    call_command('cron', 'daily')

    assert mocked_get.call_count == 7

    assert not RegionModel.objects.exists()

@mock.patch('passerelle.utils.Request.get')

def test_base_adresse_addresses(mocked_get, app, base_adresse):

    assert endpoint == '/base-adresse/base-adresse/addresses'

    resp = app.get(endpoint, params={'q': 'plop'}, status=200)

    data = resp.json['data'][0]

    assert data['lat'] == '47.474633'

@mock.patch('passerelle.utils.Request.get')

def test_base_adresse_addresses_data_change(mocked_get, app, base_adresse):

    # one user selects an address

    resp = app.get(endpoint, params={'q': 'plop'}, status=200)

    # another requests the same while upstream data has been updated

    new_content = json.loads(FAKED_CONTENT)

    new_content['features'][0]['properties']['label'] = 'changed'

    resp = app.get(endpoint, params={'q': 'plop'}, status=200)

    # first user saves the form, data should not have changed

import freezegun

import mock

import pytest

import xmlschema

from passerelle.contrib.caluire_axel import schemas

from passerelle.contrib.caluire_axel.models import CaluireAxel, Link

from passerelle.contrib.utils.axel import AxelError, OperationResult, json_date_format, xml_date_format

@pytest.fixture

def resource(db):

        CaluireAxel, slug='test', wsdl_url='http://example.net/AXEL_WS/AxelWS.php?wsdl'

    )

import mock

import pytest

from django.core.files.storage import default_storage

from django.test import override_settings

from httmock import HTTMock

from passerelle.apps.cartads_cs.models import CartaDSCS, CartaDSDossier, CartaDSFile

from passerelle.base.models import Job

from passerelle.compat import json_loads

@pytest.fixture

def connector(db):

        CartaDSCS,

        title='Test',

        slug='test',

import mock

import pytest

from django.core.files.storage import default_storage

from lxml import etree

from lxml import objectify as xobject

from passerelle.apps.cityweb.cityweb import DateType

from passerelle.apps.cityweb.models import CityWeb

from passerelle.utils.jsonresponse import APIError

@pytest.fixture

def setup(db):

PAYLOAD = [

from django.contrib.contenttypes.models import ContentType

from django.utils.six.moves.urllib import parse as urlparse

from requests.exceptions import HTTPError

from passerelle.apps.clicrdv.models import ClicRdv

from passerelle.base.models import AccessRight, ApiUser

@pytest.fixture

from django.utils.encoding import force_bytes, force_text

from django.utils.six.moves.urllib import error as urllib2

from mock import Mock, call

from passerelle.apps.cmis.models import CmisConnector

from passerelle.base.models import AccessRight, ApiUser, ResourceLog

def b64encode(content):

import base64

import pytest

from django.contrib.contenttypes.models import ContentType

from django.core.exceptions import ValidationError

from django.utils.encoding import force_text

from passerelle.apps.cryptor.models import CryptedFile, Cryptor

from passerelle.base.models import AccessRight, ApiUser

def test_cryptor_restricted_access(app, cryptor):

    assert endpoint == '/cryptor/test/file-encrypt'

    resp = app.get(endpoint, status=405)

    resp = app.post_json(endpoint, params={"foo": "bar"}, status=403)

    assert resp.json['err'] == 1

    assert 'PermissionDenied' in resp.json['err_class']

    assert endpoint == '/cryptor/test/file-decrypt/uuid'

    resp = app.post_json(endpoint, params={"foo": "bar"}, status=405)

    resp = app.get(endpoint, status=403)

        codename='can_decrypt', apiuser=api, resource_type=obj_type, resource_pk=cryptor.pk

    )

    for bad_payload in (

        'error',

        {"foo": "bar"},

        resp = app.post_json(endpoint, params=bad_payload, status=400)

        assert resp.json['err'] == 1

    endpoint = endpoint + '/bad-uuid'

    resp = app.get(endpoint, status=404)

    )

    # encrypt

    content = force_text(base64.b64encode(b'this is foo and bar'))

    payload = {"file": {"filename": "foo.txt", "content_type": "text/plain", "content": content}}

    assert resp.json['err_desc'] == 'missing public key'

    # decrypt

    endpoint = endpoint + '/' + str(cfile.uuid)

    resp = app.get(endpoint, status=200)

    assert resp.content_type == 'text/plain'

from django.utils.six import StringIO

from django.utils.six.moves.urllib.parse import urlencode

from django.utils.timezone import now

from passerelle.apps.csvdatasource.models import CsvDataSource, Query, TableRow, upload_to

from passerelle.base.models import AccessRight, ApiUser

from passerelle.compat import json_loads

data = """121;69981;DELANOUE;Eliot;H

525;6;DANIEL WILLIAMS;Shanone;F

import mock

import pytest

from django.utils.encoding import force_text

from zeep.exceptions import Fault as WebFault

from zeep.exceptions import TransportError

from passerelle.contrib.dpark.models import DPark, Pairing

from passerelle.utils.jsonresponse import APIError

SLUG = 'test'

WSDL_URL = 'https://example.net/dpark?wsdl'

import httmock

import pytest

from Cryptodome.Cipher import DES

from Cryptodome.Util.Padding import pad, unpad

from django.utils.encoding import force_bytes

from passerelle.apps.esirius.models import ESirius

from passerelle.utils.jsonresponse import APIError

CREATE_APPOINTMENT_PAYLOAD = {

    'beginDate': '2021-02-24',

@pytest.fixture

def connector(db):

        ESirius.objects.create(

            slug='test', secret_id='xxx', secret_key='yyy', base_url='https://esirius.example.net'

        )

def get_endpoint(name):

@pytest.mark.freeze_time('2021-01-26 15:13:6.880')  # epoch + 1611673986.88 s

import mock

import pytest

from passerelle.contrib.esirius_swi.models import ESiriusSwi

from passerelle.contrib.esirius_swi.utils import compute_status

@pytest.fixture

def connector(db):

        ESiriusSwi.objects.create(

            slug='test', base_url='https://example.org/eSirius/webservices/sitewaitingindicator/'

        )

import mock

import pytest

from passerelle.apps.feeds.models import Feed

FEED_EXAMPLE = u"""<?xml version="1.0" encoding="UTF-8"?>

@pytest.fixture

def connector(db):

@mock.patch('passerelle.utils.Request.get')

def test_feed(mocked_get, app, connector):

    assert endpoint == '/feeds/test/json'

    resp = app.get(endpoint)

    assert resp.json['data']['feed']['title']

from urllib.parse import parse_qs, urlparse, urlunparse

import pytest

from passerelle.apps.franceconnect_data.models import Resource

@pytest.fixture

def fc(db):

        Resource,

        slug='test',

        fc_platform_slug='prod',

import mock

import pytest

from django.contrib.contenttypes.models import ContentType

from passerelle.base.models import AccessRight, ApiUser

from passerelle.contrib.gdema.models import Gdema

def test_gdema_services_and_typologies(app, gdema):

    assert endpoint == '/gdema/test/referentiel'

    with mock.patch('passerelle.utils.Request.get') as requests_get:

        resp = app.get(endpoint + '/service/', status=200)

        assert requests_get.call_count == 1

        assert requests_get.call_args[0][0] == 'https://gdema.example.net/api/referentiel/service'

def test_gdema_referentiel(app, gdema):

    assert endpoint == '/gdema/test/referentiel'

    with mock.patch('passerelle.utils.Request.get') as requests_get:

        resp = app.get(endpoint + '/civility/', status=200)

        assert requests_get.call_count == 1

        assert requests_get.call_args[0][0] == 'https://gdema.example.net/api/referentiel/civility'

        assert resp.json['data'][0]['text'] == 'Monsieur'

    with mock.patch('passerelle.utils.Request.get') as requests_get:

        resp = app.get(endpoint + '/nothing/', status=200)

        assert requests_get.call_count == 1

        assert requests_get.call_args[0][0] == 'https://gdema.example.net/api/referentiel/nothing'

def test_gdema_get_request(app, gdema):

    assert endpoint == '/gdema/test/get-request'

    with mock.patch('passerelle.utils.Request.get') as requests_get:

        resp = app.get(endpoint + '/1/', status=200)

        assert requests_get.call_count == 1

        assert requests_get.call_args[0][0] == 'https://gdema.example.net/api/request/1'

        assert resp.json['data']['Handler']['CivilityId'] == '0'

        assert resp.json['data']['ReceptDate'] == '2006-12-13T00:00:00+01:00'

    assert endpoint == '/gdema/test/get-request-state'

    with mock.patch('passerelle.utils.Request.get') as requests_get:

        resp = app.get(endpoint + '/1/', status=200)

        assert requests_get.call_count == 1

        assert requests_get.call_args[0][0] == 'https://gdema.example.net/api/request/1/state'

def test_gdema_create_request(app, gdema):

    assert endpoint == '/gdema/test/create-request'

    with mock.patch('passerelle.utils.Request.post') as requests_post:

        resp = app.post_json(endpoint + '?raise=1', params=CREATE_INPUT, status=200)

        assert requests_post.call_count == 1

        assert requests_post.call_args[0][0] == 'https://gdema.example.net/api/request/create'

import mock

import pytest

from django.contrib.contenttypes.models import ContentType

from django.urls import reverse

from passerelle.apps.api_particulier.models import APIParticulier

from passerelle.apps.arcgis.models import ArcGIS

from passerelle.apps.mdel.models import MDEL

)

from passerelle.contrib.stub_invoices.models import StubInvoicesConnector

from passerelle.utils.api import endpoint

@pytest.fixture

def mdel(db):

@pytest.fixture

def arcgis(db):

    instance.set_log_level('DEBUG')

    return instance

@mock.patch('passerelle.utils.Request.get')

def test_proxy_logger(mocked_get, caplog, app, arcgis):

    payload = open(os.path.join(os.path.dirname(__file__), 'data', 'nancy_arcgis', 'sigresponse.json')).read()

    # simple logger

    arcgis.log_evel = 'DEBUG'

@mock.patch('requests.Session.send')

def test_proxy_logger_transaction_id(mocked_send, app, arcgis):

    payload = open(os.path.join(os.path.dirname(__file__), 'data', 'nancy_arcgis', 'sigresponse.json')).read()

    arcgis.log_evel = 'DEBUG'

    arcgis.base_url = 'https://example.net/'

    arcgis.save()

@mock.patch('passerelle.utils.Request.patch')

def test_proxy_logger_on_405(mocked_patch, caplog, app, arcgis):