1361 lines
50 KiB
Python
1361 lines
50 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
|
|
from rowers import c2stuff
|
|
from rowers import stravastuff
|
|
from rowers import polarstuff
|
|
import urllib
|
|
import json
|
|
|
|
import rowers.utils as utils
|
|
|
|
|
|
from django.db import transaction
|
|
import rowers.garmin_stuff as gs
|
|
|
|
@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,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,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):
|
|
data = c2stuff.createc2workoutdata(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.c2stuff.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.c2stuff.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.c2stuff.requests.post', side_effect=mocked_requests)
|
|
@patch('rowers.c2stuff.requests.get', side_effect=mocked_requests)
|
|
@patch('rowers.c2stuff.requests.session', side_effect=mocked_requests)
|
|
def test_c2_auto_import(self, mock_get, mock_post,MockSession):
|
|
self.r.sporttracks_auto_export = True
|
|
self.r.save()
|
|
res = c2stuff.get_c2_workouts(self.r)
|
|
self.assertEqual(res,1)
|
|
|
|
res = c2stuff.get_c2_workouts(self.r,do_async=False)
|
|
self.assertEqual(res,1)
|
|
|
|
|
|
|
|
@patch('rowers.c2stuff.requests.post', side_effect=mocked_requests)
|
|
@patch('rowers.c2stuff.requests.get', side_effect=mocked_requests)
|
|
def test_c2_upload(self, mock_get, mock_post):
|
|
response = self.c.get('/rowers/workout/'+encoded1+'/c2uploadw/')
|
|
|
|
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.c2stuff.requests.post', side_effect=mocked_requests)
|
|
@patch('rowers.c2stuff.requests.get', side_effect=mocked_requests)
|
|
def test_c2_list(self, mock_get, mock_post):
|
|
response = self.c.get('/rowers/workout/c2list',follow=True)
|
|
|
|
self.assertEqual(response.status_code,200)
|
|
|
|
@patch('rowers.c2stuff.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/',follow=True)
|
|
|
|
expected_url = reverse('workout_c2import_view')
|
|
|
|
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.c2stuff.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/c2list/'
|
|
|
|
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.c2stuff.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/c2list/'
|
|
|
|
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.c2stuff.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/c2list/'
|
|
|
|
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,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.c2stuff.requests.post', side_effect=mocked_requests)
|
|
@patch('rowers.c2stuff.requests.get', side_effect=mocked_requests)
|
|
@patch('rowers.c2stuff.Session',side_effect=mocked_requests)
|
|
def test_c2_list(self, mock_get, mock_post, mock_Session):
|
|
response = self.c.get('/rowers/workout/c2list',follow=True)
|
|
|
|
self.assertEqual(response.status_code,200)
|
|
|
|
@patch('rowers.c2stuff.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,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.nkstuff.requests.get', side_effect=mocked_requests)
|
|
@patch('rowers.nkstuff.requests.post', side_effect=mocked_requests)
|
|
def test_nk_list(self, mock_get, mockpost):
|
|
result = rowers.nkstuff.rower_nk_token_refresh(self.u)
|
|
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.nkstuff.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.nkstuff.requests.get', side_effect=mocked_requests)
|
|
@patch('rowers.nkstuff.requests.post', side_effect=mocked_requests)
|
|
def test_nk_get_workouts(self, mock_get, mockpost):
|
|
result = rowers.nkstuff.nk_open(self.u)
|
|
self.assertEqual(result,"TA3n1vrNjuQJWw0TdCDHnjSmrjIPULhTlejMIWqq")
|
|
response = self.c.get('/rowers/workout/nkimport/all/',follow=True)
|
|
|
|
expected = reverse('workouts_view')
|
|
|
|
self.assertRedirects(response,
|
|
expected_url=expected,
|
|
status_code=302,target_status_code=200)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
@patch('rowers.nkstuff.requests.get', side_effect=mocked_requests)
|
|
@patch('rowers.nkstuff.requests.post', side_effect=mocked_requests)
|
|
@patch('rowers.nkimportutils.requests.session', side_effect=mocked_session)
|
|
@patch('rowers.dataprep.getsmallrowdata_db', side_effect=mocked_getsmallrowdata_db)
|
|
def test_nk_import(self, mock_get, mock_post,
|
|
mocked_session,
|
|
mocked_getsmallrowdata_db):
|
|
|
|
result = rowers.nkstuff.rower_nk_token_refresh(self.u)
|
|
response = self.c.get('/rowers/workout/nkimport/469',follow=True)
|
|
|
|
expected_url = reverse('workout_nkimport_view')
|
|
|
|
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)
|
|
#self.assertEqual(w.inboard,0.89)
|
|
#self.assertEqual(w.oarlength,2.87)
|
|
|
|
@patch('rowers.nkstuff.requests.get', side_effect=mocked_requests)
|
|
@patch('rowers.nkstuff.requests.post', side_effect=mocked_requests)
|
|
@patch('rowers.nkimportutils.requests.session', side_effect=mocked_session)
|
|
@patch('rowers.dataprep.getsmallrowdata_db', side_effect=mocked_getsmallrowdata_db)
|
|
def test_nk_import_impeller(self, mock_get, mock_post,
|
|
mocked_session,
|
|
mocked_getsmallrowdata_db):
|
|
|
|
result = rowers.nkstuff.rower_nk_token_refresh(self.u)
|
|
response = self.c.get('/rowers/workout/nkimport/404',follow=True)
|
|
|
|
expected_url = reverse('workout_nkimport_view')
|
|
|
|
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,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.polarstuff.requests.post', side_effect=mocked_requests)
|
|
@patch('rowers.polarstuff.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()
|
|
|
|
res = polarstuff.get_polar_workouts(self.r.user)
|
|
self.assertEqual(len(res),2)
|
|
|
|
@patch('rowers.polarstuff.requests.post', side_effect=mocked_requests)
|
|
@patch('rowers.polarstuff.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.polarstuff.requests.post', side_effect=mocked_requests)
|
|
@patch('rowers.polarstuff.requests.get', side_effect=mocked_requests)
|
|
def test_polar_notifications(self, mock_get, mock_post):
|
|
data = polarstuff.get_polar_notifications()
|
|
|
|
self.assertEqual(data[0]['user-id'],475)
|
|
|
|
response = polarstuff.get_all_new_workouts(data)
|
|
self.assertEqual(response,1)
|
|
|
|
@patch('rowers.polarstuff.requests.post', side_effect=mocked_requests)
|
|
@patch('rowers.polarstuff.requests.get', side_effect=mocked_requests)
|
|
def test_polar_get_workout(self, mock_get, mock_post):
|
|
transaction_id = 240522162
|
|
id = 1937529874
|
|
|
|
response = polarstuff.get_polar_workout(self.u, 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,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.rp3stuff.requests.get', side_effect=mocked_requests)
|
|
@patch('rowers.rp3stuff.requests.post', side_effect=mocked_requests)
|
|
@patch('rowers.dataprep.getsmallrowdata_db', side_effect=mocked_getsmallrowdata_db)
|
|
def test_rp3_import(self, mock_get, mockpost,
|
|
mocked_getsmallrowdata_db):
|
|
|
|
response = self.c.get('/rowers/workout/rp3import/591621',follow=True)
|
|
|
|
expected_url = reverse('workout_rp3import_view')
|
|
|
|
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.rp3stuff.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,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.stravastuff.requests.post', side_effect=mocked_requests)
|
|
@patch('rowers.stravastuff.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':stravastuff.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.stravastuff.requests.post', side_effect=mocked_requests)
|
|
@patch('rowers.stravastuff.requests.get', side_effect=mocked_requests)
|
|
@patch('rowers.stravastuff.stravalib.Client',side_effect=MockStravalibClient)
|
|
def test_workout_strava_upload(self, mock_get, mock_post,MockStravalibClient):
|
|
w = Workout.objects.get(id=1)
|
|
res = stravastuff.workout_strava_upload(self.r.user,w,asynchron=True)
|
|
self.assertEqual(res[1],-1)
|
|
res = stravastuff.workout_strava_upload(self.r.user,w,asynchron=False)
|
|
|
|
self.assertEqual(len(res[0]),43)
|
|
|
|
@patch('rowers.stravastuff.requests.post', side_effect=mocked_requests)
|
|
@patch('rowers.stravastuff.requests.get', side_effect=mocked_requests)
|
|
@patch('rowers.stravastuff.stravalib.Client',side_effect=MockStravalibClient)
|
|
def test_strava_upload(self, mock_get, mock_post,MockStravalibClient):
|
|
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.stravastuff.requests.get', side_effect=mocked_requests)
|
|
@patch('rowers.stravastuff.requests.post', side_effect=mocked_requests)
|
|
def test_strava_list(self, mock_get, mockpost):
|
|
result = rowers.stravastuff.rower_strava_token_refresh(self.u)
|
|
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.stravastuff.requests.post', side_effect=mocked_requests)
|
|
@patch('rowers.dataprep.getsmallrowdata_db')
|
|
def test_strava_import(self, mock_get, mock_post,
|
|
mocked_getsmallrowdata_db):
|
|
|
|
response = self.c.get('/rowers/workout/stravaimport/12',follow=True)
|
|
expected_url = reverse('workout_stravaimport_view')
|
|
|
|
self.assertRedirects(response,
|
|
expected_url=expected_url,
|
|
status_code=301,target_status_code=200)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
@patch('rowers.stravastuff.requests.post', side_effect=mocked_requests)
|
|
def test_strava_callback(self, mock_post):
|
|
response = self.c.get('/stravacall_back?code=absdef23&scope=read',follow=True)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
@patch('rowers.stravastuff.requests.post', side_effect=mocked_requests)
|
|
def test_strava_token_refresh(self, mock_post):
|
|
result = rowers.stravastuff.rower_strava_token_refresh(self.u)
|
|
self.assertEqual(result,"987654321234567898765432123456789")
|
|
|
|
|
|
|
|
#@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,surveydone=True,
|
|
gdproptindate=timezone.now()
|
|
)
|
|
|
|
|
|
self.r.sporttrackstoken = '12'
|
|
self.r.sporttracksrefreshtoken = '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.sporttracksstuff.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.sporttracksstuff.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.sporttracksstuff.requests.post', side_effect=mocked_requests)
|
|
@patch('rowers.sporttracksstuff.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.sporttracksstuff.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.imports.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.imports.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.imports.requests.get', side_effect=mocked_requests)
|
|
def test_sporttracks_import_all(self, mock_get):
|
|
|
|
response = self.c.get('/rowers/workout/sporttracksimport/all/',follow=True)
|
|
|
|
expected_url = reverse('workouts_view')
|
|
|
|
self.assertRedirects(response,
|
|
expected_url=expected_url,
|
|
status_code=302,target_status_code=200)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
@patch('rowers.dataprep.create_engine')
|
|
def test_strokedata(self, mocked_sqlalchemy):
|
|
with open('rowers/tests/testdata/sporttrackstestdata.txt','r') as infile:
|
|
data = json.load(infile)
|
|
|
|
from rowers.sporttracksstuff import add_workout_from_data
|
|
|
|
res = add_workout_from_data(self.u,1,data,data)
|
|
|
|
@patch('rowers.dataprep.create_engine')
|
|
def test_strokedatanohr(self, mocked_sqlalchemy):
|
|
with open('rowers/tests/testdata/sporttrackstestnohr.txt','r') as infile:
|
|
data = json.load(infile)
|
|
|
|
from rowers.sporttracksstuff import add_workout_from_data
|
|
|
|
|
|
|
|
res = add_workout_from_data(self.u,1,data,data)
|
|
|
|
|
|
|
|
#@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,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.imports.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.tpstuff.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.tpstuff.requests.post', side_effect=mocked_requests)
|
|
@patch('rowers.tpstuff.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)
|