2022-06-20 17:32:16 +02:00
|
|
|
# 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
|
|
|
|
|
2022-09-15 15:00:28 +02:00
|
|
|
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),
|
|
|
|
]
|
|
|
|
|
2022-06-20 17:32:16 +02:00
|
|
|
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
|
2022-07-13 12:08:05 +02:00
|
|
|
|
2022-06-20 17:32:16 +02:00
|
|
|
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
|
|
|
|
|
|
|
|
|
2022-08-05 13:12:15 +02:00
|
|
|
@pytest.mark.parametrize('immat,sent_immat', [('747-xT 01', '01XT0747'), ('FD-734-hR', 'FD734HR')])
|
2022-06-20 17:32:16 +02:00
|
|
|
@mock.patch('passerelle.utils.Request.post', side_effect=(TOKEN, FINITION))
|
2022-08-05 13:12:15 +02:00
|
|
|
def test_get_vehicle_theorical_finition(mocked_post, app, conn, immat, sent_immat):
|
2022-06-20 17:32:16 +02:00
|
|
|
url = reverse(
|
|
|
|
'generic-endpoint',
|
|
|
|
kwargs={'connector': 'sivin', 'endpoint': 'consulterfinitiontheoriqueparimmat', 'slug': conn.slug},
|
|
|
|
)
|
2022-08-05 13:12:15 +02:00
|
|
|
resp = app.get(url, params={'apikey': 'sivinkey', 'immat': immat}).json
|
2022-06-20 17:32:16 +02:00
|
|
|
assert mocked_post.call_count == 2
|
2022-08-05 13:12:15 +02:00
|
|
|
assert mocked_post.mock_calls[-1].kwargs['json'] == {'immat': sent_immat}
|
2022-06-20 17:32:16 +02:00
|
|
|
assert not resp['err']
|
2022-09-15 15:00:28 +02:00
|
|
|
assert 'is_euro5' in resp['data']
|
|
|
|
resp['data'].pop('is_euro5')
|
2022-06-20 17:32:16 +02:00
|
|
|
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'
|
|
|
|
)
|
2022-09-15 15:00:28 +02:00
|
|
|
|
|
|
|
|
|
|
|
@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
|