537 lines
20 KiB
Python
537 lines
20 KiB
Python
# passerelle - uniform access to multiple data sources and services
|
|
# Copyright (C) 2021 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 base64
|
|
import json
|
|
import os
|
|
import urllib.parse
|
|
|
|
import httmock
|
|
import pytest
|
|
from django.db import connection
|
|
|
|
from passerelle.base.models import AccessRight
|
|
from passerelle.contrib.toulouse_foederis.models import Document, Resource
|
|
|
|
from .utils import make_resource, mock_url
|
|
|
|
|
|
def get_json_content(name):
|
|
with open(f'tests/data/toulouse_foederis/{name}.json') as fd:
|
|
return json.load(fd)
|
|
|
|
|
|
HTTP_MOCKS = {
|
|
'type-emploi': {
|
|
'path': r'^/.*/data/type_emploi$',
|
|
'query': 'viewIntegrationName=api_publik',
|
|
'content': get_json_content('type_emploi'),
|
|
},
|
|
'origine-candidature': {
|
|
'path': r'^/.*/data/origine_candidature$',
|
|
'query': 'viewIntegrationName=api_publik',
|
|
'content': get_json_content('origine_candidature'),
|
|
},
|
|
'civilite': {
|
|
'path': r'^/.*/data/civilite$',
|
|
'query': 'viewIntegrationName=api_publik',
|
|
'content': get_json_content('civilite'),
|
|
},
|
|
'nationalite': {
|
|
'path': r'^/.*/data/nationalite1$',
|
|
'query': 'viewIntegrationName=api_publik',
|
|
'content': get_json_content('nationalite1'),
|
|
},
|
|
'situation-actuelle': {
|
|
'path': r'^/.*/data/situation_actuelle$',
|
|
'query': 'viewIntegrationName=api_publik',
|
|
'content': get_json_content('situation_actuelle'),
|
|
},
|
|
'domaine-emploi': {
|
|
'path': r'^/.*/data/domaine_emploi$',
|
|
'query': 'viewIntegrationName=api_publik',
|
|
'content': get_json_content('domaine_emploi'),
|
|
},
|
|
'sous-domaine-emploi': {
|
|
'path': r'^/.*/data/sous_domaine_emploi$',
|
|
'query': 'viewIntegrationName=api_publik',
|
|
'content': get_json_content('sous_domaine_emploi'),
|
|
},
|
|
'emploi': {
|
|
'path': r'^/.*/data/emploi$',
|
|
'query': 'viewIntegrationName=api_publik',
|
|
'content': get_json_content('emploi'),
|
|
},
|
|
'niveau-diplome': {
|
|
'path': r'^/.*/data/niveau_diplome1$',
|
|
'query': 'viewIntegrationName=api_publik',
|
|
'content': get_json_content('niveau_diplome1'),
|
|
},
|
|
'habilitation': {
|
|
'path': r'^.*/data/habilitation$',
|
|
'query': 'viewIntegrationName=api_publik',
|
|
'content': get_json_content('habilitation'),
|
|
},
|
|
'annonce': {
|
|
'path': r'^.*/data/annonce$',
|
|
'query': 'viewIntegrationName=api_publik',
|
|
'content': get_json_content('annonce'),
|
|
},
|
|
'pdf': {
|
|
'path': r'^.*/data/annonce/[0-9]+/fields/pdf_ddr$',
|
|
'query': 'viewIntegrationName=api_publik',
|
|
'content': get_json_content('pdf'),
|
|
},
|
|
}
|
|
|
|
APIKEY = '111c11ee-e1b1-11f1-11ce-11d11af1a111-1111-111111'
|
|
|
|
|
|
@pytest.fixture
|
|
def http_mock():
|
|
for annonce_r14848258 in [3450229, 3782122, 3782130, 4005534, 4005526]:
|
|
HTTP_MOCKS['html-fields-%s' % annonce_r14848258] = {
|
|
'path': r'^.*/data/demande_de_personnel$',
|
|
'query': '&'.join(
|
|
[
|
|
'filterName=id',
|
|
'filterValue=%s' % annonce_r14848258,
|
|
'fieldList=missions' + urllib.parse.quote(',') + 'profil_requis',
|
|
'viewIntegrationName=api_publik',
|
|
]
|
|
),
|
|
# give same html field contents for all annonces (only test first one here)
|
|
'content': get_json_content('demande_de_personnel'),
|
|
}
|
|
|
|
handlers = []
|
|
|
|
for defn in HTTP_MOCKS.values():
|
|
|
|
def make_handler(path, content, query=None):
|
|
@httmock.urlmatch(path=path, query=query)
|
|
def handler(url, request):
|
|
assert request.headers['api-key'] == APIKEY
|
|
return json.dumps(content)
|
|
|
|
return handler
|
|
|
|
handlers.append(make_handler(defn['path'], defn['content'], query=defn['query']))
|
|
|
|
@httmock.urlmatch()
|
|
def error_handler(url, request):
|
|
assert False, 'should not be reached'
|
|
|
|
with httmock.HTTMock(*handlers, error_handler) as mock:
|
|
yield mock
|
|
|
|
|
|
@pytest.fixture
|
|
def resource(db, settings, caplog):
|
|
return make_resource(
|
|
Resource,
|
|
title='Foederis',
|
|
slug='foederis',
|
|
description='Foederis',
|
|
url='https://passerelle.cutm-publik-preprod.nfrance.com/foederis/',
|
|
api_key=APIKEY,
|
|
)
|
|
|
|
|
|
def test_document_delete(resource, http_mock):
|
|
resource.hourly()
|
|
document = Document.objects.filter(external_id__startswith='announce-').first()
|
|
pdf_path = document.pdf.path
|
|
assert os.path.exists(pdf_path)
|
|
document.delete()
|
|
assert not os.path.exists(pdf_path)
|
|
|
|
|
|
class TestHourly:
|
|
def test_hourly(self, resource, http_mock, caplog):
|
|
resource.hourly()
|
|
assert 'Created announce 4229013' in caplog.text
|
|
assert 'Referentials updated.' in caplog.text
|
|
assert resource.last_update_announces
|
|
assert resource.last_update_referentiels
|
|
|
|
caplog.clear()
|
|
resource.hourly()
|
|
assert 'Created' not in caplog.text
|
|
assert 'Updated' not in caplog.text
|
|
assert 'Referentials updated.' in caplog.text
|
|
|
|
caplog.clear()
|
|
annonce = get_json_content('annonce')
|
|
annonce['results'][0]['intitule_annonce'] = 'COIN'
|
|
with mock_url(url=r'.*annonce$', response=json.dumps(annonce)):
|
|
resource.hourly()
|
|
assert 'Created' not in caplog.text
|
|
assert 'Updated announce 3450231' in caplog.text
|
|
assert 'Referentials updated.' in caplog.text
|
|
|
|
# remove announce deleted from foederis
|
|
assert Document.objects.filter(external_id__startswith='announce-').count() == 5
|
|
annonce = get_json_content('annonce')
|
|
del annonce['results'][4]
|
|
with mock_url(url=r'.*annonce$', response=json.dumps(annonce)):
|
|
resource.hourly()
|
|
assert Document.objects.filter(external_id__startswith='announce-').count() == 4
|
|
|
|
def test_loaded_pdf(self, resource, http_mock):
|
|
resource.hourly()
|
|
assert Document.objects.filter(external_id__startswith='announce-', pdf='').count() == 0
|
|
document = Document.objects.get(external_id='announce-4229013')
|
|
with document.pdf.open(mode='rb') as fd:
|
|
assert fd.read() == base64.b64decode(
|
|
HTTP_MOCKS['pdf']['content']['results'][0]['pdf_ddr']['fileData']
|
|
)
|
|
|
|
def test_error_500(self, resource, caplog):
|
|
with mock_url(status_code=500):
|
|
resource.hourly()
|
|
assert 'Service is unavailable' in caplog.text
|
|
|
|
def test_not_json(self, resource, caplog):
|
|
with mock_url(status_code=200):
|
|
resource.hourly()
|
|
assert 'Service is unavailable' in caplog.text
|
|
|
|
def test_json_code_is_not_200(self, resource, caplog):
|
|
with mock_url(status_code=200, response='{"code": "x"}'):
|
|
resource.hourly()
|
|
assert 'Service is unavailable' in caplog.text
|
|
|
|
def test_data_annonce_error_500(self, resource, http_mock, caplog):
|
|
with mock_url(url=r'/.*annonce.*', status_code=500):
|
|
resource.hourly()
|
|
assert 'Service is unavailable' in caplog.text
|
|
|
|
def test_pdf_error_500(self, resource, http_mock, caplog):
|
|
with mock_url(url=r'/.*pdf_ddr.*', status_code=500):
|
|
resource.hourly()
|
|
assert 'Service is unavailable' in caplog.text
|
|
|
|
def test_html_fields_error_500(self, resource, http_mock, caplog):
|
|
with mock_url(url=r'/.*demande_de_personnel.*', status_code=500):
|
|
resource.hourly()
|
|
assert 'Service is unavailable' in caplog.text
|
|
|
|
|
|
class TestEndpoints:
|
|
@pytest.fixture(autouse=True)
|
|
def resource(self, resource, http_mock):
|
|
resource.hourly()
|
|
return resource
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def freezer(self, freezer):
|
|
freezer.move_to('2022-04-20T12:00:00')
|
|
return freezer
|
|
|
|
def test_data_sources(self, app):
|
|
for name in [
|
|
'origine-candidature',
|
|
'civilite',
|
|
'nationalite',
|
|
'situation-actuelle',
|
|
'type-emploi',
|
|
'domaine-emploi',
|
|
'sous-domaine-emploi',
|
|
'emploi',
|
|
'niveau-diplome',
|
|
'habilitation',
|
|
]:
|
|
response = app.get(f'/toulouse-foederis/foederis/ds/{name}/')
|
|
assert response.json['err'] == 0
|
|
assert response.json['last_update']
|
|
assert {d['text'] for d in response.json['data']} == {
|
|
d['name'] for d in HTTP_MOCKS[name]['content']['results']
|
|
}
|
|
|
|
def test_data_source_with_parent(self, app):
|
|
response = app.get('/toulouse-foederis/foederis/ds/emploi/?parent=5776416')
|
|
assert response.json['err'] == 0
|
|
assert response.json['last_update']
|
|
assert {d['text'] for d in response.json['data']} == {
|
|
'CHARGE OU CHARGEE DU DEVELOPPEMENT TERRITORIAL',
|
|
'DEVELOPPEUR OU DEVELOPPEUSE ECONOMIQUE',
|
|
'REFERENT OU REFERENTE DE QUARTIER',
|
|
'CHEF OU CHEFFE DE PROJET ACTIONS EUROPEENNES ET INTERNATIONALES',
|
|
}
|
|
|
|
def test_announce(self, app):
|
|
response = app.get('/toulouse-foederis/foederis/announce/')
|
|
content = response.json
|
|
assert content['err'] == 0
|
|
assert len(content['data_sources']) == 10
|
|
data = content['data']
|
|
assert len(data) == 5
|
|
|
|
def setof(name):
|
|
return {x[name] for x in data}
|
|
|
|
assert setof('id') == {'3782242', '4005540', '3866743', '4229013', '3450231'}
|
|
assert setof('text') == {
|
|
'AGENT OU AGENTE D ENTRETIEN EN CRECHE',
|
|
'Agent-e de collecte',
|
|
'ELAGUEUR',
|
|
'JARDINIER OU JARDINIERE',
|
|
'TEST PUBLIK',
|
|
}
|
|
assert setof('categorie') == {'', 'C-AGENTS EXECUTION', 'A-DIRECTION/CONCEPTION/ENCADT'}
|
|
assert setof('type_emploi') == {'Permanent', 'Temporaire'}
|
|
assert setof('filiere') == {'', 'FILIERE TECHNIQUE'}
|
|
assert setof('date') == {'2022-04-11', '2022-04-06', '2022-03-17', '2022-04-19', '2022-04-13'}
|
|
assert setof('date_fin_publication') == {'2022-05-11', '2022-04-25', '2022-12-31', '2022-04-30'}
|
|
|
|
# check html fields
|
|
assert '<br>Pilotage' in data[0]['description']
|
|
assert '<br>Connaissances' in data[0]['profil']
|
|
|
|
# check anti-chronological order on 'date'
|
|
assert list(sorted(setof('date'), reverse=True)) == list(x['date'] for x in data)
|
|
|
|
def test_announce_before_the_date(self, app, freezer):
|
|
freezer.move_to('2022-04-13')
|
|
response = app.get('/toulouse-foederis/foederis/announce/')
|
|
assert len(response.json['data']) == 4
|
|
|
|
def test_announce_after_the_date(self, app, freezer):
|
|
freezer.move_to('2022-05-01')
|
|
response = app.get('/toulouse-foederis/foederis/announce/')
|
|
assert len(response.json['data']) == 3
|
|
|
|
def test_announce_query_id(self, app):
|
|
response = app.get('/toulouse-foederis/foederis/announce/?id=4005540')
|
|
assert len(response.json['data']) == 1
|
|
assert response.json['data'][0]['id'] == '4005540'
|
|
|
|
def test_announce_query_q(self, app):
|
|
response = app.get('/toulouse-foederis/foederis/announce/?q=agent')
|
|
|
|
def setof(name):
|
|
return {x[name] for x in response.json['data']}
|
|
|
|
assert len(response.json['data']) == 2
|
|
assert setof('text') == {'AGENT OU AGENTE D ENTRETIEN EN CRECHE', 'Agent-e de collecte'}
|
|
|
|
def test_announce_query_type_emploi(self, app):
|
|
response = app.get('/toulouse-foederis/foederis/announce/?type_emploi=Permanent')
|
|
|
|
def setof(name):
|
|
return {x[name] for x in response.json['data']}
|
|
|
|
assert len(response.json['data']) == 4
|
|
assert setof('text') == {
|
|
'AGENT OU AGENTE D ENTRETIEN EN CRECHE',
|
|
'ELAGUEUR',
|
|
'JARDINIER OU JARDINIERE',
|
|
'TEST PUBLIK',
|
|
}
|
|
|
|
def test_announce_query_categorie(self, app):
|
|
response = app.get('/toulouse-foederis/foederis/announce/?categorie=C-AGENTS%20EXECUTION')
|
|
|
|
def setof(name):
|
|
return {x[name] for x in response.json['data']}
|
|
|
|
assert len(response.json['data']) == 3
|
|
assert setof('text') == {
|
|
'AGENT OU AGENTE D ENTRETIEN EN CRECHE',
|
|
'Agent-e de collecte',
|
|
'JARDINIER OU JARDINIERE',
|
|
}
|
|
|
|
def test_announce_query_filiere(self, app):
|
|
response = app.get('/toulouse-foederis/foederis/announce/?filiere=FILIERE%20TECHNIQUE')
|
|
|
|
def setof(name):
|
|
return {x[name] for x in response.json['data']}
|
|
|
|
assert len(response.json['data']) == 4
|
|
assert setof('text') == {
|
|
'AGENT OU AGENTE D ENTRETIEN EN CRECHE',
|
|
'Agent-e de collecte',
|
|
'JARDINIER OU JARDINIERE',
|
|
'TEST PUBLIK',
|
|
}
|
|
|
|
def test_announce_query_collectivite(self, app):
|
|
response = app.get('/toulouse-foederis/foederis/announce/?collectivite=MAIRIE%20DE%20TOULOUSE')
|
|
|
|
def setof(name):
|
|
return {x[name] for x in response.json['data']}
|
|
|
|
assert len(response.json['data']) == 3
|
|
assert setof('text') == {
|
|
'AGENT OU AGENTE D ENTRETIEN EN CRECHE',
|
|
'ELAGUEUR',
|
|
'JARDINIER OU JARDINIERE',
|
|
}
|
|
|
|
def test_announce_query_type_emploi_and_q(self, app):
|
|
response = app.get('/toulouse-foederis/foederis/announce/?type_emploi=Permanent&q=agen')
|
|
|
|
def setof(name):
|
|
return {x[name] for x in response.json['data']}
|
|
|
|
assert len(response.json['data']) == 1
|
|
assert setof('text') == {
|
|
'AGENT OU AGENTE D ENTRETIEN EN CRECHE',
|
|
}
|
|
|
|
def test_announce_pdf(self, resource, app):
|
|
response = app.get('/toulouse-foederis/foederis/announce/')
|
|
url = response.json['data'][0]['pdf_url']
|
|
assert url == 'http://testserver/toulouse-foederis/foederis/announce/4229013/pdf/'
|
|
# verify access is public
|
|
AccessRight.objects.all().delete()
|
|
response = app.get(url)
|
|
assert response.content.startswith(b'%PDF-1.4')
|
|
app.get('/toulouse-foederis/foederis/announce/111/pdf/', status=404)
|
|
assert response.headers['content-type'] == 'application/pdf'
|
|
|
|
def test_create_application(self, resource, app):
|
|
@httmock.urlmatch(path=r'^.*/data/candidature$')
|
|
def handler(url, request):
|
|
assert request.headers['content-type'] == 'application/json'
|
|
assert request.headers['api-key'] == APIKEY
|
|
payload = json.loads(request.body)
|
|
assert payload == {
|
|
"type_de_candidature": "E",
|
|
"annonce": 524522,
|
|
"R60284409": 170013,
|
|
"firstName": "John",
|
|
"lastName": "Doe",
|
|
"sexe": "H",
|
|
"date_de_naissance": "1985-03-06",
|
|
"R1249730": 93421,
|
|
"date_fin_autorisation_de_travail": "2023-05-09",
|
|
"rqth": "N",
|
|
"date_fin_rqth": "2023-05-08",
|
|
"permis_de_conduire": "A,B",
|
|
"fimo": "O",
|
|
"Date_delivrance_fimo": "2023-05-07",
|
|
"date_fin_validite_fimo": "2023-05-08",
|
|
"R1258320": 1258319,
|
|
"collectivite_agent": "Mairie de Toulouse",
|
|
"date_debut_disponibilite": "2023-05-02",
|
|
"date_fin_disponibilite": "2023-05-01",
|
|
"pretentions_salariales": "1000",
|
|
"adresse": "12 Sesame Street",
|
|
"code_postal": "77710",
|
|
"ville": "Nemours",
|
|
"telephone": "+33 636656565",
|
|
"email": "csechet@entrouvert.com",
|
|
"date_debut_contrat": "2023-05-06",
|
|
"date_fin_contrat": "2023-05-04",
|
|
"complement_information_candidature": "I need money.",
|
|
"R1261279": 1561049,
|
|
"accord_RGPD": True,
|
|
"R1249707": 157193,
|
|
"R60845221": 5776395,
|
|
"R60845244": 5776394,
|
|
"temps_de_travail_souhaite": "TC",
|
|
"duree_du_contrat_de_stage_apprentissage": "2h",
|
|
"ecole_centre_de_formation_mission_loc": "Ecole de la vie",
|
|
"intitule_diplome_vise": "BE",
|
|
"specialite_diplome": "Curling",
|
|
"R1249737": 1124022,
|
|
"dernier_diplome_obtenu": "BAC",
|
|
"derniere_classe_suivie": "Terminale",
|
|
}
|
|
|
|
return httmock.response(200, json.dumps({"code": 200, "results": [{"id": 42}]}))
|
|
|
|
@httmock.urlmatch()
|
|
def error_handler(url, request):
|
|
assert False, 'should not be reached'
|
|
|
|
with httmock.HTTMock(handler, error_handler):
|
|
response = app.post_json(
|
|
'/toulouse-foederis/foederis/create-application',
|
|
params={
|
|
"additional_informations": "I need money.",
|
|
"address": "12 Sesame Street",
|
|
"address_complement": "",
|
|
"agent_collectivity": "Mairie de Toulouse",
|
|
"aimed_diploma_level": "1124022",
|
|
"announce_id": "0524522",
|
|
"availability_end_date": "2023-05-01",
|
|
"availability_start_date": "2023-05-02",
|
|
"birth_date": "1985-03-06",
|
|
"city": "Nemours",
|
|
"civility": "170013",
|
|
"contract_end_date": "2023-05-04",
|
|
"contract_start_date": "2023-05-06",
|
|
"current_situation": "1258319",
|
|
"desired_work_time": "TC",
|
|
"diploma_name": "BE",
|
|
"diploma_speciality": "Curling",
|
|
"driving_license": "A,B",
|
|
"email": "csechet@entrouvert.com",
|
|
"fimo": "o",
|
|
"fimo_delivrance_date": "2023-05-07",
|
|
"fimo_end_validity_date": "2023-05-08",
|
|
"first_name": "John",
|
|
"gender": "H",
|
|
"internship_duration": "2h",
|
|
"job_family": "5776394",
|
|
"job_realm": "5776395",
|
|
"job_type": "157193",
|
|
"last_course_taken": "Terminale",
|
|
"last_name": "Doe",
|
|
"last_obtained_diploma": "BAC",
|
|
"nationality": "93421",
|
|
"origin": "1561049",
|
|
"origin_precisions": "",
|
|
"phone": "+33 636656565",
|
|
"rgpd_agreement": "tRuE",
|
|
"rqth": False,
|
|
"rqth_end_date": "2023-05-08",
|
|
"salary_expectations": "1000",
|
|
"school_name": "Ecole de la vie",
|
|
"type": "E",
|
|
"work_authorization_end_date": "2023-05-09",
|
|
"zip": "77710",
|
|
},
|
|
)
|
|
|
|
assert response.json["data"]["application_id"] == 42
|
|
|
|
|
|
def test_migration_0003_no_null_no_charfield(migration):
|
|
with connection.cursor() as cur:
|
|
cur.execute('SET CONSTRAINTS ALL IMMEDIATE')
|
|
apps = migration.before([('toulouse_foederis', '0001_initial')])
|
|
Document = apps.get_model('toulouse_foederis', 'Document')
|
|
Resource = apps.get_model('toulouse_foederis', 'Resource')
|
|
resource = Resource.objects.create(
|
|
title='Foederis',
|
|
slug='foederis',
|
|
description='Foederis',
|
|
url='https://passerelle.cutm-publik-preprod.nfrance.com/foederis/',
|
|
api_key=APIKEY,
|
|
)
|
|
doc_before = Document.objects.create(resource=resource, external_id='1', data=None, text=None, pdf='')
|
|
|
|
apps = migration.apply([('toulouse_foederis', '0003_no_null_no_charfield')])
|
|
Document = apps.get_model('toulouse_foederis', 'Document')
|
|
doc_after = Document.objects.get(pk=doc_before.pk)
|
|
assert doc_after.data == {}
|
|
assert doc_after.text == ''
|