diff --git a/fittrackee/tests/users/test_export_data.py b/fittrackee/tests/users/test_export_data.py new file mode 100644 index 00000000..ef75d8d1 --- /dev/null +++ b/fittrackee/tests/users/test_export_data.py @@ -0,0 +1,335 @@ +import os +import secrets +from unittest.mock import Mock, call, patch + +from flask import Flask + +from fittrackee.users.export_data import UserDataExporter +from fittrackee.users.models import User +from fittrackee.workouts.models import Sport, Workout + +from ..mixins import CallArgsMixin +from ..utils import random_string +from ..workouts.utils import post_a_workout + + +class TestUserDataExporterGetData: + def test_it_return_serialized_user_as_info_info( + self, app: Flask, user_1: User + ) -> None: + exporter = UserDataExporter(user_1) + + user_data = exporter.get_user_info() + + assert user_data == user_1.serialize(user_1) + + def test_it_returns_empty_list_when_user_has_no_workouts( + self, app: Flask, user_1: User + ) -> None: + exporter = UserDataExporter(user_1) + + workouts_data = exporter.get_user_workouts_data() + + assert workouts_data == [] + + def test_it_returns_data_for_workout_without_gpx( + self, + app: Flask, + user_1: User, + sport_1_cycling: Sport, + workout_cycling_user_1: Workout, + ) -> None: + exporter = UserDataExporter(user_1) + + workouts_data = exporter.get_user_workouts_data() + + assert workouts_data == [ + { + 'id': workout_cycling_user_1.short_id, + 'sport_id': sport_1_cycling.id, + 'sport_label': sport_1_cycling.label, + 'title': workout_cycling_user_1.title, + 'creation_date': workout_cycling_user_1.creation_date, + 'modification_date': workout_cycling_user_1.modification_date, + 'workout_date': workout_cycling_user_1.workout_date, + 'duration': str(workout_cycling_user_1.duration), + 'pauses': None, + 'moving': str(workout_cycling_user_1.moving), + 'distance': float(workout_cycling_user_1.distance), + 'min_alt': None, + 'max_alt': None, + 'descent': None, + 'ascent': None, + 'max_speed': float(workout_cycling_user_1.max_speed), + 'ave_speed': float(workout_cycling_user_1.ave_speed), + 'gpx': None, + 'records': [ + record.serialize() + for record in workout_cycling_user_1.records + ], + 'segments': [], + 'weather_start': None, + 'weather_end': None, + 'notes': workout_cycling_user_1.notes, + } + ] + + def test_it_returns_data_for_workout_with_gpx( + self, + app: Flask, + user_1: User, + sport_1_cycling: Sport, + gpx_file: str, + ) -> None: + _, workout_short_id = post_a_workout(app, gpx_file) + workout = Workout.query.first() + exporter = UserDataExporter(user_1) + + workouts_data = exporter.get_user_workouts_data() + + assert workouts_data == [ + { + 'id': workout.short_id, + 'sport_id': sport_1_cycling.id, + 'sport_label': sport_1_cycling.label, + 'title': workout.title, + 'creation_date': workout.creation_date, + 'modification_date': workout.modification_date, + 'workout_date': workout.workout_date, + 'duration': str(workout.duration), + 'pauses': None, + 'moving': str(workout.moving), + 'distance': float(workout.distance), + 'min_alt': float(workout.min_alt), + 'max_alt': float(workout.max_alt), + 'descent': float(workout.descent), + 'ascent': float(workout.ascent), + 'max_speed': float(workout.max_speed), + 'ave_speed': float(workout.ave_speed), + 'gpx': workout.gpx.split('/')[-1], + 'records': [record.serialize() for record in workout.records], + 'segments': [ + segment.serialize() for segment in workout.segments + ], + 'weather_start': None, + 'weather_end': None, + 'notes': workout.notes, + } + ] + + def test_it_stores_only_user_workouts( + self, + app: Flask, + user_1: User, + user_2: User, + sport_1_cycling: Sport, + workout_cycling_user_1: Workout, + workout_cycling_user_2: Workout, + ) -> None: + exporter = UserDataExporter(user_1) + + workouts_data = exporter.get_user_workouts_data() + + assert [data["id"] for data in workouts_data] == [ + workout_cycling_user_1.short_id + ] + + def test_export_data_generates_json_file_in_user_directory( + self, + app: Flask, + user_1: User, + ) -> None: + data = {"foo": "bar"} + export = UserDataExporter(user_1) + user_directory = os.path.join( + app.config['UPLOAD_FOLDER'], 'exports', str(user_1.id) + ) + os.makedirs(user_directory, exist_ok=True) + file_name = random_string() + + export.export_data(data, file_name) + + assert ( + os.path.isfile(os.path.join(user_directory, f"{file_name}.json")) + is True + ) + + def test_it_returns_json_file_path( + self, + app: Flask, + user_1: User, + ) -> None: + data = {"foo": "bar"} + exporter = UserDataExporter(user_1) + user_directory = os.path.join( + app.config['UPLOAD_FOLDER'], 'exports', str(user_1.id) + ) + file_name = random_string() + + file_path = exporter.export_data(data, file_name) + + assert file_path == os.path.join(user_directory, f"{file_name}.json") + + +class TestUserDataExporterArchive(CallArgsMixin): + @patch.object(secrets, 'token_urlsafe') + @patch.object(UserDataExporter, 'export_data') + @patch('fittrackee.users.export_data.ZipFile') + def test_it_calls_export_data_twice( + self, + zipfile_mock: Mock, + export_data: Mock, + secrets_mock: Mock, + app: Flask, + user_1: User, + ) -> None: + exporter = UserDataExporter(user_1) + + exporter.generate_archive() + + export_data.assert_has_calls( + [ + call(exporter.get_user_info(), 'user_data'), + call(exporter.get_user_workouts_data(), 'workouts_data'), + ] + ) + + @patch.object(secrets, 'token_urlsafe') + @patch.object(UserDataExporter, 'export_data') + @patch('fittrackee.users.export_data.ZipFile') + def test_it_calls_zipfile_with_expected_patch( + self, + zipfile_mock: Mock, + export_data: Mock, + secrets_mock: Mock, + app: Flask, + user_1: User, + ) -> None: + exporter = UserDataExporter(user_1) + token_urlsafe = random_string() + secrets_mock.return_value = token_urlsafe + expected_path = os.path.join( + app.config['UPLOAD_FOLDER'], + 'exports', + str(user_1.id), + f"archive_{token_urlsafe}.zip", + ) + + exporter.generate_archive() + + zipfile_mock.assert_called_once_with(expected_path, 'w') + + @patch.object(secrets, 'token_urlsafe') + @patch.object(UserDataExporter, 'export_data') + @patch('fittrackee.users.export_data.ZipFile') + def test_it_calls_zipfile_for_each_json_file( + self, + zipfile_mock: Mock, + export_data: Mock, + secrets_mock: Mock, + app: Flask, + user_1: User, + ) -> None: + exporter = UserDataExporter(user_1) + token_urlsafe = random_string() + secrets_mock.return_value = token_urlsafe + export_data.side_effect = [call('user_info'), call('workouts_data')] + + exporter.generate_archive() + + # fmt: off + zipfile_mock.return_value.__enter__\ + .return_value.write.assert_has_calls( + [ + call(call('user_info'), 'user_data.json'), + call(call('workouts_data'), 'user_workouts_data.json'), + ] + ) + # fmt: on + + @patch.object(secrets, 'token_urlsafe') + @patch.object(UserDataExporter, 'export_data') + @patch('fittrackee.users.export_data.ZipFile') + def test_it_calls_zipfile_for_gpx_file( + self, + zipfile_mock: Mock, + export_data: Mock, + secrets_mock: Mock, + app: Flask, + user_1: User, + sport_1_cycling: Sport, + gpx_file: str, + ) -> None: + _, workout_short_id = post_a_workout(app, gpx_file) + workout = Workout.query.first() + expected_path = os.path.join( + app.config['UPLOAD_FOLDER'], + workout.gpx, + ) + exporter = UserDataExporter(user_1) + + exporter.generate_archive() + + # fmt: off + zipfile_mock.return_value.__enter__.\ + return_value.write.assert_has_calls( + [ + call(expected_path, f"gpx/{workout.gpx.split('/')[-1]}"), + ] + ) + # fmt: on + + @patch.object(secrets, 'token_urlsafe') + @patch.object(UserDataExporter, 'export_data') + @patch('fittrackee.users.export_data.ZipFile') + def test_it_calls_zipfile_for_profile_image_when_exists( + self, + zipfile_mock: Mock, + export_data: Mock, + secrets_mock: Mock, + app: Flask, + user_1: User, + sport_1_cycling: Sport, + gpx_file: str, + ) -> None: + user_1.picture = random_string() + expected_path = os.path.join( + app.config['UPLOAD_FOLDER'], + user_1.picture, + ) + exporter = UserDataExporter(user_1) + + with patch( + 'fittrackee.users.export_data.os.path.isfile', return_value=True + ): + exporter.generate_archive() + + # fmt: off + zipfile_mock.return_value.__enter__.\ + return_value.write.assert_has_calls( + [ + call(expected_path, user_1.picture.split('/')[-1]), + ] + ) + # fmt: on + + @patch.object(secrets, 'token_urlsafe') + def test_it_test_it_generates_a_zip_archive( + self, + secrets_mock: Mock, + app: Flask, + user_1: User, + ) -> None: + token_urlsafe = random_string() + secrets_mock.return_value = token_urlsafe + expected_path = os.path.join( + app.config['UPLOAD_FOLDER'], + 'exports', + str(user_1.id), + f"archive_{token_urlsafe}.zip", + ) + exporter = UserDataExporter(user_1) + + exporter.generate_archive() + + assert os.path.isfile(expected_path) diff --git a/fittrackee/users/export_data.py b/fittrackee/users/export_data.py new file mode 100644 index 00000000..bbe4d154 --- /dev/null +++ b/fittrackee/users/export_data.py @@ -0,0 +1,86 @@ +import json +import os +import secrets +from typing import Dict, List, Optional, Union +from zipfile import ZipFile + +from fittrackee.files import get_absolute_file_path + +from .models import User + + +class UserDataExporter: + """ + generates a zip archive with: + - user info from database (json file) + - data from database for all workouts if exist (json file) + - profile picture file if exists + - gpx files if exist + """ + + def __init__(self, user: User) -> None: + self.user = user + self.export_directory = get_absolute_file_path( + os.path.join('exports', str(self.user.id)) + ) + os.makedirs(self.export_directory, exist_ok=True) + self.workouts_directory = get_absolute_file_path( + os.path.join('workouts', str(self.user.id)) + ) + + def get_user_info(self) -> Dict: + return self.user.serialize(self.user) + + def get_user_workouts_data(self) -> List[Dict]: + workouts_data = [] + for workout in self.user.workouts: + workout_data = workout.get_workout_data() + workout_data["sport_label"] = workout.sport.label + workout_data["gpx"] = ( + workout.gpx.split('/')[-1] if workout.gpx else None + ) + workouts_data.append(workout_data) + return workouts_data + + def export_data(self, data: Union[Dict, List], name: str) -> str: + """export data in existing user upload directory""" + json_object = json.dumps(data, indent=4, default=str) + file_path = os.path.join(self.export_directory, f"{name}.json") + with open(file_path, "w") as export_file: + export_file.write(json_object) + return file_path + + def generate_archive(self) -> Optional[str]: + try: + user_data_file_name = self.export_data( + self.get_user_info(), "user_data" + ) + workout_data_file_name = self.export_data( + self.get_user_workouts_data(), "workouts_data" + ) + zip_file = f"archive_{secrets.token_urlsafe(15)}.zip" + zip_path = os.path.join(self.export_directory, zip_file) + with ZipFile(zip_path, 'w') as zip_object: + zip_object.write(user_data_file_name, "user_data.json") + zip_object.write( + workout_data_file_name, "user_workouts_data.json" + ) + if self.user.picture: + picture_path = get_absolute_file_path(self.user.picture) + if os.path.isfile(picture_path): + zip_object.write( + picture_path, self.user.picture.split('/')[-1] + ) + for file in os.listdir(self.workouts_directory): + if os.path.isfile( + os.path.join(self.workouts_directory, file) + ) and file.endswith('.gpx'): + zip_object.write( + os.path.join(self.workouts_directory, file), + f"gpx/{file}", + ) + + file_exists = os.path.exists(zip_path) + return zip_file if file_exists else None + except Exception: + return None diff --git a/fittrackee/workouts/models.py b/fittrackee/workouts/models.py index 76252603..4421cbd1 100644 --- a/fittrackee/workouts/models.py +++ b/fittrackee/workouts/models.py @@ -192,6 +192,33 @@ class Workout(BaseModel): def short_id(self) -> str: return encode_uuid(self.uuid) + def get_workout_data(self) -> Dict: + return { + 'id': self.short_id, # WARNING: client use uuid as id + 'sport_id': self.sport_id, + 'title': self.title, + 'creation_date': self.creation_date, + 'modification_date': self.modification_date, + 'workout_date': self.workout_date, + 'duration': str(self.duration) if self.duration else None, + 'pauses': str(self.pauses) if self.pauses else None, + 'moving': str(self.moving) if self.moving else None, + 'distance': float(self.distance) if self.distance else None, + 'min_alt': float(self.min_alt) if self.min_alt else None, + 'max_alt': float(self.max_alt) if self.max_alt else None, + 'descent': float(self.descent) + if self.descent is not None + else None, + 'ascent': float(self.ascent) if self.ascent is not None else None, + 'max_speed': float(self.max_speed) if self.max_speed else None, + 'ave_speed': float(self.ave_speed) if self.ave_speed else None, + 'records': [record.serialize() for record in self.records], + 'segments': [segment.serialize() for segment in self.segments], + 'weather_start': self.weather_start, + 'weather_end': self.weather_end, + 'notes': self.notes, + } + def serialize(self, params: Optional[Dict] = None) -> Dict: date_from = params.get('from') if params else None date_to = params.get('to') if params else None @@ -282,41 +309,21 @@ class Workout(BaseModel): .order_by(Workout.workout_date.asc()) .first() ) - return { - 'id': self.short_id, # WARNING: client use uuid as id - 'user': self.user.username, - 'sport_id': self.sport_id, - 'title': self.title, - 'creation_date': self.creation_date, - 'modification_date': self.modification_date, - 'workout_date': self.workout_date, - 'duration': str(self.duration) if self.duration else None, - 'pauses': str(self.pauses) if self.pauses else None, - 'moving': str(self.moving) if self.moving else None, - 'distance': float(self.distance) if self.distance else None, - 'min_alt': float(self.min_alt) if self.min_alt else None, - 'max_alt': float(self.max_alt) if self.max_alt else None, - 'descent': float(self.descent) - if self.descent is not None - else None, - 'ascent': float(self.ascent) if self.ascent is not None else None, - 'max_speed': float(self.max_speed) if self.max_speed else None, - 'ave_speed': float(self.ave_speed) if self.ave_speed else None, - 'with_gpx': self.gpx is not None, - 'bounds': [float(bound) for bound in self.bounds] - if self.bounds - else [], # noqa - 'previous_workout': previous_workout.short_id - if previous_workout - else None, # noqa - 'next_workout': next_workout.short_id if next_workout else None, - 'segments': [segment.serialize() for segment in self.segments], - 'records': [record.serialize() for record in self.records], - 'map': self.map_id if self.map else None, - 'weather_start': self.weather_start, - 'weather_end': self.weather_end, - 'notes': self.notes, - } + + workout = self.get_workout_data() + workout["next_workout"] = ( + next_workout.short_id if next_workout else None + ) + workout["previous_workout"] = ( + previous_workout.short_id if previous_workout else None + ) + workout["bounds"] = ( + [float(bound) for bound in self.bounds] if self.bounds else [] + ) + workout["user"] = self.user.username + workout["map"] = self.map_id if self.map else None + workout["with_gpx"] = self.gpx is not None + return workout @classmethod def get_user_workout_records(