import os import secrets from datetime import datetime, timedelta from typing import Optional, Tuple from unittest.mock import Mock, call, patch from flask import Flask from fittrackee import db from fittrackee.users.export_data import ( UserDataExporter, clean_user_data_export, export_user_data, generate_user_data_archives, ) from fittrackee.users.models import User, UserDataExport from fittrackee.workouts.models import Sport, Workout from ..mixins import CallArgsMixin from ..utils import random_int, 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, user_2: 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, user_2: 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, user_2: 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, user_2: 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_does_not_call_zipfile_for_another_user_gpx_file( self, zipfile_mock: Mock, export_data: Mock, secrets_mock: Mock, app: Flask, user_1: User, user_2: 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_2) exporter.generate_archive() # fmt: off assert ( call(expected_path, f"gpx/{workout.gpx.split('/')[-1]}") not in zipfile_mock.return_value.__enter__. return_value.write.call_args_list ) # 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, user_2: 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') @patch.object(UserDataExporter, 'export_data') @patch('fittrackee.users.export_data.ZipFile') def test_it_does_not_call_zipfile_for_another_user_profile_image( self, zipfile_mock: Mock, export_data: Mock, secrets_mock: Mock, app: Flask, user_1: User, user_2: 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_2) with patch( 'fittrackee.users.export_data.os.path.isfile', return_value=True ): exporter.generate_archive() # fmt: off assert ( call(expected_path, user_1.picture.split('/')[-1]) not in zipfile_mock.return_value.__enter__. return_value.write.call_args_list ) # 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) @patch('fittrackee.users.export_data.appLog') @patch.object(UserDataExporter, 'generate_archive') class TestExportUserData: def test_it_logs_error_if_not_request_for_given_id( self, generate_archive: Mock, logger_mock: Mock, app: Flask, ) -> None: request_id = random_int() export_user_data(export_request_id=request_id) logger_mock.error.assert_called_once_with( f"No export to process for id '{request_id}'" ) def test_it_logs_error_if_request_already_processed( self, generate_archive: Mock, logger_mock: Mock, app: Flask, user_1: User, ) -> None: export_request = UserDataExport(user_id=user_1.id) db.session.add(export_request) export_request.completed = True db.session.commit() export_user_data(export_request_id=export_request.id) logger_mock.info.assert_called_once_with( f"Export id '{export_request.id}' already processed" ) def test_it_updates_export_request_when_export_is_successful( self, generate_archive_mock: Mock, logger_mock: Mock, data_export_email_mock: Mock, app: Flask, user_1: User, ) -> None: export_request = UserDataExport(user_id=user_1.id) db.session.add(export_request) db.session.commit() archive_name = random_string() generate_archive_mock.return_value = (random_string(), archive_name) archive_size = random_int() with patch( 'fittrackee.users.export_data.os.path.getsize', return_value=archive_size, ): export_user_data(export_request_id=export_request.id) assert export_request.completed is True assert export_request.updated_at is not None assert export_request.file_name == archive_name assert export_request.file_size == archive_size def test_it_updates_export_request_when_export_fails( self, generate_archive_mock: Mock, logger_mock: Mock, app: Flask, user_1: User, ) -> None: export_request = UserDataExport(user_id=user_1.id) db.session.add(export_request) db.session.commit() generate_archive_mock.return_value = (None, None) export_user_data(export_request_id=export_request.id) assert export_request.completed is True assert export_request.updated_at is not None assert export_request.file_name is None assert export_request.file_size is None def test_it_does_not_call_data_export_email_when_export_failed( self, generate_archive_mock: Mock, logger_mock: Mock, data_export_email_mock: Mock, app: Flask, user_1: User, ) -> None: export_request = UserDataExport(user_id=user_1.id) db.session.add(export_request) db.session.commit() generate_archive_mock.return_value = (None, None) export_user_data(export_request_id=export_request.id) data_export_email_mock.send.assert_not_called() def test_it_calls_data_export_email_when_export_is_successful( self, generate_archive_mock: Mock, logger_mock: Mock, data_export_email_mock: Mock, app: Flask, user_1: User, ) -> None: export_request = UserDataExport(user_id=user_1.id) db.session.add(export_request) db.session.commit() archive_name = random_string() generate_archive_mock.return_value = (random_string(), archive_name) archive_size = random_int() with patch( 'fittrackee.users.export_data.os.path.getsize', return_value=archive_size, ): export_user_data(export_request_id=export_request.id) data_export_email_mock.send.assert_called_once_with( { 'language': 'en', 'email': user_1.email, }, { 'username': user_1.username, 'account_url': 'http://0.0.0.0:5000/profile/edit/account', 'fittrackee_url': 'http://0.0.0.0:5000', }, ) class UserDataExportTestCase: @staticmethod def create_user_request( user: User, days: int = 0, completed: bool = True ) -> UserDataExport: user_data_export = UserDataExport( user_id=user.id, created_at=datetime.now() - timedelta(days=days), ) db.session.add(user_data_export) user_data_export.completed = completed db.session.commit() return user_data_export def generate_archive( self, user: User ) -> Tuple[UserDataExport, Optional[str]]: user_data_export = self.create_user_request(user, days=7) exporter = UserDataExporter(user) archive_path, archive_file_name = exporter.generate_archive() user_data_export.file_name = archive_file_name user_data_export.file_size = random_int() db.session.commit() return user_data_export, archive_path class TestCleanUserDataExport(UserDataExportTestCase): def test_it_returns_0_when_no_export_requests(self, app: Flask) -> None: counts = clean_user_data_export(days=7) assert counts["deleted_requests"] == 0 def test_it_returns_0_when_export_request_is_not_completed( self, app: Flask, user_1: User ) -> None: self.create_user_request(user_1, days=7, completed=False) counts = clean_user_data_export(days=7) assert counts["deleted_requests"] == 0 def test_it_returns_0_when_export_request_created_less_than_given_days( self, app: Flask, user_1: User ) -> None: self.create_user_request(user_1, days=1) counts = clean_user_data_export(days=7) assert counts["deleted_requests"] == 0 def test_it_returns_export_requests_created_more_than_given_days_count( self, app: Flask, user_1: User, user_2: User ) -> None: self.create_user_request(user_1, days=7) self.create_user_request(user_2, days=7) counts = clean_user_data_export(days=7) assert counts["deleted_requests"] == 2 def test_it_returns_counts( self, app: Flask, user_1: User, user_2: User, user_3: User ) -> None: user_1_data_export, archive_path = self.generate_archive(user_1) user_2_data_export, archive_path = self.generate_archive(user_2) self.create_user_request(user_3, days=7) counts = clean_user_data_export(days=7) assert counts["deleted_requests"] == 3 assert counts["deleted_archives"] == 2 assert counts["freed_space"] == ( user_1_data_export.file_size + user_2_data_export.file_size ) def test_it_deletes_archive( self, app: Flask, user_1: User, user_2: User ) -> None: _, archive_path = self.generate_archive(user_1) clean_user_data_export(days=7) assert os.path.exists(archive_path) is False # type: ignore def test_it_deletes_requests( self, app: Flask, user_1: User, user_2: User ) -> None: self.generate_archive(user_1) clean_user_data_export(days=7) assert ( UserDataExport.query.filter_by(user_id=user_1.id).first() is None ) class TestGenerateUsersArchives(UserDataExportTestCase): def test_it_returns_0_when_no_request(self, app: Flask) -> None: count = generate_user_data_archives(max_count=1) assert count == 0 def test_it_returns_0_when_request_request_completed( self, app: Flask, user_1: User ) -> None: self.create_user_request(user_1, completed=True) count = generate_user_data_archives(max_count=1) assert count == 0 def test_it_returns_count_when_archive_is_generated_user_archive( self, app: Flask, user_1: User ) -> None: self.create_user_request(user_1, completed=False) count = generate_user_data_archives(max_count=1) assert count == 1 @patch.object(secrets, 'token_urlsafe') def test_it_generates_user_archive( self, secrets_mock: Mock, app: Flask, user_1: User ) -> None: token_urlsafe = random_string() secrets_mock.return_value = token_urlsafe archive_path = os.path.join( app.config['UPLOAD_FOLDER'], 'exports', str(user_1.id), f"archive_{token_urlsafe}.zip", ) self.create_user_request(user_1, completed=False) generate_user_data_archives(max_count=1) assert os.path.exists(archive_path) is True # type: ignore def test_it_generates_max_count_of_archives( self, app: Flask, user_1: User, user_2: User, user_3: User ) -> None: self.create_user_request(user_3, completed=False) self.create_user_request(user_1, completed=False) self.create_user_request(user_2, completed=False) count = generate_user_data_archives(max_count=2) assert count == 2 assert ( UserDataExport.query.filter_by(user_id=user_1.id).first().completed is True ) assert ( UserDataExport.query.filter_by(user_id=user_2.id).first().completed is False ) assert ( UserDataExport.query.filter_by(user_id=user_3.id).first().completed is True )