637 lines
20 KiB
Python
637 lines
20 KiB
Python
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,
|
|
) -> 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)
|
|
|
|
|
|
@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
|
|
)
|