from __future__ import absolute_import from __future__ import division from __future__ import print_function from __future__ import unicode_literals from django.db import transaction #from __future__ import print_function from .statements import * nu = datetime.datetime.now() import numpy as np import rowers from rowers import dataprep from rowers import tasks import urllib import json import rowers.utils as utils import rowers.rojabo_stuff as rojabo_stuff from rowers.integrations import * from django.db import transaction import rowers.integrations.strava as strava from rowers.nkimportutils import * @pytest.mark.django_db @override_settings(TESTING=True) class IntervalsObjects(DjangoTestCase): def setUp(self): self.c = Client() self.u = User.objects.create_user('john', 'sander@ds.ds', 'koeinsloot') self.u.first_name = 'John' self.u.last_name = 'Sander' self.u.save() self.r = Rower.objects.create(user=self.u,gdproptin=True, ftpset=True,surveydone=True, gdproptindate=timezone.now() ) self.r.save() self.c.login(username='john',password='koeinsloot') self.nu = datetime.datetime.now() filename = 'rowers/tests/testdata/testdata.csv' rr = rrower(hrmax=self.r.max,hrut2=self.r.ut2, hrut1=self.r.ut1,hrat=self.r.at, hrtr=self.r.tr,hran=self.r.an,ftp=self.r.ftp) row = rdata(csvfile=filename,rower=rr) totaldist = row.df['cum_dist'].max() totaltime = row.df['TimeStamp (sec)'].max()-row.df['TimeStamp (sec)'].min() totaltime = totaltime+row.df.loc[:,' ElapsedTime (sec)'].iloc[0] hours = int(totaltime/3600.) minutes = int((totaltime - 3600.*hours)/60.) seconds = int(totaltime - 3600.*hours - 60.*minutes) tenths = int(10*(totaltime - 3600.*hours - 60.*minutes - seconds)) duration = "%s:%s:%s.%s" % (hours,minutes,seconds,tenths) workoutdate = row.rowdatetime.strftime('%Y-%m-%d') workoutstarttime = row.rowdatetime.strftime('%H:%M:%S') self.w = Workout.objects.create( name='testworkout',workouttype='water', user=self.r,date=self.nu.strftime('%Y-%m-%d'), starttime=workoutstarttime, startdatetime=row.rowdatetime, duration=duration,distance=totaldist, csvfilename=filename ) self.u2 = User.objects.create_user('john2', 'ba@ds.ds', 'koeinsloot2') self.u2.first_name = 'John' self.u2.last_name = 'Sander2' self.u2.save() self.r2 = Rower.objects.create(user=self.u2,gdproptin=True, ftpset=True,surveydone=True, gdproptindate=timezone.now() ) self.w2 = Workout.objects.create( name='testworkout',workouttype='water', user=self.r2,date=self.nu.strftime('%Y-%m-%d'), starttime=workoutstarttime, startdatetime=row.rowdatetime, duration=duration,distance=totaldist, csvfilename=filename ) @patch('rowers.tasks.requests.get', side_effect=mocked_requests) @patch('rowers.tasks.requests.post', side_effect=mocked_requests) def test_intervals_import(self, mock_get, mock_post): res = tasks.handle_intervals_getworkout(self.r, '12345', 'i101042071') self.assertEqual(res,1) @pytest.mark.django_db @override_settings(TESTING=True) class RojaboObjects(DjangoTestCase): def setUp(self): self.c = Client() self.u = User.objects.create_user('john', 'sander@ds.ds', 'koeinsloot') self.u.first_name = 'John' self.u.last_name = 'Sander' self.u.save() self.r = Rower.objects.create(user=self.u,gdproptin=True, ftpset=True,surveydone=True, gdproptindate=timezone.now() ) self.r.rojabo_token = '12' self.r.rojabo_refreshtoken = 'ab' self.r.rojabo_tokenexpirydate = arrow.get(datetime.datetime.now()+datetime.timedelta(days=1)).datetime self.r.defaulttimezone = 'Europe/Prague' self.r.rowerplan = 'pro' self.r.save() self.u2 = User.objects.create_user('john2', 'sander@ds2.ds', 'stierinsloot') self.u2.first_name = 'Sander' self.u2.last_name = 'John' self.u2.save() self.r2 = Rower.objects.create(user=self.u2,gdproptin=True, ftpset=True,surveydone=True, gdproptindate=timezone.now() ) self.r2.rojabo_token = '12' self.r2.rojabo_refreshtoken = 'ab' self.r2.rojabo_tokenexpirydate = arrow.get(datetime.datetime.now()+datetime.timedelta(days=1)).datetime self.r2.defaulttimezone = 'Europe/Prague' self.r2.rowerplan = 'basic' self.r2.save() self.nu = datetime.datetime.now() @patch('rowers.rojabo_stuff.requests.get', side_effect=mocked_requests) @patch('rowers.rojabo_stuff.requests.post', side_effect=mocked_requests) def test_rojabo_import(self, mock_get, mock_post): self.c.login(username='john',password='koeinsloot') url = '/rowers/session/rojaboimport/' response = self.c.get(url) self.assertEqual(response.status_code, 200) form_data = { 'csrfmiddlewaretoken': ['xalRrf8y7P2Hhobges9QC4aKCWg4l06gexDc8g2DeuVExgadfk4YhV8oSE1Yu43U'], 'sessions': ['Import selected sessions'], 'sessionid': ['862172'] } response = self.c.post(url,form_data) self.assertEqual(response.status_code,200) pss = PlannedSession.objects.filter(manager=self.r.user) self.assertEqual(pss.count(),1) ps = pss[0] self.assertEqual(ps.approximate_duration,57) self.assertEqual(len(ps.steps['steps']),20) @patch('rowers.rojabo_stuff.requests.get', side_effect=mocked_requests) @patch('rowers.rojabo_stuff.requests.post', side_effect=mocked_requests) def test_rojabo_import_denied(self, mock_get, mock_post): self.c.login(username='john2',password='stierinsloot') url = '/rowers/session/rojaboimport/' response = self.c.get(url) self.assertEqual(response.status_code, 302) form_data = { 'csrfmiddlewaretoken': ['xalRrf8y7P2Hhobges9QC4aKCWg4l06gexDc8g2DeuVExgadfk4YhV8oSE1Yu43U'], 'sessions': ['Import selected sessions'], 'sessionid': ['862172'] } response = self.c.post(url,form_data) self.assertEqual(response.status_code,302) @pytest.mark.django_db @override_settings(TESTING=True) class C2Objects(DjangoTestCase): def setUp(self): self.c = Client() self.u = User.objects.create_user('john', 'sander@ds.ds', 'koeinsloot') self.u.first_name = 'John' self.u.last_name = 'Sander' self.u.save() self.r = Rower.objects.create(user=self.u,gdproptin=True, ftpset=True,surveydone=True, gdproptindate=timezone.now() ) self.r.c2token = '12' self.r.c2refreshtoken = 'ab' self.r.tokenexpirydate = arrow.get(datetime.datetime.now()+datetime.timedelta(days=1)).datetime self.r.defaulttimezone = 'Europe/Prague' self.r.save() self.c.login(username='john',password='koeinsloot') self.nu = datetime.datetime.now() filename = 'rowers/tests/testdata/testdata.csv' rr = rrower(hrmax=self.r.max,hrut2=self.r.ut2, hrut1=self.r.ut1,hrat=self.r.at, hrtr=self.r.tr,hran=self.r.an,ftp=self.r.ftp) row = rdata(csvfile=filename,rower=rr) totaldist = row.df['cum_dist'].max() totaltime = row.df['TimeStamp (sec)'].max()-row.df['TimeStamp (sec)'].min() totaltime = totaltime+row.df.loc[:,' ElapsedTime (sec)'].iloc[0] self.totaltime = totaltime hours = int(totaltime/3600.) minutes = int((totaltime - 3600.*hours)/60.) seconds = int(totaltime - 3600.*hours - 60.*minutes) tenths = int(10*(totaltime - 3600.*hours - 60.*minutes - seconds)) duration = "%s:%s:%s.%s" % (hours,minutes,seconds,tenths) workoutdate = row.rowdatetime.strftime('%Y-%m-%d') workoutstarttime = row.rowdatetime.strftime('%H:%M:%S') self.w = Workout.objects.create( name='testworkout',workouttype='water', user=self.r,date=self.nu.strftime('%Y-%m-%d'), starttime=workoutstarttime, startdatetime=row.rowdatetime, duration=duration,distance=totaldist, csvfilename=filename, timezone='Europe/Amsterdam' ) self.u2 = User.objects.create_user('john2', 'ba@ds.ds', 'koeinsloot2') self.u2.first_name = 'John' self.u2.last_name = 'Sander2' self.u2.save() self.r2 = Rower.objects.create(user=self.u2,gdproptin=True, ftpset=True,surveydone=True, gdproptindate=timezone.now() ) self.w2 = Workout.objects.create( name='testworkout',workouttype='water', user=self.r2,date=self.nu.strftime('%Y-%m-%d'), starttime=workoutstarttime, startdatetime=row.rowdatetime, duration=duration,distance=totaldist, csvfilename=filename ) def test_timezone_c2(self): c2integration = C2Integration(self.u) data = c2integration.createworkoutdata(self.w) wenddtime = self.w.startdatetime+datetime.timedelta(seconds=self.totaltime) t1 = arrow.get(wenddtime).timestamp() t2 = arrow.get(data['date']).timestamp() diff = np.abs(t1-t2+7200) if diff>=2: print(arrow.get(wenddtime)) print(arrow.get(data['date'])) print('Diff ',diff) self.assertEqual(data['timezone'],'Europe/Amsterdam') self.assertTrue(diff<2) #self.assertEqual(data['date'],wenddtime.strftime('%Y-%m-%d %H:%M:%S')) @patch('rowers.integrations.c2.Session', side_effect=mocked_requests) def test_c2_callback(self, mock_Session): response = self.c.get('/call_back?code=dsdoij232s',follow=True) self.assertEqual(response.status_code, 200) @patch('rowers.integrations.c2.Session', side_effect=mocked_requests) def test_c2_token_refresh(self, mock_Session): response = self.c.get('/rowers/me/c2refresh/',follow=True) self.assertEqual(response.status_code, 200) @patch('rowers.integrations.c2.requests.post', side_effect=mocked_requests) @patch('rowers.integrations.c2.requests.get', side_effect=mocked_requests) @patch('rowers.integrations.c2.requests.session', side_effect=mocked_requests) def test_c2_auto_import(self, mock_get, mock_post,MockSession): self.r.c2_auto_export = True self.r.save() c2integration = C2Integration(self.u) res = c2integration.get_workouts() self.assertEqual(res,50) res = c2integration.get_workouts() self.assertEqual(res,0) @patch('rowers.integrations.c2.requests.post', side_effect=mocked_requests) @patch('rowers.integrations.c2.requests.get', side_effect=mocked_requests) def test_c2_upload(self, mock_get, mock_post): url = '/rowers/workout/'+encoded1+'/c2uploadw/' response = self.c.get(url) self.assertRedirects(response, expected_url = '/rowers/workout/'+encoded1+'/edit/', status_code=302,target_status_code=200) self.assertEqual(response.url, '/rowers/workout/'+encoded1+'/edit/') self.assertEqual(response.status_code, 302) @patch('rowers.integrations.c2.requests.post', side_effect=mocked_requests) @patch('rowers.integrations.c2.requests.get', side_effect=mocked_requests) def test_c2_upload2(self, mock_get, mock_post): url = '/rowers/workout/'+encoded2+'/c2uploadw/' response = self.c.get(url) self.assertRedirects(response, expected_url = '/rowers/list-workouts/', status_code=302,target_status_code=200) self.assertEqual(response.url, '/rowers/list-workouts/') self.assertEqual(response.status_code, 302) @patch('rowers.integrations.c2.requests.post', side_effect=mocked_requests) @patch('rowers.integrations.c2.requests.get', side_effect=mocked_requests) def test_c2_list(self, mock_get, mock_post): response = self.c.get('/rowers/workout/c2import',follow=True) self.assertEqual(response.status_code,200) @patch('rowers.integrations.c2.requests.get', side_effect=mocked_requests) @patch('rowers.dataprep.create_engine') def test_c2_import(self, mock_get, mocked_sqlalchemys): response = self.c.get('/rowers/workout/c2import/12/',follow=True) expected_url = reverse('workout_import_view',kwargs={'source':'c2'}) self.assertRedirects(response, expected_url=expected_url, status_code=302,target_status_code=200) self.assertEqual(response.status_code, 200) def test_c2_import_54517340(self): with open('rowers/tests/testdata/c2_54517340.json','r') as infile: data = json.load(infile) ( startdatetime, starttime, workoutdate, duration, starttimeunix, timezone ) = utils.get_startdatetime_from_c2data(data) self.assertEqual(str(timezone),'America/Los_Angeles') df = tasks.df_from_summary(data) self.assertEqual(len(df),8) got = arrow.get(startdatetime).isoformat() want = arrow.get('2021-05-23 09:11:37.100000-07:00').isoformat() self.assertEqual(got, want) self.assertEqual(workoutdate,'2021-05-23') def test_c2_import_54583351(self): with open('rowers/tests/testdata/c2_54583351.json','r') as infile: data = json.load(infile) ( startdatetime, starttime, workoutdate, duration, starttimeunix, timezone ) = utils.get_startdatetime_from_c2data(data) self.assertEqual(str(timezone),'UTC') got = arrow.get(startdatetime).isoformat() want = arrow.get('2021-05-26 08:59:34.000000+00:00').isoformat() self.assertEqual(got, want) self.assertEqual(workoutdate,'2021-05-26') def test_c2_import_54838157(self): with open('rowers/tests/testdata/c2_54838157.json','r') as infile: data = json.load(infile) ( startdatetime, starttime, workoutdate, duration, starttimeunix, timezone ) = utils.get_startdatetime_from_c2data(data) self.assertEqual(str(timezone),'America/Los_Angeles') got = arrow.get(startdatetime).isoformat() want = arrow.get('2021-06-06 17:14:41.400000-07:00').isoformat() self.assertEqual(got, want) self.assertEqual(workoutdate,'2021-06-06') def test_c2_import_54744009(self): with open('rowers/tests/testdata/c2_54744009.json','r') as infile: data = json.load(infile) ( startdatetime, starttime, workoutdate, duration, starttimeunix, timezone ) = utils.get_startdatetime_from_c2data(data) self.assertEqual(str(timezone),'America/Los_Angeles') got = arrow.get(startdatetime).isoformat() want = arrow.get('2021-06-02 07:51:14.600000-07:00').isoformat() self.assertEqual(got, want) self.assertEqual(workoutdate,'2021-06-02') self.assertEqual(duration,'00:49:39.4') def test_c2_import_54933639(self): with open('rowers/tests/testdata/c2_54933639.json','r') as infile: data = json.load(infile) ( startdatetime, starttime, workoutdate, duration, starttimeunix, timezone ) = utils.get_startdatetime_from_c2data(data) self.assertEqual(str(timezone),'America/Los_Angeles') got = arrow.get(startdatetime).isoformat() want = arrow.get('2021-06-10 19:22:02.600000-07:00').isoformat() self.assertEqual(got, want) self.assertEqual(workoutdate,'2021-06-10') want = arrow.get('2021-06-10 19:22:02.600000-07:00').timestamp() got = 1623378122.6 self.assertEqual(got, want) got = arrow.get(startdatetime).timestamp() self.assertEqual(got, want) @patch('rowers.integrations.c2.requests.get', side_effect=mocked_requests) @patch('rowers.dataprep.create_engine') @patch('rowers.tasks.requests.session', side_effect=mocked_session) def test_c2_import_tz(self, mock_get, mocked_sqlalchemy, mock_session): response = self.c.get('/rowers/workout/c2import/22/',follow=True) expected_url = '/rowers/workout/c2import/' self.assertRedirects(response, expected_url=expected_url, status_code=302,target_status_code=200) self.assertEqual(response.status_code, 200) with open('rowers/tests/testdata/c2_timezone.json','r') as infile: data = json.load(infile) timezone = str(utils.get_timezone_from_c2data(data['data'])) self.assertEqual(timezone,'Europe/Prague') @patch('rowers.integrations.c2.requests.get', side_effect=mocked_requests) @patch('rowers.dataprep.create_engine') def test_c2_import_tz3(self, mock_get, mocked_sqlalchemy): response = self.c.get('/rowers/workout/c2import/32/',follow=True) expected_url = '/rowers/workout/c2import/' self.assertRedirects(response, expected_url=expected_url, status_code=302,target_status_code=200) self.assertEqual(response.status_code, 200) with open('rowers/tests/testdata/c2_timezonebad.json','r') as infile: data = json.load(infile) timezone = str(utils.get_timezone_from_c2data(data['data'])) self.assertEqual(timezone,'UTC') @patch('rowers.tasks.requests.get', side_effect=mocked_requests) @patch('rowers.dataprep.create_engine') @patch('rowers.integrations.c2.requests.session', side_effect=mocked_requests) def test_c2_import_tz2(self, mock_get, mocked_sqlalchemy, MockSession): response = self.c.get('/rowers/workout/c2import/31/',follow=True) expected_url = '/rowers/workout/c2import/' result = tasks.handle_c2_getworkout(self.r.user.id,self.r.c2token,31,self.r.defaulttimezone) self.assertRedirects(response, expected_url=expected_url, status_code=302,target_status_code=200) self.assertEqual(response.status_code, 200) with open('rowers/tests/testdata/c2_timezone2.json','r') as infile: data = json.load(infile) timezone = str(utils.get_timezone_from_c2data(data['data'])) self.assertEqual(timezone,'Europe/Amsterdam') startdatetime,starttime,workoutdate,duration,starttimeunix,timezone = utils.get_startdatetime_from_c2data(data['data']) self.assertEqual(starttime,"20:04:55") @patch('rowers.tasks.requests.get',side_effect=mocked_requests) @patch('rowers.tasks.requests.post',side_effect=mocked_requests) @patch('rowers.tasks.requests.session', side_effect=mocked_requests) def test_handle_c2_get_workouts(self, mock_get, mockpost,MockSession): with open('rowers/tests/testdata/c2workoutlist.txt','r') as infile: c2workoutdata = json.load(infile) alldata = {} for item in c2workoutdata['data']: alldata[item['id']] = item res = tasks.handle_c2_async_workout(alldata,self.u.id,self.r.c2token,33991243,0,self.r.defaulttimezone) self.assertEqual(res,1) @override_settings(TESTING=True) class C2ObjectsTokenExpired(DjangoTestCase): def setUp(self): self.c = Client() self.u = User.objects.create_user('john', 'sander@ds.ds', 'koeinsloot') self.u.first_name = 'John' self.u.last_name = 'Sander' self.u.save() self.r = Rower.objects.create(user=self.u,gdproptin=True, ftpset=True,surveydone=True, gdproptindate=timezone.now() ) self.r.c2token = '12' self.r.c2refreshtoken = 'ab' self.r.tokenexpirydate = arrow.get(datetime.datetime.now()-datetime.timedelta(days=1)).datetime self.r.save() self.c.login(username='john',password='koeinsloot') self.nu = datetime.datetime.now() filename = 'rowers/tests/testdata/testdata.csv' rr = rrower(hrmax=self.r.max,hrut2=self.r.ut2, hrut1=self.r.ut1,hrat=self.r.at, hrtr=self.r.tr,hran=self.r.an,ftp=self.r.ftp) row = rdata(csvfile=filename,rower=rr) totaldist = row.df['cum_dist'].max() totaltime = row.df['TimeStamp (sec)'].max()-row.df['TimeStamp (sec)'].min() totaltime = totaltime+row.df.loc[:,' ElapsedTime (sec)'].iloc[0] hours = int(totaltime/3600.) minutes = int((totaltime - 3600.*hours)/60.) seconds = int(totaltime - 3600.*hours - 60.*minutes) tenths = int(10*(totaltime - 3600.*hours - 60.*minutes - seconds)) duration = "%s:%s:%s.%s" % (hours,minutes,seconds,tenths) workoutdate = row.rowdatetime.strftime('%Y-%m-%d') workoutstarttime = row.rowdatetime.strftime('%H:%M:%S') self.w = Workout.objects.create( name='testworkout',workouttype='water', user=self.r,date=self.nu.strftime('%Y-%m-%d'), starttime=workoutstarttime, startdatetime=row.rowdatetime, duration=duration,distance=totaldist, csvfilename=filename ) @patch('rowers.integrations.c2.requests.post', side_effect=mocked_requests) @patch('rowers.integrations.c2.requests.get', side_effect=mocked_requests) @patch('rowers.integrations.c2.Session',side_effect=mocked_requests) def test_c2_list(self, mock_get, mock_post, mock_Session): response = self.c.get('/rowers/workout/c2import/',follow=True) self.assertEqual(response.status_code,200) @patch('rowers.integrations.c2.requests.get', side_effect=mocked_requests) @patch('rowers.dataprep.create_engine') def test_c2_import(self, mock_get, mocked_sqlalchemy): response = self.c.get('/rowers/workout/c2import/12/') self.assertEqual(response.status_code, 302) #@pytest.mark.django_db @override_settings(TESTING=True) class NKObjects(DjangoTestCase): def setUp(self): self.c = Client() self.u = User.objects.create_user('john', 'sander@ds.ds', 'koeinsloot') self.u.first_name = 'John' self.u.last_name = 'Sander' self.u.save() self.r = Rower.objects.create(user=self.u,gdproptin=True, ftpset=True,surveydone=True, gdproptindate=timezone.now() ) self.r.nktoken = '12' self.r.nkrefreshtoken = '123' self.r.nktokenexpirydate = arrow.get(datetime.datetime.now()-datetime.timedelta(days=1)).datetime self.r.save() self.c.login(username='john',password='koeinsloot') self.nu = datetime.datetime.now() filename = 'rowers/tests/testdata/testdata.csv' rr = rrower(hrmax=self.r.max,hrut2=self.r.ut2, hrut1=self.r.ut1,hrat=self.r.at, hrtr=self.r.tr,hran=self.r.an,ftp=self.r.ftp) row = rdata(csvfile=filename,rower=rr) totaldist = row.df['cum_dist'].max() totaltime = row.df['TimeStamp (sec)'].max()-row.df['TimeStamp (sec)'].min() totaltime = totaltime+row.df.loc[:,' ElapsedTime (sec)'].iloc[0] hours = int(totaltime/3600.) minutes = int((totaltime - 3600.*hours)/60.) seconds = int(totaltime - 3600.*hours - 60.*minutes) tenths = int(10*(totaltime - 3600.*hours - 60.*minutes - seconds)) duration = "%s:%s:%s.%s" % (hours,minutes,seconds,tenths) workoutdate = row.rowdatetime.strftime('%Y-%m-%d') workoutstarttime = row.rowdatetime.strftime('%H:%M:%S') self.w = Workout.objects.create( name='testworkout',workouttype='water', user=self.r,date=self.nu.strftime('%Y-%m-%d'), starttime=workoutstarttime, startdatetime=row.rowdatetime, duration=duration,distance=totaldist, csvfilename=filename ) self.u2 = User.objects.create_user('john2', 'ba@ds.ds', 'koeinsloot2') self.u2.first_name = 'John' self.u2.last_name = 'Sander2' self.u2.save() self.r2 = Rower.objects.create(user=self.u2,gdproptin=True, ftpset=True,surveydone=True, gdproptindate=timezone.now() ) self.w2 = Workout.objects.create( name='testworkout',workouttype='water', user=self.r2,date=self.nu.strftime('%Y-%m-%d'), starttime=workoutstarttime, startdatetime=row.rowdatetime, duration=duration,distance=totaldist, csvfilename=filename ) @patch('rowers.dataprep.create_engine') @patch('rowers.dataprep.read_data',side_effect=mocked_read_data) def test_nk_intervals(self,mocked_sqlalchemy,mocked_read_data): with open('rowers/tests/testdata/carlos_workout.json','r') as f: workoutdata = json.load(f) with open('rowers/tests/testdata/carlos_strokes.json','r') as f: strokedata = json.load(f) df = strokeDataToDf(strokedata, seatIndex=1) csvfilename = 'media/test_nk.csv.gz' df.to_csv(csvfilename, index_label='index',compression='gzip') summary = get_nk_allstats(workoutdata, df) self.assertEqual(len(summary),324) form_data = {"secret": settings.UPLOAD_SERVICE_SECRET, "user": 1, "file": csvfilename, "title": "JustGo-17256M", "workouttype": "water", "boattype": "1x", "nkid": 1206081, "inboard": 0.86, "oarlength": 2.84, "summary": summary, "oarlockfirmware": "001.01", "elapsedTime": 5349.0, "totalDistance": 17256.13, "useImpeller": False, "seatNumber": 1 } url = reverse('workout_upload_api') response = self.c.post(url,json.dumps(form_data),HTTP_HOST='127.0.0.1:4533', content_type='application/json') self.assertEqual(response.status_code,200) id = int(response.json()['id']) # should also test if workout is created w = Workout.objects.get(id=id) self.assertEqual(len(w.summary),324) @patch('rowers.integrations.nk.requests.get', side_effect=mocked_requests) @patch('rowers.integrations.nk.requests.post', side_effect=mocked_requests) def test_nk_list(self, mock_get, mockpost): integration = NKIntegration(self.u) result = integration.token_refresh() self.assertEqual(result,"TA3n1vrNjuQJWw0TdCDHnjSmrjIPULhTlejMIWqq") response = self.c.get('/rowers/workout/nkimport/') self.assertEqual(response.status_code,200) @patch('rowers.tasks.requests.get', side_effect=mocked_requests) @patch('rowers.tasks.requests.post', side_effect=mocked_requests) @patch('rowers.nkimportutils.requests.session', side_effect=mocked_session) def test_handle_nk_get_workouts(self, mock_get,mockpost, mocked_session): with open('rowers/tests/testdata/nk_list.json','r') as f: data = json.load(f) alldata = {} for item in data: alldata[item['id']] = item res = tasks.handle_nk_async_workout( alldata,self.r.user.id,self.r.nktoken,469,0,'UTC' ) self.assertTrue(res>0) @patch('rowers.integrations.nk.requests.post', side_effect=mocked_requests) def notest_nk_callback(self, mock_post): response = self.c.get('/nk_callback?code=absdef23&scope=read',follow=True) self.assertEqual(response.status_code, 200) @patch('rowers.integrations.nk.requests.get', side_effect=mocked_requests) @patch('rowers.integrations.nk.requests.post', side_effect=mocked_requests) def test_nk_get_workouts(self, mock_get, mockpost): integration = NKIntegration(self.u) result = integration.open() self.assertEqual(result,"TA3n1vrNjuQJWw0TdCDHnjSmrjIPULhTlejMIWqq") response = self.c.get('/rowers/workout/nkimport/?selectallnew=true',follow=True) expected = reverse('workouts_view') self.assertEqual(response.status_code, 200) @patch('rowers.integrations.nk.requests.get', side_effect=mocked_requests) @patch('rowers.integrations.nk.requests.post', side_effect=mocked_requests) @patch('rowers.nkimportutils.requests.session', side_effect=mocked_session) @patch('rowers.dataprep.read_data', side_effect=mocked_read_data) def test_nk_import(self, mock_get, mock_post, mocked_session, mocked_read_data): integration = NKIntegration(self.u) result = integration.token_refresh() response = self.c.get('/rowers/workout/nkimport/469',follow=True) expected_url = reverse('workout_import_view',kwargs={'source':'nk'}) self.assertRedirects(response, expected_url=expected_url, status_code=301,target_status_code=200) self.assertEqual(response.status_code, 200) w = Workout.objects.get(id=1) startdate = nu.date() enddate = (nu+datetime.timedelta(days=3)).date() form_data = { 'startdate': startdate, 'enddate': enddate, 'workoutid':['469'] } expected_url = reverse('workouts_view') response = self.c.post('/rowers/workout/nkimport/',form_data) self.assertRedirects(response, expected_url=expected_url, status_code=302,target_status_code=200) @patch('rowers.integrations.nk.requests.get', side_effect=mocked_requests) @patch('rowers.integrations.nk.requests.post', side_effect=mocked_requests) @patch('rowers.nkimportutils.requests.session', side_effect=mocked_session) @patch('rowers.dataprep.read_data', side_effect=mocked_read_data) def test_nk_import_task(self, mock_get, mock_post, mocked_session, mocked_read_data): alldata = { 469: {'elapsedTime': 3901900, 'totalDistanceImp': 0, 'totalDistanceGps': 12155.47, 'avgPaceImp': 0, 'avgPaceGps': 160771.704180064, 'avgStrokeRate': 20.5, 'distStrokeImp': 0, 'distStrokeGps': 9.8, 'avgHeartRate': 137, 'totalStrokeCount': 1237, 'totalCalories': 0, 'avgCalHour': 0, 'avgSpeedGps': 3.11, 'avgSpeedImp': 0, 'avgPower': 149.822294022617, 'avgCatch': -55.4211356466877, 'avgSlip': 7.16365131578947, 'avgFinish': 45.2515772870662, 'avgWash': 14.2442434210526, 'avgForceAvg': 236.360883280757, 'avgWork': 441.713731825525, 'avgForceMax': 464.872082018927, 'avgMaxForceAngle': -15.198738170347, 'startGpsLat': 52.2253125, 'startGpsLon': 6.8455583, 'id': 469, 'name': 'JustGo-12155M', 'type': 0, 'speedInput': 0, 'startTime': 1617102784995, 'endTime': 1617106686914, 'location': 'N/A', 'deviceId': 514, 'intervals': [{'elapsedTime': 3901900, 'totalDistanceImp': 0, 'totalDistanceGps': 12155.4697, 'avgPaceImp': 0, 'avgPaceGps': 160771.704180064, 'avgStrokeRate': 20.5, 'distStrokeImp': 0, 'distStrokeGps': 9.8, 'avgHeartRate': 137, 'totalStrokeCount': 1237, 'totalCalories': 0, 'avgCalHour': 0, 'avgSpeedGps': 3.11, 'avgSpeedImp': 0, 'avgPower': 149.822294022617, 'avgCatch': -55.4211356466877, 'avgSlip': 7.16365131578947, 'avgFinish': 45.2515772870662, 'avgWash': 14.2442434210526, 'avgForceAvg': 236.360883280757, 'avgWork': 441.713731825525, 'avgForceMax': 464.872082018927, 'avgMaxForceAngle': -15.198738170347, 'startGpsLat': 52.2253125, 'startGpsLon': 6.8455583, 'id': 622, 'sessionId': 469, 'startTime': 1617102784995, 'intervalNumber': 1, 'sessionStrokeStartIndex': 12545, 'sessionStrokeEndIndex': 15425, 'sessionStrokeCount': 1613}], 'oarlockSessions': [{'id': 69, 'sessionId': 469, 'boatName': 'DOLF1', 'seatNumber': 1, 'portStarboard': 1, 'oarLength': 287, 'oarInboardLength': 89}]} } res = tasks.handle_nk_async_workout(alldata,self.r.user.id,self.r.nktoken,469, 0,'UTC',debug=True) self.assertEqual(res,1) @patch('rowers.integrations.nk.requests.get', side_effect=mocked_requests) @patch('rowers.integrations.nk.requests.post', side_effect=mocked_requests) @patch('rowers.nkimportutils.requests.session', side_effect=mocked_session) @patch('rowers.dataprep.read_data', side_effect=mocked_read_data) def test_nk_import_impeller(self, mock_get, mock_post, mocked_session, mocked_read_data): integration = NKIntegration(self.u) result = integration.token_refresh() response = self.c.get('/rowers/workout/nkimport/404',follow=True) expected_url = reverse('workout_import_view',kwargs={'source':'nk'}) self.assertRedirects(response, expected_url=expected_url, status_code=301,target_status_code=200) self.assertEqual(response.status_code, 200) response = self.c.get('/rowers/workout/'+encoded1+'/otwuseimpeller/', follow=True) self.assertRedirects(response, expected_url='/rowers/workout/'+encoded1+'/edit/', status_code=302,target_status_code=200) w = Workout.objects.get(id=1) #self.assertTrue(w.impeller) @override_settings(TESTING=True) class PolarObjects(DjangoTestCase): def setUp(self): self.c = Client() self.u = User.objects.create_user('john', 'sander@ds.ds', 'koeinsloot') self.u.first_name = 'John' self.u.last_name = 'Sander' self.u.save() self.r = Rower.objects.create(user=self.u,gdproptin=True, ftpset=True,surveydone=True, gdproptindate=timezone.now() ) self.r.polartoken = '12' self.r.polarrefreshtoken = '123' self.r.polaruserid = 475 self.r.rowerplan = 'pro' self.r.polar_auto_import = True self.r.polartokenexpirydate = arrow.get(datetime.datetime.now()+datetime.timedelta(days=100)).datetime self.r.save() self.c.login(username='john',password='koeinsloot') self.nu = datetime.datetime.now() filename = 'rowers/tests/testdata/testdata.csv' rr = rrower(hrmax=self.r.max,hrut2=self.r.ut2, hrut1=self.r.ut1,hrat=self.r.at, hrtr=self.r.tr,hran=self.r.an,ftp=self.r.ftp) row = rdata(csvfile=filename,rower=rr) totaldist = row.df['cum_dist'].max() totaltime = row.df['TimeStamp (sec)'].max()-row.df['TimeStamp (sec)'].min() totaltime = totaltime+row.df.loc[:,' ElapsedTime (sec)'].iloc[0] hours = int(totaltime/3600.) minutes = int((totaltime - 3600.*hours)/60.) seconds = int(totaltime - 3600.*hours - 60.*minutes) tenths = int(10*(totaltime - 3600.*hours - 60.*minutes - seconds)) duration = "%s:%s:%s.%s" % (hours,minutes,seconds,tenths) workoutdate = row.rowdatetime.strftime('%Y-%m-%d') workoutstarttime = row.rowdatetime.strftime('%H:%M:%S') self.w = Workout.objects.create( name='testworkout',workouttype='water', user=self.r,date=self.nu.strftime('%Y-%m-%d'), starttime=workoutstarttime, startdatetime=row.rowdatetime, duration=duration,distance=totaldist, csvfilename=filename ) self.u2 = User.objects.create_user('john2', 'ba@ds.ds', 'koeinsloot2') self.u2.first_name = 'John' self.u2.last_name = 'Sander2' self.u2.save() self.r2 = Rower.objects.create(user=self.u2,gdproptin=True, ftpset=True,surveydone=True, gdproptindate=timezone.now() ) self.w2 = Workout.objects.create( name='testworkout',workouttype='water', user=self.r2,date=self.nu.strftime('%Y-%m-%d'), starttime=workoutstarttime, startdatetime=row.rowdatetime, duration=duration,distance=totaldist, csvfilename=filename ) @patch('rowers.integrations.polar.requests.post', side_effect=mocked_requests) @patch('rowers.integrations.polar.requests.get', side_effect=mocked_requests) def test_polar_auto_import(self, mock_get, mock_post): self.r.polar_auto_import = True self.r.save() integration = PolarIntegration(self.r.user) res = integration.get_workouts(self.r.user) self.assertEqual(res,1) @patch('rowers.integrations.polar.requests.post', side_effect=mocked_requests) @patch('rowers.integrations.polar.requests.get', side_effect=mocked_requests) def test_polar_callback(self, mock_get, mock_post): response = self.c.get('/polarflowcallback?code=abcdef&state=12sdss',follow=True) self.assertEqual(response.status_code,200) @patch('rowers.integrations.polar.requests.post', side_effect=mocked_requests) @patch('rowers.integrations.polar.requests.get', side_effect=mocked_requests) def test_polar_notifications(self, mock_get, mock_post): integration = PolarIntegration(self.r.user) data = integration.get_notifications() self.assertEqual(data[0]['user-id'],475) response = integration.get_workouts() self.assertEqual(response,1) @patch('rowers.integrations.polar.requests.post', side_effect=mocked_requests) @patch('rowers.integrations.polar.requests.get', side_effect=mocked_requests) def test_polar_get_workout(self, mock_get, mock_post): transaction_id = 240522162 id = 1937529874 integration = PolarIntegration(self.u) response = integration.get_workout(id, transaction_id) self.assertEqual(len(response),14836) #@pytest.mark.django_db @override_settings(TESTING=True) class RP3Objects(DjangoTestCase): def setUp(self): self.c = Client() self.u = User.objects.create_user('john', 'sander@ds.ds', 'koeinsloot') self.u.first_name = 'John' self.u.last_name = 'Sander' self.u.save() self.r = Rower.objects.create(user=self.u,gdproptin=True, ftpset=True,surveydone=True, gdproptindate=timezone.now() ) self.r.rp3token = '12' self.r.rp3refreshtoken = '123' self.r.rp3tokenexpirydate = arrow.get(datetime.datetime.now()-datetime.timedelta(days=1)).datetime self.r.save() self.c.login(username='john',password='koeinsloot') self.nu = datetime.datetime.now() filename = 'rowers/tests/testdata/testdata.csv' rr = rrower(hrmax=self.r.max,hrut2=self.r.ut2, hrut1=self.r.ut1,hrat=self.r.at, hrtr=self.r.tr,hran=self.r.an,ftp=self.r.ftp) row = rdata(csvfile=filename,rower=rr) totaldist = row.df['cum_dist'].max() totaltime = row.df['TimeStamp (sec)'].max()-row.df['TimeStamp (sec)'].min() totaltime = totaltime+row.df.loc[:,' ElapsedTime (sec)'].iloc[0] hours = int(totaltime/3600.) minutes = int((totaltime - 3600.*hours)/60.) seconds = int(totaltime - 3600.*hours - 60.*minutes) tenths = int(10*(totaltime - 3600.*hours - 60.*minutes - seconds)) duration = "%s:%s:%s.%s" % (hours,minutes,seconds,tenths) workoutdate = row.rowdatetime.strftime('%Y-%m-%d') workoutstarttime = row.rowdatetime.strftime('%H:%M:%S') self.w = Workout.objects.create( name='testworkout',workouttype='water', user=self.r,date=self.nu.strftime('%Y-%m-%d'), starttime=workoutstarttime, startdatetime=row.rowdatetime, duration=duration,distance=totaldist, csvfilename=filename ) self.u2 = User.objects.create_user('john2', 'ba@ds.ds', 'koeinsloot2') self.u2.first_name = 'John' self.u2.last_name = 'Sander2' self.u2.save() self.r2 = Rower.objects.create(user=self.u2,gdproptin=True, ftpset=True,surveydone=True, gdproptindate=timezone.now() ) self.w2 = Workout.objects.create( name='testworkout',workouttype='water', user=self.r2,date=self.nu.strftime('%Y-%m-%d'), starttime=workoutstarttime, startdatetime=row.rowdatetime, duration=duration,distance=totaldist, csvfilename=filename ) @patch('rowers.integrations.rp3.requests.get', side_effect=mocked_requests) @patch('rowers.integrations.rp3.requests.post', side_effect=mocked_requests) @patch('rowers.dataprep.read_data', side_effect=mocked_read_data) def test_rp3_import(self, mock_get, mockpost, mocked_read_data): response = self.c.get('/rowers/workout/rp3import/591621',follow=True) expected_url = reverse('workout_import_view',kwargs={'source':'rp3'}) self.assertRedirects(response, expected_url=expected_url, status_code=301,target_status_code=200) self.assertEqual(response.status_code, 200) w = Workout.objects.get(id=1) @patch('rowers.tasks.requests.get',side_effect=mocked_requests) @patch('rowers.tasks.requests.post',side_effect=mocked_requests) @patch('rowers.tasks.requests.session', side_effect=mocked_requests) def test_handle_rp3_get_workouts(self, mock_get, mockpost,MockSession): userid = self.u.id rp3token = 'ap' rp3id = '591621' startdatetime = timezone.now()-datetime.timedelta(days=30) max_attempts = 2 res = tasks.handle_rp3_async_workout(userid,rp3token,rp3id,startdatetime,max_attempts) self.assertEqual(res,1) @patch('rowers.integrations.rp3.requests.post', side_effect=mocked_requests) def notest_rp3_callback(self, mock_post): response = self.c.get('/rp3_callback?code=absdef23&scope=read',follow=True) self.assertEqual(response.status_code, 200) #@pytest.mark.django_db @override_settings(TESTING=True) class StravaObjects(DjangoTestCase): def setUp(self): self.c = Client() self.u = User.objects.create_user('john', 'sander@ds.ds', 'koeinsloot') self.u.first_name = 'John' self.u.last_name = 'Sander' self.u.save() self.r = Rower.objects.create(user=self.u,gdproptin=True, ftpset=True,surveydone=True, gdproptindate=timezone.now(), strava_auto_import=True, strava_auto_delete=True, ) self.r.stravatoken = '12' self.r.stravarefreshtoken = '123' self.r.stravatokenexpirydate = arrow.get(datetime.datetime.now()-datetime.timedelta(days=1)).datetime self.r.strava_owner_id = 4 self.r.save() self.c.login(username='john',password='koeinsloot') self.nu = datetime.datetime.now() filename = 'rowers/tests/testdata/testdata.csv' rr = rrower(hrmax=self.r.max,hrut2=self.r.ut2, hrut1=self.r.ut1,hrat=self.r.at, hrtr=self.r.tr,hran=self.r.an,ftp=self.r.ftp) row = rdata(csvfile=filename,rower=rr) totaldist = row.df['cum_dist'].max() totaltime = row.df['TimeStamp (sec)'].max()-row.df['TimeStamp (sec)'].min() totaltime = totaltime+row.df.loc[:,' ElapsedTime (sec)'].iloc[0] hours = int(totaltime/3600.) minutes = int((totaltime - 3600.*hours)/60.) seconds = int(totaltime - 3600.*hours - 60.*minutes) tenths = int(10*(totaltime - 3600.*hours - 60.*minutes - seconds)) duration = "%s:%s:%s.%s" % (hours,minutes,seconds,tenths) workoutdate = row.rowdatetime.strftime('%Y-%m-%d') workoutstarttime = row.rowdatetime.strftime('%H:%M:%S') self.w = Workout.objects.create( name='testworkout',workouttype='water', user=self.r,date=self.nu.strftime('%Y-%m-%d'), starttime=workoutstarttime, startdatetime=row.rowdatetime, duration=duration,distance=totaldist, csvfilename=filename,uploadedtostrava=123, ) self.u2 = User.objects.create_user('john2', 'ba@ds.ds', 'koeinsloot2') self.u2.first_name = 'John' self.u2.last_name = 'Sander2' self.u2.save() self.r2 = Rower.objects.create(user=self.u2,gdproptin=True, ftpset=True,surveydone=True, gdproptindate=timezone.now() ) self.w2 = Workout.objects.create( name='testworkout',workouttype='water', user=self.r2,date=self.nu.strftime('%Y-%m-%d'), starttime=workoutstarttime, startdatetime=row.rowdatetime, duration=duration,distance=totaldist, csvfilename=filename ) @patch('rowers.integrations.strava.requests.post', side_effect=mocked_requests) @patch('rowers.integrations.strava.requests.get', side_effect=mocked_requests) def test_strava_webhook(self, mock_get, mock_post): url = reverse('strava_webhook_view') params = { 'hub.challenge':'aap', 'hub.verify_token':strava.webhookverification, } url2 = url+'?'+urllib.parse.urlencode(params) response = self.c.get(url2) self.assertEqual(response.status_code,200) data = { 'aspect_type':'create', 'object_id': 245, 'object_type':'activity', 'owner_id': self.r.strava_owner_id, 'event_time': arrow.get(datetime.datetime.now()).timestamp(), } raw_data = json.dumps(data) response = self.c.generic('POST', url, raw_data) self.assertEqual(response.status_code,200) data = { 'aspect_type':'update', 'updates':{ 'title':faker.word(), 'type': 'Ride', }, 'object_id': 123, 'object_type':'activity', 'owner_id': self.r.strava_owner_id, 'event_time': arrow.get(datetime.datetime.now()).timestamp(), } raw_data = json.dumps(data) response = self.c.generic('POST', url, raw_data) self.assertEqual(response.status_code,200) data = { 'aspect_type':'delete', 'object_id': 245, 'object_type':'activity', 'owner_id': self.r.strava_owner_id, 'event_time': arrow.get(datetime.datetime.now()).timestamp(), } raw_data = json.dumps(data) response = self.c.generic('POST', url, raw_data) self.assertEqual(response.status_code,200) @patch('rowers.integrations.strava.requests.post', side_effect=mocked_requests) @patch('rowers.integrations.strava.requests.get', side_effect=mocked_requests) def test_workout_strava_upload(self, mock_get, mock_post): w = Workout.objects.get(id=1) integration = StravaIntegration(self.r.user) res = integration.workout_export(w) self.assertEqual(res,1) @patch('rowers.integrations.strava.requests.post', side_effect=mocked_requests) @patch('rowers.integrations.strava.requests.get', side_effect=mocked_requests) def test_strava_upload2(self, mock_get, mock_post): url = '/rowers/workout/'+encoded2+'/stravauploadw/' response = self.c.get(url) self.assertRedirects(response, expected_url = '/rowers/list-workouts/', status_code=302,target_status_code=200) self.assertEqual(response.url, '/rowers/list-workouts/') self.assertEqual(response.status_code, 302) @patch('rowers.integrations.strava.requests.post', side_effect=mocked_requests) @patch('rowers.integrations.strava.requests.get', side_effect=mocked_requests) def test_strava_upload(self, mock_get, mock_post): response = self.c.get('/rowers/workout/'+encoded1+'/stravauploadw/') self.assertRedirects(response, expected_url = '/rowers/workout/'+encoded1+'/edit/', status_code=302,target_status_code=200) self.assertEqual(response.url, '/rowers/workout/'+encoded1+'/edit/') self.assertEqual(response.status_code, 302) @patch('rowers.integrations.strava.requests.get', side_effect=mocked_requests) @patch('rowers.integrations.strava.requests.post', side_effect=mocked_requests) def test_strava_list(self, mock_get, mockpost): integration = StravaIntegration(self.u) result = integration.token_refresh() self.assertEqual(result,"987654321234567898765432123456789") response = self.c.get('/rowers/workout/stravaimport/') self.assertEqual(response.status_code,200) @patch('rowers.utils.requests.get', side_effect=mocked_requests) @patch('rowers.integrations.strava.requests.post', side_effect=mocked_requests) @patch('rowers.dataprep.read_data') def test_strava_import(self, mock_get, mock_post, mocked_read_data): response = self.c.get('/rowers/workout/stravaimport/12',follow=True) expected_url = reverse('workout_import_view',kwargs={'source':'strava'}) self.assertRedirects(response, expected_url=expected_url, status_code=301,target_status_code=200) self.assertEqual(response.status_code, 200) @patch('rowers.integrations.integrations.requests.post', side_effect=mocked_requests) @patch('rowers.integrations.strava.requests.get', side_effect=mocked_requests) def test_strava_callback(self, mock_post, mock_get): response = self.c.get('/stravacall_back?code=absdef23&scope=read',follow=True) self.assertEqual(response.status_code, 200) #@pytest.mark.django_db @override_settings(TESTING=True) class STObjects(DjangoTestCase): def setUp(self): self.c = Client() self.u = User.objects.create_user('john', 'sander@ds.ds', 'koeinsloot') self.u.first_name = 'John' self.u.last_name = 'Sander' self.u.save() self.r = Rower.objects.create(user=self.u,gdproptin=True, ftpset=True,surveydone=True, gdproptindate=timezone.now() ) self.r.sporttrackstoken = '12' self.r.sporttroacksrefreshtoken = '12' self.r.sporttrackstokenexpirydate = arrow.get(datetime.datetime.now()+datetime.timedelta(days=1)).datetime self.r.save() self.c.login(username='john',password='koeinsloot') self.nu = datetime.datetime.now() filename = 'rowers/tests/testdata/testdata.csv' rr = rrower(hrmax=self.r.max,hrut2=self.r.ut2, hrut1=self.r.ut1,hrat=self.r.at, hrtr=self.r.tr,hran=self.r.an,ftp=self.r.ftp) row = rdata(csvfile=filename,rower=rr) totaldist = row.df['cum_dist'].max() totaltime = row.df['TimeStamp (sec)'].max()-row.df['TimeStamp (sec)'].min() totaltime = totaltime+row.df.loc[:,' ElapsedTime (sec)'].iloc[0] hours = int(totaltime/3600.) minutes = int((totaltime - 3600.*hours)/60.) seconds = int(totaltime - 3600.*hours - 60.*minutes) tenths = int(10*(totaltime - 3600.*hours - 60.*minutes - seconds)) duration = "%s:%s:%s.%s" % (hours,minutes,seconds,tenths) workoutdate = row.rowdatetime.strftime('%Y-%m-%d') workoutstarttime = row.rowdatetime.strftime('%H:%M:%S') self.w = Workout.objects.create( name='testworkout',workouttype='water', user=self.r,date=self.nu.strftime('%Y-%m-%d'), starttime=workoutstarttime, startdatetime=row.rowdatetime, duration=duration,distance=totaldist, csvfilename=filename ) self.u2 = User.objects.create_user('john2', 'ba@ds.ds', 'koeinsloot2') self.u2.first_name = 'John' self.u2.last_name = 'Sander2' self.u2.save() self.r2 = Rower.objects.create(user=self.u2,gdproptin=True, ftpset=True,surveydone=True, gdproptindate=timezone.now() ) self.w2 = Workout.objects.create( name='testworkout',workouttype='water', user=self.r2,date=self.nu.strftime('%Y-%m-%d'), starttime=workoutstarttime, startdatetime=row.rowdatetime, duration=duration,distance=totaldist, csvfilename=filename ) @patch('rowers.integrations.sporttracks.requests.post', side_effect=mocked_requests) def test_sporttracks_callback(self, mock_post): response = self.c.get('/sporttracks_callback?code=dsdoij232s',follow=True) self.assertEqual(response.status_code, 200) @patch('rowers.integrations.sporttracks.requests.post', side_effect=mocked_requests) def test_sporttracks_token_refresh(self, mock_post): response = self.c.get('/rowers/me/sporttracksrefresh/',follow=True) self.assertEqual(response.status_code, 200) @patch('rowers.integrations.sporttracks.requests.post', side_effect=mocked_requests) @patch('rowers.integrations.sporttracks.requests.get', side_effect=mocked_requests) def test_sporttracks_upload(self, mock_get, mock_post): response = self.c.get('/rowers/workout/'+encoded1+'/sporttracksuploadw/') self.assertRedirects(response, expected_url = '/rowers/workout/'+encoded1+'/edit/', status_code=302,target_status_code=200) self.assertEqual(response.url, '/rowers/workout/'+encoded1+'/edit/') self.assertEqual(response.status_code, 302) @patch('rowers.integrations.sporttracks.requests.post', side_effect=mocked_requests) @patch('rowers.integrations.sporttracks.requests.get', side_effect=mocked_requests) def test_tp_upload2(self, mock_get, mock_post): url = '/rowers/workout/'+encoded2+'/sporttracksuploadw/' response = self.c.get(url) self.assertRedirects(response, expected_url = '/rowers/list-workouts/', status_code=302,target_status_code=200) self.assertEqual(response.url, '/rowers/list-workouts/') self.assertEqual(response.status_code, 302) @patch('rowers.integrations.sporttracks.requests.get', side_effect=mocked_requests) def test_sporttracks_list(self, mock_get): response = self.c.get('/rowers/workout/sporttracksimport',follow=True) self.assertEqual(response.status_code,200) @patch('rowers.integrations.sporttracks.requests.get', side_effect=mocked_requests) def test_sporttracks_import(self, mock_get): response = self.c.get('/rowers/workout/sporttracksimport/12/',follow=True) self.assertRedirects(response, expected_url='/rowers/workout/'+encoded2+'/edit/', status_code=302,target_status_code=200) self.assertEqual(response.status_code, 200) @patch('rowers.integrations.sporttracks.requests.get', side_effect=mocked_requests) def test_sporttracks_import(self, mock_get): response = self.c.get('/rowers/workout/sporttracksimport/13/',follow=True) expected_url = '/rowers/workout/sporttracksimport/' self.assertRedirects(response, expected_url=expected_url, status_code=302,target_status_code=200) self.assertEqual(response.status_code, 200) @patch('rowers.integrations.sporttracks.requests.get', side_effect=mocked_requests) def test_sporttracks_import_all(self, mock_get): response = self.c.get('/rowers/workout/sporttracksimport/?selectallnew=true') self.assertEqual(response.status_code, 200) #@pytest.mark.django_db @override_settings(TESTING=True) class TPObjects(DjangoTestCase): def setUp(self): self.c = Client() self.u = User.objects.create_user('john', 'sander@ds.ds', 'koeinsloot') self.u.first_name = 'John' self.u.last_name = 'Sander' self.u.save() self.r = Rower.objects.create(user=self.u,gdproptin=True, ftpset=True,surveydone=True, gdproptindate=timezone.now() ) self.r.tptoken = '12' self.r.tprefreshtoken = '12' self.r.tptokenexpirydate = arrow.get(datetime.datetime.now()+datetime.timedelta(days=1)).datetime self.r.save() self.c.login(username='john',password='koeinsloot') self.nu = datetime.datetime.now() filename = 'rowers/tests/testdata/testdata.csv' rr = rrower(hrmax=self.r.max,hrut2=self.r.ut2, hrut1=self.r.ut1,hrat=self.r.at, hrtr=self.r.tr,hran=self.r.an,ftp=self.r.ftp) row = rdata(csvfile=filename,rower=rr) totaldist = row.df['cum_dist'].max() totaltime = row.df['TimeStamp (sec)'].max()-row.df['TimeStamp (sec)'].min() totaltime = totaltime+row.df.loc[:,' ElapsedTime (sec)'].iloc[0] hours = int(totaltime/3600.) minutes = int((totaltime - 3600.*hours)/60.) seconds = int(totaltime - 3600.*hours - 60.*minutes) tenths = int(10*(totaltime - 3600.*hours - 60.*minutes - seconds)) duration = "%s:%s:%s.%s" % (hours,minutes,seconds,tenths) workoutdate = row.rowdatetime.strftime('%Y-%m-%d') workoutstarttime = row.rowdatetime.strftime('%H:%M:%S') self.w = Workout.objects.create( name='testworkout',workouttype='water', user=self.r,date=self.nu.strftime('%Y-%m-%d'), starttime=workoutstarttime, startdatetime=row.rowdatetime, duration=duration,distance=totaldist, csvfilename=filename ) self.u2 = User.objects.create_user('john2', 'ba@ds.ds', 'koeinsloot2') self.u2.first_name = 'John' self.u2.last_name = 'Sander2' self.u2.save() self.r2 = Rower.objects.create(user=self.u2,gdproptin=True, ftpset=True,surveydone=True, gdproptindate=timezone.now() ) self.w2 = Workout.objects.create( name='testworkout',workouttype='water', user=self.r2,date=self.nu.strftime('%Y-%m-%d'), starttime=workoutstarttime, startdatetime=row.rowdatetime, duration=duration,distance=totaldist, csvfilename=filename ) @patch('rowers.integrations.trainingpeaks.requests.post', side_effect=mocked_requests) def test_tp_callback(self, mock_post): response = self.c.get('/tp_callback?code=dsdoij232s',follow=True) self.assertEqual(response.status_code, 200) @patch('rowers.integrations.trainingpeaks.requests.post', side_effect=mocked_requests) def test_tp_token_refresh(self, mock_post): response = self.c.get('/rowers/me/tprefresh/',follow=True) self.assertEqual(response.status_code, 200) @patch('rowers.integrations.trainingpeaks.requests.post', side_effect=mocked_requests) @patch('rowers.integrations.trainingpeaks.requests.get', side_effect=mocked_requests) def test_tp_upload(self, mock_get, mock_post): url = '/rowers/workout/'+encoded1+'/tpuploadw/' response = self.c.get(url) self.assertRedirects(response, expected_url = '/rowers/workout/'+encoded1+'/edit/', status_code=302,target_status_code=200) self.assertEqual(response.url, '/rowers/workout/'+encoded1+'/edit/') self.assertEqual(response.status_code, 302) @patch('rowers.integrations.trainingpeaks.requests.post', side_effect=mocked_requests) @patch('rowers.integrations.trainingpeaks.requests.get', side_effect=mocked_requests) def test_tp_upload2(self, mock_get, mock_post): url = '/rowers/workout/'+encoded2+'/tpuploadw/' response = self.c.get(url) self.assertRedirects(response, expected_url = '/rowers/list-workouts/', status_code=302,target_status_code=200) self.assertEqual(response.url, '/rowers/list-workouts/') self.assertEqual(response.status_code, 302)