diff --git a/fittrackee/tests/workouts/test_workouts_api_1_post.py b/fittrackee/tests/workouts/test_workouts_api_1_post.py index e149bec2..103549e5 100644 --- a/fittrackee/tests/workouts/test_workouts_api_1_post.py +++ b/fittrackee/tests/workouts/test_workouts_api_1_post.py @@ -217,6 +217,22 @@ def assert_workout_data_wo_gpx(data: Dict) -> None: assert records[3]['value'] == 10.0 +def assert_files_are_deleted(app: Flask, user: User) -> None: + upload_directory = os.path.join( + app.config["UPLOAD_FOLDER"], f"workouts/{user.id}" + ) + assert ( + len( + [ + name + for name in os.listdir(upload_directory) + if os.path.isfile(os.path.join(upload_directory, name)) + ] + ) + == 0 + ) + + class TestPostWorkoutWithGpx(ApiTestCaseMixin, CallArgsMixin): def test_it_returns_error_if_user_is_not_authenticated( self, app: Flask, sport_1_cycling: Sport, gpx_file: str @@ -800,6 +816,56 @@ class TestPostWorkoutWithGpx(ApiTestCaseMixin, CallArgsMixin): ) assert 'data' not in data + def test_it_cleans_uploaded_file_on_gpx_processing_error( + self, app: Flask, user_1: User, sport_1_cycling: Sport, gpx_file: str + ) -> None: + client, auth_token = self.get_test_client_and_auth_token( + app, user_1.email + ) + + with patch( + 'fittrackee.workouts.utils.workouts.generate_map', + side_effect=Exception(), + ): + client.post( + '/api/workouts', + data=dict( + file=(BytesIO(str.encode(gpx_file)), 'example.gpx'), + data='{"sport_id": 1}', + ), + headers=dict( + content_type='multipart/form-data', + Authorization=f'Bearer {auth_token}', + ), + ) + + assert_files_are_deleted(app, user_1) + + def test_it_cleans_uploaded_file_and_static_map_on_segments_creation_error( + self, app: Flask, user_1: User, sport_1_cycling: Sport, gpx_file: str + ) -> None: + client, auth_token = self.get_test_client_and_auth_token( + app, user_1.email + ) + + with patch( + 'fittrackee.workouts.utils.workouts.create_segment', + side_effect=ValueError(), + ): + client.post( + '/api/workouts', + data=dict( + file=(BytesIO(str.encode(gpx_file)), 'example.gpx'), + data='{"sport_id": 1}', + ), + headers=dict( + content_type='multipart/form-data', + Authorization=f'Bearer {auth_token}', + ), + ) + + assert_files_are_deleted(app, user_1) + @pytest.mark.parametrize( 'client_scope, can_access', [ @@ -1399,6 +1465,31 @@ class TestPostWorkoutWithZipArchive(ApiTestCaseMixin): ) assert 'data' not in data + def test_it_cleans_uploaded_file_on_error( + self, app: Flask, user_1: User, sport_1_cycling: Sport + ) -> None: + client, auth_token = self.get_test_client_and_auth_token( + app, user_1.email + ) + file_path = os.path.join(app.root_path, 'tests/files/gpx_test.zip') + # 'gpx_test.zip' contains 3 gpx files (same data) and 1 non-gpx file + with open(file_path, 'rb') as zip_file, patch( + 'fittrackee.workouts.utils.workouts.generate_map', + side_effect=Exception(), + ): + client.post( + '/api/workouts', + data=dict( + file=(zip_file, 'gpx_test.zip'), data='{"sport_id": 1}' + ), + headers=dict( + content_type='multipart/form-data', + Authorization=f'Bearer {auth_token}', + ), + ) + + assert_files_are_deleted(app, user_1) + class TestPostAndGetWorkoutWithGpx(ApiTestCaseMixin): def workout_assertion( diff --git a/fittrackee/workouts/utils/workouts.py b/fittrackee/workouts/utils/workouts.py index f033177a..5872af11 100644 --- a/fittrackee/workouts/utils/workouts.py +++ b/fittrackee/workouts/utils/workouts.py @@ -12,7 +12,7 @@ from sqlalchemy import exc from werkzeug.datastructures import FileStorage from werkzeug.utils import secure_filename -from fittrackee import db +from fittrackee import appLog, db from fittrackee.files import get_absolute_file_path from fittrackee.users.models import User, UserSportPreference @@ -279,12 +279,26 @@ def get_new_file_path( return file_path +def delete_files( + absolute_gpx_filepath: Optional[str], absolute_map_filepath: Optional[str] +) -> None: + try: + if absolute_gpx_filepath and os.path.exists(absolute_gpx_filepath): + os.remove(absolute_gpx_filepath) + if absolute_map_filepath and os.path.exists(absolute_map_filepath): + os.remove(absolute_map_filepath) + except Exception: + appLog.error('Unable to delete files after processing error.') + + def process_one_gpx_file( params: Dict, filename: str, stopped_speed_threshold: float ) -> Workout: """ Get all data from a gpx file to create a workout with map image """ + absolute_gpx_filepath = None + absolute_map_filepath = None try: gpx_data, map_data, weather_data = get_gpx_info( params['file_path'], stopped_speed_threshold @@ -314,8 +328,10 @@ def process_one_gpx_file( absolute_map_filepath = get_absolute_file_path(map_filepath) generate_map(absolute_map_filepath, map_data) except (gpxpy.gpx.GPXXMLSyntaxException, TypeError) as e: + delete_files(absolute_gpx_filepath, absolute_map_filepath) raise WorkoutException('error', 'error during gpx file parsing', e) except Exception as e: + delete_files(absolute_gpx_filepath, absolute_map_filepath) raise WorkoutException('error', 'error during gpx processing', e) try: @@ -337,6 +353,7 @@ def process_one_gpx_file( db.session.commit() return new_workout except (exc.IntegrityError, ValueError) as e: + delete_files(absolute_gpx_filepath, absolute_map_filepath) raise WorkoutException('fail', 'Error during workout save.', e)