From 482891f2fd23be7f1836ad98023a306ce2e5a940 Mon Sep 17 00:00:00 2001 From: "gerome.perrin" Date: Wed, 1 Jul 2026 09:31:54 +0200 Subject: [PATCH 1/7] [fix] handle journeys error properly --- c2corg_api/views/navitia.py | 19 ++++++++++--------- 1 file changed, 10 insertions(+), 9 deletions(-) diff --git a/c2corg_api/views/navitia.py b/c2corg_api/views/navitia.py index 12771d46a..a423d4447 100644 --- a/c2corg_api/views/navitia.py +++ b/c2corg_api/views/navitia.py @@ -14,7 +14,7 @@ from c2corg_api.views.waypoint import build_reachable_waypoints_query from c2corg_api.views.route import build_reachable_route_query_with_waypoints from shapely.geometry import Point -from pyramid.httpexceptions import HTTPBadRequest, HTTPInternalServerError +from pyramid.httpexceptions import HTTPBadRequest, HTTPInternalServerError, HTTPNotFound, HTTPUnauthorized from pyramid.response import Response from cornice.resource import resource, view from c2corg_api.views import cors_policy, to_json_dict @@ -751,20 +751,21 @@ def handle_navitia_response(response): Raises HTTP* exceptions otherwise. """ - if response.status_code == 401: - raise HTTPInternalServerError("Authentication error with Navitia API") + if response.status_code == 400: + raise HTTPBadRequest("invalid_param") - elif response.status_code == 400: - raise HTTPBadRequest("Invalid parameters for Navitia API") + elif response.status_code == 401: + raise HTTPUnauthorized("auth_error") elif response.status_code == 404: - # 404: unable to find an object, no response - # This should not raise any exception error, just return None - return None + raise HTTPNotFound(response.json()['error']['id']) + + elif response.status_code == 500: + raise HTTPInternalServerError("navitia_internal_error") elif not response.ok: raise HTTPInternalServerError( - "Navitia API error: %d " % response.status_code + "unknown_error" ) return response.json() From 80730d6555e91df12e730620f056f33bead196df Mon Sep 17 00:00:00 2001 From: "gerome.perrin" Date: Thu, 2 Jul 2026 09:20:03 +0200 Subject: [PATCH 2/7] [fix] fix navitia unit test with recent changes regarding journeys error handling --- c2corg_api/tests/views/test_navitia.py | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/c2corg_api/tests/views/test_navitia.py b/c2corg_api/tests/views/test_navitia.py index c38df16cd..db5ad9794 100644 --- a/c2corg_api/tests/views/test_navitia.py +++ b/c2corg_api/tests/views/test_navitia.py @@ -2,7 +2,7 @@ from shapely.geometry import Polygon import json import requests -from pyramid.httpexceptions import HTTPBadRequest +from pyramid.httpexceptions import HTTPBadRequest, HTTPNotFound, HTTPUnauthorized from unittest.mock import call from c2corg_api.tests import BaseTestCase from c2corg_api.views import to_json_dict @@ -1207,7 +1207,7 @@ def test_handle_navitia_response_401(self): mock_response = mock.Mock() mock_response.status_code = 401 - with self.assertRaises(HTTPInternalServerError): + with self.assertRaises(HTTPUnauthorized): handle_navitia_response(mock_response) def test_handle_navitia_response_400(self): @@ -1220,18 +1220,18 @@ def test_handle_navitia_response_400(self): def test_handle_navitia_response_404_no_journey(self): mock_response = mock.Mock() mock_response.status_code = 404 - mock_response.json.return_value = {"error": {"id": "no_origin"}} + mock_response.json.return_value = {"error": {"id": "no_solution"}} + + with self.assertRaises(HTTPNotFound): + handle_navitia_response(mock_response) - result = handle_navitia_response(mock_response) - self.assertIsNone(result) def test_handle_navitia_response_404_other_error(self): mock_response = mock.Mock() mock_response.status_code = 404 - mock_response.json.return_value = {"error": {"id": "some_other_error"}} - - result = handle_navitia_response(mock_response) - self.assertIsNone(result) + mock_response.json.return_value = {"error": {"id": "unknown_object"}} + with self.assertRaises(HTTPNotFound): + handle_navitia_response(mock_response) def test_handle_navitia_response_other_error(self): mock_response = mock.Mock() From 4cd3ed3d07a203b1b234a4722aba9e5d66cc0509 Mon Sep 17 00:00:00 2001 From: "gerome.perrin" Date: Thu, 2 Jul 2026 09:20:51 +0200 Subject: [PATCH 3/7] [feat] add unit tests for stoparea and waypoint_stoparea --- c2corg_api/tests/views/test_stopareas.py | 98 ++++++++++ .../tests/views/test_waypoint_stoparea.py | 167 ++++++++++++++++++ c2corg_api/views/waypoint_stoparea.py | 17 +- 3 files changed, 281 insertions(+), 1 deletion(-) create mode 100644 c2corg_api/tests/views/test_stopareas.py create mode 100644 c2corg_api/tests/views/test_waypoint_stoparea.py diff --git a/c2corg_api/tests/views/test_stopareas.py b/c2corg_api/tests/views/test_stopareas.py new file mode 100644 index 000000000..54c37cb8f --- /dev/null +++ b/c2corg_api/tests/views/test_stopareas.py @@ -0,0 +1,98 @@ +from c2corg_api.models.stoparea import Stoparea +from c2corg_api.tests.views import BaseDocumentTestRest + + +class TestStopareaRest(BaseDocumentTestRest): + + def setUp(self): # noqa + self.set_prefix_and_model( + "/stopareas", "sa", Stoparea, None, None) + BaseDocumentTestRest.setUp(self) + self._add_test_data() + + def _add_test_data(self): + # Create a stoparea for testing + stoparea = Stoparea( + stoparea_id=1, + navitia_id='nav1', + stoparea_name='Stop Area 1', + line='line1', + operator='operator1', + ) + + self.session.add(stoparea) + self.session.flush() + + def test_collection_get(self): + """Test getting list of stopareas""" + response = self.app.get('/stopareas', status=200) + result = response.json + + assert result['total_results'] >= 0 + assert isinstance(result['documents'], list) + + def test_get_stoparea_not_found(self): + """Test getting a stoparea that doesn't exist""" + response = self.app.get('/stopareas/999999', status=404) + result = response.json + + assert result['error'] == 'Stoparea not found' + + def test_get_stoparea_found(self): + """Test getting a stoparea that exists""" + response = self.app.get('/stopareas/1', status=200) + result = response.json + + assert result['id'] == 1 + assert result['navitia_id'] == 'nav1' + assert result['stoparea_name'] == 'Stop Area 1' + assert result['line'] == 'line1' + assert result['operator'] == 'operator1' + + +class TestStopareaInfoRest(BaseDocumentTestRest): + + def setUp(self): # noqa + self.set_prefix_and_model( + "/stopareas", "sa", Stoparea, None, None) + BaseDocumentTestRest.setUp(self) + self._add_test_data() + + def _add_test_data(self): + # Create a stoparea for testing + stoparea = Stoparea( + stoparea_id=1, + navitia_id='nav1', + stoparea_name='Stop Area 1', + line='line1', + operator='operator1' + ) + + self.session.add(stoparea) + self.session.flush() + + def test_get_info_stoparea_not_found(self): + """Test getting info for a stoparea that doesn't exist""" + response = self.app.get('/stopareas/999999/fr/info', status=404) + result = response.json + + assert result['error'] == 'Stoparea not found' + + def test_get_info_stoparea_found(self): + """Test getting info for a stoparea that exists""" + response = self.app.get('/stopareas/1/fr/info', status=200) + result = response.json + + assert result['stoparea_id'] == 1 + assert result['attributes']['navitia_id'] == 'nav1' + assert result['attributes']['stoparea_name'] == 'Stop Area 1' + assert result['attributes']['line'] == 'line1' + assert result['attributes']['operator'] == 'operator1' + + def test_get_info_stoparea_lang_not_found(self): + """Test getting info for a stoparea that doesn't exist""" + response = self.app.get('/stopareas/1/invalid/info', status=400) + result = response.json + + assert result['errors'][0]['description'] == 'invalid lang' + diff --git a/c2corg_api/tests/views/test_waypoint_stoparea.py b/c2corg_api/tests/views/test_waypoint_stoparea.py new file mode 100644 index 000000000..d0fec1b17 --- /dev/null +++ b/c2corg_api/tests/views/test_waypoint_stoparea.py @@ -0,0 +1,167 @@ +from c2corg_api.models.waypoint import ( + Waypoint, WaypointLocale) +from c2corg_api.models.stoparea import Stoparea +from c2corg_api.models.waypoint_stoparea import WaypointStoparea + +from c2corg_api.tests.views import BaseDocumentTestRest + + +class TestWaypointStopareaRest(BaseDocumentTestRest): + + def setUp(self): # noqa + self.set_prefix_and_model( + "/waypoints_stopareas", "ws", WaypointStoparea, None, None) + BaseDocumentTestRest.setUp(self) + self._add_test_data() + + def _add_test_data(self): + # Create waypoints for testing + waypoint1 = Waypoint( + waypoint_type='summit', + elevation=3779 + ) + + locale_en = WaypointLocale( + lang='en', title='Mont Pourri', access='y') + waypoint1.locales.append(locale_en) + + self.session.add(waypoint1) + self.session.flush() + self.waypoint1 = waypoint1 + + waypoint2 = Waypoint( + waypoint_type='summit', + elevation=3000 + ) + + locale_en2 = WaypointLocale( + lang='en', title='Another Summit', access='y') + waypoint2.locales.append(locale_en2) + + self.session.add(waypoint2) + self.session.flush() + self.waypoint2 = waypoint2 + + # Create stopareas for testing + stoparea1 = Stoparea( + stoparea_id=1, + navitia_id='nav1', + stoparea_name='Stop Area 1', + line='line1', + operator='operator1' + ) + + self.session.add(stoparea1) + self.session.flush() + self.stoparea1 = stoparea1 + + stoparea2 = Stoparea( + stoparea_id=2, + navitia_id='nav2', + stoparea_name='Stop Area 2', + line='line2', + operator='operator2' + ) + + self.session.add(stoparea2) + self.session.flush() + self.stoparea2 = stoparea2 + + # Create waypoint-stoparea associations + waypoint_stoparea1 = WaypointStoparea( + waypoint_stoparea_id=1, + waypoint_id=waypoint1.document_id, + stoparea_id=stoparea1.stoparea_id, + distance=100.0 + ) + + self.session.add(waypoint_stoparea1) + self.session.flush() + self.waypoint_stoparea1 = waypoint_stoparea1 + + waypoint_stoparea2 = WaypointStoparea( + waypoint_stoparea_id=2, + waypoint_id=waypoint1.document_id, + stoparea_id=stoparea2.stoparea_id, + distance=200.0 + ) + + self.session.add(waypoint_stoparea2) + self.session.flush() + self.waypoint_stoparea2 = waypoint_stoparea2 + + def test_get_stopareas_by_waypoint(self): + """Test getting stopareas for a waypoint""" + response = self.app.get('/waypoints/{}/stopareas'.format( + self.waypoint1.document_id), status=200) + result = response.json + + self.assertEqual(result['waypoint_id'], self.waypoint1.document_id) + self.assertEqual(len(result['stopareas']), 2) + + stoparea1 = result['stopareas'][0] + self.assertEqual(stoparea1['id'], 1) + self.assertEqual(stoparea1['navitia_id'], 'nav1') + self.assertEqual(stoparea1['stoparea_name'], 'Stop Area 1') + self.assertEqual(stoparea1['line'], 'line1') + self.assertEqual(stoparea1['operator'], 'operator1') + self.assertEqual(stoparea1['distance'], 100.0) + + stoparea2 = result['stopareas'][1] + self.assertEqual(stoparea2['id'], 2) + self.assertEqual(stoparea2['navitia_id'], 'nav2') + self.assertEqual(stoparea2['stoparea_name'], 'Stop Area 2') + self.assertEqual(stoparea2['line'], 'line2') + self.assertEqual(stoparea2['operator'], 'operator2') + self.assertEqual(stoparea2['distance'], 200.0) + + def test_get_stopareas_by_waypoint_not_found(self): + """Test getting stopareas for a waypoint that doesn't exist""" + # We'll test that it returns an empty array instead of 404 + response = self.app.get('/waypoints/999999/stopareas', status=200) + result = response.json + + self.assertEqual(result['waypoint_id'], 999999) + self.assertEqual(result['stopareas'], []) + + def test_get_is_reachable_true(self): + """Test checking if a waypoint is reachable (has stopareas)""" + response = self.app.get('/waypoints/{}/isReachable'.format( + self.waypoint1.document_id), status=200) + result = response.json + + self.assertTrue(result) + + def test_get_is_reachable_false(self): + """Test checking if a waypoint is not reachable (no stopareas)""" + response = self.app.get('/waypoints/{}/isReachable'.format( + self.waypoint2.document_id), status=200) + result = response.json + + self.assertFalse(result) + + def test_get_info(self): + """Test getting info for a waypoint-stoparea""" + + response = self.app.get( + '/waypoints_stopareas/1/en/info', status=200) + result = response.json + + self.assertEqual(result['waypoint_stoparea_id'], self.waypoint_stoparea1.waypoint_stoparea_id) + self.assertEqual(result['attributes']['distance'], 100.0) + self.assertEqual(result['attributes']['waypoint_id'], self.waypoint1.document_id) + self.assertEqual(result['attributes']['stoparea_id'], 1) + + def test_get_info_not_found(self): + """Test getting info for a waypoint-stoparea that doesn't exist""" + response = self.app.get( + '/waypoints_stopareas/999999/en/info', status=404) + self.assertEqual(response.json_body, {'error': 'Waypoint Stoparea not found'}) + + def test_get_info_lang_not_found(self): + """Test getting info for a waypoint-stoparea that exist with an invalid lang""" + response = self.app.get( + '/waypoints_stopareas/1/invalid/info', status=400) + + # Updated to match actual JSON response format + self.assertEqual(response.json_body['errors'][0]['description'], "invalid lang") \ No newline at end of file diff --git a/c2corg_api/views/waypoint_stoparea.py b/c2corg_api/views/waypoint_stoparea.py index e446d451a..dea011efa 100644 --- a/c2corg_api/views/waypoint_stoparea.py +++ b/c2corg_api/views/waypoint_stoparea.py @@ -28,7 +28,22 @@ class WaypointStopareaInfoRest(DocumentInfoRest): @view(validators=[validate_id, validate_lang]) def get(self): - return self._get_document_info(schema_waypoint_stoparea), + waypoint_stoparea_id = self.request.matchdict['id'] + waypoint_stoparea = DBSession.query(WaypointStoparea).filter_by( + waypoint_stoparea_id=waypoint_stoparea_id).first() + + if not waypoint_stoparea: + self.request.response.status = 404 + return {'error': 'Waypoint Stoparea not found'} + + return { + 'waypoint_stoparea_id': waypoint_stoparea.waypoint_stoparea_id, + 'attributes': { + 'stoparea_id': waypoint_stoparea.stoparea_id, + 'waypoint_id': waypoint_stoparea.waypoint_id, + 'distance': waypoint_stoparea.distance, + } + } def validate_waypoint_id(request, *args, **kwargs): From 5523080a3c635476a0da2ba2b9ab9745c28fffb8 Mon Sep 17 00:00:00 2001 From: "gerome.perrin" Date: Thu, 2 Jul 2026 11:54:01 +0200 Subject: [PATCH 4/7] [fix] fix flake8 lint errors --- c2corg_api/tests/views/test_navitia.py | 3 +- c2corg_api/tests/views/test_stopareas.py | 17 +++--- .../tests/views/test_waypoint_stoparea.py | 60 ++++++++++++------- c2corg_api/views/navitia.py | 3 +- c2corg_api/views/waypoint_stoparea.py | 3 +- 5 files changed, 51 insertions(+), 35 deletions(-) diff --git a/c2corg_api/tests/views/test_navitia.py b/c2corg_api/tests/views/test_navitia.py index db5ad9794..ca44b7088 100644 --- a/c2corg_api/tests/views/test_navitia.py +++ b/c2corg_api/tests/views/test_navitia.py @@ -2,7 +2,8 @@ from shapely.geometry import Polygon import json import requests -from pyramid.httpexceptions import HTTPBadRequest, HTTPNotFound, HTTPUnauthorized +from pyramid.httpexceptions import (HTTPBadRequest, HTTPNotFound, + HTTPUnauthorized) from unittest.mock import call from c2corg_api.tests import BaseTestCase from c2corg_api.views import to_json_dict diff --git a/c2corg_api/tests/views/test_stopareas.py b/c2corg_api/tests/views/test_stopareas.py index 54c37cb8f..8ae1fa7ff 100644 --- a/c2corg_api/tests/views/test_stopareas.py +++ b/c2corg_api/tests/views/test_stopareas.py @@ -19,7 +19,7 @@ def _add_test_data(self): line='line1', operator='operator1', ) - + self.session.add(stoparea) self.session.flush() @@ -27,7 +27,7 @@ def test_collection_get(self): """Test getting list of stopareas""" response = self.app.get('/stopareas', status=200) result = response.json - + assert result['total_results'] >= 0 assert isinstance(result['documents'], list) @@ -35,14 +35,14 @@ def test_get_stoparea_not_found(self): """Test getting a stoparea that doesn't exist""" response = self.app.get('/stopareas/999999', status=404) result = response.json - + assert result['error'] == 'Stoparea not found' def test_get_stoparea_found(self): """Test getting a stoparea that exists""" response = self.app.get('/stopareas/1', status=200) result = response.json - + assert result['id'] == 1 assert result['navitia_id'] == 'nav1' assert result['stoparea_name'] == 'Stop Area 1' @@ -67,7 +67,7 @@ def _add_test_data(self): line='line1', operator='operator1' ) - + self.session.add(stoparea) self.session.flush() @@ -75,14 +75,14 @@ def test_get_info_stoparea_not_found(self): """Test getting info for a stoparea that doesn't exist""" response = self.app.get('/stopareas/999999/fr/info', status=404) result = response.json - + assert result['error'] == 'Stoparea not found' def test_get_info_stoparea_found(self): """Test getting info for a stoparea that exists""" response = self.app.get('/stopareas/1/fr/info', status=200) result = response.json - + assert result['stoparea_id'] == 1 assert result['attributes']['navitia_id'] == 'nav1' assert result['attributes']['stoparea_name'] == 'Stop Area 1' @@ -93,6 +93,5 @@ def test_get_info_stoparea_lang_not_found(self): """Test getting info for a stoparea that doesn't exist""" response = self.app.get('/stopareas/1/invalid/info', status=400) result = response.json - - assert result['errors'][0]['description'] == 'invalid lang' + assert result['errors'][0]['description'] == 'invalid lang' diff --git a/c2corg_api/tests/views/test_waypoint_stoparea.py b/c2corg_api/tests/views/test_waypoint_stoparea.py index d0fec1b17..bb98bbcad 100644 --- a/c2corg_api/tests/views/test_waypoint_stoparea.py +++ b/c2corg_api/tests/views/test_waypoint_stoparea.py @@ -20,11 +20,11 @@ def _add_test_data(self): waypoint_type='summit', elevation=3779 ) - + locale_en = WaypointLocale( lang='en', title='Mont Pourri', access='y') waypoint1.locales.append(locale_en) - + self.session.add(waypoint1) self.session.flush() self.waypoint1 = waypoint1 @@ -33,11 +33,11 @@ def _add_test_data(self): waypoint_type='summit', elevation=3000 ) - + locale_en2 = WaypointLocale( lang='en', title='Another Summit', access='y') waypoint2.locales.append(locale_en2) - + self.session.add(waypoint2) self.session.flush() self.waypoint2 = waypoint2 @@ -50,7 +50,7 @@ def _add_test_data(self): line='line1', operator='operator1' ) - + self.session.add(stoparea1) self.session.flush() self.stoparea1 = stoparea1 @@ -62,7 +62,7 @@ def _add_test_data(self): line='line2', operator='operator2' ) - + self.session.add(stoparea2) self.session.flush() self.stoparea2 = stoparea2 @@ -74,7 +74,7 @@ def _add_test_data(self): stoparea_id=stoparea1.stoparea_id, distance=100.0 ) - + self.session.add(waypoint_stoparea1) self.session.flush() self.waypoint_stoparea1 = waypoint_stoparea1 @@ -85,7 +85,7 @@ def _add_test_data(self): stoparea_id=stoparea2.stoparea_id, distance=200.0 ) - + self.session.add(waypoint_stoparea2) self.session.flush() self.waypoint_stoparea2 = waypoint_stoparea2 @@ -95,10 +95,10 @@ def test_get_stopareas_by_waypoint(self): response = self.app.get('/waypoints/{}/stopareas'.format( self.waypoint1.document_id), status=200) result = response.json - + self.assertEqual(result['waypoint_id'], self.waypoint1.document_id) self.assertEqual(len(result['stopareas']), 2) - + stoparea1 = result['stopareas'][0] self.assertEqual(stoparea1['id'], 1) self.assertEqual(stoparea1['navitia_id'], 'nav1') @@ -120,7 +120,7 @@ def test_get_stopareas_by_waypoint_not_found(self): # We'll test that it returns an empty array instead of 404 response = self.app.get('/waypoints/999999/stopareas', status=200) result = response.json - + self.assertEqual(result['waypoint_id'], 999999) self.assertEqual(result['stopareas'], []) @@ -129,7 +129,7 @@ def test_get_is_reachable_true(self): response = self.app.get('/waypoints/{}/isReachable'.format( self.waypoint1.document_id), status=200) result = response.json - + self.assertTrue(result) def test_get_is_reachable_false(self): @@ -137,7 +137,7 @@ def test_get_is_reachable_false(self): response = self.app.get('/waypoints/{}/isReachable'.format( self.waypoint2.document_id), status=200) result = response.json - + self.assertFalse(result) def test_get_info(self): @@ -146,22 +146,38 @@ def test_get_info(self): response = self.app.get( '/waypoints_stopareas/1/en/info', status=200) result = response.json - - self.assertEqual(result['waypoint_stoparea_id'], self.waypoint_stoparea1.waypoint_stoparea_id) - self.assertEqual(result['attributes']['distance'], 100.0) - self.assertEqual(result['attributes']['waypoint_id'], self.waypoint1.document_id) + + self.assertEqual( + result['waypoint_stoparea_id'], + self.waypoint_stoparea1.waypoint_stoparea_id + ) + self.assertEqual( + result['attributes']['distance'], + 100.0 + ) + self.assertEqual( + result['attributes']['waypoint_id'], + self.waypoint1.document_id + ) self.assertEqual(result['attributes']['stoparea_id'], 1) def test_get_info_not_found(self): """Test getting info for a waypoint-stoparea that doesn't exist""" response = self.app.get( '/waypoints_stopareas/999999/en/info', status=404) - self.assertEqual(response.json_body, {'error': 'Waypoint Stoparea not found'}) - + self.assertEqual( + response.json_body, + {'error': 'Waypoint Stoparea not found'} + ) + def test_get_info_lang_not_found(self): - """Test getting info for a waypoint-stoparea that exist with an invalid lang""" + """Test getting info for a waypoint-stoparea + that exist with an invalid lang""" response = self.app.get( '/waypoints_stopareas/1/invalid/info', status=400) - + # Updated to match actual JSON response format - self.assertEqual(response.json_body['errors'][0]['description'], "invalid lang") \ No newline at end of file + self.assertEqual( + response.json_body['errors'][0]['description'], + "invalid lang" + ) diff --git a/c2corg_api/views/navitia.py b/c2corg_api/views/navitia.py index a423d4447..1636f7609 100644 --- a/c2corg_api/views/navitia.py +++ b/c2corg_api/views/navitia.py @@ -14,7 +14,8 @@ from c2corg_api.views.waypoint import build_reachable_waypoints_query from c2corg_api.views.route import build_reachable_route_query_with_waypoints from shapely.geometry import Point -from pyramid.httpexceptions import HTTPBadRequest, HTTPInternalServerError, HTTPNotFound, HTTPUnauthorized +from pyramid.httpexceptions import (HTTPBadRequest, HTTPInternalServerError, + HTTPNotFound, HTTPUnauthorized) from pyramid.response import Response from cornice.resource import resource, view from c2corg_api.views import cors_policy, to_json_dict diff --git a/c2corg_api/views/waypoint_stoparea.py b/c2corg_api/views/waypoint_stoparea.py index dea011efa..45b17aa2b 100644 --- a/c2corg_api/views/waypoint_stoparea.py +++ b/c2corg_api/views/waypoint_stoparea.py @@ -5,8 +5,7 @@ from sqlalchemy import func, exists -from c2corg_api.models.waypoint_stoparea import ( - WaypointStoparea, schema_waypoint_stoparea) +from c2corg_api.models.waypoint_stoparea import WaypointStoparea from c2corg_api.views.document import ( make_validator_create, make_validator_update) From 4c357f00ade237284ac674c5264cd5273b31b128 Mon Sep 17 00:00:00 2001 From: Leo Gourdin Date: Thu, 2 Jul 2026 16:29:37 +0200 Subject: [PATCH 5/7] [scripts] recollement: add a guard in case max duration wasn't respected by the API --- scripts/public_transport/get_public_transports.sh | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/scripts/public_transport/get_public_transports.sh b/scripts/public_transport/get_public_transports.sh index 1e90b10db..e83338fed 100755 --- a/scripts/public_transport/get_public_transports.sh +++ b/scripts/public_transport/get_public_transports.sh @@ -294,6 +294,11 @@ for ((k=1; k<=nb_waypoints; k++)); do walk_duration=$(echo "$journey_response" | jq -r '.journeys[0].duration // 0') distance_km=$(awk "BEGIN {printf \"%.2f\", ($walk_duration * $WALKING_SPEED) / 1000}") + if (( $(echo "$walk_duration > $DURATION" | bc -l) )); then + log "Walk duration returned by the API journey suggestion ($walk_duration) was greater than $DURATION (= $MAX_DISTANCE_WAYPOINT_TO_STOPAREA / $WALKING_SPEED) for stop $stop_name, skipping." + continue + fi + if [[ -n "${INSERTED_STOP_AREAS[$stop_id]+x}" ]]; then # Already inserted in this run: reference via navitia_id subquery echo "INSERT INTO guidebook.waypoints_stopareas (stoparea_id, waypoint_id, distance) From 76cb022677f21454997936e7d6c6c45d1bfc636b Mon Sep 17 00:00:00 2001 From: Leo Gourdin Date: Thu, 2 Jul 2026 18:10:42 +0200 Subject: [PATCH 6/7] [rebase] smart-origin on master --- c2corg_api/tests/views/test_navitia.py | 1 - 1 file changed, 1 deletion(-) diff --git a/c2corg_api/tests/views/test_navitia.py b/c2corg_api/tests/views/test_navitia.py index ca44b7088..05c3f28db 100644 --- a/c2corg_api/tests/views/test_navitia.py +++ b/c2corg_api/tests/views/test_navitia.py @@ -1226,7 +1226,6 @@ def test_handle_navitia_response_404_no_journey(self): with self.assertRaises(HTTPNotFound): handle_navitia_response(mock_response) - def test_handle_navitia_response_404_other_error(self): mock_response = mock.Mock() mock_response.status_code = 404 From ebaab6d478d9a1280b1c930223118c8e9dbc175c Mon Sep 17 00:00:00 2001 From: "gerome.perrin" Date: Fri, 3 Jul 2026 12:08:05 +0200 Subject: [PATCH 7/7] [fix] fix itinevert + codacy issues (cherry picked from commit 93bebb80fdb59d386ceef810b2adec13521ce962) --- c2corg_api/tests/views/test_stopareas.py | 12 +++--- .../tests/views/test_waypoint_stoparea.py | 15 ++++---- c2corg_api/views/navitia.py | 37 +++++++++++-------- 3 files changed, 34 insertions(+), 30 deletions(-) diff --git a/c2corg_api/tests/views/test_stopareas.py b/c2corg_api/tests/views/test_stopareas.py index 8ae1fa7ff..198f37bcc 100644 --- a/c2corg_api/tests/views/test_stopareas.py +++ b/c2corg_api/tests/views/test_stopareas.py @@ -24,7 +24,7 @@ def _add_test_data(self): self.session.flush() def test_collection_get(self): - """Test getting list of stopareas""" + """Test getting list of stopareas.""" response = self.app.get('/stopareas', status=200) result = response.json @@ -32,14 +32,14 @@ def test_collection_get(self): assert isinstance(result['documents'], list) def test_get_stoparea_not_found(self): - """Test getting a stoparea that doesn't exist""" + """Test getting a stoparea that doesn't exist.""" response = self.app.get('/stopareas/999999', status=404) result = response.json assert result['error'] == 'Stoparea not found' def test_get_stoparea_found(self): - """Test getting a stoparea that exists""" + """Test getting a stoparea that exists.""" response = self.app.get('/stopareas/1', status=200) result = response.json @@ -72,14 +72,14 @@ def _add_test_data(self): self.session.flush() def test_get_info_stoparea_not_found(self): - """Test getting info for a stoparea that doesn't exist""" + """Test getting info for a stoparea that doesn't exist.""" response = self.app.get('/stopareas/999999/fr/info', status=404) result = response.json assert result['error'] == 'Stoparea not found' def test_get_info_stoparea_found(self): - """Test getting info for a stoparea that exists""" + """Test getting info for a stoparea that exists.""" response = self.app.get('/stopareas/1/fr/info', status=200) result = response.json @@ -90,7 +90,7 @@ def test_get_info_stoparea_found(self): assert result['attributes']['operator'] == 'operator1' def test_get_info_stoparea_lang_not_found(self): - """Test getting info for a stoparea that doesn't exist""" + """Test getting info for a stoparea that doesn't exist.""" response = self.app.get('/stopareas/1/invalid/info', status=400) result = response.json diff --git a/c2corg_api/tests/views/test_waypoint_stoparea.py b/c2corg_api/tests/views/test_waypoint_stoparea.py index bb98bbcad..ae102711b 100644 --- a/c2corg_api/tests/views/test_waypoint_stoparea.py +++ b/c2corg_api/tests/views/test_waypoint_stoparea.py @@ -91,7 +91,7 @@ def _add_test_data(self): self.waypoint_stoparea2 = waypoint_stoparea2 def test_get_stopareas_by_waypoint(self): - """Test getting stopareas for a waypoint""" + """Test getting stopareas for a waypoint.""" response = self.app.get('/waypoints/{}/stopareas'.format( self.waypoint1.document_id), status=200) result = response.json @@ -116,7 +116,7 @@ def test_get_stopareas_by_waypoint(self): self.assertEqual(stoparea2['distance'], 200.0) def test_get_stopareas_by_waypoint_not_found(self): - """Test getting stopareas for a waypoint that doesn't exist""" + """Test getting stopareas for a waypoint that doesn't exist.""" # We'll test that it returns an empty array instead of 404 response = self.app.get('/waypoints/999999/stopareas', status=200) result = response.json @@ -125,7 +125,7 @@ def test_get_stopareas_by_waypoint_not_found(self): self.assertEqual(result['stopareas'], []) def test_get_is_reachable_true(self): - """Test checking if a waypoint is reachable (has stopareas)""" + """Test checking if a waypoint is reachable (has stopareas).""" response = self.app.get('/waypoints/{}/isReachable'.format( self.waypoint1.document_id), status=200) result = response.json @@ -133,7 +133,7 @@ def test_get_is_reachable_true(self): self.assertTrue(result) def test_get_is_reachable_false(self): - """Test checking if a waypoint is not reachable (no stopareas)""" + """Test checking if a waypoint is not reachable (no stopareas).""" response = self.app.get('/waypoints/{}/isReachable'.format( self.waypoint2.document_id), status=200) result = response.json @@ -141,7 +141,7 @@ def test_get_is_reachable_false(self): self.assertFalse(result) def test_get_info(self): - """Test getting info for a waypoint-stoparea""" + """Test getting info for a waypoint-stoparea.""" response = self.app.get( '/waypoints_stopareas/1/en/info', status=200) @@ -162,7 +162,7 @@ def test_get_info(self): self.assertEqual(result['attributes']['stoparea_id'], 1) def test_get_info_not_found(self): - """Test getting info for a waypoint-stoparea that doesn't exist""" + """Test getting info for a waypoint-stoparea that doesn't exist.""" response = self.app.get( '/waypoints_stopareas/999999/en/info', status=404) self.assertEqual( @@ -171,8 +171,7 @@ def test_get_info_not_found(self): ) def test_get_info_lang_not_found(self): - """Test getting info for a waypoint-stoparea - that exist with an invalid lang""" + """Test getting info with an invalid lang.""" response = self.app.get( '/waypoints_stopareas/1/invalid/info', status=400) diff --git a/c2corg_api/views/navitia.py b/c2corg_api/views/navitia.py index 1636f7609..99e1626f0 100644 --- a/c2corg_api/views/navitia.py +++ b/c2corg_api/views/navitia.py @@ -578,24 +578,29 @@ def is_wp_journey_reachable(waypoint, journey_params): else: url = BASE_URL + "/journeys" - json_response = navitia_get( - url, - params=journey_params, - headers={"Authorization": api_key}, - ) - - # no_origin / no_destination - if json_response is None: - return False + try: + json_response = navitia_get( + url, + params=journey_params, + headers={"Authorization": api_key}, + ) + # no_origin / no_destination + if json_response is None: + return False - # Business logic only - for journey in json_response.get("journeys", []): - journey_day = int(journey["departure_date_time"][6:8]) - param_day = int(journey_params["datetime"][6:8]) - if journey_day == param_day: - return True + # Business logic only + for journey in json_response.get("journeys", []): + journey_day = int(journey["departure_date_time"][6:8]) + param_day = int(journey_params["datetime"][6:8]) + if journey_day == param_day: + return True - return False + return False + except Exception as e: + if (str(e) == "no_solution"): + return False + else: + raise HTTPInternalServerError(e) def get_navitia_isochrone(isochrone_params):