FitTrackee/fittrackee/tests/users/test_users_export_data.py

708 lines
22 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,
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
)