passerelle/tests/test_toulouse_foederis.py

703 lines
27 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 cgi
import json
import os
import urllib.parse
from io import BytesIO
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 = {
'origine-candidature': {
'path': r'^/.*/data/origine_candidature$',
'query': 'viewIntegrationName=api_publik&count=200',
'content': get_json_content('origine_candidature'),
},
'civilite': {
'path': r'^/.*/data/civilite$',
'query': 'viewIntegrationName=api_publik&count=200',
'content': get_json_content('civilite'),
},
'nationalite': {
'path': r'^/.*/data/nationalite1$',
'query': 'viewIntegrationName=api_publik&count=200',
'content': get_json_content('nationalite1'),
},
'situation-actuelle': {
'path': r'^/.*/data/situation_actuelle$',
'query': 'viewIntegrationName=api_publik&count=200',
'content': get_json_content('situation_actuelle'),
},
'domaine-emploi': {
'path': r'^/.*/data/domaine_emploi$',
'query': 'viewIntegrationName=api_publik&count=200',
'content': get_json_content('domaine_emploi'),
},
'sous-domaine-emploi': {
'path': r'^/.*/data/sous_domaine_emploi$',
'query': 'viewIntegrationName=api_publik&count=200',
'content': get_json_content('sous_domaine_emploi'),
},
'emploi': {
'path': r'^/.*/custom/emploi$',
'query': 'viewIntegrationName=api_publik',
'content': get_json_content('emploi'),
},
'niveau-diplome': {
'path': r'^/.*/data/niveau_diplome1$',
'query': 'viewIntegrationName=api_publik&count=200',
'content': get_json_content('niveau_diplome1'),
},
'habilitation': {
'path': r'^.*/data/habilitation$',
'query': 'viewIntegrationName=api_publik&count=200',
'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',
'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=6131210')
assert response.json['err'] == 0
assert response.json['last_update']
assert {d['text'] for d in response.json['data']} == {
"CHARGE OU CHARGEE D'OPERATIONS AMENAGEMENT",
'CHARGE OU CHARGEE DE MOBILITE',
'CHARGE OU CHARGEE DE LA SECURITE ROUTIERE',
'CHARGE OU CHARGEE D ANIMATION PLUI-H ET RLPI',
'CHARGE OU CHARGEE PLANIFICATION PROGRAMMATION',
"INSTRUCTEUR OU INSTRUCTRICE DES AUTORISATIONS D'URBANISME, ENSEIGNES ET TLPE",
}
def test_announce(self, app):
response = app.get('/toulouse-foederis/foederis/announce/')
content = response.json
assert content['err'] == 0
assert len(content['data_sources']) == 9
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'}
assert setof('offer_id') == {4005534, 3782130, 3782122, 3450229, 4005526}
# 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):
expected_payload = {
'type_de_candidature': 'E',
'candidature_spontane': 'O',
'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,
'type_emploi_souhaite': 'Emploi saisonnier',
'R60845221': [5776395, 5776396],
'R60845244': [5776394, 5776395],
'R15017962': [],
'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',
}
application_data = {
'additional_informations': 'I need money.',
'address': '12 Sesame Street',
'address_complement': '',
'agent_collectivity': 'Mairie de Toulouse',
'aimed_diploma_level': '1124022',
'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_families': ['5776394', '5776395'],
'job_realms': ['5776395', '5776396'],
'job_types': 'Emploi saisonnier',
'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',
}
@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 == expected_payload
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=application_data,
)
assert response.json['data']['application_id'] == 42
announce_id = '9876543'
response = app.post_json(
'/toulouse-foederis/foederis/create-application',
params=application_data | {'announce_id': announce_id},
status=404,
)
expected_payload.update(
{'annonce': int(announce_id), 'candidature_spontane': 'N', 'R14846954': 'ID_OFFRE'}
)
external_id = f'announce-{announce_id}'
resource.documents.create(external_id=external_id, data={'offer_id': 'ID_OFFRE'})
response = app.post_json(
'/toulouse-foederis/foederis/create-application',
params=application_data | {'announce_id': announce_id},
)
@pytest.mark.parametrize(
'phone_number,prefix',
[('+33 6 36 65 65 65', '33'), ('06 36 65 65 65', '33'), ('+596 6 36 65 65 65', '596')],
)
def test_create_application_phone_format(self, resource, app, phone_number, prefix):
@httmock.urlmatch(path=r'^.*/data/candidature$')
def handler(url, request):
payload = json.loads(request.body)
assert payload['telephone'] == f'+{prefix} 636656565'
return httmock.response(200, json.dumps({'code': 200, 'results': [{'id': 42}]}))
with httmock.HTTMock(handler):
response = app.post_json(
'/toulouse-foederis/foederis/create-application',
params={'phone': phone_number},
)
assert response.json['data']['application_id'] == 42
def test_create_application_phone_error(self, resource, app):
response = app.post_json(
'/toulouse-foederis/foederis/create-application',
params={'phone': 'mille sabords'},
)
assert response.json['err'] == 1
assert 'Couldn\'t recognize provided phone number' in response.json['err_desc']
def test_create_application_empty_phone(self, resource, app):
@httmock.urlmatch(path=r'^.*/data/candidature$')
def handler(url, request):
payload = json.loads(request.body)
assert 'telephone' not in payload
return httmock.response(200, json.dumps({'code': 200, 'results': [{'id': 42}]}))
with httmock.HTTMock(handler):
response = app.post_json(
'/toulouse-foederis/foederis/create-application',
params={'phone': ''},
)
assert response.json['err'] == 0
assert response.json['data']['application_id'] == 42
def test_attach_file(self, resource, app):
@httmock.urlmatch(path=r'^.*/data/candidature/424242/fields/cv$')
def handler(url, request):
assert request.headers['content-type'].startswith('multipart/form-data')
assert request.headers['api-key'] == APIKEY
_, headers = cgi.parse_header(request.headers['content-type'])
headers['boundary'] = bytes(headers['boundary'], 'utf-8')
headers['CONTENT-LENGTH'] = request.headers['Content-Length']
payload = cgi.parse_multipart(BytesIO(request.body), headers)
assert payload == {
'contentType': ['application/pdf'],
'value': ['base 64 content'],
'fileName': ['cv.pdf'],
}
return httmock.response(200, json.dumps({'code': 200, 'results': ['Field updated']}))
@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/attach-file',
params={
'application_id': '424242',
'name': 'cv',
'file': {
'content_type': 'application/pdf',
'content': 'base 64 content',
'filename': 'cv.pdf',
},
},
)
assert response.json['err'] == 0
def test_attach_degree(self, resource, app):
@httmock.urlmatch(path=r'^.*/data/diplome2$')
def create_degree_handler(url, request):
assert request.headers['content-type'] == 'application/json'
assert request.headers['api-key'] == APIKEY
payload = json.loads(request.body)
assert payload == {
'intitule_diplome': 'DUT anarchisme',
'R1258215': '424242',
}
return httmock.response(200, json.dumps({'code': 200, 'results': [{'id': 'DEGREE_ID'}]}))
@httmock.urlmatch(path=r'^.*/data/diplome2/DEGREE_ID/fields/justificatif_diplome$')
def degree_file_handler(url, request):
assert request.headers['content-type'].startswith('multipart/form-data')
assert request.headers['api-key'] == APIKEY
_, headers = cgi.parse_header(request.headers['content-type'])
headers['boundary'] = bytes(headers['boundary'], 'utf-8')
headers['CONTENT-LENGTH'] = request.headers['Content-Length']
payload = cgi.parse_multipart(BytesIO(request.body), headers)
assert payload == {
'contentType': ['application/pdf'],
'value': ['base 64 content'],
'fileName': ['cv.pdf'],
}
return httmock.response(200, json.dumps({'code': 200, 'results': [{'id': 'DEGREE_ID'}]}))
@httmock.urlmatch()
def error_handler(url, request):
assert False, 'should not be reached'
with httmock.HTTMock(create_degree_handler, degree_file_handler, error_handler):
response = app.post_json(
'/toulouse-foederis/foederis/attach-degree',
params={
'application_id': '424242',
'name': 'DUT anarchisme',
'file': {
'content_type': 'application/pdf',
'content': 'base 64 content',
'filename': 'cv.pdf',
},
},
)
assert response.json['err'] == 0
with httmock.HTTMock(create_degree_handler, degree_file_handler, error_handler):
response = app.post_json(
'/toulouse-foederis/foederis/attach-degree',
params={
'application_id': '424242',
'name': 'DUT anarchisme',
'file': None,
},
)
assert response.json['err'] == 0
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 == ''