entrouvert
/
passerelle
15
0
Fork 0
Code Issues Pull Requests 14 Releases Activity
passerelle/tests/test_sivin.py

240 lines
8.2 KiB
Python
Raw Permalink Blame History

# Copyright (C) 2022 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 json
from unittest import mock
import pytest
from django.contrib.contenttypes.models import ContentType
from django.urls import reverse
from requests.exceptions import ReadTimeout
import tests.utils
from passerelle.apps.sivin.models import Resource
from passerelle.base.models import AccessRight, ApiUser
pytestmark = pytest.mark.django_db
EURO5_INDEXES = [
('0555*0651C Euro 4', False),
('0555*0651G Euro 5', True),
('0555*0651G EURO 5', True),
('0555*0874G (EURO 5)', True),
('134/2014EURO4', False),
('134/2014EURO5', True),
('175/2007*195/2013EURO6', True),
('200/55*2008/74EURO5', True),
('2005/55*51euro5', True),
('2006/51G-EUROV (G)', True),
('2006/96euro4', False),
('59/2009*2016/1718EURO6', True),
('595/2009*2018/932DEUROVI', True),
('595/2009*2018/932EURO', False),
('595/2009*627/2014EUROIV', False),
('70/220*1999/102EURO3', False),
('EURO 3', False),
('EURO III', False),
('EURO2', False),
('EURO5G', True),
('EURO6B', True),
('2001/100A', False),
]
VEHICLE_DETAILS = {
'carrosserie': 'BERLINE',
'clEnvironPrf': '70/220 2001/100EURO3',
'codifVin': 'VF7FCKFVB26857835',
'genreVCG': 'VP',
'immatSiv': 'FS032GM',
'genreVPrf': 'VP',
'date1erCir': '2003-11-21',
'nSiren': '000000000',
}
VEHICLE_THEORICAL_FINITION = {
'carrosserie': 'COMBISPACE',
'clEnvironPrf': '2001/100A',
'codifVin': 'VF7GJRHYK93204774',
'genreVCG': 'VP',
'immatSiv': '01XT0747',
'genreVPrf': 'VP',
'date1erCir': '2004-11-19',
'nSiren': '000000000',
}
EXPIRED_TOKEN_MESSAGE = (
'<ams:fault xmlns:ams="http://wso2.org/apimanager/security">'
'<ams:code>900901</ams:code><ams:message>Invalid Credentials</ams:message>'
'<ams:description>Invalid Credentials. Make sure you have provided the correct security credentials</ams:description>'
'</ams:fault>'
)
TOKEN_401 = tests.utils.FakedResponse(
content='{"error_description": "Client Authentication failed.", "error": "invalid_client"}',
status_code=401,
headers={'Content-Type': 'application/json'},
)
TOKEN = tests.utils.FakedResponse(
content='{"access_token": "token_value", "scope": "default", "token_type": "Bearer", "expires_in": 3600}',
status_code=200,
headers={'Content-Type': 'application/json'},
)
EXPIRED_TOKEN = tests.utils.FakedResponse(
content=EXPIRED_TOKEN_MESSAGE,
status_code=500,
)
VIN = tests.utils.FakedResponse(
content=json.dumps(VEHICLE_DETAILS),
status_code=200,
headers={'Content-Type': 'application/json'},
)
VEHICLES = tests.utils.FakedResponse(
content='{"vins":["VF1FB30A511331122"]}', status_code=200, headers={'Content-Type': 'application/json'}
)
FINITION = tests.utils.FakedResponse(
content=json.dumps(VEHICLE_THEORICAL_FINITION),
status_code=200,
headers={'Content-Type': 'application/json'},
)
@pytest.fixture
def conn():
api_user = ApiUser.objects.create(username='sivin', keytype='API', key='sivinkey')
connector = Resource.objects.create(
title='Test', slug='test', consumer_key='key', consumer_secret='secret', environment='test'
)
obj_type = ContentType.objects.get_for_model(Resource)
AccessRight.objects.create(
codename='can_access', apiuser=api_user, resource_type=obj_type, resource_pk=connector.pk
)
return connector
@mock.patch('passerelle.utils.Request.post')
def test_no_api_key(mocked_post, app, conn):
url = reverse(
'generic-endpoint',
kwargs={'connector': 'sivin', 'endpoint': 'consultervehiculeparvin', 'slug': conn.slug},
)
app.get(url, params={'vin': 'vin'}, status=403)
assert mocked_post.call_count == 0
@mock.patch('passerelle.utils.Request.post')
def test_wrong_credentials(mocked_post, app, conn):
mocked_post.return_value = TOKEN_401
url = reverse(
'generic-endpoint',
kwargs={'connector': 'sivin', 'endpoint': 'consultervehiculeparvin', 'slug': conn.slug},
)
resp = app.get(url, params={'apikey': 'sivinkey', 'vin': 'VF1BA0E0514143067'}).json
assert mocked_post.call_count == 1
assert mocked_post.call_args[0][0] == 'https://api.rec.sivin.fr/token'
assert resp['err']
assert resp['err_desc'] == 'Failed to get token. Error: 401'
@mock.patch('passerelle.utils.Request.post')
def test_get_token(mocked_post, app, conn):
mocked_post.return_value = TOKEN
conn.get_token()
assert mocked_post.call_count == 1
assert mocked_post.call_args[0][0] == 'https://api.rec.sivin.fr/token'
# token is in cache so no more http hits
conn.get_token()
assert mocked_post.call_count == 1
@mock.patch('passerelle.utils.Request.post', side_effect=(TOKEN, VIN))
def test_get_details_by_vin(mocked_post, app, conn):
url = reverse(
'generic-endpoint',
kwargs={'connector': 'sivin', 'endpoint': 'consultervehiculeparvin', 'slug': conn.slug},
)
resp = app.get(url, params={'apikey': 'sivinkey', 'vin': 'VF1BA0E0514143067'}).json
assert mocked_post.call_count == 2
assert not resp['err']
assert resp['data'] == VEHICLE_DETAILS
@mock.patch('passerelle.utils.Request.post', side_effect=(TOKEN, VEHICLES))
def test_get_vehicles_by_siren(mocked_post, app, conn):
url = reverse(
'generic-endpoint',
kwargs={'connector': 'sivin', 'endpoint': 'consulterflotteparsiren', 'slug': conn.slug},
)
resp = app.get(url, params={'apikey': 'sivinkey', 'siren': '000399634'}).json
assert mocked_post.call_count == 2
assert not resp['err']
for item in resp['data']:
assert 'id' in item
assert 'text' in item
@mock.patch('passerelle.utils.Request.post', side_effect=(TOKEN, EXPIRED_TOKEN))
def test_get_with_expired_token(mocked_post, app, conn):
url = reverse(
'generic-endpoint',
kwargs={'connector': 'sivin', 'endpoint': 'consulterflotteparsiren', 'slug': conn.slug},
)
resp = app.get(url, params={'apikey': 'sivinkey', 'siren': '000399634'}).json
assert mocked_post.call_count == 2
assert resp['err']
assert resp['err_desc'] == EXPIRED_TOKEN_MESSAGE
@pytest.mark.parametrize('immat,sent_immat', [('747-xT 01', '01XT0747'), ('FD-734-hR', 'FD734HR')])
@mock.patch('passerelle.utils.Request.post', side_effect=(TOKEN, FINITION))
def test_get_vehicle_theorical_finition(mocked_post, app, conn, immat, sent_immat):
url = reverse(
'generic-endpoint',
kwargs={'connector': 'sivin', 'endpoint': 'consulterfinitiontheoriqueparimmat', 'slug': conn.slug},
)
resp = app.get(url, params={'apikey': 'sivinkey', 'immat': immat}).json
assert mocked_post.call_count == 2
assert mocked_post.mock_calls[-1].kwargs['json'] == {'immat': sent_immat}
assert not resp['err']
assert 'is_euro5' in resp['data']
resp['data'].pop('is_euro5')
assert resp['data'] == VEHICLE_THEORICAL_FINITION
@mock.patch('passerelle.utils.Request.post', side_effect=ReadTimeout('timeout'))
def test_connection_timeout(mocked_post, app, conn):
url = reverse(
'generic-endpoint',
kwargs={'connector': 'sivin', 'endpoint': 'consulterflotteparsiren', 'slug': conn.slug},
)
resp = app.get(url, params={'apikey': 'sivinkey', 'siren': '000399634'}).json
assert mocked_post.call_count == 1
assert resp['err']
assert (
resp['err_desc']
== 'failed to call https://api.rec.sivin.fr/sivin/v2/consulterflotteparsiren: timeout'
)
@pytest.mark.parametrize('clEnvironPrf, is_euro5', EURO5_INDEXES)
def test_compute_euro_index(app, conn, clEnvironPrf, is_euro5):
assert conn.is_euro5(clEnvironPrf) == is_euro5
View Git Blame Copy Permalink