From cf31c1e75000962a209e1537fd3958abda613aea Mon Sep 17 00:00:00 2001 From: Sander Roosendaal Date: Thu, 29 Feb 2024 18:08:46 +0100 Subject: [PATCH] better tests --- rowers/tasks.py | 44 +- rowers/tests/test_aworkouts.py | 1607 ++++++++++++++++---------------- rowers/tests/test_urls.py | 22 + rowers/views/apiviews.py | 50 +- rowers/views/workoutviews.py | 6 +- 5 files changed, 894 insertions(+), 835 deletions(-) diff --git a/rowers/tasks.py b/rowers/tasks.py index dac27df8..388af488 100644 --- a/rowers/tasks.py +++ b/rowers/tasks.py @@ -94,7 +94,7 @@ SITE_URL_DEV = CFG['site_url'] PROGRESS_CACHE_SECRET = CFG['progress_cache_secret'] try: SETTINGS_NAME = CFG['settings_name'] -except KeyError: +except KeyError: # pragma: no cover SETTINGS_NAME = 'rowsandall_ap.settings' try: @@ -321,7 +321,7 @@ def summaryfromsplitdata(splitdata, data, filename, sep='|', workouttype='rower' return sums, sa, results @app.task -def handle_post_workout_api(uploadoptions, debug=False, **kwargs): +def handle_post_workout_api(uploadoptions, debug=False, **kwargs): # pragma: no cover session = requests.session() newHeaders = {'Content-type': 'application/json', 'Accept': 'text/plain'} session.headers.update(newHeaders) @@ -335,14 +335,14 @@ def handle_post_workout_api(uploadoptions, debug=False, **kwargs): @app.task -def handle_remove_workouts_team(ws, t, debug=False, **kwargs): +def handle_remove_workouts_team(ws, t, debug=False, **kwargs): # pragma: no cover for w in ws: w.team.remove(t) return 1 @app.task -def handle_add_workouts_team(ws, t, debug=False, **kwargs): +def handle_add_workouts_team(ws, t, debug=False, **kwargs): # pragma: no cover for w in ws: w.team.add(t) @@ -350,7 +350,7 @@ def handle_add_workouts_team(ws, t, debug=False, **kwargs): return 1 def uploadactivity(access_token, filename, description='', - name='Rowsandall.com workout'): + name='Rowsandall.com workout'): # pragma: no cover data_gz = BytesIO() try: @@ -393,7 +393,7 @@ def uploadactivity(access_token, filename, description='', @app.task -def check_tp_workout_id(workout, location, attempts=5, debug=False, **kwargs): +def check_tp_workout_id(workout, location, attempts=5, debug=False, **kwargs): # pragma: no cover authorizationstring = str('Bearer ' + workout.user.tptoken) headers = {'Authorization': authorizationstring, 'user-agent': 'sanderroosendaal', @@ -411,7 +411,7 @@ def check_tp_workout_id(workout, location, attempts=5, debug=False, **kwargs): return 1 @app.task -def handle_workout_tp_upload(w, thetoken, tcxfilename, debug=False, **kwargs): +def handle_workout_tp_upload(w, thetoken, tcxfilename, debug=False, **kwargs): # pragma: no cover tpid = 0 r = w.user if not tcxfilename: @@ -443,7 +443,7 @@ def handle_workout_tp_upload(w, thetoken, tcxfilename, debug=False, **kwargs): return tpid @app.task -def instroke_static(w, metric, debug=False, **kwargs): +def instroke_static(w, metric, debug=False, **kwargs): # pragma: no cover f1 = w.csvfilename[6:-4] rowdata = rdata(csvfile=w.csvfilename) @@ -512,7 +512,7 @@ def handle_c2_sync(workoutid, url, headers, data, debug=False, **kwargs): try: workout = Workout.objects.get(id=workoutid) - except Workout.DoesNotExist: + except Workout.DoesNotExist: # pragma: no cover dologging('c2_log.log','failed for c2id {c2id}'.format(c2id=c2id)) return 0 @@ -531,7 +531,7 @@ def handle_c2_sync(workoutid, url, headers, data, debug=False, **kwargs): return 1 -def splitstdata(lijst): +def splitstdata(lijst): # pragma: no cover t = [] latlong = [] while len(lijst) >= 2: @@ -543,7 +543,7 @@ def splitstdata(lijst): @app.task def handle_sporttracks_workout_from_data(user, importid, source, - workoutsource, debug=False, **kwargs): + workoutsource, debug=False, **kwargs): # pragma: no cover r = user.rower authorizationstring = str('Bearer ' + r.sporttrackstoken) @@ -812,7 +812,7 @@ def handle_strava_sync(stravatoken, if not failed: try: workout = Workout.objects.get(id=workoutid) - except Workout.DoesNotExist: + except Workout.DoesNotExist: # pragma: no cover return 0 workout.uploadedtostrava = res.id @@ -1562,7 +1562,7 @@ def handle_calctrimp(id, try: workout = Workout.objects.get(id=id) - except Workout.DoesNotExist: + except Workout.DoesNotExist: # pragma: no cover dologging('metrics.log','Could not find workout {id}'.format(id=id)) return 0 @@ -2037,7 +2037,7 @@ def handle_sendemail_expired(useremail, userfirstname, userlastname, expireddate return 1 @app.task -def handle_sendemail_newftp(rower,power,mode, **kwargs): +def handle_sendemail_newftp(rower,power,mode, **kwargs): # pragma: no cover subject = "You may want to update your FTP on rowsandall.com" from_email = 'Rowsandall ' @@ -2069,7 +2069,7 @@ def handle_sendemail_breakthrough(workoutid, useremail, lastname = '' - if surname: + if surname: # pragma: no cover lastname = userlastname tablevalues = [ @@ -2122,7 +2122,7 @@ def handle_sendemail_hard(workoutid, useremail, ] lastname = '' - if surname: + if surname: # pragma: no cover lastname = userlastname # send email with attachment @@ -3414,7 +3414,7 @@ def handle_nk_async_workout(alldata, userid, nktoken, nkid, delaysec, defaulttim jsonData = response.json() try: strokeData = jsonData[str(nkid)] - except KeyError: + except KeyError: # pragma: no cover dologging('nklog.log','Could not find strokeData') return 0 @@ -3424,7 +3424,7 @@ def handle_nk_async_workout(alldata, userid, nktoken, nkid, delaysec, defaulttim if oarlockSessions: oarlocksession = oarlockSessions[0] seatNumber = oarlocksession['seatNumber'] - except KeyError: + except KeyError: # pragma: no cover pass df = strokeDataToDf(strokeData, seatIndex=seatNumber) @@ -3439,14 +3439,14 @@ def handle_nk_async_workout(alldata, userid, nktoken, nkid, delaysec, defaulttim workoutid, error = add_workout_from_data(userid, nkid, data, df) # dologging('nklog.log','NK Workout ID {id}'.format(id=workoutid)) - if workoutid == 0: + if workoutid == 0: # pragma: no cover return 0 try: workout = Workout.objects.get(id=workoutid) newnkid = workout.uploadedtonk sr = create_or_update_syncrecord(workout.user, workout, nkid=newnkid) - except Workout.DoesNotExist: + except Workout.DoesNotExist: # pragma: no cover pass return workoutid @@ -3775,7 +3775,7 @@ def handle_c2_async_workout(alldata, userid, c2token, c2id, delaysec, return workoutid @app.task -def fetch_rojabo_session(id,alldata,userid,rowerid,debug=False, **kwargs): +def fetch_rojabo_session(id,alldata,userid,rowerid,debug=False, **kwargs): # pragma: no cover try: item = alldata[id] except KeyError: @@ -3880,7 +3880,7 @@ def fetch_strava_workout(stravatoken, oauth_data, stravaid, csvfilename, userid, wsize = round(5./dt) velo2 = ewmovingaverage(velo, wsize) - except ValueError: + except ValueError: # pragma: no cover velo2 = velo if coords is not None: diff --git a/rowers/tests/test_aworkouts.py b/rowers/tests/test_aworkouts.py index 3f5f8f24..06da2171 100644 --- a/rowers/tests/test_aworkouts.py +++ b/rowers/tests/test_aworkouts.py @@ -32,858 +32,889 @@ def create_image(storage, filename, size=(100, 100), image_mode='RGB', image_for return storage.save(filename, image_file) class ListWorkoutTest(TestCase): - def setUp(self): - self.u = UserFactory() + 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='coach') - self.r = Rower.objects.create(user=self.u, - birthdate=faker.profile()['birthdate'], - gdproptin=True, ftpset=True,surveydone=True, - gdproptindate=timezone.now(), - rowerplan='coach') + self.c = Client() + self.user_workouts = WorkoutFactory.create_batch(len(workouttypes), user=self.r) + i = 0 + for workouttype in workouttypes: + self.user_workouts[i].workouttype = workouttype[0] + self.user_workouts[i].save() + i = i+1 - self.c = Client() - self.user_workouts = WorkoutFactory.create_batch(len(workouttypes), user=self.r) - i = 0 - for workouttype in workouttypes: - self.user_workouts[i].workouttype = workouttype[0] - self.user_workouts[i].save() - i = i+1 + self.factory = RequestFactory() + self.password = faker.word() + self.u.set_password(self.password) + self.u.save() - 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 - 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.getsmallrowdata_db') + @patch('rowers.dataprep.myqueue') + def test_list_workouts(self, mocked_sqlalchemy, + mocked_getsmallrowdata_db, + mocked_myqueue): - @patch('rowers.dataprep.create_engine') - @patch('rowers.dataprep.getsmallrowdata_db') - @patch('rowers.dataprep.myqueue') - def test_list_workouts(self, mocked_sqlalchemy, - mocked_getsmallrowdata_db, - mocked_myqueue): + login = self.c.login(username=self.u.username, password=self.password) + self.assertTrue(login) - 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.getsmallrowdata_db') + @patch('rowers.dataprep.myqueue') + def test_bulk_workouts(self, mocked_sqlalchemy, + mocked_getsmallrowdata_db, + 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) - url = reverse('workouts_view') - 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)], + } - url2 = url+'?page=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) - 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) + form_data = { + 'workouts': ['1','2'], + 'action': ['remove'], + } + response = self.c.post(expected_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() + 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='coach') + self.r = Rower.objects.create(user=self.u, + birthdate=faker.profile()['birthdate'], + gdproptin=True, ftpset=True,surveydone=True, + gdproptindate=timezone.now(), + rowerplan='coach') - 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() + 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') + 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', - ) + self.wwater = WorkoutFactory(user=self.r, + csvfilename=result['filename'], + starttime=result['starttime'], + startdatetime=result['startdatetime'], + duration=result['duration'], + distance=result['totaldist'], + workouttype = 'water', + ) -# self.comments = CommentFactory.create_batch(5, -# user = self.u, -# created = timezone.now, -# workout = self.wwater) + 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/quiske_in_stroke.csv') + 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', + ) - 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/erg2.csv') - result = get_random_file(filename='rowers/tests/testdata/erg1.csv') + self.werg2 = WorkoutFactory(user=self.r, + csvfilename=result['filename'], + starttime=result['starttime'], + startdatetime=result['startdatetime'], + duration=result['duration'], + distance=result['totaldist'], + workouttype = 'rower', + ) - 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') - 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', - ) + 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') + 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: + 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.getsmallrowdata_db') - @patch('rowers.middleware.myqueue') - def test_forcecurve(self, mocked_sqlalchemy, mocked_getsmallrowdata_db, - 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)}) + @patch('rowers.dataprep.create_engine') + @patch('rowers.dataprep.getsmallrowdata_db') + @patch('rowers.middleware.myqueue') + def test_forcecurve(self, mocked_sqlalchemy, mocked_getsmallrowdata_db, + mocked_myqueue): + login = self.c.login(username=self.u.username, password=self.password) + self.assertTrue(login) - response = self.c.get(url) - self.assertEqual(response.status_code,200) + 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' - } + form_data = { + 'workstrokesonly': True, + 'plottype': 'scatter' + } - response = self.c.post(url,form_data) - self.assertEqual(response.status_code,200) + response = self.c.post(url,form_data) + self.assertEqual(response.status_code,200) - @patch('rowers.dataprep.create_engine') - @patch('rowers.dataprep.getsmallrowdata_db') - @pytest.mark.django_db(transaction=True) - def test_joins(self, mocked_sqlalchemy, mocked_getsmallrowdata_db): - with transaction.atomic(): - login = self.c.login(username=self.u.username, password=self.password) - self.assertTrue(login) + @patch('rowers.dataprep.create_engine') + @patch('rowers.dataprep.getsmallrowdata_db') + @pytest.mark.django_db(transaction=True) + def test_joins(self, mocked_sqlalchemy, mocked_getsmallrowdata_db): + 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'], + } - url = reverse('workouts_join_select') - response = self.c.get(url) - self.assertEqual(response.status_code,200) + form = DateRangeForm(form_data) + self.assertTrue(form.is_valid()) - d1 = self.werg1.date-datetime.timedelta(days=2) - d2 = self.werg2.date+datetime.timedelta(days=2) + form = TrendFlexModalForm(form_data) + self.assertTrue(form.is_valid()) - form_data = { - 'startdate': d1.strftime('%Y-%m-%d'), - 'enddate': d2.strftime('%Y-%m-%d'), - 'modality':'water', - 'waterboattype':['1x'], - } + response = self.c.post(url,form_data) + self.assertEqual(response.status_code,200) - form = DateRangeForm(form_data) - self.assertTrue(form.is_valid()) + url = reverse('workouts_join_view') + + response = self.c.get(url,follow=True) + self.assertEqual(response.status_code,200) - form = TrendFlexModalForm(form_data) - self.assertTrue(form.is_valid()) + expected_url = reverse('workouts_join_select') + self.assertRedirects(response, + expected_url=expected_url, + status_code=302, + target_status_code=200) - response = self.c.post(url,form_data) - self.assertEqual(response.status_code,200) + form_data = { + 'workout_name': 'joined', + 'set_private': False, + 'workouts': [str(self.werg1.id),str(self.werg2.id),str(self.werg3.id)], + } - url = reverse('workouts_join_view') + form = WorkoutMultipleCompareForm(form_data) + self.assertTrue(form.is_valid()) - response = self.c.get(url,follow=True) - self.assertEqual(response.status_code,200) + form = WorkoutJoinParamForm(form_data) + self.assertTrue(form.is_valid()) - expected_url = reverse('workouts_join_select') - self.assertRedirects(response, - expected_url=expected_url, - status_code=302, - target_status_code=200) + response = self.c.post(url,form_data,follow=True) + self.assertEqual(response.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 = { + 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.getsmallrowdata_db') - @patch('rowers.weather.requests.get', side_effect=mocked_requests) - def test_waterworkout_view(self, - mocked_sqlalchemy, mocked_getsmallrowdata_db, 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) - - url = reverse('workout_downloadmetar_view',kwargs={ - 'id': encoder.encode_hex(self.wwater.id), - 'airportcode': 'LKHO' - } - ) - - response = self.c.get(url,follow=True) - self.assertEqual(response.status_code,200) - - url = reverse('workout_downloadwind_view',kwargs={ - 'id': encoder.encode_hex(self.wwater.id), - } - ) - - response = self.c.get(url,follow=True) - 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.getsmallrowdata_db') - def test_smoothen(self, mocked_sqlalchemy, mocked_getsmallrowdata_db): - 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.getsmallrowdata_db') - def test_windform(self, mocked_sqlalchemy, mocked_getsmallrowdata_db): - login = self.c.login(username=self.u.username, password=self.password) - self.assertTrue(login) - - - url = reverse('workout_wind_view',kwargs={'id':encoder.encode_hex(self.wwater.id)}) - - response = self.c.get(url) - self.assertEqual(response.status_code,200) - - form_data = { - 'dist1':1000, - 'dist2':2000, - 'vwind1':2.0, - 'vwind2':1.4, - 'windunit':'m', - 'winddirection1': 0, - 'winddirection2': 90, - } - - form = UpdateWindForm(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.getsmallrowdata_db') - def test_streamform(self, mocked_sqlalchemy, mocked_getsmallrowdata_db): - login = self.c.login(username=self.u.username, password=self.password) - self.assertTrue(login) - - - url = reverse('workout_stream_view',kwargs={'id':encoder.encode_hex(self.wwater.id)}) - - response = self.c.get(url) - self.assertEqual(response.status_code,200) - - form_data = { - 'dist1':1000, - 'dist2':2000, - 'stream1':2.0, - 'stream2':1.4, - 'streamunit':'m', - } - - form = UpdateStreamForm(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.getsmallrowdata_db') - @patch('rowers.middleware.myqueue') - def test_setpowerform(self, mocked_sqlalchemy, mocked_getsmallrowdata_db, - mocked_myqueue): - login = self.c.login(username=self.u.username, password=self.password) - self.assertTrue(login) - - url = reverse('instroke_view',kwargs={'id':encoder.encode_hex(self.winstroke.id)}) - - response = self.c.get(url) - self.assertEqual(response.status_code,200) - - rowdata = rowingdata.rowingdata(csvfile=self.winstroke.csvfilename) - instrokemetrics = rowdata.get_instroke_columns() - - self.assertTrue(len(instrokemetrics)>0) - - url = reverse('instroke_chart', - kwargs={ - 'id':encoder.encode_hex(self.winstroke.id), - 'metric':instrokemetrics[0], - }) - url2 = reverse(self.r.defaultlandingpage,kwargs={'id':encoder.encode_hex(self.winstroke.id)}) - response = self.c.get(url) - self.assertRedirects(response, - expected_url=url2, - status_code=302,target_status_code=200) - - @patch('rowers.dataprep.create_engine') - @patch('rowers.dataprep.getsmallrowdata_db') - def test_setpowerform(self, mocked_sqlalchemy, mocked_getsmallrowdata_db): - login = self.c.login(username=self.u.username, password=self.password) - self.assertTrue(login) - - - url = reverse('workout_otwsetpower_view',kwargs={'id':encoder.encode_hex(self.wwater.id)}) - - response = self.c.get(url) - self.assertEqual(response.status_code,200) - - form_data = { - 'quick_calc':True, - 'boattype': '1x', - 'weightvalue': 75.0, - 'boatbrand':'maas', - } - - form = AdvancedWorkoutForm(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('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.getsmallrowdata_db') - def test_commentview(self, mocked_sqlalchemy, mocked_getsmallrowdata_db): - 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.getsmallrowdata_db') - def test_mapview(self, mocked_sqlalchemy, mocked_getsmallrowdata_db): - 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.getsmallrowdata_db') - def notworking_test_workout_image(self, mocked_sqlalchemy, - mocked_getsmallrowdata_db): - 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, - } - - #form = ImageForm(form_data, file_data) - #if not form.is_valid(): - # print form.errors - #self.assertTrue(form.is_valid()) - - response = self.c.post(url,form_data,format='multipart',follow=True) - - expected_url = reverse(self.r.defaultlandingpage, + }) + + self.assertRedirects(response, + expected_url=expected_url, + status_code=302,target_status_code=200) + + + @patch('rowers.dataprep.create_engine') + @patch('rowers.dataprep.getsmallrowdata_db') + @patch('rowers.weather.requests.get', side_effect=mocked_requests) + def test_waterworkout_view(self, + mocked_sqlalchemy, mocked_getsmallrowdata_db, 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) + + url = reverse('workout_downloadmetar_view',kwargs={ + 'id': encoder.encode_hex(self.wwater.id), + 'airportcode': 'LKHO' + } + ) + + response = self.c.get(url,follow=True) + self.assertEqual(response.status_code,200) + + url = reverse('workout_downloadwind_view',kwargs={ + 'id': encoder.encode_hex(self.wwater.id), + } + ) + + response = self.c.get(url,follow=True) + 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.getsmallrowdata_db') + def test_smoothen(self, mocked_sqlalchemy, mocked_getsmallrowdata_db): + 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.getsmallrowdata_db') + def test_windform(self, mocked_sqlalchemy, mocked_getsmallrowdata_db): + login = self.c.login(username=self.u.username, password=self.password) + self.assertTrue(login) + + + url = reverse('workout_wind_view',kwargs={'id':encoder.encode_hex(self.wwater.id)}) + + response = self.c.get(url) + self.assertEqual(response.status_code,200) + + form_data = { + 'dist1':1000, + 'dist2':2000, + 'vwind1':2.0, + 'vwind2':1.4, + 'windunit':'m', + 'winddirection1': 0, + 'winddirection2': 90, + } + + form = UpdateWindForm(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.getsmallrowdata_db') + def test_streamform(self, mocked_sqlalchemy, mocked_getsmallrowdata_db): + login = self.c.login(username=self.u.username, password=self.password) + self.assertTrue(login) + + + url = reverse('workout_stream_view',kwargs={'id':encoder.encode_hex(self.wwater.id)}) + + response = self.c.get(url) + self.assertEqual(response.status_code,200) + + form_data = { + 'dist1':1000, + 'dist2':2000, + 'stream1':2.0, + 'stream2':1.4, + 'streamunit':'m', + } + + form = UpdateStreamForm(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.getsmallrowdata_db') + @patch('rowers.middleware.myqueue') + def test_setpowerform(self, mocked_sqlalchemy, mocked_getsmallrowdata_db, + mocked_myqueue): + login = self.c.login(username=self.u.username, password=self.password) + self.assertTrue(login) + + url = reverse('instroke_view',kwargs={'id':encoder.encode_hex(self.winstroke.id)}) + + response = self.c.get(url) + self.assertEqual(response.status_code,200) + + rowdata = rowingdata.rowingdata(csvfile=self.winstroke.csvfilename) + instrokemetrics = rowdata.get_instroke_columns() + + self.assertTrue(len(instrokemetrics)>0) + + url = reverse('instroke_chart', + kwargs={ + 'id':encoder.encode_hex(self.winstroke.id), + 'metric':instrokemetrics[0], + }) + url2 = reverse(self.r.defaultlandingpage,kwargs={'id':encoder.encode_hex(self.winstroke.id)}) + response = self.c.get(url) + self.assertRedirects(response, + expected_url=url2, + status_code=302,target_status_code=200) + + @patch('rowers.dataprep.create_engine') + @patch('rowers.dataprep.getsmallrowdata_db') + def test_setpowerform(self, mocked_sqlalchemy, mocked_getsmallrowdata_db): + login = self.c.login(username=self.u.username, password=self.password) + self.assertTrue(login) + + + url = reverse('workout_otwsetpower_view',kwargs={'id':encoder.encode_hex(self.wwater.id)}) + + response = self.c.get(url) + self.assertEqual(response.status_code,200) + + form_data = { + 'quick_calc':True, + 'boattype': '1x', + 'weightvalue': 75.0, + 'boatbrand':'maas', + } + + form = AdvancedWorkoutForm(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('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.getsmallrowdata_db') + def test_commentview(self, mocked_sqlalchemy, mocked_getsmallrowdata_db): + 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.getsmallrowdata_db') + def test_mapview(self, mocked_sqlalchemy, mocked_getsmallrowdata_db): + 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.getsmallrowdata_db') + def notworking_test_workout_image(self, mocked_sqlalchemy, + mocked_getsmallrowdata_db): + 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) + self.assertRedirects(response, + expected_url=expected_url, + status_code=302,target_status_code=200) + + @patch('rowers.dataprep.create_engine') + @patch('rowers.dataprep.getsmallrowdata_db') + @patch('rowers.dataprep.getrowdata_db',side_effect=mocked_getrowdata_db) + @pytest.mark.django_db(transaction=True) + def test_workout_split(self, mocked_sqlalchemy, mocked_getsmallrowdata_db, + 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.getsmallrowdata_db',side_effect=mocked_getsmallrowdata_forfusion) + def test_workout_fusion(self, mocked_sqlalchemy, mocked_getrowdata_db, + mocked_getsmallrowdata_db): + 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) + + + def test_remove_power_view(self): + login = self.c.login(username=self.u.username, password=self.password) + self.assertTrue(login) + + url = reverse('remove_power_view',kwargs={'id':encoder.encode_hex(self.wwater.id)}) + url2 = reverse(self.r.defaultlandingpage, + kwargs={ + 'id':encoder.encode_hex(self.wwater.id) + } + ) + response = self.c.get(url) + self.assertRedirects(response, + expected_url=url2, + status_code=302, + target_status_code=200 + ) - @patch('rowers.dataprep.create_engine') - @patch('rowers.dataprep.getsmallrowdata_db') - @patch('rowers.dataprep.getrowdata_db',side_effect=mocked_getrowdata_db) - @pytest.mark.django_db(transaction=True) - def test_workout_split(self, mocked_sqlalchemy, mocked_getsmallrowdata_db, - 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.getsmallrowdata_db',side_effect=mocked_getsmallrowdata_forfusion) - def test_workout_fusion(self, mocked_sqlalchemy, mocked_getrowdata_db, - mocked_getsmallrowdata_db): - 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)}) + @patch('rowers.dataprep.create_engine') + @patch('rowers.dataprep.getsmallrowdata_db') + @patch('rowers.dataprep.get_video_data',side_effect=mocked_videodata) + def test_workout_video_view(self, mocked_sqlalchemy, mocked_getsmallrowdata_db, + 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)}) - response = self.c.get(url) - self.assertEqual(response.status_code,200) + 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', + } - form_data = { - 'posneg':'pos', - 'offset': datetime.timedelta(seconds=0), - 'columns': ['power','hr' - ] - } + 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 = FusionMetricChoiceForm(form_data,instance=self.werg2copy) + form_data = { + 'url': yturl, + 'delay': -119, + 'groups': ['basic'], + 'name': 'Video A', + 'save_button': 'Save', + } - self.assertTrue(form.is_valid()) + response = self.c.post(url2,form_data) + self.assertEqual(response.status_code,200) - response = self.c.post(url,form_data,follow=True) - 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.getsmallrowdata_db') + @patch('rowers.dataprep.get_video_data',side_effect=mocked_videodata) + def test_workout_video_view_erg(self, mocked_sqlalchemy, mocked_getsmallrowdata_db, + 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) - def test_remove_power_view(self): - login = self.c.login(username=self.u.username, password=self.password) - self.assertTrue(login) - - url = reverse('remove_power_view',kwargs={'id':encoder.encode_hex(self.wwater.id)}) - url2 = reverse(self.r.defaultlandingpage, - kwargs={ - 'id':encoder.encode_hex(self.wwater.id) - } - ) - response = self.c.get(url) - self.assertRedirects(response, - expected_url=url2, - status_code=302, - target_status_code=200 - ) - - @patch('rowers.dataprep.create_engine') - @patch('rowers.dataprep.getsmallrowdata_db') - @patch('rowers.dataprep.get_video_data',side_effect=mocked_videodata) - def test_workout_video_view(self, mocked_sqlalchemy, mocked_getsmallrowdata_db, + @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_getsmallrowdata_db, mocked_videodata): - login = self.c.login(username=self.u.username, password=self.password) - self.assertTrue(login) + 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)}) + 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") + } - yturl = 'https://youtu.be/6UzaWm1Ybrw' + url2 = reverse('workout_video_create_view',kwargs={'id':encoder.encode_hex(1)}) - 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.getsmallrowdata_db') - @patch('rowers.dataprep.get_video_data',side_effect=mocked_videodata) - def test_workout_video_view_erg(self, mocked_sqlalchemy, mocked_getsmallrowdata_db, - 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_getsmallrowdata_db, - 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) + 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.getsmallrowdata_db') - def test_editsummaryview(self, mocked_sqlalchemy, mocked_getsmallrowdata_db): - login = self.c.login(username=self.u.username, password=self.password) - self.assertTrue(login) + @patch('rowers.dataprep.create_engine') + @patch('rowers.dataprep.getsmallrowdata_db') + def test_editsummaryview(self, mocked_sqlalchemy, mocked_getsmallrowdata_db): + 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) - 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.getsmallrowdata_db') - def test_workout_delete(self, mocked_sqlalchemy, mocked_getsmallrowdata_db): - 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) + @patch('rowers.dataprep.create_engine') + @patch('rowers.dataprep.getsmallrowdata_db') + def test_workout_delete(self, mocked_sqlalchemy, mocked_getsmallrowdata_db): + 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) diff --git a/rowers/tests/test_urls.py b/rowers/tests/test_urls.py index 87da9b7d..c798445f 100644 --- a/rowers/tests/test_urls.py +++ b/rowers/tests/test_urls.py @@ -7,6 +7,7 @@ from __future__ import unicode_literals from .statements import * nu = datetime.datetime.now() +import rowers.courses as courses tested = [ @@ -27,6 +28,14 @@ class URLTests(TestCase): r = Rower.objects.create(user=u,rowerplan='coach',gdproptin=True, ftpset=True, gdproptindate=timezone.now()) self.c = Client() + cs = courses.kmltocourse('rowers/tests/testdata/thyro.kml') + course = cs[0] + cname = course['name'] + cnotes = course['description'] + polygons = course['polygons'] + self.ThyroBaantje = courses.createcourse(r,cname,polygons,notes=cnotes) + self.ThyroBaantje.save() + self.nu = datetime.datetime.now() filename = 'rowers/tests/testdata/testdata.csv' @@ -168,6 +177,19 @@ class URLTests(TestCase): '/rowers/workout/upload/team/', '/rowers/workouts-join/', '/rowers/workouts-join-select/', + '/rowers/api/courses/', + '/rowers/api/courses/?name=Brno', + '/rowers/api/courses/?course_distance=2000', + '/rowers/api/courses/?course_distance=aap', + '/rowers/api/courses/?distance_from=52&latitude=42&longitude=7', + '/rowers/api/courses/?distance_from=50&latitude=42&longitude=-aap', + '/rowers/api/courses/', + '/rowers/api/courses/1/', + '/rowers/list-workouts/?selectworkouts=true', + '/rowers/api/courses/kml/', + '/rowers/api/courses/kml/?id=1', + '/rowers/api/courses/kml?id=1&id=4/', + '/rowers/api/courses/kml/?id=aap', ] diff --git a/rowers/views/apiviews.py b/rowers/views/apiviews.py index d7173ea3..c6f60c98 100644 --- a/rowers/views/apiviews.py +++ b/rowers/views/apiviews.py @@ -27,7 +27,7 @@ class XMLParser(BaseParser): dologging("apilog.log", "XML Parser") try: s = ET.parse(stream).getroot() - except Exception as e: + except Exception as e: # pragma: no cover dologging("apilog.log",e) return HttpResponse(status=500) return s @@ -231,11 +231,11 @@ def strokedataform_v2(request, id=0): def part_of_day(hour): if 5 <= hour < 12: return "Morning" - elif 12 <= hour < 18: + elif 12 <= hour < 18: # pragma: no cover return "Afternoon" - elif 18 <= hour < 24: + elif 18 <= hour < 24: # pragma: no cover return "Evening" - else: + else: # pragma: no cover return "Night" # KML API views @@ -254,7 +254,7 @@ Optional, not for CN @api_view(["GET"]) @permission_classes([AllowAny]) def course_list(request): - if request.method != 'GET': + if request.method != 'GET': # pragma: no cover dologging('apilog.log','{m} request to KML endpoint'.format(m=request.method)) return HttpResponseNotAllowed("Method not supported") @@ -281,7 +281,7 @@ def course_list(request): newlist = [] for c in courses: distance = geo_distance(float(latitude), float(longitude), c.coord[0], c.coord[1])[0] - if distance < float(distance_from): + if distance < float(distance_from): # pragma: no cover newlist.append(c) courses = newlist except ValueError: @@ -307,13 +307,13 @@ def course_list(request): @api_view(["GET"]) @permission_classes([AllowAny]) def get_crewnerd_kml(request,id=0): - if request.method != 'GET': + if request.method != 'GET': # pragma: no cover dologging('apilog.log','{m} request to CrewNerd KML endpoint'.format(m=request.method)) return HttpResponseNotAllowed("Method not supported") try: c = GeoCourse.objects.get(id=id) - except GeoCourse.DoesNotExist: + except GeoCourse.DoesNotExist: # pragma: no cover raise Http404("This course does not exist") kml = coursetokml(c, cn=True) @@ -324,14 +324,20 @@ def get_crewnerd_kml(request,id=0): @api_view(["GET"]) @permission_classes([AllowAny]) def get_crewnerd_multiple(request): - if request.method != 'GET': + if request.method != 'GET': # pragma: no cover dologging('apilog.log','{m} request to CrewNerd KML endpoint'.format(m=request.method)) return HttpResponseNotAllowed("Method not supported") ids = request.GET.get('id') if ids is not None: tdict = dict(request.GET.lists()) - ids = [int(id) for id in tdict['id']] + idsnew = [] + for id in tdict['id']: + try: + idsnew.append(int(id)) + except ValueError: + pass + ids = idsnew else: gcs = GeoCourse.objects.all() ids = [c.id for c in gcs] @@ -351,11 +357,11 @@ def strokedata_tcx(request): """ Upload a TCX file through API """ - if request.method != 'POST': + if request.method != 'POST': # pragma: no cover dologging('apilog.log','GET request to TCX endpoint') - return HttpResponseNotAllowed("Method not supported") # pragma: no cover + return HttpResponseNotAllowed("Method not supported") - if 'application/xml' not in request.content_type.lower(): + if 'application/xml' not in request.content_type.lower(): # pragma: no cover dologging('apilog.log','POST data not application/xml, request to TCX endpoint') dologging('apilog.log', request.content_type.lower()) return HttpResponseNotAllowed("Need application/xml") @@ -382,7 +388,7 @@ def strokedata_tcx(request): lap_duration_seconds = float(lap_duration_node.text) total_duration += lap_duration_seconds - except Exception as e: + except Exception as e: # pragma: no cover dologging('apilog.log','TCX') dologging('apilog.log',e) return HttpResponseNotAllowed("Could not parse TCX data") @@ -436,7 +442,7 @@ def strokedata_tcx(request): "workout id": workoutid, "status": "success", }) - except Exception as e: + except Exception as e: # pragma: no cover dologging('apilog.log','TCX API endpoint') dologging('apilog.log',e) return HttpResponse(status=500) @@ -489,7 +495,7 @@ def strokedatajson_v3(request): title = request.data.get('name','') try: elapsedTime = request.data['elapsedTime'] - except KeyError: + except KeyError: # pragma: no cover try: duration = request.data['duration'] try: @@ -501,7 +507,7 @@ def strokedatajson_v3(request): return HttpResponse("Missing Elapsed Time", status=400) try: totalDistance = request.data['distance'] - except KeyError: + except KeyError: # pragma: no cover return HttpResponse("Missing Total Distance", status=400) timeZone = request.data.get('timezone','UTC') workouttype = request.data.get('workouttype','rower') @@ -522,12 +528,12 @@ def strokedatajson_v3(request): df = pd.DataFrame() try: strokes = request.data['strokes'] - except KeyError: + except KeyError: # pragma: no cover return HttpResponse("No Stroke Data in JSON", status=400) try: df = pd.DataFrame(strokes['data']) - except KeyError: + except KeyError: # pragma: no cover try: df = pd.DataFrame(request.data['strokedata']) except: @@ -538,7 +544,7 @@ def strokedatajson_v3(request): status, comment, data = api_get_dataframe(startdatetime, df) - if status != 200: + if status != 200: # pragma: no cover return HttpResponse(comment, status=status) @@ -648,7 +654,7 @@ def strokedatajson_v2(request, id): try: for d in request.data['data']: dologging('apilog.log',json.dumps(d)) - except KeyError: + except KeyError: # pragma: no cover try: for d in request.data['strokedata']: dologging('apilog.log',json.dumps(d)) @@ -672,7 +678,7 @@ def strokedatajson_v2(request, id): df.sort_index(inplace=True) status, comment, data = api_get_dataframe(row.startdatetime, df) - if status != 200: + if status != 200: # pragma: no cover return HttpResponse(comment, status=status) r = getrower(request.user) diff --git a/rowers/views/workoutviews.py b/rowers/views/workoutviews.py index 7aa5b6f5..32bf82cf 100644 --- a/rowers/views/workoutviews.py +++ b/rowers/views/workoutviews.py @@ -1664,7 +1664,7 @@ def course_compare_view(request, id=0): labeldict = { int(w.id): w.__str__() for w in workouts } - except: + except: # pragma: no cover labeldict = {} res = interactive_multiple_compare_chart(workoutids, xparam, yparam, @@ -1846,7 +1846,7 @@ def virtualevent_compare_view(request, id=0): labeldict = { int(w.id): w.__str__() for w in workouts } - except: + except: # pragma: no cover labeldict = {} res = interactive_multiple_compare_chart(workoutids, xparam, yparam, @@ -2085,7 +2085,7 @@ def workouts_bulk_actions(request): destination=destination)) url = reverse('workouts_view') return HttpResponseRedirect(url) - else: + else: # pragma: no cover if len(workouts) == 0: url = reverse(workouts_view) return HttpResponseRedirect(url)