349 lines
12 KiB
Python
349 lines
12 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 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'),
|
|
},
|
|
'categorie': {
|
|
'path': r'^/.*/data/categorie1$',
|
|
'query': 'viewIntegrationName=api_publik',
|
|
'content': get_json_content('categorie1'),
|
|
},
|
|
'filiere': {
|
|
'path': r'^.*/data/Filiere$',
|
|
'query': 'viewIntegrationName=api_publik',
|
|
'content': get_json_content('Filiere'),
|
|
},
|
|
'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 ['type-emploi', 'categorie', 'filiere']:
|
|
response = app.get(f'/toulouse-foederis/foederis/ds/{name}/')
|
|
assert response.json['err'] == 0
|
|
assert response.json['last_update']
|
|
assert {d['id'] for d in response.json['data']} == {
|
|
d['name'] for d in HTTP_MOCKS[name]['content']['results']
|
|
}
|
|
|
|
def test_announce(self, app):
|
|
response = app.get('/toulouse-foederis/foederis/announce/')
|
|
content = response.json
|
|
assert content['err'] == 0
|
|
assert len(content['data_sources']) == 3
|
|
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'
|