sander/rowsandall
Private
Public Access
1
0
Fork 0
Code Activity

better tests

Browse Source
This commit is contained in:
Sander Roosendaal
2024-02-29 18:08:46 +01:00
parent 4fd98fbbab
commit cf31c1e750
5 changed files with 894 additions and 835 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
rowers/tasks.py
View File

@@ -94,7 +94,7 @@ SITE_URL_DEV = CFG['site_url']
PROGRESS_CACHE_SECRET = CFG['progress_cache_secret']
try:
SETTINGS_NAME = CFG['settings_name']
except KeyError:
except KeyError: # pragma: no cover
SETTINGS_NAME = 'rowsandall_ap.settings'
try:
@@ -321,7 +321,7 @@ def summaryfromsplitdata(splitdata, data, filename, sep='|', workouttype='rower'
return sums, sa, results
@app.task
def handle_post_workout_api(uploadoptions, debug=False, **kwargs):
def handle_post_workout_api(uploadoptions, debug=False, **kwargs): # pragma: no cover
session = requests.session()
newHeaders = {'Content-type': 'application/json', 'Accept': 'text/plain'}
session.headers.update(newHeaders)
@@ -335,14 +335,14 @@ def handle_post_workout_api(uploadoptions, debug=False, **kwargs):
@app.task
def handle_remove_workouts_team(ws, t, debug=False, **kwargs):
def handle_remove_workouts_team(ws, t, debug=False, **kwargs): # pragma: no cover
for w in ws:
w.team.remove(t)
return 1
@app.task
def handle_add_workouts_team(ws, t, debug=False, **kwargs):
def handle_add_workouts_team(ws, t, debug=False, **kwargs): # pragma: no cover
for w in ws:
w.team.add(t)
@@ -350,7 +350,7 @@ def handle_add_workouts_team(ws, t, debug=False, **kwargs):
return 1
def uploadactivity(access_token, filename, description='',
name='Rowsandall.com workout'):
name='Rowsandall.com workout'): # pragma: no cover
data_gz = BytesIO()
try:
@@ -393,7 +393,7 @@ def uploadactivity(access_token, filename, description='',
@app.task
def check_tp_workout_id(workout, location, attempts=5, debug=False, **kwargs):
def check_tp_workout_id(workout, location, attempts=5, debug=False, **kwargs): # pragma: no cover
authorizationstring = str('Bearer ' + workout.user.tptoken)
headers = {'Authorization': authorizationstring,
'user-agent': 'sanderroosendaal',
@@ -411,7 +411,7 @@ def check_tp_workout_id(workout, location, attempts=5, debug=False, **kwargs):
return 1
@app.task
def handle_workout_tp_upload(w, thetoken, tcxfilename, debug=False, **kwargs):
def handle_workout_tp_upload(w, thetoken, tcxfilename, debug=False, **kwargs): # pragma: no cover
tpid = 0
r = w.user
if not tcxfilename:
@@ -443,7 +443,7 @@ def handle_workout_tp_upload(w, thetoken, tcxfilename, debug=False, **kwargs):
return tpid
@app.task
def instroke_static(w, metric, debug=False, **kwargs):
def instroke_static(w, metric, debug=False, **kwargs): # pragma: no cover
f1 = w.csvfilename[6:-4]
rowdata = rdata(csvfile=w.csvfilename)
@@ -512,7 +512,7 @@ def handle_c2_sync(workoutid, url, headers, data, debug=False, **kwargs):
try:
workout = Workout.objects.get(id=workoutid)
except Workout.DoesNotExist:
except Workout.DoesNotExist: # pragma: no cover
dologging('c2_log.log','failed for c2id {c2id}'.format(c2id=c2id))
return 0
@@ -531,7 +531,7 @@ def handle_c2_sync(workoutid, url, headers, data, debug=False, **kwargs):
return 1
def splitstdata(lijst):
def splitstdata(lijst): # pragma: no cover
t = []
latlong = []
while len(lijst) >= 2:
@@ -543,7 +543,7 @@ def splitstdata(lijst):
@app.task
def handle_sporttracks_workout_from_data(user, importid, source,
workoutsource, debug=False, **kwargs):
workoutsource, debug=False, **kwargs): # pragma: no cover
r = user.rower
authorizationstring = str('Bearer ' + r.sporttrackstoken)
@@ -812,7 +812,7 @@ def handle_strava_sync(stravatoken,
if not failed:
try:
workout = Workout.objects.get(id=workoutid)
except Workout.DoesNotExist:
except Workout.DoesNotExist: # pragma: no cover
return 0
workout.uploadedtostrava = res.id
@@ -1562,7 +1562,7 @@ def handle_calctrimp(id,
try:
workout = Workout.objects.get(id=id)
except Workout.DoesNotExist:
except Workout.DoesNotExist: # pragma: no cover
dologging('metrics.log','Could not find workout {id}'.format(id=id))
return 0
@@ -2037,7 +2037,7 @@ def handle_sendemail_expired(useremail, userfirstname, userlastname, expireddate
return 1
@app.task
def handle_sendemail_newftp(rower,power,mode, **kwargs):
def handle_sendemail_newftp(rower,power,mode, **kwargs): # pragma: no cover
subject = "You may want to update your FTP on rowsandall.com"
from_email = 'Rowsandall <info@rowsandall.com>'
@@ -2069,7 +2069,7 @@ def handle_sendemail_breakthrough(workoutid, useremail,
lastname = ''
if surname:
if surname: # pragma: no cover
lastname = userlastname
tablevalues = [
@@ -2122,7 +2122,7 @@ def handle_sendemail_hard(workoutid, useremail,
]
lastname = ''
if surname:
if surname: # pragma: no cover
lastname = userlastname
# send email with attachment
@@ -3414,7 +3414,7 @@ def handle_nk_async_workout(alldata, userid, nktoken, nkid, delaysec, defaulttim
jsonData = response.json()
try:
strokeData = jsonData[str(nkid)]
except KeyError:
except KeyError: # pragma: no cover
dologging('nklog.log','Could not find strokeData')
return 0
@@ -3424,7 +3424,7 @@ def handle_nk_async_workout(alldata, userid, nktoken, nkid, delaysec, defaulttim
if oarlockSessions:
oarlocksession = oarlockSessions[0]
seatNumber = oarlocksession['seatNumber']
except KeyError:
except KeyError: # pragma: no cover
pass
df = strokeDataToDf(strokeData, seatIndex=seatNumber)
@@ -3439,14 +3439,14 @@ def handle_nk_async_workout(alldata, userid, nktoken, nkid, delaysec, defaulttim
workoutid, error = add_workout_from_data(userid, nkid, data, df)
# dologging('nklog.log','NK Workout ID {id}'.format(id=workoutid))
if workoutid == 0:
if workoutid == 0: # pragma: no cover
return 0
try:
workout = Workout.objects.get(id=workoutid)
newnkid = workout.uploadedtonk
sr = create_or_update_syncrecord(workout.user, workout, nkid=newnkid)
except Workout.DoesNotExist:
except Workout.DoesNotExist: # pragma: no cover
pass
return workoutid
@@ -3775,7 +3775,7 @@ def handle_c2_async_workout(alldata, userid, c2token, c2id, delaysec,
return workoutid
@app.task
def fetch_rojabo_session(id,alldata,userid,rowerid,debug=False, **kwargs):
def fetch_rojabo_session(id,alldata,userid,rowerid,debug=False, **kwargs): # pragma: no cover
try:
item = alldata[id]
except KeyError:
@@ -3880,7 +3880,7 @@ def fetch_strava_workout(stravatoken, oauth_data, stravaid, csvfilename, userid,
wsize = round(5./dt)
velo2 = ewmovingaverage(velo, wsize)
except ValueError:
except ValueError: # pragma: no cover
velo2 = velo
if coords is not None: