1500 lines
57 KiB
Python
1500 lines
57 KiB
Python
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.garmin_stuff as gs
|
|
import rowers.integrations.strava as strava
|
|
from rowers.nkimportutils import *
|
|
|
|
@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 = 'plan'
|
|
self.r.save()
|
|
self.c.login(username='john',password='koeinsloot')
|
|
|
|
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):
|
|
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)
|
|
|
|
|
|
@pytest.mark.django_db
|
|
@override_settings(TESTING=True)
|
|
class GarminObjects(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.garmintoken = 'dfdzf'
|
|
self.r.garminrefreshtoken = 'fsls'
|
|
self.r.rowerplan = 'plan'
|
|
self.r.save()
|
|
self.c.login(username='john',password='koeinsloot')
|
|
|
|
self.nu = datetime.datetime.now()
|
|
|
|
startdate = nu.date()
|
|
enddate = (nu+datetime.timedelta(days=3)).date()
|
|
preferreddate = startdate
|
|
|
|
self.ps_trimp = SessionFactory(
|
|
startdate=startdate,enddate=enddate,
|
|
sessiontype='test',
|
|
sessionmode = 'TRIMP',
|
|
criterium = 'none',
|
|
sessionvalue = 77,
|
|
sessionunit='none',
|
|
preferreddate=preferreddate,
|
|
manager=self.u,
|
|
)
|
|
|
|
self.ps_trimp.interval_string = '10min+4x1000m@200W/20sec+2000m@24spm+10min'
|
|
self.ps_trimp.save()
|
|
|
|
|
|
def tearDown(self):
|
|
ws = Workout.objects.filter(user=self.r)
|
|
for w in ws:
|
|
w.delete()
|
|
|
|
def test_garmin_push_summaries(self):
|
|
|
|
with open('rowers/tests/testdata/garminsummarydata.txt','r') as f:
|
|
data = json.load(f)
|
|
response = self.c.post('/rowers/garmin/summaries/',json.dumps(data),
|
|
content_type="application/json")
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
#response = self.c.get('/rowers/workout/'+encoded1+'/', follow=True)
|
|
#self.assertEqual(response.status_code, 200)
|
|
|
|
ws = Workout.objects.filter(user=self.r)
|
|
self.assertEqual(ws.count(),3)
|
|
|
|
def test_garmin_push_details3(self):
|
|
with open('rowers/tests/testdata/garmindetail3.txt','r') as f:
|
|
data = json.load(f)
|
|
response = self.c.post('/rowers/garmin/activities/',json.dumps(data),
|
|
content_type='application/json')
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
ws = Workout.objects.filter(user=self.r)
|
|
self.assertEqual(ws.count(),1)
|
|
|
|
data,w = dataprep.getrowdata_db(id=ws[0].id)
|
|
|
|
self.assertEqual(len(data),515)
|
|
|
|
def test_garmin_push_details4(self):
|
|
with open('rowers/tests/testdata/garmindetail4.txt','r') as f:
|
|
data = json.load(f)
|
|
response = self.c.post('/rowers/garmin/activities/',json.dumps(data),
|
|
content_type='application/json')
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
ws = Workout.objects.filter(user=self.r)
|
|
self.assertEqual(ws.count(),1)
|
|
|
|
data,w = dataprep.getrowdata_db(id=ws[0].id)
|
|
|
|
self.assertEqual(len(data),18)
|
|
|
|
def test_garmin_push_details2(self):
|
|
with open('rowers/tests/testdata/garmindetail2.txt','r') as f:
|
|
data = json.load(f)
|
|
response = self.c.post('/rowers/garmin/activities/',json.dumps(data),
|
|
content_type='application/json')
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
ws = Workout.objects.filter(user=self.r)
|
|
self.assertEqual(ws.count(),3)
|
|
|
|
data,w = dataprep.getrowdata_db(id=ws[0].id)
|
|
self.assertEqual(len(data),451)
|
|
|
|
def test_garmin_push_details1(self):
|
|
with open('rowers/tests/testdata/garmindetail1.txt','r') as f:
|
|
data = json.load(f)
|
|
response = self.c.post('/rowers/garmin/activities/',json.dumps(data),
|
|
content_type='application/json')
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
response = self.c.get('/rowers/workout/'+encoded1+'/', follow=True)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
ws = Workout.objects.filter(user=self.r)
|
|
self.assertEqual(ws.count(),2)
|
|
|
|
data,w = dataprep.getrowdata_db(id=ws[0].id)
|
|
self.assertEqual(len(data),2)
|
|
|
|
def test_garmin_deregistration(self):
|
|
data = {"deregistrations":[{"userAccessToken":"dfdzf"}]}
|
|
response = self.c.post('/rowers/garmin/deregistration/',json.dumps(data),
|
|
content_type='application/json')
|
|
|
|
self.assertEqual(response.status_code,200)
|
|
|
|
@patch('rowers.tasks.OAuth1Session',side_effect=mocked_requests)
|
|
@patch('rowers.tasks.requests.session', side_effect=mocked_requests)
|
|
def test_handle_get_garmin_file(self, MockSession, MockOAuth1Session):
|
|
client_id = 'garmin'
|
|
client_secret = 'noot'
|
|
garmintoken = 'mies'
|
|
garminrefreshtoken = 'jet'
|
|
userid = self.r.user.id
|
|
url = 'fake_url'
|
|
filetype = 'fit'
|
|
|
|
res = tasks.handle_get_garmin_file(
|
|
client_id,client_secret,garmintoken,garminrefreshtoken,userid,url,filetype
|
|
)
|
|
|
|
self.assertEqual(res,1)
|
|
|
|
@patch('rowers.garmin_stuff.OAuth1Session')
|
|
def notest_garmin_callback(self,MockOAuth1Session):
|
|
with transaction.atomic():
|
|
response = self.c.get('/garmin_callback/?oauth_token=528ea5d9-1163-434d-b172-f428c5d9f522&oauth_verifier=LW33ZMBP8H')
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
@patch('rowers.garmin_stuff.requests.get',side_effect=mocked_requests)
|
|
def test_garmin_can_export_session(self,mock_get):
|
|
result = gs.garmin_can_export_session(self.u)
|
|
self.assertTrue(result)
|
|
|
|
def test_ps_to_garmin(self):
|
|
res = gs.ps_to_garmin(self.ps_trimp,self.r)
|
|
self.assertTrue(len(json.dumps(res))>500)
|
|
|
|
@patch('rowers.garmin_stuff.requests.get',side_effect=mocked_requests)
|
|
@patch('rowers.garmin_stuff.requests.post',side_effect=mocked_requests)
|
|
def test_garmin_session_create(self,mock_get,mock_post):
|
|
res = gs.garmin_session_create(self.ps_trimp,self.u)
|
|
self.assertEqual(res,1212)
|
|
|
|
@patch('rowers.garmin_stuff.requests.get',side_effect=mocked_requests)
|
|
@patch('rowers.garmin_stuff.requests.post',side_effect=mocked_requests)
|
|
def test_togarmin_view(self,mock_get,mock_post):
|
|
url = reverse('plannedsession_togarmin_view',kwargs={'id':self.ps_trimp.id})
|
|
response = self.c.get(url,follow=True)
|
|
|
|
self.assertEqual(response.status_code,200)
|
|
|
|
|
|
@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'
|
|
)
|
|
|
|
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_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
|
|
)
|
|
|
|
@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
|
|
)
|
|
|
|
@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
|
|
)
|
|
|
|
@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,
|
|
)
|
|
|
|
@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_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
|
|
)
|
|
|
|
@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.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
|
|
)
|
|
|
|
@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)
|