Produits Entr'ouvert » Passerelle

from httmock import HTTMock, response, urlmatch

@urlmatch(netloc='^entreprise.api.gouv.fr$', path='^/v2/etablissements/')

def api_entreprise_etablissements(url, request):

    return response(200, ETABLISSEMENTS_RESPONSE, request=request)

@urlmatch(netloc='^entreprise.api.gouv.fr$', path='^/v2/entreprises/')

def api_entreprise_entreprises(url, request):

    return response(200, ENTREPRISES_RESPONSE, request=request)

@urlmatch(netloc='^entreprise.api.gouv.fr$', path='^/v2/associations/')

def api_entreprise_associations(url, request):

    return response(200, ASSOCIATIONS_RESPONSE, request=request)

@urlmatch(netloc='^entreprise.api.gouv.fr$', path='^/v2/extraits_rcs_infogreffe/')

def api_entreprise_extraits_rcs(url, request):

    return response(200, EXTRAITS_RCS_RESPONSE, request=request)

@urlmatch(netloc='^entreprise.api.gouv.fr$', path='^/v2/documents_associations/')

def api_entreprise_documents_associations(url, request):

    return response(200, DOCUMENTS_ASSOCIATION_RESPONSE, request=request)

@urlmatch(netloc='^apientreprise.fr$', path='^/attestations/')

def api_entreprise_document_association(url, request):

    return response(200, DOCUMENT_ASSOCIATION_RESPONSE, request=request)

@urlmatch(netloc='^entreprise.api.gouv.fr$', path='^/v2/effectifs_annuels_acoss_covid/')

def effectifs_annuels_acoss_covid(url, request):

    return response(200, EFFECTIFS_ANNUELS_ACOSS_COVID, request=request)

@urlmatch(netloc='^entreprise.api.gouv.fr$', path='^/v2/effectifs_mensuels_acoss_covid/.*/entreprise')

def entreprise_effectifs_mensuels_acoss_covid(url, request):

    return response(200, ENTREPRISE_EFFECTIFS_MENSUELS_ACOSS_COVID, request=request)

@urlmatch(netloc='^entreprise.api.gouv.fr$', path='^/v2/effectifs_mensuels_acoss_covid/.*/etablissement')

def etablisssment_effectifs_mensuels_acoss_covid(url, request):

    return response(200, ETABLISSEMENT_EFFECTIFS_MENSUELS_ACOSS_COVID, request=request)

@urlmatch(netloc='^entreprise.api.gouv.fr$', path='^/v2/exercices')

def api_entreprise_exercices(url, request):

    return response(200, EXERCICES_RESPONSE, request=request)

@urlmatch(netloc='^entreprise.api.gouv.fr$')

def api_entreprise_error_500(url, request):

    return response(500, 'bad error happened', request=request)

@urlmatch(netloc='^entreprise.api.gouv.fr$')

def api_entreprise_connection_error(url, request):

    raise requests.RequestException('connection timed-out')

@urlmatch(netloc='^entreprise.api.gouv.fr$')

def api_entreprise_error_not_json(url, request):

    return response(200, 'simple text', request=request)

@urlmatch(netloc='^entreprise.api.gouv.fr$')

def api_entreprise_error_not_found(url, request):

    return response(404, {'error': 'not_found', 'message': 'Page not found'}, request=request)

@pytest.yield_fixture

def mock_api_entreprise():

    with HTTMock(

        api_entreprise_etablissements,

        api_entreprise_entreprises,

        api_entreprise_associations,

        api_entreprise_extraits_rcs,

        api_entreprise_associations,

        api_entreprise_documents_associations,

        api_entreprise_document_association,

        effectifs_annuels_acoss_covid,

        entreprise_effectifs_mensuels_acoss_covid,

        etablisssment_effectifs_mensuels_acoss_covid,

        api_entreprise_exercices,

    ):

        yield None

def test_entreprises_endpoint(app, resource, mock_api_entreprise):

    response = app.get('/api-entreprise/test/entreprises/443170139/', params=REQUEST_PARAMS)

def test_simple_match_mandataire_social(app, resource, mock_api_entreprise):

    url = '/api-entreprise/test/match_mandataire_social/443170139/'

    response = app.get(url, params=params)

    # successful matching

    data = response.json['data']

    assert data['nom'] == 'HISQUIN'

    assert data['prenom'] == 'FRANCOIS'

    assert data['date_naissance'] == '1965-01-27'

    assert data['type'] == 'PP'

    params['method'] = 'unkwown'

    response = app.get(url, params=params)

    # unsupported method

    assert response.json['err']

    assert 'data' not in response.json

    params['method'] = 'simple'

    params.pop('first_name')

    response = app.get(url, params=params, status=400)

    # missing parameter

    assert not response.json['data']

    assert 'missing parameter' in response.json['err_desc']

    params['first_name'] = 'JOHN'

    response = app.get(url, params=params)

    # matching failed

    assert not response.json['err']

    assert not response.json['data']

def test_levenshtein_match_mandataire_social(app, resource, mock_api_entreprise):

    url = '/api-entreprise/test/match_mandataire_social/443170139/'

    response = app.get(url, params=params)

    # successful matching

    data = response.json['data']

    assert data['nom'] == 'HISQUIN'

    assert data['prenom'] == 'FRANCOIS'

    assert data['date_naissance'] == '1965-01-27'

    assert data['type'] == 'PP'

    params['first_name'] = 'JOHN'

    response = app.get(url, params=params)

    # matching failed

    assert not response.json['err']

    assert not response.json['data']

    params['first_name'] = 'francoi'

    params['last_name'] = 'hisquinn'

    response = app.get(url, params=params)

    # successful matching within distance bracket

    data = response.json['data']

    assert data['nom'] == 'HISQUIN'

    assert data['prenom'] == 'FRANCOIS'

    assert data['date_naissance'] == '1965-01-27'

    assert data['type'] == 'PP'

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

def test_entreprises_endpoint_include_private(mocked_get, app, resource, mock_api_entreprise):

    mocked_get.return_value = FakedResponse(content='{}', status_code=200)

    request_params = REQUEST_PARAMS.copy()

    app.get('/api-entreprise/test/entreprises/443170139/', params=request_params)

    assert 'non_diffusables' not in mocked_get.mock_calls[0].kwargs['data']

    request_params['include_private'] = True

    app.get('/api-entreprise/test/entreprises/443170139/', params=request_params)

    assert mocked_get.mock_calls[1].kwargs['data']['non_diffusables'] == 'true'

def test_etablissements_endpoint(app, resource, mock_api_entreprise):

    response = app.get('/api-entreprise/test/etablissements/44317013900036/', params=REQUEST_PARAMS)

def test_associations_endpoint(app, resource, mock_api_entreprise):

    response = app.get('/api-entreprise/test/associations/443170139/', params=REQUEST_PARAMS)

def test_documents_associations_endpoint(app, resource, mock_api_entreprise):

    response = app.get('/api-entreprise/test/documents_associations/443170139/', params=REQUEST_PARAMS)

def test_associations_last_document_of_type_endpoint(app, resource, mock_api_entreprise):

    params = REQUEST_PARAMS

    params['document_type'] = 'Statuts'

    response = app.get('/api-entreprise/test/document_association/443170139/get-last/', params=params)

    assert 'data' in response.json

    data = response.json['data']

    assert data['timestamp'] == '1500660325'

    params['document_type'] = 'Liste des dirigeants'

    response = app.get('/api-entreprise/test/document_association/443170139/get-last/', params=params)

    assert not response.json['data']

def test_extraits_rcs(app, resource, mock_api_entreprise, freezer):

    response = app.get('/api-entreprise/test/extraits_rcs/443170139/', params=REQUEST_PARAMS)

def test_document_association(app, resource, mock_api_entreprise, freezer):

    response = app.get('/api-entreprise/test/documents_associations/443170139/', params=REQUEST_PARAMS)

    assert 'data' in response.json

    data = response.json['data']

    assert len(data) == 3

    document = data[0]

    assert 'url' in document

    app.get(document['url'], params={'context': 'MSP', 'object': 'demand', 'recipient': 'siret'}, status=200)

    # try to get document with wrong signature

    wrong_url = document['url'] + "wrong/"

    app.get(wrong_url, status=404)

    # try expired url

    freezer.move_to(timezone.now() + timezone.timedelta(days=8))

    app.get(document['url'], status=404)

def test_effectifs_annuels_acoss_covid(app, resource, mock_api_entreprise, freezer):

    response = app.get('/api-entreprise/test/effectifs_annuels_acoss_covid/418166096/', params=REQUEST_PARAMS)

def test_entreprise_effectifs_mensuels_acoss_covid(app, resource, mock_api_entreprise, freezer):

    response = app.get(

        '/api-entreprise/test/entreprise_effectifs_mensuels_acoss_covid/2019/02/418166096/',

        params=REQUEST_PARAMS,

    )

def test_etablissment_effectifs_mensuels_acoss_covid(app, resource, mock_api_entreprise, freezer):

    response = app.get(

        '/api-entreprise/test/etablissement_effectifs_mensuels_acoss_covid/2019/02/418166096/',

        params=REQUEST_PARAMS,

    )

def test_exercices(app, resource, mock_api_entreprise, freezer):

    response = app.get('/api-entreprise/test/exercices/418166096/', params=REQUEST_PARAMS)

def test_error_500(app, resource, mock_api_entreprise):

    with HTTMock(api_entreprise_error_500):

def test_no_json_error(app, resource, mock_api_entreprise):

    with HTTMock(api_entreprise_error_not_json):

def test_error_404(app, resource, mock_api_entreprise):

    with HTTMock(api_entreprise_error_not_found):

def test_connection_error(app, resource, mock_api_entreprise):

    with HTTMock(api_entreprise_connection_error):

-