diff --git a/c2corg_api/tests/views/test_navitia.py b/c2corg_api/tests/views/test_navitia.py index c38df16cd..05c3f28db 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 +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 +1208,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 +1221,17 @@ 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"}} - result = handle_navitia_response(mock_response) - self.assertIsNone(result) + 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 - 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() diff --git a/c2corg_api/tests/views/test_stopareas.py b/c2corg_api/tests/views/test_stopareas.py new file mode 100644 index 000000000..198f37bcc --- /dev/null +++ b/c2corg_api/tests/views/test_stopareas.py @@ -0,0 +1,97 @@ +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..ae102711b --- /dev/null +++ b/c2corg_api/tests/views/test_waypoint_stoparea.py @@ -0,0 +1,182 @@ +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 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" + ) diff --git a/c2corg_api/views/navitia.py b/c2corg_api/views/navitia.py index 12771d46a..99e1626f0 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 +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 @@ -577,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): @@ -751,20 +757,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() diff --git a/c2corg_api/views/waypoint_stoparea.py b/c2corg_api/views/waypoint_stoparea.py index e446d451a..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) @@ -28,7 +27,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): 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)