entrouvert
/
passerelle
15
0
Fork 0
Code Issues Pull Requests 15 Releases Activity

api_entreprise: use v3 in entreprises endpoint (#72961)
gitea-wip/passerelle/pipeline/pr-main This commit looks good Details
gitea/passerelle/pipeline/head Something is wrong with the build of this commit Details

This commit is contained in:
Emmanuel Cazenave 2023-01-05 12:28:54 +01:00
parent 5ec431d6c0
commit e919d4e695
2 changed files with 167 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
passerelle/apps/api_entreprise/models.py
View File

@ -334,14 +334,54 @@ class APIEntreprise(BaseResource):
'context': CONTEXT_PARAM,
'recipient': RECIPIENT_PARAM,
'include_private': {'description': _('Include private informations'), 'example_value': 'true'},
'mandataires': {'description': _('Include mandataires informations'), 'example_value': 'true'},
},
)
def entreprises(self, request, siren, include_private=False, **kwargs):
def entreprises(self, request, siren, include_private=False, mandataires=False, **kwargs):
if len(siren) != 9:
raise APIError(_('invalid SIREN length (must be 9 characters)'))
unite_legale_url = 'v3/insee/sirene/unites_legales/diffusibles/%s' % siren
if include_private:
kwargs['non_diffusables'] = True
return self.get('v2/entreprises/%s/' % siren, **kwargs)
unite_legale_url = 'v3/insee/sirene/unites_legales/%s' % siren
data = self.get(unite_legale_url, raw=True, **kwargs).get('data')
forme_juridique = data.get('forme_juridique', {}).get('libelle', '')
forme_juridique_code = data.get('forme_juridique', {}).get('code', '')
data['forme_juridique'] = forme_juridique
data['forme_juridique_code'] = forme_juridique_code
naf = data.get('activite_principale', {}).get('code', '')
data['naf_entreprise'] = naf.replace('.', '')
data['naf_point_entreprise'] = naf
data['libelle_naf_entreprise'] = data.get('activite_principale', {}).get('libelle', '')
data['raison_sociale'] = data.get('personne_morale_attributs', {}).get('raison_sociale', '')
data['tranche_effectif_salarie_entreprise'] = data.get('tranche_effectif_salarie', {})
raw_siege_data = self.get('%s/siege_social' % unite_legale_url, raw=True, **kwargs)
siege_data = raw_siege_data.get('data')
adresse = siege_data.get('adresse', {})
adresse['nom_voie'] = adresse.get('libelle_voie', '')
adresse['localite'] = adresse.get('libelle_commune', '')
adresse['code_insee_localite'] = adresse.get('code_commune', '')
for i in range(1, 8):
adresse['l%s' % i] = adresse.get('acheminement_postal', {}).get('l%s' % i)
siege_data['adresse'] = adresse
meta = raw_siege_data.get('meta')
siege_data['date_mise_a_jour'] = meta.get('date_derniere_mise_a_jour', '')
mandataires_sociaux = []
if mandataires:
mandataires_data = self.get(
'v3/infogreffe/rcs/unites_legales/%s/mandataires_sociaux' % siren, raw=True, **kwargs
).get('data')
for mandataire in mandataires_data:
for key in ('nom', 'prenom', 'fonction'):
if key not in mandataire:
mandataire[key] = ''
mandataires_sociaux.append(mandataire)
data['mandataires_sociaux'] = mandataires_sociaux
return {'data': {'entreprise': data, 'etablissement_siege': siege_data}}
@endpoint(
perm='can_access',

137
tests/test_api_entreprise.py
View File

@ -149,6 +149,82 @@ ENTREPRISES_RESPONSE = {
"gateway_error": False,
}
UNITES_LEGALES_RESPONSE = {
"data": {
"siren": "418166096",
"siret_siege_social": "41816609600051",
"numero_tva_intracommunautaire": "FR16418166096",
"forme_juridique": {
"libelle": "SA à directoire (s.a.i.)",
"code": "5699",
},
"activite_principale": {
"code": "62.02A",
"libelle": "Conseil en systèmes et logiciels informatiques",
},
"personne_morale_attributs": {
"raison_sociale": "OCTO-TECHNOLOGY",
},
"tranche_effectif_salarie": {
"de": 200,
"a": 249,
"code": "31",
"date_reference": "2014",
"intitule": "200 à 249 salariés",
},
"date_creation": 891381600,
"categorie_entreprise": "PME",
"diffusable_commercialement": True,
}
}
UNITES_LEGALES_SIEGE_RESPONSE = {
"data": {
"siret": "41816609600051",
"adresse": {
"numero_voie": "50",
"type_voie": "AV",
"libelle_voie": "DES CHAMPS ELYSEES",
"code_postal": "75008",
"libelle_commune": "PARIS 8",
"code_commune": "75108",
"acheminement_postal": {
"l1": "OCTO TECHNOLOGY",
"l4": "50 AVENUE DES CHAMPS ELYSEES",
"l6": "75008 PARIS",
"l7": "FRANCE",
},
},
},
"meta": {
"date_derniere_mise_a_jour": 1449183600,
},
}
RCS_UNITES_LEGALES_MANDATAIRES_SOCIAUX_RESPONSE = {
"data": [
{
"nom": "HISQUIN",
"prenom": "FRANCOIS",
"fonction": "PRESIDENT DU DIRECTOIRE",
"dirigeant": True,
"date_naissance": "1965-01-27",
"raison_sociale": "",
"identifiant": "",
"type": "PP",
},
{
"fonction": "COMMISSAIRE AUX COMPTES SUPPLEANT",
"dirigeant": True,
"date_naissance": "",
"date_naissance_timestamp": 0,
"raison_sociale": "BCRH & ASSOCIES - SOCIETE A RESPONSABILITE LIMITEE A ASSOCIE UNIQUE",
"identifiant": "490092574",
"type": "PM",
},
]
}
EXTRAITS_RCS_RESPONSE = {
"data": {
"siren": "418166096",
@ -292,8 +368,23 @@ def test_endpoint_with_no_params(mocked_get, app, resource):
def test_entreprises_endpoint(app, resource):
with responses.RequestsMock() as rsps:
rsps.get('https://entreprise.api.gouv.fr/v2/entreprises/443170139/', json=ENTREPRISES_RESPONSE)
response = app.get('/api-entreprise/test/entreprises/443170139/', params=REQUEST_PARAMS)
rsps.get(
'https://entreprise.api.gouv.fr/v3/insee/sirene/unites_legales/diffusibles/443170139',
json=UNITES_LEGALES_RESPONSE,
)
rsps.get(
'https://entreprise.api.gouv.fr/v3/insee/sirene/unites_legales/diffusibles/443170139/siege_social',
json=UNITES_LEGALES_SIEGE_RESPONSE,
)
rsps.get(
'https://entreprise.api.gouv.fr/v3/infogreffe/rcs/unites_legales/443170139/mandataires_sociaux',
json=RCS_UNITES_LEGALES_MANDATAIRES_SOCIAUX_RESPONSE,
)
request_params = REQUEST_PARAMS.copy()
request_params['mandataires'] = True
response = app.get('/api-entreprise/test/entreprises/443170139/', params=request_params)
data = response.json['data']
assert data['entreprise']['categorie_entreprise'] == 'PME'
assert data['entreprise']['numero_tva_intracommunautaire'] == 'FR16418166096'
@ -304,6 +395,7 @@ def test_entreprises_endpoint(app, resource):
assert data['entreprise']['date_creation'] == '1998-03-31'
assert data['entreprise']['diffusable_commercialement'] is True
assert data['entreprise']['naf_entreprise'] == '6202A'
assert data['entreprise']['naf_point_entreprise'] == '62.02A'
assert data['entreprise']['libelle_naf_entreprise'] == 'Conseil en systèmes et logiciels informatiques'
assert data['entreprise']['raison_sociale'] == 'OCTO-TECHNOLOGY'
assert data['entreprise']['tranche_effectif_salarie_entreprise']['intitule'] == '200 à 249 salariés'
@ -420,16 +512,33 @@ def test_levenshtein_match_mandataire_social(app, resource):
def test_entreprises_endpoint_include_private(app, resource):
with responses.RequestsMock() as rsps:
rsps.get('https://entreprise.api.gouv.fr/v2/entreprises/443170139/', json=ENTREPRISES_RESPONSE)
rsps.get(
'https://entreprise.api.gouv.fr/v3/insee/sirene/unites_legales/443170139',
json=UNITES_LEGALES_RESPONSE,
)
rsps.get(
'https://entreprise.api.gouv.fr/v3/insee/sirene/unites_legales/443170139/siege_social',
json=UNITES_LEGALES_SIEGE_RESPONSE,
)
request_params = REQUEST_PARAMS.copy()
app.get('/api-entreprise/test/entreprises/443170139/', params=request_params)
assert 'non_diffusables' not in rsps.calls[-1].request.body
request_params['include_private'] = True
app.get('/api-entreprise/test/entreprises/443170139/', params=request_params)
assert 'non_diffusables=true' in rsps.calls[-1].request.body
# the mocked URLS do not contain 'diffusibles'
# responses would raise an error if they were not called
def test_entreprises_endpoint_no_mandataires(app, resource):
with responses.RequestsMock() as rsps:
rsps.get(
'https://entreprise.api.gouv.fr/v3/insee/sirene/unites_legales/diffusibles/443170139',
json=UNITES_LEGALES_RESPONSE,
)
rsps.get(
'https://entreprise.api.gouv.fr/v3/insee/sirene/unites_legales/diffusibles/443170139/siege_social',
json=UNITES_LEGALES_SIEGE_RESPONSE,
)
response = app.get('/api-entreprise/test/entreprises/443170139/', params=REQUEST_PARAMS)
assert response.json['data']['entreprise']['mandataires_sociaux'] == []
def test_etablissements_endpoint(app, resource):
@ -646,7 +755,9 @@ def test_exercices(app, resource):
def test_error_500(app, resource):
with responses.RequestsMock() as rsps:
rsps.get(
'https://entreprise.api.gouv.fr/v2/entreprises/443170139/', body='bad error happened', status=500
'https://entreprise.api.gouv.fr/v3/insee/sirene/unites_legales/diffusibles/443170139',
body='bad error happened',
status=500,
)
response = app.get('/api-entreprise/test/entreprises/443170139/', params=REQUEST_PARAMS)
assert response.status_code == 200
@ -658,7 +769,7 @@ def test_error_500(app, resource):
def test_no_json_error(app, resource):
with responses.RequestsMock() as rsps:
rsps.get(
'https://entreprise.api.gouv.fr/v2/entreprises/443170139/',
'https://entreprise.api.gouv.fr/v3/insee/sirene/unites_legales/diffusibles/443170139',
body='simple text',
)
response = app.get('/api-entreprise/test/entreprises/443170139/', params=REQUEST_PARAMS)
@ -673,7 +784,7 @@ def test_no_json_error(app, resource):
def test_error_404(app, resource):
with responses.RequestsMock() as rsps:
rsps.get(
'https://entreprise.api.gouv.fr/v2/entreprises/443170139/',
'https://entreprise.api.gouv.fr/v3/insee/sirene/unites_legales/diffusibles/443170139',
json={'error': 'not_found', 'message': 'Page not found'},
status=404,
)
@ -686,7 +797,7 @@ def test_error_404(app, resource):
def test_connection_error(app, resource):
with responses.RequestsMock() as rsps:
rsps.get(
'https://entreprise.api.gouv.fr/v2/entreprises/443170139/',
'https://entreprise.api.gouv.fr/v3/insee/sirene/unites_legales/diffusibles/443170139',
body=requests.RequestException('connection timed-out'),
)
response = app.get('/api-entreprise/test/entreprises/443170139/', params=REQUEST_PARAMS)