FitTrackee/fittrackee/tests/users/test_users_export_data.py

417 lines
13 KiB
Python
Raw Normal View History

2023-03-01 12:10:21 +01:00
import os
import secrets
from unittest.mock import Mock, call, patch
from flask import Flask
from fittrackee import db
from fittrackee.users.export_data import UserDataExporter, export_user_data
from fittrackee.users.models import User, UserDataExport
2023-03-01 12:10:21 +01:00
from fittrackee.workouts.models import Sport, Workout
from ..mixins import CallArgsMixin
from ..utils import random_int, random_string
2023-03-01 12:10:21 +01:00
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,
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