from __future__ import absolute_import from __future__ import division from __future__ import print_function from __future__ import unicode_literals from .statements import * from django.db import transaction nu = datetime.datetime.now() from django.core.files import File from io import BytesIO from PIL import Image from io import StringIO from rowers.mytypes import workouttypes def create_image(storage, filename, size=(100, 100), image_mode='RGB', image_format='PNG'): """ Generate a test image, returning the filename that it was saved as. If ``storage`` is ``None``, the BytesIO containing the image data will be passed instead. """ data = BytesIO() Image.new(image_mode, size).save(data, image_format) data.seek(0) if not storage: return data image_file = ContentFile(data.read()) return storage.save(filename, image_file) class ListWorkoutTest(TestCase): def setUp(self): self.u = UserFactory() self.r = Rower.objects.create(user=self.u, birthdate=faker.profile()['birthdate'], gdproptin=True, ftpset=True,surveydone=True, gdproptindate=timezone.now(), rowerplan='pro') self.c = Client() self.user_workouts = WorkoutFactory.create_batch(5, user=self.r) self.factory = RequestFactory() self.password = faker.word() self.u.set_password(self.password) self.u.save() def tearDown(self): for workout in self.user_workouts: try: os.remove(workout.csvfilename) except (IOError, FileNotFoundError,OSError): pass @patch('rowers.dataprep.create_engine') @patch('rowers.dataprep.read_data') @patch('rowers.dataprep.myqueue') def test_list_workouts(self, mocked_sqlalchemy, mocked_read_data, mocked_myqueue): login = self.c.login(username=self.u.username, password=self.password) self.assertTrue(login) url = reverse('workouts_view') response = self.c.get(url) self.assertEqual(response.status_code,200) url2 = url+'?page=2' response = self.c.get(url2) self.assertEqual(response.status_code,200) url3 = url+'?page=3' response = self.c.get(url3) self.assertEqual(response.status_code,200) @patch('rowers.dataprep.create_engine') @patch('rowers.dataprep.read_data') @patch('rowers.dataprep.myqueue') def test_bulk_workouts(self, mocked_sqlalchemy, mocked_read_data, mocked_myqueue): login = self.c.login(username=self.u.username, password=self.password) self.assertTrue(login) url = '/rowers/list-workouts/?selectworkouts=true' response = self.c.get(url) self.assertEqual(response.status_code,200) form_data = { 'action': ['remove'], 'selectworkouts': ['Submit'], 'workoutid': [encoder.encode_hex(1), encoder.encode_hex(2)], } response = self.c.post(url, form_data, follow=True) expected_url = reverse('workouts_bulk_actions') self.assertRedirects(response, expected_url=expected_url, status_code=302, target_status_code=200) form_data = { 'workouts': ['1','2'], 'action': ['export'], 'destination': 'strava', } response = self.c.post(expected_url, form_data, follow=True) url = expected_url expected_url = reverse('workouts_view') self.assertRedirects(response, expected_url=expected_url, status_code=302, target_status_code=200) form_data = { 'workouts': ['1','2'], 'action': ['remove'], } response = self.c.post(url, form_data, follow=True) expected_url = reverse('workouts_view') self.assertRedirects(response, expected_url=expected_url, status_code=302, target_status_code=200) @override_settings(TESTING=True) class WorkoutViewTest(TestCase): def setUp(self): self.u = UserFactory() self.r = Rower.objects.create(user=self.u, birthdate=faker.profile()['birthdate'], gdproptin=True, ftpset=True,surveydone=True, gdproptindate=timezone.now(), rowerplan='pro') self.c = Client() self.c2 = Client() # for anonymous self.user_workouts = WorkoutFactory.create_batch(5, user=self.r) self.factory = RequestFactory() self.password = faker.word() self.u.set_password(self.password) self.u.save() result = get_random_file(filename='rowers/tests/testdata/onwater2.csv') self.wwater = WorkoutFactory(user=self.r, csvfilename=result['filename'], starttime=result['starttime'], startdatetime=result['startdatetime'], duration=result['duration'], distance=result['totaldist'], workouttype = 'water', ) result = get_random_file(filename='rowers/tests/testdata/quiske_in_stroke.csv') self.winstroke = WorkoutFactory(user=self.r, csvfilename=result['filename'], starttime=result['starttime'], startdatetime=result['startdatetime'], duration=result['duration'], distance=result['totaldist'], workouttype = 'water', ) result = get_random_file(filename='rowers/tests/testdata/erg1.csv') self.werg1 = WorkoutFactory(user=self.r, csvfilename=result['filename'], starttime=result['starttime'], startdatetime=result['startdatetime'], duration=result['duration'], distance=result['totaldist'], workouttype = 'rower', ) result = get_random_file(filename='rowers/tests/testdata/erg2.csv') self.werg2 = WorkoutFactory(user=self.r, csvfilename=result['filename'], starttime=result['starttime'], startdatetime=result['startdatetime'], duration=result['duration'], distance=result['totaldist'], workouttype = 'rower', ) result = get_random_file(filename='rowers/tests/testdata/erg2.csv') self.werg2copy = WorkoutFactory(user=self.r, csvfilename=result['filename'], starttime=result['starttime'], startdatetime=result['startdatetime'], duration=result['duration'], distance=result['totaldist'], workouttype = 'rower', ) result = get_random_file(filename='rowers/tests/testdata/erg3.csv') self.werg3 = WorkoutFactory(user=self.r, csvfilename=result['filename'], starttime=result['starttime'], startdatetime=result['startdatetime'], duration=result['duration'], distance=result['totaldist'], workouttype = 'rower', ) def tearDown(self): vs = VideoAnalysis.objects.all() for v in vs: v.delete() @patch('rowers.dataprep.create_engine') @patch('rowers.dataprep.read_data') @patch('rowers.middleware.myqueue') def test_forcecurve(self, mocked_sqlalchemy, mocked_read_data, mocked_myqueue): login = self.c.login(username=self.u.username, password=self.password) self.assertTrue(login) url = reverse('workout_forcecurve_view',kwargs={'id':encoder.encode_hex(self.wwater.id)}) response = self.c.get(url) self.assertEqual(response.status_code,200) form_data = { 'workstrokesonly': True, 'plottype': 'scatter' } response = self.c.post(url,form_data) self.assertEqual(response.status_code,200) @patch('rowers.dataprep.create_engine') @patch('rowers.dataprep.read_data') @patch('rowers.middleware.myqueue') def test_resample(self, mocked_sqlalchemy, mocked_read_data, mocked_myqueue): login = self.c.login(username=self.u.username, password=self.password) self.assertTrue(login) url = reverse('workout_resample_view',kwargs={'id':encoder.encode_hex(self.wwater.id)}) response = self.c.get(url) self.assertEqual(response.status_code,200) form_data = { 'resamplechoice': 'copy', } response = self.c.post(url,form_data) self.assertEqual(response.status_code,200) form_data = { 'resamplechoice': 'overwrite', } response = self.c.post(url,form_data) self.assertEqual(response.status_code,200) @patch('rowers.dataprep.create_engine') @patch('rowers.dataprep.read_data') @pytest.mark.django_db(transaction=True) def test_joins(self, mocked_sqlalchemy, mocked_read_data): with transaction.atomic(): login = self.c.login(username=self.u.username, password=self.password) self.assertTrue(login) url = reverse('workouts_join_select') response = self.c.get(url) self.assertEqual(response.status_code,200) d1 = self.werg1.date-datetime.timedelta(days=2) d2 = self.werg2.date+datetime.timedelta(days=2) form_data = { 'startdate': d1.strftime('%Y-%m-%d'), 'enddate': d2.strftime('%Y-%m-%d'), 'modality':'water', 'waterboattype':['1x'], } form = DateRangeForm(form_data) self.assertTrue(form.is_valid()) form = TrendFlexModalForm(form_data) self.assertTrue(form.is_valid()) response = self.c.post(url,form_data) self.assertEqual(response.status_code,200) url = reverse('workouts_join_view') response = self.c.get(url,follow=True) self.assertEqual(response.status_code,200) expected_url = reverse('workouts_join_select') self.assertRedirects(response, expected_url=expected_url, status_code=302, target_status_code=200) form_data = { 'workout_name': 'joined', 'set_private': False, 'workouts': [str(self.werg1.id),str(self.werg2.id),str(self.werg3.id)], } form = WorkoutMultipleCompareForm(form_data) self.assertTrue(form.is_valid()) form = WorkoutJoinParamForm(form_data) self.assertTrue(form.is_valid()) response = self.c.post(url,form_data,follow=True) self.assertEqual(response.status_code,200) expected_url = reverse(self.r.defaultlandingpage, kwargs = { 'id':encoder.encode_hex(12) }) self.assertRedirects(response, expected_url=expected_url, status_code=302,target_status_code=200) @patch('rowers.dataprep.create_engine') @patch('rowers.dataprep.read_data') @patch('rowers.weather.requests.get', side_effect=mocked_requests) def test_waterworkout_view(self, mocked_sqlalchemy, mocked_read_data, mocked_requests): login = self.c.login(username=self.u.username, password=self.password) self.assertTrue(login) url = reverse('workout_view',kwargs={'id':encoder.encode_hex(self.wwater.id)}) response = self.c.get(url) self.assertEqual(response.status_code,200) # Stacked Flex Chart url = reverse('workout_flexchart_stacked_view',kwargs={ 'id': encoder.encode_hex(self.wwater.id), }) response = self.c.get(url,follow=True) self.assertEqual(response.status_code,200) form_data = { 'xaxis':'time', 'yaxis1':'power', 'yaxis2':'pace', 'yaxis3':'hr', 'yaxis4':'spm' } response = self.c.post(url,form_data) self.assertEqual(response.status_code,200) url = reverse('workout_erase_column_view',kwargs={'id':encoder.encode_hex(self.wwater.id),'column':'hr'}) expected_url = reverse('workout_data_view',kwargs={'id':encoder.encode_hex(self.wwater.id)}) response = self.c.get(url,follow=True) self.assertEqual(response.status_code,200) form_data = {} response = self.c.post(url,form_data,follow=True) self.assertRedirects(response,expected_url=expected_url,status_code=302,target_status_code=200) url = expected_url form_data = { 'cols':['hr','spm'] } form = DataFrameColumnsForm(form_data) self.assertTrue(form.is_valid()) response = self.c.post(url,form_data) self.assertEqual(response.status_code,200) @patch('rowers.dataprep.create_engine') @patch('rowers.dataprep.read_data') def test_smoothen(self, mocked_sqlalchemy, mocked_read_data): login = self.c.login(username=self.u.username, password=self.password) self.assertTrue(login) url = reverse('workout_smoothenpace_view',kwargs={'id':encoder.encode_hex(self.wwater.id)}) response = self.c.get(url,follow=True) self.assertEqual(response.status_code,200) expected_url = reverse(self.r.defaultlandingpage, kwargs = { 'id':encoder.encode_hex(self.wwater.id) } ) self.assertRedirects(response, expected_url=expected_url, status_code=302,target_status_code=200) url = reverse('workout_undo_smoothenpace_view',kwargs={'id':encoder.encode_hex(self.wwater.id)}) response = self.c.get(url,follow=True) self.assertEqual(response.status_code,200) expected_url = reverse(self.r.defaultlandingpage, kwargs = { 'id':encoder.encode_hex(self.wwater.id) } ) self.assertRedirects(response, expected_url=expected_url, status_code=302,target_status_code=200) @patch('rowers.dataprep.create_engine') @patch('rowers.dataprep.read_data') def test_commentview(self, mocked_sqlalchemy, mocked_read_data): login = self.c.login(username=self.u.username, password=self.password) self.assertTrue(login) url = reverse('workout_comment_view',kwargs={'id':encoder.encode_hex(self.wwater.id)}) response = self.c.get(url) self.assertEqual(response.status_code,200) form_data = { 'comment': faker.text() } form = WorkoutCommentForm(form_data) self.assertTrue(form.is_valid()) response = self.c.post(url,form_data,follow=True) self.assertEqual(response.status_code,200) url = reverse('workout_unsubscribe_view',kwargs={'id':encoder.encode_hex(self.wwater.id)}) response = self.c.get(url) self.assertEqual(response.status_code,200) @patch('rowers.dataprep.create_engine') @patch('rowers.dataprep.read_data') def test_mapview(self, mocked_sqlalchemy, mocked_read_data): login = self.c.login(username=self.u.username, password=self.password) self.assertTrue(login) url = reverse('workout_map_view',kwargs={'id':encoder.encode_hex(self.wwater.id)}) response = self.c.get(url) self.assertEqual(response.status_code,200) @patch('rowers.dataprep.create_engine') @patch('rowers.dataprep.read_data') def notworking_test_workout_image(self, mocked_sqlalchemy, mocked_read_data): login = self.c.login(username=self.u.username, password=self.password) self.assertTrue(login) url = reverse('workout_uploadimage_view', kwargs={'id':encoder.encode_hex(self.werg1.id)}) response = self.c.get(url) self.assertEqual(response.status_code,200) with open('rowers/tests/testdata/lofoten.jpg','rb') as screenshot_file: file_data = {'file':screenshot_file} form_data = { 'file':screenshot_file, } response = self.c.post(url,form_data,format='multipart',follow=True) expected_url = reverse(self.r.defaultlandingpage, kwargs={'id':encoder.encode_hex(self.werg1.id)}) self.assertRedirects(response, expected_url=expected_url, status_code=302,target_status_code=200) @patch('rowers.dataprep.create_engine') @patch('rowers.dataprep.read_data') @patch('rowers.dataprep.getrowdata_db',side_effect=mocked_getrowdata_db) @pytest.mark.django_db(transaction=True) def test_workout_split(self, mocked_sqlalchemy, mocked_read_data, mocked_getrowdata_db): with transaction.atomic(): login = self.c.login(username=self.u.username, password=self.password) self.assertTrue(login) url = reverse('workout_split_view',kwargs={'id':encoder.encode_hex(self.werg1.id)}) response = self.c.get(url) self.assertEqual(response.status_code,200) form_data = { 'splittime': '2:00', 'splitmode': ['keep original','keep first','keep second'] } form = WorkoutSplitForm(form_data) self.assertTrue(form.is_valid()) response = self.c.post(url, form_data,follow=True) self.assertEqual(response.status_code,200) @patch('rowers.dataprep.create_engine') @patch('rowers.dataprep.getrowdata_db',side_effect=mocked_getrowdata_db) @patch('rowers.dataprep.read_data',side_effect=mocked_getsmallrowdata_forfusion) def test_workout_fusion(self, mocked_sqlalchemy, mocked_getrowdata_db, mocked_read_data): login = self.c.login(username=self.u.username, password=self.password) self.assertTrue(login) url = reverse('workout_fusion_view', kwargs={'id1':encoder.encode_hex(self.werg2.id), 'id2':encoder.encode_hex(self.werg2copy.id)}) response = self.c.get(url) self.assertEqual(response.status_code,200) form_data = { 'posneg':'pos', 'offset': datetime.timedelta(seconds=0), 'columns': ['power','hr' ] } form = FusionMetricChoiceForm(form_data,instance=self.werg2copy) self.assertTrue(form.is_valid()) response = self.c.post(url,form_data,follow=True) self.assertEqual(response.status_code,200) @patch('rowers.dataprep.create_engine') @patch('rowers.dataprep.read_data') @patch('rowers.dataprep.get_video_data',side_effect=mocked_videodata) def test_workout_video_view(self, mocked_sqlalchemy, mocked_read_data, mocked_videodata): login = self.c.login(username=self.u.username, password=self.password) self.assertTrue(login) url = reverse('workout_video_create_view',kwargs={'id':encoder.encode_hex(self.wwater.id)}) url2 = reverse('workout_video_view',kwargs={'id':encoder.encode_hex(1)}) yturl = 'https://youtu.be/6UzaWm1Ybrw' response = self.c.get(url) self.assertEqual(response.status_code,200) form_data = { 'url': yturl, 'delay': -119, 'groups': ['basic'], 'name': 'Video A', 'save_button':'Save', } response = self.c.post(url,form_data) self.assertRedirects(response, expected_url=url2, status_code=302, target_status_code=200) response = self.c.get(url2) self.assertEqual(response.status_code,200) form_data = { 'url': yturl, 'delay': -119, 'groups': ['basic'], 'name': 'Video A', 'save_button': 'Save', } response = self.c.post(url2,form_data) self.assertEqual(response.status_code,200) url = reverse('workout_video_view_mini',kwargs={'id':encoder.encode_hex(1)}) response = self.c.get(url) self.assertEqual(response.status_code,200) response = self.c.post(url,form_data) self.assertEqual(response.status_code,200) # delete video url = reverse('video_delete',kwargs={'pk':1}) response = self.c.get(url) self.assertEqual(response.status_code,200) response = self.c.post(url,{},follow=True) expected_url = reverse('workout_edit_view',kwargs={'id':encoder.encode_hex(self.wwater.id)}) self.assertRedirects(response,expected_url=expected_url,status_code=302,target_status_code=200) @patch('rowers.dataprep.create_engine') @patch('rowers.dataprep.read_data') @patch('rowers.dataprep.get_video_data',side_effect=mocked_videodata) def test_workout_video_view_erg(self, mocked_sqlalchemy, mocked_read_data, mocked_videodata): login = self.c.login(username=self.u.username, password=self.password) self.assertTrue(login) url = reverse('workout_video_create_view',kwargs={'id':encoder.encode_hex(self.werg1.id)}) url2 = reverse('workout_video_view',kwargs={'id':encoder.encode_hex(1)}) yturl = 'https://youtu.be/6UzaWm1Ybrw' response = self.c.get(url) self.assertEqual(response.status_code,200) form_data = { 'url': yturl, 'delay': -119, 'groups': ['basic'], 'name': 'Video A', 'save_button':'Save', } response = self.c.post(url,form_data) self.assertRedirects(response, expected_url=url2, status_code=302, target_status_code=200) response = self.c.get(url2) self.assertEqual(response.status_code,200) form_data = { 'url': yturl, 'delay': -119, 'groups': ['basic'], 'name': 'Video A', 'save_button': 'Save', } response = self.c.post(url2,form_data) self.assertEqual(response.status_code,200) url = reverse('workout_video_view_mini',kwargs={'id':encoder.encode_hex(1)}) response = self.c.get(url) self.assertEqual(response.status_code,200) response = self.c.post(url,form_data) self.assertEqual(response.status_code,200) url = reverse('workout_video_view_mini',kwargs={'id':encoder.encode_hex(1)}) response = self.c2.get(url) self.assertEqual(response.status_code,200) url = reverse('workout_video_view',kwargs={'id':encoder.encode_hex(1)}) response = self.c2.get(url) self.assertEqual(response.status_code,200) @patch('rowers.dataprep.create_engine') @patch('rowers.dataprep.getrowdata_db',side_effect=mocked_getrowdata_db) @patch('rowers.dataprep.get_video_data',side_effect=mocked_videodata) def test_video_selectworkout(self, mocked_sqlalchemy, mocked_read_data, mocked_videodata): login = self.c.login(username=self.u.username, password=self.password) self.assertTrue(login) url = reverse('video_selectworkout') response = self.c.get(url) self.assertEqual(response.status_code,200) d1 = self.wwater.date-datetime.timedelta(days=24) d2 = self.wwater.date+datetime.timedelta(days=24) form_data = { 'workout':'1', 'startdate': d1.strftime("%Y-%m-%d"), 'enddate': d2.strftime("%Y-%m-%d") } url2 = reverse('workout_video_create_view',kwargs={'id':encoder.encode_hex(1)}) response = self.c.post(url,form_data) self.assertRedirects(response, expected_url=url2, status_code=302, target_status_code=200) @patch('rowers.dataprep.create_engine') @patch('rowers.dataprep.read_data') def test_editsummaryview(self, mocked_sqlalchemy, mocked_read_data): login = self.c.login(username=self.u.username, password=self.password) self.assertTrue(login) url = reverse('workout_summary_edit_view',kwargs={'id':encoder.encode_hex(self.wwater.id)}) response = self.c.get(url) self.assertEqual(response.status_code,200) form_data = { 'ruptures':'ruptures', } response = self.c.post(url,form_data) self.assertEqual(response.status_code,200) form_data = { 'intervalstring':'4x2min/1min', } form = SummaryStringForm(form_data) self.assertTrue(form.is_valid()) response = self.c.post(url,form_data) self.assertEqual(response.status_code,200) form_data = { 'powerorpace':'power', 'value_pace':'2:23', 'value_power':'200', 'value_work':'400', 'value_spm':'20', 'activeminutesmin':'0', 'activeminutesmax':'60', 'savepowerpaceform':True, } response = self.c.post(url,form_data) self.assertEqual(response.status_code,200) form_data = { 'savestringform':'4x2min/1min', } response = self.c.post(url,form_data) self.assertEqual(response.status_code,200) form_data = { 'selector':'power', 'power': 200, 'pace': '2:30', 'work': 400, 'activeminutesmin':'0', 'activeminutesmax': '60', } form = PowerIntervalUpdateForm(form_data) self.assertTrue(form.is_valid()) response = self.c.post(url,form_data) self.assertEqual(response.status_code,200) @patch('rowers.dataprep.create_engine') @patch('rowers.dataprep.read_data') def test_workout_delete(self, mocked_sqlalchemy, mocked_read_data): login = self.c.login(username=self.u.username, password=self.password) self.assertTrue(login) url = reverse('workout_delete',kwargs={'pk':self.wwater.id}) response = self.c.get(url) self.assertEqual(response.status_code,200) post_data = {} response = self.c.post(url,follow=True) expected_url = reverse('workouts_view') self.assertRedirects(response, expected_url=expected_url, status_code=302,target_status_code=200)