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

more coverage related

Browse Source
This commit is contained in:
Sander Roosendaal
2021-04-23 14:18:13 +02:00
parent 99dfb171b5
commit e19158926d
10 changed files with 383 additions and 297 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
rowers/dataprep.py
View File

@@ -92,7 +92,7 @@ import sqlalchemy as sa
import sys
import rowers.utils as utils
import rowers.datautils as datautils
from rowers.utils import lbstoN,myqueue,is_ranking_piece,wavg
from rowers.utils import lbstoN,myqueue,wavg
from timezonefinder import TimezoneFinder

128
rowers/dataprepnodjango.py
View File

@@ -38,24 +38,24 @@ from timezonefinder import TimezoneFinder
try:
user = DATABASES['default']['USER']
except KeyError:
except KeyError: # pragma: no cover
user = ''
try:
password = DATABASES['default']['PASSWORD']
except KeyError:
except KeyError: # pragma: no cover
password = ''
try:
database_name = DATABASES['default']['NAME']
except KeyError:
except KeyError: # pragma: no cover
database_name = ''
try:
host = DATABASES['default']['HOST']
except KeyError:
except KeyError: # pragma: no cover
host = ''
try:
port = DATABASES['default']['PORT']
except KeyError:
except KeyError: # pragma: no cover
port = ''
database_url = 'mysql://{user}:{password}@{host}:{port}/{database_name}'.format(
@@ -113,7 +113,7 @@ def strfdelta(tdelta):
try:
minutes,seconds = divmod(tdelta.seconds,60)
tenths = int(tdelta.microseconds/1e5)
except AttributeError:
except AttributeError: # pragma: no cover
minutes,seconds = divmod(tdelta.view(np.int64),60e9)
seconds,rest = divmod(seconds,1e9)
tenths = int(rest/1e8)
@@ -137,13 +137,13 @@ def nicepaceformat(values):
def timedeltaconv(x):
if not np.isnan(x):
dt = datetime.timedelta(seconds=x)
else:
else: # pragma: no cover
dt = datetime.timedelta(seconds=350.)
return dt
def rdata(file,rower=rrower()):
def rdata(file,rower=rrower()): # pragma: no cover
try:
res = rrdata(csvfile=file,rower=rower)
except IOError:
@@ -161,7 +161,7 @@ from rowers.metrics import dtypes
# Creates C2 stroke data
def create_c2_stroke_data_db(
distance,duration,workouttype,
workoutid,starttimeunix,csvfilename,debug=False):
workoutid,starttimeunix,csvfilename,debug=False): # pragma: no cover
nr_strokes = int(distance/10.)
@@ -247,7 +247,7 @@ def add_c2_stroke_data_db(strokedata,workoutid,starttimeunix,csvfilename,
nr_rows = len(unixtime)
try:
try: # pragma: no cover
latcoord = strokedata.loc[:,'lat']
loncoord = strokedata.loc[:,'lon']
except:
@@ -264,12 +264,12 @@ def add_c2_stroke_data_db(strokedata,workoutid,starttimeunix,csvfilename,
try:
spm = strokedata.loc[:,'spm']
except KeyError:
except KeyError: # pragma: no cover
spm = 0*dist2
try:
hr = strokedata.loc[:,'hr']
except KeyError:
except KeyError: # pragma: no cover
hr = 0*spm
pace = strokedata.loc[:,'p']/10.
@@ -278,7 +278,7 @@ def add_c2_stroke_data_db(strokedata,workoutid,starttimeunix,csvfilename,
velo = 500./pace
power = 2.8*velo**3
if workouttype == 'bike':
if workouttype == 'bike': # pragma: no cover
velo = 1000./pace
@@ -320,13 +320,13 @@ def add_c2_stroke_data_db(strokedata,workoutid,starttimeunix,csvfilename,
try:
data = dataprep(df,id=workoutid,bands=False,debug=debug)
except:
except: # pragma: no cover
return 0
return data
def handle_nonpainsled(f2,fileformat,summary=''):
def handle_nonpainsled(f2,fileformat,summary=''): # pragma: no cover
oarlength = 2.89
inboard = 0.88
# handle RowPro:
@@ -413,19 +413,19 @@ def delete_strokedata(id,debug=False):
dirname = 'media/strokedata_{id}.parquet.gz'.format(id=id)
try:
shutil.rmtree(dirname)
except FileNotFoundError:
except FileNotFoundError: # pragma: no cover
pass
def update_strokedata(id,df,debug=False):
delete_strokedata(id,debug=debug)
if debug:
if debug: # pragma: no cover # pragma: no cover
print("updating ",id)
rowdata = dataprep(df,id=id,bands=True,barchart=True,otwpower=True,
debug=debug)
return rowdata
def update_empower(id, inboard, oarlength, boattype, df, f1, debug=False):
def update_empower(id, inboard, oarlength, boattype, df, f1, debug=False): # pragma: no cover
corr_factor = 1.0
if 'x' in boattype:
@@ -452,11 +452,11 @@ def update_empower(id, inboard, oarlength, boattype, df, f1, debug=False):
if success:
delete_strokedata(id,debug=debug)
if debug:
if debug: # pragma: no cover
print("updated ",id)
print("correction ",corr_factor)
else:
if debug:
if debug: # pragma: no cover
print("not updated ",id)
@@ -469,7 +469,7 @@ def update_empower(id, inboard, oarlength, boattype, df, f1, debug=False):
return success
def testdata(time,distance,pace,spm):
def testdata(time,distance,pace,spm): # pragma: no cover
t1 = np.issubdtype(time,np.number)
t2 = np.issubdtype(distance,np.number)
t3 = np.issubdtype(pace,np.number)
@@ -486,7 +486,7 @@ def getsmallrowdata_db(columns,ids=[],debug=False):
df = pd.DataFrame()
if len(ids)>1:
if len(ids)>1: # pragma: no cover
for id, f in zip(ids,csvfilenames):
try:
df = pd.read_parquet(f,columns=columns,engine='pyarrow')
@@ -503,16 +503,16 @@ def getsmallrowdata_db(columns,ids=[],debug=False):
elif len(ids)==1:
try:
df = pd.read_parquet(csvfilenames[0],columns=columns,engine='pyarrow')
except (OSError,IndexError):
except (OSError,IndexError): # pragma: no cover
df = pd.DataFrame()
else:
else: # pragma: no cover
df = pd.DataFrame()
return df
def update_workout_field_sql(workoutid,fieldname,value,debug=False):
if debug:
if debug: # pragma: no cover # pragma: no cover
engine = create_engine(database_url_debug, echo=False)
else:
engine = create_engine(database_url, echo=False)
@@ -530,7 +530,7 @@ def update_workout_field_sql(workoutid,fieldname,value,debug=False):
return 1
def update_c2id_sql(id,c2id):
def update_c2id_sql(id,c2id): # pragma: no cover
engine = create_engine(database_url, echo=False)
table = 'rowers_workout'
@@ -548,7 +548,7 @@ def update_c2id_sql(id,c2id):
def read_cols_df_sql(ids,columns,debug=False):
def read_cols_df_sql(ids,columns,debug=False): # pragma: no cover
columns = list(columns)+['distance','spm']
columns = [x for x in columns if x != 'None']
columns = list(set(columns))
@@ -579,7 +579,7 @@ def read_cols_df_sql(ids,columns,debug=False):
return df
def read_df_sql(id,debug=False):
def read_df_sql(id,debug=False): # pragma: no cover
try:
f = 'media/strokedata_{id}.parquet.gz'.format(id=id)
df = pd.read_parquet(f)
@@ -590,8 +590,8 @@ def read_df_sql(id,debug=False):
return df
def getcpdata_sql(rower_id,table='cpdata',debug=False):
if debug:
def getcpdata_sql(rower_id,table='cpdata',debug=False): # pragma: no cover
if debug: # pragma: no cover
engine = create_engine(database_url_debug, echo=False)
else:
engine = create_engine(database_url, echo=False)
@@ -605,8 +605,8 @@ def getcpdata_sql(rower_id,table='cpdata',debug=False):
return df
def deletecpdata_sql(rower_id,table='cpdata',debug=False):
if debug:
def deletecpdata_sql(rower_id,table='cpdata',debug=False): # pragma: no cover
if debug: # pragma: no cover
engine = create_engine(database_url_debug, echo=False)
else:
engine = create_engine(database_url, echo=False)
@@ -618,15 +618,15 @@ def deletecpdata_sql(rower_id,table='cpdata',debug=False):
with engine.connect() as conn, conn.begin():
try:
result = conn.execute(query)
except:
except: # pragma: no cover
print("Database locked")
conn.close()
engine.dispose()
def delete_agegroup_db(age,sex,weightcategory,debug=False):
if debug:
if debug: # pragma: no cover
engine = create_engine(database_url_debug, echo=False)
else:
else: # pragma: no cover
engine = create_engine(database_url, echo=False)
query = sa.text('DELETE from {table} WHERE age={age} and weightcategory = {weightcategory} and sex={sex};'.format(
@@ -638,7 +638,7 @@ def delete_agegroup_db(age,sex,weightcategory,debug=False):
with engine.connect() as conn, conn.begin():
try:
result = conn.execute(query)
except:
except: # pragma: no cover
print("Database locked")
conn.close()
engine.dispose()
@@ -664,7 +664,7 @@ def update_agegroup_db(age,sex,weightcategory,wcdurations,wcpower,
df.replace([np.inf,-np.inf],np.nan,inplace=True)
df.dropna(axis=0,inplace=True)
if debug:
if debug: # pragma: no cover # pragma: no cover
engine = create_engine(database_url_debug, echo=False)
else:
engine = create_engine(database_url, echo=False)
@@ -690,7 +690,7 @@ def updatecpdata_sql(rower_id,delta,cp,table='cpdata',distance=pd.Series([]),deb
if not distance.empty:
df['distance'] = distance
if debug:
if debug: # pragma: no cover
engine = create_engine(database_url_debug, echo=False)
else:
engine = create_engine(database_url, echo=False)
@@ -706,7 +706,7 @@ def updatecpdata_sql(rower_id,delta,cp,table='cpdata',distance=pd.Series([]),deb
def smalldataprep(therows,xparam,yparam1,yparam2):
def smalldataprep(therows,xparam,yparam1,yparam2): # pragma: no cover
df = pd.DataFrame()
if yparam2 == 'None':
yparam2 = 'power'
@@ -749,8 +749,8 @@ def smalldataprep(therows,xparam,yparam1,yparam2):
def dataprep(rowdatadf,id=0,bands=True,barchart=True,otwpower=True,
empower=True,debug=False,inboard=0.88,forceunit='lbs'):
if rowdatadf.empty:
if debug:
if rowdatadf.empty: # pragma: no cover
if debug: # pragma: no cover
print("empty")
return 0
@@ -776,7 +776,7 @@ def dataprep(rowdatadf,id=0,bands=True,barchart=True,otwpower=True,
drivelength = rowdatadf.loc[:,' DriveLength (meters)']
try:
workoutstate = rowdatadf.loc[:,' WorkoutState']
except KeyError:
except KeyError: # pragma: no cover
workoutstate = 0*hr
peakforce = rowdatadf.loc[:,' PeakDriveForce (lbs)']
@@ -789,18 +789,18 @@ def dataprep(rowdatadf,id=0,bands=True,barchart=True,otwpower=True,
recoverytime = rowdatadf.loc[:,' StrokeRecoveryTime (ms)']
rhythm = 100.*drivetime/(recoverytime+drivetime)
rhythm = rhythm.fillna(value=0)
except:
except: # pragma: no cover
rhythm = 0.0*forceratio
f = rowdatadf['TimeStamp (sec)'].diff().mean()
if f != 0:
try:
windowsize = 2*(int(10./(f)))+1
except ValueError:
except ValueError: # pragma: no cover
windowsize = 1
else:
else: # pragma: no cover
windowsize = 1
if windowsize <= 3:
if windowsize <= 3: # pragma: no cover
windowsize = 5
if windowsize > 3 and windowsize<len(hr):
@@ -811,7 +811,7 @@ def dataprep(rowdatadf,id=0,bands=True,barchart=True,otwpower=True,
try:
t2 = t.fillna(method='ffill').apply(lambda x: timedeltaconv(x))
except TypeError:
except TypeError: # pragma: no cover
t2 = 0*t
@@ -819,19 +819,19 @@ def dataprep(rowdatadf,id=0,bands=True,barchart=True,otwpower=True,
try:
drivespeed = drivelength/rowdatadf[' DriveTime (ms)']*1.0e3
except KeyError:
except KeyError: # pragma: no cover
drivespeed = 0.0*rowdatadf['TimeStamp (sec)']
except TypeError:
except TypeError: # pragma: no cover
drivespeed = 0.0*rowdatadf['TimeStamp (sec)']
drivespeed = drivespeed.fillna(value=0)
try:
driveenergy = rowdatadf['driveenergy']
except KeyError:
except KeyError: # pragma: no cover
if forceunit == 'lbs':
driveenergy = drivelength*averageforce*lbstoN
else:
else: # pragma: no cover
drivenergy = drivelength*averageforce
distance = rowdatadf.loc[:,'cum_dist']
@@ -882,7 +882,7 @@ def dataprep(rowdatadf,id=0,bands=True,barchart=True,otwpower=True,
try:
tel = rowdatadf.loc[:,' ElapsedTime (sec)']
except KeyError:
except KeyError: # pragma: no cover
rowdatadf[' ElapsedTime (sec)'] = rowdatadf['TimeStamp (sec)']
@@ -918,7 +918,7 @@ def dataprep(rowdatadf,id=0,bands=True,barchart=True,otwpower=True,
arclength = (inboard-0.05)*(np.radians(finish)-np.radians(catch))
if arclength.mean()>0:
if arclength.mean()>0: # pragma: no cover
drivelength = arclength
elif drivelength.mean() == 0:
drivelength = driveenergy/(averageforce*4.44822)
@@ -931,7 +931,7 @@ def dataprep(rowdatadf,id=0,bands=True,barchart=True,otwpower=True,
try:
totalangle = finish-catch
effectiveangle = finish-wash-catch-slip
except ValueError:
except ValueError: # pragma: no cover
totalangle = 0*t
effectiveangle = 0*t
@@ -939,39 +939,39 @@ def dataprep(rowdatadf,id=0,bands=True,barchart=True,otwpower=True,
if windowsize > 3 and windowsize<len(slip):
try:
wash = savgol_filter(wash,windowsize,3)
except TypeError:
except TypeError: # pragma: no cover
pass
try:
slip = savgol_filter(slip,windowsize,3)
except TypeError:
except TypeError: # pragma: no cover
pass
try:
catch = savgol_filter(catch,windowsize,3)
except TypeError:
except TypeError: # pragma: no cover
pass
try:
finish = savgol_filter(finish,windowsize,3)
except TypeError:
except TypeError: # pragma: no cover
pass
try:
peakforceangle = savgol_filter(peakforceangle,windowsize,3)
except TypeError:
except TypeError: # pragma: no cover
pass
try:
driveenergy = savgol_filter(driveenergy,windowsize,3)
except TypeError:
except TypeError: # pragma: no cover
pass
try:
drivelength = savgol_filter(drivelength,windowsize,3)
except TypeError:
except TypeError: # pragma: no cover
pass
try:
totalangle = savgol_filter(totalangle,windowsize,3)
except TypeError:
except TypeError: # pragma: no cover
pass
try:
effectiveangle = savgol_filter(effectiveangle,windowsize,3)
except TypeError:
except TypeError: # pragma: no cover
pass
velo = 500./p
@@ -993,7 +993,7 @@ def dataprep(rowdatadf,id=0,bands=True,barchart=True,otwpower=True,
data['totalangle'] = totalangle
data['effectiveangle'] = effectiveangle
data['efficiency'] = efficiency
except ValueError:
except ValueError: # pragma: no cover
pass
if otwpower:

40
rowers/forms.py
View File

@@ -51,10 +51,10 @@ class FlexibleDecimalField(forms.DecimalField):
pass
try:
dot_index = value.index('.')
except ValueError:
except ValueError: # pragma: no cover
pass
if value:
if comma_index > dot_index:
if comma_index > dot_index: # pragma: no cover
value = value.replace('.', '').replace(',', '.')
return super(FlexibleDecimalField, self).to_python(value)
@@ -362,7 +362,7 @@ class WorkFlowLeftPanelForm(forms.Form):
js = ['/admin/jsi18n/']
def __init__(self, *args, **kwargs):
def __init__(self, *args, **kwargs): # pragma: no cover
if 'instance' in kwargs:
r = kwargs.pop('instance')
panels = r.workflowleftpanel
@@ -395,7 +395,7 @@ class WorkFlowMiddlePanelForm(forms.Form):
}
js = ['/admin/jsi18n/']
def __init__(self, *args, **kwargs):
def __init__(self, *args, **kwargs): # pragma: no cover
if 'instance' in kwargs:
r = kwargs.pop('instance')
panels = r.workflowmiddlepanel
@@ -500,18 +500,18 @@ class UploadOptionsForm(forms.Form):
choices3 = [(0,'---')]
noregistrations = []
for ra in VirtualRace.objects.filter(registration_closure__gt=timezone.now(),sessiontype='race'):
for ra in VirtualRace.objects.filter(registration_closure__gt=timezone.now(),sessiontype='race'): # pragma: no cover
rs = VirtualRaceResult.objects.filter(race = ra,userid=r.id)
if rs.count()==0:
noregistrations.append((-ra.id,ra.name))
for ra in VirtualRace.objects.filter(registration_closure__gt=timezone.now(),sessiontype='indoorrace'):
for ra in VirtualRace.objects.filter(registration_closure__gt=timezone.now(),sessiontype='indoorrace'): # pragma: no cover
rs = IndoorVirtualRaceResult.objects.filter(race = ra,userid=r.id)
if rs.count()==0:
noregistrations.append((-ra.id,ra.name))
choices = choices3+choices1+choices2+noregistrations
if int(raceid) in [r.id for r in races]:
if int(raceid) in [r.id for r in races]: # pragma: no cover
therace = VirtualRace.objects.get(id=raceid)
self.fields['raceid'].initial = therace.id
if therace.sessiontype == 'race':
@@ -526,7 +526,7 @@ class UploadOptionsForm(forms.Form):
choices = [(r.id,str(r)) for r in registrations]
choices = choices+[(0,'---')]
if races:
if races: # pragma: no cover
self.fields['submitrace'].choices = choices
else:
del self.fields['submitrace']
@@ -920,7 +920,7 @@ class RegistrationFormUniqueEmail(RegistrationFormTermsOfService):
Validate that the supplied email address is unique for the
site.
"""
if User.objects.filter(email__iexact=self.cleaned_data['email']):
if User.objects.filter(email__iexact=self.cleaned_data['email']): # pragma: no cover
raise forms.ValidationError("This email address is already in use. Please supply a different email address.")
return self.cleaned_data['email']
@@ -949,7 +949,7 @@ class RegistrationFormSex(RegistrationFormUniqueEmail):
def clean_birthdate(self):
dob = self.cleaned_data['birthdate']
age = (timezone.now() - dob).days/365
if age < 16:
if age < 16: # pragma: no cover
raise forms.ValidationError('Must be at least 16 years old to register')
return self.cleaned_data['birthdate']
@@ -970,7 +970,7 @@ class RegistrationFormSex(RegistrationFormUniqueEmail):
# Time field supporting microseconds. Not used, I believe.
class MyTimeField(forms.TimeField):
def __init__(self, *args, **kwargs):
def __init__(self, *args, **kwargs): # pragma: no cover
super(MyTimeField, self).__init__(*args, **kwargs)
supports_microseconds = True
@@ -1085,7 +1085,7 @@ class StatsOptionsForm(forms.Form):
def __init__(self, *args, **kwargs):
def __init__(self, *args, **kwargs): # pragma: no cover
super(StatsOptionsForm, self).__init__(*args,**kwargs)
for type in mytypes.checktypes:
@@ -1128,7 +1128,7 @@ class PlanSelectForm(forms.Form):
class CourseSelectForm(forms.Form):
course = forms.ModelChoiceField(queryset=GeoCourse.objects.filter())
def __init__(self, *args, **kwargs):
def __init__(self, *args, **kwargs): # pragma: no cover
course = kwargs.pop('course',None)
manager = kwargs.pop('manager',None)
super(CourseSelectForm,self).__init__(*args,**kwargs)
@@ -1380,7 +1380,7 @@ class FusionMetricChoiceForm(ModelForm):
if df.loc[:,label].std() == 0:
try:
formaxlabels2.pop(label)
except KeyError:
except KeyError: # pragma: no cover
pass
metricchoices = list(sorted(formaxlabels2.items(), key = lambda x:x[1]))
@@ -1511,7 +1511,7 @@ class RaceResultFilterForm(forms.Form):
if len(theboatclasses)<= 1:
del self.fields['boatclass']
else:
else: # pragma: no cover
boatclasschoices = []
for choice in self.fields['boatclass'].choices:
if choice[0] in theboatclasses:
@@ -1522,7 +1522,7 @@ class RaceResultFilterForm(forms.Form):
try:
theboattypees = [record.boattype for record in records]
theboattypees = list(set(theboattypees))
except AttributeError:
except AttributeError: # pragma: no cover
theboattypees = []
if len(theboattypees)<= 1:
@@ -1540,7 +1540,7 @@ class RaceResultFilterForm(forms.Form):
if len(theweightcategoryes)<= 1:
del self.fields['weightcategory']
else:
else: # pragma: no cover
weightcategorychoices = []
for choice in self.fields['weightcategory'].choices:
if choice[0] in theweightcategoryes:
@@ -1553,7 +1553,7 @@ class RaceResultFilterForm(forms.Form):
if len(theadaptivecategoryes)<= 1:
del self.fields['adaptivecategory']
else:
else: # pragma: no cover
adaptivecategorychoices = []
for choice in self.fields['adaptivecategory'].choices:
if choice[0] in theadaptivecategoryes:
@@ -1622,7 +1622,7 @@ class PlannedSessionTeamForm(forms.Form):
self.fields['team'].queryset = Team.objects.filter(manager=user)
def clean(self):
if any(self.errors):
if any(self.errors): # pragma: no cover
return
cd = self.cleaned_data
@@ -1651,7 +1651,7 @@ def get_countries():
countries = VirtualRace.objects.order_by('country').values_list('country').distinct()
countries = tuple([(c[0],c[0]) for c in countries])
countries = countries+(('All','All'),)
except:
except: # pragma: no cover
countries = (('All','All'))
return countries

35
rowers/garmin_stuff.py
View File

@@ -179,6 +179,11 @@ def step_to_garmin(step,order=0):
durationtype = step['dict']['durationType']
durationvalue = step['dict']['durationValue']
durationvaluetype = ''
try:
intensity = step['dict']['intensity']
except KeyError:
intensity = None
#durationvaluetype = ''
if durationtype == 'Time':
durationtype = 'TIME'
durationvalue = int(durationvalue/1000.)
@@ -217,20 +222,40 @@ def step_to_garmin(step,order=0):
elif durationtype == 'Reps':
durationtype = 'REPS'
try:
targetType = step['dict']['targetType']
except KeyError:
targetType = None
try:
targetValue = step['dict']['targetValue']
except KeyError:
targetValue = None
try:
targetValueLow = step['dict']['targetValueLow']
except KeyError:
targetValueLow = None
try:
targetValueHigh = step['dict']['targetValueHigh'],
except KeyError:
targetValueHigh = None
out = {
'type': step['type'],
'stepOrder':order,
'repeatType':step['type'],
'repeatValue':step['repeatValue'],
'intensity':step['dict']['intensity'],
'intensity':intensity,
'description':step['dict']['wkt_step_name'],
'durationType':durationtype,
'durationValue':durationvalue,
'durationValueType':durationvaluetype,
'targetType':step['dict']['targetType'],
'targetValue':step['dict']['targetValue'],
'targetValueLow':step['dict']['targetValueLow'],
'targetValueHigh':step['dict']['targetValueHigh'],
'targetType':targetType,
'targetValue':targetValue,
'targetValueLow':targetValueLow,
'targetValueHigh':targetValueHigh,
}
try:
steps = step['steps']

2
rowers/models.py
View File

@@ -117,8 +117,6 @@ class UserFullnameChoiceField(forms.ModelChoiceField):
def label_from_instance(self,obj):
return obj.get_full_name()
class PlannedSessionStepField(models.TextField):
pass
def get_file_path(instance, filename):
ext = filename.split('.')[-1]

122
rowers/templatetags/rowerfilters.py
View File

@@ -47,11 +47,11 @@ from six import string_types
@register.filter
def isfollower(user,id):
if user.is_anonymous:
if user.is_anonymous: # pragma: no cover
return True
try:
race = VirtualRace.objects.get(id=id)
except VirtualRace.DoesNotExist:
except VirtualRace.DoesNotExist: # pragma: no cover
return False
followers = VirtualRaceFollower.objects.filter(race=race,user=user)
@@ -83,14 +83,14 @@ def steptostring(steps):
def verbose(s):
try:
return favanalysisdict[s]
except KeyError:
except KeyError: # pragma: no cover
return ''
@register.filter
def icon(s):
try:
return favanalysisicons[s]
except KeyError:
except KeyError: # pragma: no cover
return 'fa-chart-line'
@register.filter
@@ -109,7 +109,7 @@ def adaptive(s):
return u
@register.filter
def nkviewerlink(workout):
def nkviewerlink(workout): # pragma: no cover
url = "{nkviewer}{nkid}".format(
nkid=workout.uploadedtonk,
nkviewer=NK_VIEWER_LOCATION)
@@ -168,23 +168,23 @@ def weight(s):
def sigdig(value, digits = 3):
try:
order = int(math.floor(math.log10(math.fabs(value))))
except (ValueError,TypeError):
except (ValueError,TypeError): # pragma: no cover
return value
# return integers as is
if value % 1 == 0:
return value
places = digits - order - 1
places = digits - order - 1 # pragma: no cover
if places > 0:
if places > 0: # pragma: no cover
fmtstr = "%%.%df" % (places)
else:
else: # pragma: no cover
fmtstr = "%.0f"
return fmtstr % (round(value, places))
return fmtstr % (round(value, places)) # pragma: no cover
@register.filter
def pickle(dc):
def pickle(dc): # pragma: no cover
s = dict()
for key, value in dc.items():
s[key] = value
@@ -223,25 +223,25 @@ def strfdelta(tdelta):
from rowers.teams import rower_get_managers
@register.filter
def alertstatspercentage(list,i):
def alertstatspercentage(list,i): # pragma: no cover
alertstats = list[i-1]
return alertstats["percentage"]
@register.filter
def alertstartdate(list,i):
def alertstartdate(list,i): # pragma: no cover
alertstats = list[i-1]
return alertstats["startdate"]
@register.filter
def alertnperiod(list,i):
def alertnperiod(list,i): # pragma: no cover
alertstats = list[i-1]
return alertstats["nperiod"]
@register.filter
def alertenddate(list,i):
def alertenddate(list,i): # pragma: no cover
alertstats = list[i-1]
return alertstats["enddate"]
@@ -255,27 +255,27 @@ def is_coach(rower,rowers):
return True
@register.filter
def waterpower(x,rower):
def waterpower(x,rower): # pragma: no cover
if rower is not None:
return int(x*(100-rower.otwslack)/100.)
return int(x)
@register.filter
def round20(x):
def round20(x): # pragma: no cover
try:
return int(20.*(1+int(int(x)/20)))
except ValueError:
return 20
@register.filter
def round100(x):
def round100(x): # pragma: no cover
try:
return int(100.*(1+int(int(x)/100)))
except ValueError:
return 100
@register.filter
def majorticks(maxval):
def majorticks(maxval): # pragma: no cover
ticks = range(1+int(maxval/100.))
newticks =[]
for t in ticks:
@@ -284,7 +284,7 @@ def majorticks(maxval):
return newticks
@register.filter
def hrmajorticks(maxval,minval):
def hrmajorticks(maxval,minval): # pragma: no cover
ticks = range(int((maxval-minval)/20.)-1)
newticks =[]
for t in ticks:
@@ -319,7 +319,7 @@ def secondstotimestring(tdelta):
@register.filter
def existing_customer(user):
if user.is_anonymous:
if user.is_anonymous: # pragma: no cover
return False
else:
return payments.is_existing_customer(user.rower)
@@ -328,7 +328,7 @@ def existing_customer(user):
def aantalcomments(workout):
try:
comments = WorkoutComment.objects.filter(workout=workout)
except:
except: # pragma: no cover
return 0
aantalcomments = len(comments)
@@ -347,7 +347,7 @@ def encode(id):
def water(workout):
try:
return workout.workouttype in otwtypes
except AttributeError:
except AttributeError: # pragma: no cover
return False
@register.filter
@@ -360,7 +360,7 @@ def spacetohtml(t):
@register.filter
def durationprint(d,dstring):
if (d == None):
if (d == None): # pragma: no cover
return d
else:
try:
@@ -421,13 +421,13 @@ def previousperiodstart(timeperiod):
@register.filter
def paceprint(d):
if (d == None):
if (d == None): # pragma: no cover
return d
else:
return strfdelta(d)
@register.filter
def deltatimeprint(d):
def deltatimeprint(d): # pragma: no cover
if (d == None):
return d
else:
@@ -437,7 +437,7 @@ def deltatimeprint(d):
def c2userid(user):
try:
thetoken = c2_open(user)
except NoTokenError:
except NoTokenError: # pragma: no cover
return 0
c2userid = c2stuff.get_userid(thetoken)
@@ -448,7 +448,7 @@ def c2userid(user):
def currency(word):
try:
amount = float(word)
except ValueError:
except ValueError: # pragma: no cover
return word
return '{amount:.2f}'.format(amount=amount)
@@ -465,11 +465,11 @@ def rkuserid(user):
return rkuserid
@register.filter
def courselength(course):
def courselength(course): # pragma: no cover
return course_length(course)
@register.filter(is_safe=True)
def jsdict(dict,key):
def jsdict(dict,key): # pragma: no cover
s = dict.get(key)
return mark_safe(json.dumps(s))
@@ -479,7 +479,7 @@ def jsdict(dict,key):
def lookup(dict, key):
try:
s = dict.get(key)
except KeyError:
except KeyError: # pragma: no cover
return None
if isinstance(s,string_types) and len(s) > 22:
@@ -490,7 +490,7 @@ def lookup(dict, key):
def lookuplong(dict, key):
try:
s = dict.get(key)
except KeyError:
except KeyError: # pragma: no cover
return None
return s
@@ -516,7 +516,7 @@ from rowers.models import PlannedSession
def is_session_manager(id,user):
try:
ps = PlannedSession.objects.get(id=id)
except PlannedSession.DoesNotExist:
except PlannedSession.DoesNotExist: # pragma: no cover
return False
return ps.manager == user
@@ -537,7 +537,7 @@ def may_edit(workout,request):
@register.filter
def mayeditplan(obj,request):
if obj is None:
if obj is None: # pragma: no cover
return False
if hasattr(obj,'plan'):
@@ -555,7 +555,7 @@ def mayeditplan(obj,request):
return mayedit
@register.filter
def iterrows(df):
def iterrows(df): # pragma: no cover
return df.iterrows()
@register.filter(name='times')
@@ -563,7 +563,7 @@ def times(number):
return range(number)
@register.simple_tag
def get_df_iloc(data,i,j):
def get_df_iloc(data,i,j): # pragma: no cover
return data.iloc(i,j)
@register.simple_tag
@@ -589,7 +589,7 @@ def is_planmember(user):
return isplanmember(user)
@register.filter
def get_age(r):
def get_age(r): # pragma: no cover
return calculate_age(r.birthdate)
@@ -612,7 +612,7 @@ def user_team1(user):
teams1 = therower.team.all()
teams2 = Team.objects.filter(manager=user)
teams = list(set(teams1).union(set(teams2)))
except TypeError:
except TypeError: # pragma: no cover
teams = []
return teams[0].id
@@ -628,7 +628,7 @@ def announcements(request):
return announcements[0:4]
@register.filter
def has_teams(user):
def has_teams(user): # pragma: no cover
try:
therower = Rower.objects.get(user=user)
teams1 = therower.team.all()
@@ -644,7 +644,7 @@ def has_teams(user):
def team_members(user):
try:
therower = Rower.objects.get(user=user)
if therower.rowerplan == 'basic':
if therower.rowerplan == 'basic': # pragma: no cover
return []
teams = Team.objects.filter(manager=user)
members = Rower.objects.filter(
@@ -653,10 +653,10 @@ def team_members(user):
"user__last_name","user__first_name"
).exclude(rowerplan='freecoach')
return [rower.user for rower in members]
except TypeError:
except TypeError: # pragma: no cover
return []
return []
return [] # pragma: no cover
@register.filter
def openactions(user):
@@ -675,7 +675,7 @@ def openactions(user):
@register.filter
def team_rowers(user):
def team_rowers(user): # pragma: no cover
try:
therower = Rower.objects.get(user=user)
if therower.rowerplan == 'basic':
@@ -696,13 +696,13 @@ from rowers.teams import coach_getcoachees
def coach_rowers(user):
if user.rower.rowerplan != 'freecoach':
thelist = [user.rower]+[c for c in coach_getcoachees(user.rower)]
else:
else: # pragma: no cover
thelist = [c for c in coach_getcoachees(user.rower)]
return thelist
@register.filter
def verbosetimeperiod(timeperiod):
def verbosetimeperiod(timeperiod): # pragma: no cover
table = {
'today':'Today',
'thisweek': 'This Week',
@@ -723,7 +723,7 @@ def verbosetimeperiod(timeperiod):
from datetime import date
@ register.filter
def future_date_only(the_date):
def future_date_only(the_date): # pragma: no cover
if the_date > date.today():
return the_date
else:
@@ -747,7 +747,7 @@ def date_dif(the_date):
return 1
if the_date:
return the_date - date.today()
else:
else: # pragma: no cover
return 1
@@ -756,38 +756,38 @@ def can_register(race,r):
return race_can_register(r,race)
@register.filter
def can_submit(race,r):
def can_submit(race,r): # pragma: no cover
return race_can_submit(r,race)
@register.filter
def race_complete(race,r):
def race_complete(race,r): # pragma: no cover
is_complete,has_registered = race_rower_status(r,race)
return is_complete
@register.filter
def past_not_registered(race,r):
def past_not_registered(race,r): # pragma: no cover
is_complete,has_registered = race_rower_status(r,race)
return not has_registered
@register.filter
def future_registered(race,r):
def future_registered(race,r): # pragma: no cover
is_complete, has_registered = race_rower_status(r,race)
is_open = race.evaluation_closure > timezone.now()
return has_registered and not is_complete and is_open
@property
def is_past_due(self):
def is_past_due(self): # pragma: no cover
return datetime.date.today() > self.date
@property
def is_not_past_due(self):
def is_not_past_due(self): # pragma: no cover
return datetime.date.today() <= self.date
@register.filter
def is_closed(race):
def is_closed(race): # pragma: no cover
return race.evaluation_closure < timezone.now()
@register.filter
def is_final(race):
def is_final(race): # pragma: no cover
return race.evaluation_closure < timezone.now()-datetime.timedelta(hours=1)
@register.filter
@@ -827,7 +827,7 @@ def teamurl(path,team):
return replaced
@register.filter
def timeurl(path,timestring):
def timeurl(path,timestring): # pragma: no cover
pattern = re.compile('\?when=w.*')
timeurl = '?when=%s' % timestring
replaced = ''
@@ -852,7 +852,7 @@ def trainingplans(rower):
return plans
@register.filter
def mesomacroid(id):
def mesomacroid(id): # pragma: no cover
try:
thismeso = TrainingMesoCycle.objects.get(id=id)
except TrainingMesoCycle.DoesNotExist:
@@ -863,7 +863,7 @@ def mesomacroid(id):
return str(theid)
@register.filter
def micromesoid(id):
def micromesoid(id): # pragma: no cover
try:
thismicro = TrainingMicroCycle.objects.get(id=id)
except TrainingMicroCycle.DoesNotExist:
@@ -875,7 +875,7 @@ def micromesoid(id):
@register.filter
def micromacroid(id):
def micromacroid(id): # pragma: no cover
try:
thismicro = TrainingMicroCycle.objects.get(id=id)
except TrainingMicroCycle.DoesNotExist:
@@ -905,7 +905,7 @@ def nextworkout(workout,user):
).order_by(
"startdatetime"
).exclude(id=workout.id)
except ValueError:
except ValueError: # pragma: no cover
return 0
if ws:
@@ -935,7 +935,7 @@ def previousworkout(workout,user):
).order_by(
"-startdatetime"
).exclude(id=workout.id)
except ValueError:
except ValueError: # pragma: no cover
return 0

51
rowers/tests/mocks.py
View File

@@ -817,7 +817,7 @@ def mocked_requests(*args, **kwargs):
self.status_code = status_code
self.ok = True
class MockOAuth1Session():
class MockOAuth1Session:
def __init__(self,*args, **kwargs):
pass
@@ -851,18 +851,46 @@ def mocked_requests(*args, **kwargs):
return MockResponse(json_data,200)
if 'garmin' in args:
return MockOAuth1Session()
if 'url' in kwargs:
if 'rp3' in kwargs['url']:
args = [kwargs['url']]
if "tofit" in kwargs['url']:
args = [kwargs['url']]
if not args:
return MockSession()
if "tofit" in args[0]:
jsonresponse = {
'name': '',
'sport': 'rowing',
'filename': '/home/sander/python/rowsandall/media/630a9e78-6d34-4eb3-8d53-4c02b2e95fff.fit',
'steps': [
{
'wkt_step_name': '0',
'stepId': 0,
'durationType': 'Distance',
'durationValue': 100000,
'intensity': 'Active'
},
{
'wkt_step_name': '1',
'stepId': 1,
'durationType': 'RepeatUntilStepsCmplt',
'targetValue': 4,
'durationValue': 0
}
]
}
return MockResponse(jsonresponse,200)
polartester = re.compile('.*?polaraccesslink\.com')
c2tester = re.compile('.*?log\.concept2\.com')
stravatester = re.compile('.*?strava\.com')
@@ -1171,3 +1199,22 @@ class MockEmailMessage:
def send(self):
return 1
class MockResponse:
def __init__(self, json_data, status_code):
self.json_data = json_data
self.status_code = status_code
self.ok = True
def json(self):
return self.json_data
class MockOAuth1Session:
def __init__(self,*args, **kwargs):
pass
def get(*args,**kwargs):
return MockStreamResponse('rowers/tests/testdata/3x250m.fit',200)
def post(*args, **kwargs):
return MockResponse({},200)

29
rowers/tests/test_plans.py
View File

@@ -8,12 +8,14 @@ from .statements import *
nu = datetime.datetime.now()
from rowers.utils import allmonths,allsundays
from rowers import garmin_stuff
import rowers.plannedsessions as plannedsessions
from django.db import transaction
from rowers.views.workoutviews import plannedsession_compare_view
from rowers.opaque import encoder
from django.utils.crypto import get_random_string
@override_settings(TESTING=True)
class TrainingPlanTest(TestCase):
@@ -93,7 +95,7 @@ class TrainingPlanTest(TestCase):
for url in urls:
if 'macrocycle' in url and 'delete' not in url:
macrourl = url
print(macrourl)
response = self.c.get(macrourl)
self.assertEqual(response.status_code,200)
@@ -121,7 +123,7 @@ class TrainingPlanTest(TestCase):
for url in urls:
if 'planbymonths' in url:
print(url)
response = self.c.get(url,follow=True)
self.assertEqual(response.status_code,200)
@@ -1872,6 +1874,29 @@ description: ""
response = self.c.post(urlshare,form_data)
self.assertEqual(response.status_code,200)
key = ShareKey.objects.create(pk=get_random_string(40),
expiration_seconds=60,
location=url
)
key.save()
url = '/rowers/access/'+key.token
response = self.c.get(url,follow=True)
self.assertEqual(response.status_code,200)
@patch('rowers.garmin_stuff.requests.post', side_effect=mocked_requests)
@patch('rowers.utils.requests.post', side_effect=mocked_requests)
@patch('rowers.garmin_stuff.OAuth1Session', side_effect=MockOAuth1Session)
def test_plannedsession_steps(self,mockpost,mock_post,MockOAuth1Session):
self.ps_trimp.interval_string = '4x1000m'
self.ps_trimp.save()
stepsdict = self.ps_trimp.steps['steps']
self.assertEqual(len(stepsdict),2)
response = garmin_stuff.ps_to_garmin(self.ps_trimp,self.r)
self.assertEqual(response.status_code,200)
def test_plannedsessions_dateform_view(self):

127
rowers/utils.py
View File

@@ -178,20 +178,11 @@ rankingdurations.append(datetime.time(minute=30))
rankingdurations.append(datetime.time(hour=1,minute=15))
rankingdurations.append(datetime.time(hour=1))
def is_ranking_piece(workout):
if workout.distance in rankingdistances:
return True
elif workout.duration in rankingdurations:
return True
return False
def range_to_color_hex(groupcols,palette='monochrome_blue'):
try:
plt = palettes[palette]
except KeyError:
except KeyError: # pragma: no cover
plt = palettes['monochrome_blue']
rgb = [colorsys.hsv_to_rgb((plt[0]+plt[1]*x)/360.,
@@ -203,7 +194,7 @@ def range_to_color_hex(groupcols,palette='monochrome_blue'):
return colors
def str2bool(v):
def str2bool(v): # pragma: no cover
return v.lower() in ("yes", "true", "t", "1")
def uniqify(seq, idfun=None):
@@ -222,11 +213,11 @@ def uniqify(seq, idfun=None):
result.append(item)
return result
def serialize_list(value,token=','):
def serialize_list(value,token=','): # pragma: no cover
assert(isinstance(value, list) or isinstance(value, tuple) or isinstance(value,np.ndarray))
return token.join([str(s) for s in value])
def deserialize_list(value,token=','):
def deserialize_list(value,token=','): # pragma: no cover
if isinstance(value, list):
return value
elif isinstance(value, np.ndarray):
@@ -316,15 +307,15 @@ def myqueue(queue,function,*args,**kwargs):
self.result = 1
self.id = 1
def revoke(self):
def revoke(self): # pragma: no cover
return 1
if settings.TESTING:
return MockJob()
elif settings.CELERY:
elif settings.CELERY: # pragma: no cover
kwargs['debug'] = True
job = function.delay(*args,**kwargs)
else:
else: # pragma: no cover
if settings.DEBUG:
kwargs['debug'] = True
@@ -335,7 +326,7 @@ def myqueue(queue,function,*args,**kwargs):
job = queue.enqueue(function,*args,**kwargs)
return job
return job # pragma: no cover
from datetime import date
@@ -361,7 +352,7 @@ def my_dict_from_instance(instance,model):
try:
verbosename = f.verbose_name
except:
except: # pragma: no cover
verbosename = f.name
get_choice = 'get_'+fname+'_display'
@@ -370,7 +361,7 @@ def my_dict_from_instance(instance,model):
else:
try:
value = getattr(instance,fname)
except AttributeError:
except AttributeError: # pragma: no cover
value = None
if f.editable and value:
@@ -390,33 +381,33 @@ def wavg(group, avg_name, weight_name):
return d.mean()
try:
return (d * w).sum() / w.sum()
except ZeroDivisionError:
except ZeroDivisionError: # pragma: no cover
return d.mean()
def totaltime_sec_to_string(totaltime,shorten=False):
if np.isnan(totaltime):
return ''
hours = int(totaltime / 3600.)
if hours > 23:
if hours > 23: # pragma: no cover
message = 'Warning: The workout duration was longer than 23 hours. '
hours = 23
minutes = int((totaltime - 3600. * hours) / 60.)
if minutes > 59:
if minutes > 59: # pragma: no cover
minutes = 59
if not message:
if not message: # pragma: no cover
message = 'Warning: there is something wrong with the workout duration'
seconds = int(totaltime - 3600. * hours - 60. * minutes)
if seconds > 59:
if seconds > 59: # pragma: no cover
seconds = 59
if not message:
if not message: # pragma: no cover
message = 'Warning: there is something wrong with the workout duration'
tenths = int(10 * (totaltime - 3600. * hours - 60. * minutes - seconds))
if tenths > 9:
if tenths > 9: # pragma: no cover
tenths = 9
if not message:
if not message: # pragma: no cover
message = 'Warning: there is something wrong with the workout duration'
duration = ""
@@ -428,7 +419,7 @@ def totaltime_sec_to_string(totaltime,shorten=False):
tenths=tenths
)
else:
if hours != 0:
if hours != 0: # pragma: no cover
duration = "{hours}:{minutes:02d}:{seconds:02d}".format(
hours=hours,
minutes=minutes,
@@ -446,7 +437,7 @@ def totaltime_sec_to_string(totaltime,shorten=False):
return duration
def iscoach(m,r):
def iscoach(m,r): # pragma: no cover
result = False
result = m in r.coaches
@@ -468,7 +459,7 @@ def ewmovingaverage(interval,window_size):
interval2 = np.vstack((i_ewma1,i_ewma2[::-1]))
interval2 = np.mean( interval2, axis=0) # average
except ValueError:
except ValueError: # pragma: no cover
interval2 = interval
return interval2
@@ -479,10 +470,10 @@ class NoTokenError(Exception):
def __init__(self,value):
self.value=value
def __str__(self):
def __str__(self): # pragma: no cover
return repr(self.value)
class ProcessorCustomerError(Exception):
class ProcessorCustomerError(Exception): # pragma: no cover
def __init__(self, value):
self.value=value
@@ -516,7 +507,7 @@ def get_strava_stream(r,metric,stravaid,series_type='time',fetchresolution='high
'Content-Type': 'application/json',
'resolution': 'medium',}
if metric == 'power':
if metric == 'power': # pragma: no cover
metric = 'watts'
url = "https://www.strava.com/api/v3/activities/{stravaid}/streams/{metric}?resolution={fetchresolution}&series_type={series_type}".format(
@@ -530,7 +521,7 @@ def get_strava_stream(r,metric,stravaid,series_type='time',fetchresolution='high
s = requests.get(url,headers=headers)
if metric=='power':
if metric=='power': # pragma: no cover
with open('data.txt', 'w') as outfile:
json.dump(s.json(), outfile)
print('saved to file')
@@ -540,7 +531,7 @@ def get_strava_stream(r,metric,stravaid,series_type='time',fetchresolution='high
try:
if data['type'] == metric:
return np.array(data['data'])
except TypeError:
except TypeError: # pragma: no cover
return None
return None
@@ -565,13 +556,13 @@ def steps_read_fit(filename,name='',sport='Custom'):
response = requests.post(url=url,headers=headers,json={'filename':filename})
if response.status_code != 200:
if response.status_code != 200: # pragma: no cover
return None
w = response.json()
try:
d = w['workout']
except KeyError:
except KeyError: # pragma: no cover
return None
return d
@@ -583,13 +574,13 @@ def steps_write_fit(steps):
response = requests.post(url=url,headers=headers,json=steps)
if response.status_code != 200:
if response.status_code != 200: # pragma: no cover
return None
w = response.json()
try:
filename = w['filename']
except KeyError:
except KeyError: # pragma: no cover
return None
return filename
@@ -599,12 +590,12 @@ def step_to_time_dist(step,avgspeed = 3.7):
distance = 0
durationtype = step['durationType']
if step['durationValue'] == 0:
if step['durationValue'] == 0: # pragma: no cover
return 0,0
try:
targettype = step['targetType']
except KeyError:
except KeyError: # pragma: no cover
targettype = 0
@@ -620,13 +611,13 @@ def step_to_time_dist(step,avgspeed = 3.7):
valuelow = step['targetValueLow']
valuehigh = step['targetValueHigh']
if value != 0:
if value != 0: # pragma: no cover
distance = seconds*value
elif valuelow != 0 and valuehigh != 0:
elif valuelow != 0 and valuehigh != 0: # pragma: no cover
distance = seconds*(valuelow+valuehigh)/2.
return seconds,distance
elif durationtype == 'Distance':
elif durationtype == 'Distance': # pragma: no cover
value = step['durationValue']
distance = value/100.
seconds = distance/avgspeed
@@ -636,21 +627,21 @@ def step_to_time_dist(step,avgspeed = 3.7):
valuelow = step['targetValueLow']
valuehigh = step['targetValueHigh']
if value != 0:
if value != 0: # pragma: no cover
seconds = distance/value
elif valuelow != 0 and valuehigh != 0:
elif valuelow != 0 and valuehigh != 0: # pragma: no cover
midspeed = (valuelow+valuehigh)/2.
seconds = distance/midspeed
return seconds, distance
elif durationtype in ['PowerLessThan','PowerGreaterThan','HrLessThan','HrGreaterThan']:
elif durationtype in ['PowerLessThan','PowerGreaterThan','HrLessThan','HrGreaterThan']: # pragma: no cover
seconds = 600
distance = seconds*avgspeed
return seconds,distance
return seconds,distance
def get_step_type(step):
def get_step_type(step): # pragma: no cover
t = 'WorkoutStep'
if step['durationType'] in ['RepeatUntilStepsCmplt','RepeatUntilHrLessThan','RepeatUntilHrGreaterThan']:
@@ -659,7 +650,7 @@ def get_step_type(step):
return t
def peel(l):
if len(l)==0:
if len(l)==0: # pragma: no cover
return None,None
if len(l)==1:
return l[0],None
@@ -667,7 +658,7 @@ def peel(l):
first = l[0]
rest = l[1:]
if first['type'] == 'Step':
if first['type'] == 'Step': # pragma: no cover
return first, rest
# repeatstep
theID = -1
@@ -765,7 +756,7 @@ def ps_dict_order(d,short=False):
factor /= multiplier.pop()
spaces = spaces[:-18]
holduntil.pop()
else:
else: # pragma: no cover
prevstep = sdict3.pop()
prevstep['string'] = prevstep['string'][18:]
prevprevstep = sdict3.pop()
@@ -799,13 +790,13 @@ def step_to_string(step,short=False):
durationtype = step['durationType']
if step['durationValue'] == 0:
if durationtype not in ['RepeatUntilStepsCmplt','RepeatUntilHrLessThan','RepeatUntilHrGreaterThan']:
if durationtype not in ['RepeatUntilStepsCmplt','RepeatUntilHrLessThan','RepeatUntilHrGreaterThan']: # pragma: no cover
return '',type, -1, -1,1
if durationtype == 'Time':
unit = 'min'
value = step['durationValue']
if value/1000. >= 3600:
if value/1000. >= 3600: # pragma: no cover
unit = 'h'
dd = timedelta(seconds=value/1000.)
#duration = humanize.naturaldelta(dd, minimum_unit="seconds")
@@ -814,7 +805,7 @@ def step_to_string(step,short=False):
unit = 'm'
value = step['durationValue']/100.
duration = int(value)
elif durationtype == 'HrLessThan':
elif durationtype == 'HrLessThan': # pragma: no cover
value = step['durationValue']
if value <= 100:
duration = 'until heart rate lower than {v}% of max'.format(v=value)
@@ -824,7 +815,7 @@ def step_to_string(step,short=False):
duration = 'until heart rate lower than {v}'.format(v=value-100)
if short:
duration = 'until HR<{v}'.format(v=value-100)
elif durationtype == 'HrGreaterThan':
elif durationtype == 'HrGreaterThan': # pragma: no cover
value = step['durationValue']
if value <= 100:
duration = 'until heart rate greater than {v}% of max'.format(v=value)
@@ -834,7 +825,7 @@ def step_to_string(step,short=False):
duration = 'until heart rate greater than {v}'.format(v=value-100)
if short:
duration = 'until HR>{v}'.format(v=value/100)
elif durationtype == 'PowerLessThan':
elif durationtype == 'PowerLessThan': # pragma: no cover
value = step['durationValue']
targetvalue = step['targetvalue']
if value <= 1000:
@@ -849,7 +840,7 @@ def step_to_string(step,short=False):
)
if short:
'until < {targetvalue} W'.format(targetvalue=targetvalue-1000)
elif durationtype == 'PowerGreaterThan':
elif durationtype == 'PowerGreaterThan': # pragma: no cover
value = step['durationValue']
targetvalue = step['targetvalue']
if value <= 1000:
@@ -864,13 +855,13 @@ def step_to_string(step,short=False):
)
if short:
duration = 'until > {targetvalue} W'.format(targetvalue=targetvalue)
elif durationtype == 'RepeatUntilStepsCmplt':
elif durationtype == 'RepeatUntilStepsCmplt': # pragma: no cover
type = 'RepeatStep'
ntimes = ': {v}x'.format(v=step['targetValue'])
repeatID = step['durationValue']
duration =ntimes
repeatValue = step['targetValue']
elif durationtype == 'RepeatUntilHrGreaterThan':
elif durationtype == 'RepeatUntilHrGreaterThan': # pragma: no cover
type = 'RepeatStep'
targetvalue = step['targetValue']
if targetvalue <= 100:
@@ -886,7 +877,7 @@ def step_to_string(step,short=False):
if short:
duration = ': untl HR>{targetvalue}'.format(targetvalue=targetvalue-100)
repeatID = step['durationValue']
elif durationtype == 'RepeatUntilHrLessThan':
elif durationtype == 'RepeatUntilHrLessThan': # pragma: no cover
type = 'RepeatStep'
targetvalue = step['targetValue']
if targetvalue <= 100:
@@ -911,7 +902,7 @@ def step_to_string(step,short=False):
except KeyError:
targettype = None
if targettype == 'HeartRate':
if targettype == 'HeartRate': # pragma: no cover
try:
value = step['targetValue']
except KeyError:
@@ -938,7 +929,7 @@ def step_to_string(step,short=False):
l = valuelow - 100,
h = valuehigh - 100,
)
elif targettype == 'Power':
elif targettype == 'Power': # pragma: no cover
try:
value = step['targetValue']
except KeyError:
@@ -969,7 +960,7 @@ def step_to_string(step,short=False):
l = valuelow-1000,
h = valuehigh-1000,
)
elif targettype == 'Speed':
elif targettype == 'Speed': # pragma: no cover
try:
value = step['targetValue']
except KeyError:
@@ -990,9 +981,9 @@ def step_to_string(step,short=False):
target = '@ {v} m/s {p}, per 500m'.format(
v=value/1000.,
p=pacestring)
if short:
if short: # pragma: no cover
target = '@ {p}'.format(p=pacestring)
elif valuelow != 0 and valuehigh != 0:
elif valuelow != 0 and valuehigh != 0: # pragma: no cover
v = valuelow/1000.
pace = 500./v
pacestringlow = to_pace(pace)
@@ -1012,7 +1003,7 @@ def step_to_string(step,short=False):
pl = pacestringlow,
ph = pacestringhigh,
)
elif targettype == 'Cadence':
elif targettype == 'Cadence': # pragma: no cover
try:
value = step['targetValue']
except KeyError:
@@ -1041,7 +1032,7 @@ def step_to_string(step,short=False):
notes = ''
try:
if len(step['description']):
if len(step['description']): # pragma: no cover
notes = ' - '+step['description']
except KeyError:
notes = ''
@@ -1093,7 +1084,7 @@ def step_to_string(step,short=False):
return s,type, nr, repeatID, repeatValue
def strfdelta(tdelta):
def strfdelta(tdelta): # pragma: no cover
try:
minutes, seconds = divmod(tdelta.seconds, 60)
tenths = int(tdelta.microseconds / 1e5)

144
rowers/views/statements.py
View File

@@ -307,14 +307,14 @@ def sharedPage(request, key):
try:
try:
shareKey = ShareKey.objects.get(pk=key)
except:
except: # pragma: no cover
raise SharifyError
if shareKey.expired:
if shareKey.expired: # pragma: no cover
raise SharifyError
func, args, kwargs = resolve(shareKey.location)
kwargs["__shared"] = True
return func(request, *args, **kwargs)
except SharifyError:
except SharifyError: # pragma: no cover
raise Http404 # or add a more detailed error page. This either means that the key doesn’t exist or is expired.
def createShareURL(request):
@@ -326,10 +326,10 @@ def createShareURL(request):
location = url)
key.save()
return render(request, 'share.html', {"key":key})
else:
else: # pragma: no cover
raise Http404
def createShareModel(request, model_id):
def createShareModel(request, model_id): # pragma: no cover
task = MyModel.objects.get(pk=model_id)
key = ShareKey.objects.create(pk=get_random_string(40),
expiration_seconds=60*60*24, # 1 day
@@ -353,7 +353,7 @@ def getfavorites(r,row):
matchworkouttypes = [workouttype,'all']
workoutsource = row.workoutsource
if 'speedcoach2' in row.workoutsource:
if 'speedcoach2' in row.workoutsource: # pragma: no cover
workoutsource = 'speedcoach2'
favorites = FavoriteChart.objects.filter(user=r,
@@ -370,7 +370,7 @@ def getfavorites(r,row):
return favorites,maxfav
def get_logo_by_pk(request,*args,**kwargs):
def get_logo_by_pk(request,*args,**kwargs): # pragma: no cover
id = kwargs['id']
return get_object_or_404(RaceLogo,pk=id)
@@ -378,7 +378,7 @@ def get_virtualevent_by_pk(request,*args,**kwargs):
id = kwargs['id']
return get_object_or_404(VirtualRace,pk=id)
def get_promember(request,*args,**kwargs):
def get_promember(request,*args,**kwargs): # pragma: no cover
return request.user
def get_course_by_pk(request,*args,**kwargs):
@@ -401,15 +401,15 @@ def get_plan_by_pk(request,*args,**kwargs):
id = kwargs['id']
return get_object_or_404(TrainingPlan,pk=id)
def get_macro_by_pk(request,*args,**kwargs):
def get_macro_by_pk(request,*args,**kwargs): # pragma: no cover
id = kwargs['id']
return get_object_or_404(TrainingMacroCycle,pk=id)
def get_meso_by_pk(request,*args,**kwargs):
def get_meso_by_pk(request,*args,**kwargs): # pragma: no cover
id = kwargs['id']
return get_object_or_404(TrainingMesoCycle,pk=id)
def get_micro_by_pk(request,*args,**kwargs):
def get_micro_by_pk(request,*args,**kwargs): # pragma: no cover
id = kwargs['id']
return get_object_or_404(TrainingMicroCycle,pk=id)
@@ -420,7 +420,7 @@ def get_workout_default_page(request,id):
r = Rower.objects.get(user=request.user)
if r.defaultlandingpage == 'workout_edit_view':
return reverse('workout_edit_view',kwargs={'id':id})
else:
else: # pragma: no cover
return reverse('workout_workflow_view',kwargs={'id':id})
def get_user_by_userid(*args,**kwargs):
@@ -436,7 +436,7 @@ def get_user_by_userid(*args,**kwargs):
u = get_object_or_404(User,pk=id)
return u
def get_user_by_id(*args,**kwargs):
def get_user_by_id(*args,**kwargs): # pragma: no cover
request = args[0]
try:
id = args[1]
@@ -448,7 +448,7 @@ def get_user_by_id(*args,**kwargs):
return get_object_or_404(User,pk=id)
def get_rower_by_id(request,id):
def get_rower_by_id(request,id): # pragma: no cover
u = User.objects.get(id=id)
return u.rower
@@ -474,14 +474,14 @@ def getrequestrower(request,rowerid=0,userid=0,notpermanent=False):
elif userid != 0:
u = User.objects.get(id=userid)
r = getrower(u)
elif request.user.is_anonymous:
elif request.user.is_anonymous: # pragma: no cover
return None
else:
r = getrower(request.user)
u = r.user
except Rower.DoesNotExist:
except Rower.DoesNotExist: # pragma: no cover: # pragma: no cover
raise Http404("Rower doesn't exist")
if r.user == request.user:
@@ -520,21 +520,21 @@ def getrequestrowercoachee(request,rowerid=0,userid=0,notpermanent=False):
elif userid != 0:
u = User.objects.get(id=userid)
r = getrower(u)
elif request.user.is_anonymous:
elif request.user.is_anonymous: # pragma: no cover
return None
else:
r = getrower(request.user)
u = r.user
except Rower.DoesNotExist:
except Rower.DoesNotExist: # pragma: no cover: # pragma: no cover
raise Http404("Rower doesn't exist")
if r.user == request.user:
request.session['rowerid'] = r.id
return r
if userid != 0 and not is_coach_user(request.user,u):
if userid != 0 and not is_coach_user(request.user,u): # pragma: no cover
request.session['rowerid'] = request.user.rower.id
raise PermissionDenied("You have no access to this user")
@@ -563,16 +563,16 @@ def getrequestplanrower(request,rowerid=0,userid=0,notpermanent=False):
elif userid != 0:
try:
u = User.objects.get(id=userid)
except User.DoesNotExist:
except User.DoesNotExist: # pragma: no cover: # pragma: no cover
raise Http404("User does not exist")
r = getrower(u)
else:
r = getrower(request.user)
except Rower.DoesNotExist:
except Rower.DoesNotExist: # pragma: no cover: # pragma: no cover
raise Http404("Rower doesn't exist")
if 'shared' in request.session and request.session['shared']:
if 'shared' in request.session and request.session['shared']: # pragma: no cover
return r
if r.user != request.user and not can_plan_user(request.user,r ):
@@ -589,12 +589,12 @@ def getrower(user):
try:
if user is None or user.is_anonymous:
return None
except AttributeError:
except AttributeError: # pragma: no cover
if User.objects.get(id=user).is_anonymous:
return None
try:
r = Rower.objects.get(user=user)
except Rower.DoesNotExist:
except Rower.DoesNotExist: # pragma: no cover:
r = Rower(user=user)
r.save()
@@ -605,7 +605,7 @@ def get_workout(id):
try:
id = encoder.decode_hex(id)
w = Workout.objects.get(id=id)
except Workout.DoesNotExist:
except Workout.DoesNotExist: # pragma: no cover:
raise Http404("Workout doesn't exist")
return w
@@ -614,15 +614,15 @@ def get_workoutuser(id,request):
try:
id = encoder.decode_hex(id)
w = Workout.objects.get(id=id)
except Workout.DoesNotExist:
except Workout.DoesNotExist: # pragma: no cover:
raise Http404("Workout doesn't exist")
if not is_workout_user(request.user,w):
if not is_workout_user(request.user,w): # pragma: no cover
raise PermissionDenied
return w
def getvalue(data):
def getvalue(data): # pragma: no cover
perc = 0
total = 1
done = 0
@@ -640,7 +640,7 @@ def getvalue(data):
return total,done,id,session_key
class SessionTaskListener(threading.Thread):
class SessionTaskListener(threading.Thread): # pragma: no cover
def __init__(self, r, channels):
threading.Thread.__init__(self)
self.redis = r
@@ -685,7 +685,7 @@ from rq.job import Job
try:
from rest_framework_swagger.views import get_swagger_view
except ImportError:
except ImportError: # pragma: no cover
pass
from rest_framework.renderers import JSONRenderer
@@ -694,7 +694,7 @@ from rest_framework.response import Response
from rowers.serializers import RowerSerializer,WorkoutSerializer
try:
from rest_framework import status,permissions,generics
except ImportError:
except ImportError: # pragma: no cover
pass
from rest_framework.decorators import api_view, renderer_classes, permission_classes
@@ -723,10 +723,10 @@ from rowers.celery import result as celery_result
# Define the API documentation
try:
schema_view = get_swagger_view(title='Rowsandall API')
except NameError:
except NameError: # pragma: no cover
pass
def remove_asynctask(request,id):
def remove_asynctask(request,id): # pragma: no cover
try:
oldtasks = request.session['async_tasks']
except KeyError:
@@ -739,7 +739,7 @@ def remove_asynctask(request,id):
request.session['async_tasks'] = newtasks
def get_job_result(jobid):
def get_job_result(jobid): # pragma: no cover
if settings.TESTING:
return None
elif settings.CELERY:
@@ -771,7 +771,7 @@ verbose_job_status = {
'submit_race': 'Checking Race Course Result',
}
def get_job_status(jobid):
def get_job_status(jobid): # pragma: no cover
if settings.TESTING:
summary = {
'status': 'failed',
@@ -836,7 +836,7 @@ def get_job_status(jobid):
return summary
def kill_async_job(request,id='aap'):
def kill_async_job(request,id='aap'): # pragma: no cover
if settings.CELERY:
job = celery_result.AsyncResult(id)
job.revoke()
@@ -853,7 +853,7 @@ def kill_async_job(request,id='aap'):
return HttpResponseRedirect(url)
@login_required()
def raise_500(request):
def raise_500(request): # pragma: no cover
if request.user.is_superuser:
raise ValueError
else:
@@ -895,7 +895,7 @@ def raise_500(request):
# return HttpResponseRedirect(url)
@csrf_exempt
def post_progress(request,id=None,value=0):
def post_progress(request,id=None,value=0): # pragma: no cover
if request.method == 'POST':
try:
secret = request.POST['secret']
@@ -924,7 +924,7 @@ def post_progress(request,id=None,value=0):
else: # request method is not POST
return HttpResponse('GET method not allowed',status=405)
def get_all_queued_jobs(userid=0):
def get_all_queued_jobs(userid=0): # pragma: no cover
r = StrictRedis()
jobs = []
@@ -969,7 +969,7 @@ def get_stored_tasks_status(request):
taskids = []
taskstatus = []
for id,func_name in reversed(taskids):
for id,func_name in reversed(taskids): # pragma: no cover
progress = 0
try:
cached_progress = cache.get(id)
@@ -1027,13 +1027,13 @@ def get_thumbnails(request,id):
try:
if charts[0]['script'] == '':
charts = []
except IndexError:
except IndexError: # pragma: no cover
charts = []
return JSONResponse(charts)
def get_blog_posts(request):
def get_blog_posts(request): # pragma: no cover
blogposts = BlogPost.objects.all().order_by("-date")
jsondata = []
@@ -1049,7 +1049,7 @@ def get_blog_posts(request):
return JSONResponse(jsondata)
def get_blog_posts_old(request):
def get_blog_posts_old(request): # pragma: no cover
try:
response = requests.get(
'https://analytics.rowsandall.com/wp-json/wp/v2/posts?per_page=3')
@@ -1116,12 +1116,12 @@ def rowhascoordinates(row):
try:
latitude = rowdata.df[' latitude']
if not latitude.std():
if not latitude.std(): # pragma: no cover
hascoordinates = 0
except (KeyError,AttributeError):
hascoordinates = 0
else:
else: # pragma: no cover
hascoordinates = 0
return hascoordinates
@@ -1130,13 +1130,13 @@ def rowhascoordinates(row):
# Wrapper around the rowingdata call to catch some exceptions
# Checks for CSV file, then for gzipped CSV file, and if all fails, returns 0
def rdata(csvfile=None,rower=rrower()):
if csvfile is None:
if csvfile is None: # pragma: no cover
return 0
try:
res = rrdata(csvfile=csvfile,rower=rower)
except pd.errors.EmptyDataError:
except pd.errors.EmptyDataError: # pragma: no cover
res = 0
except (IOError, IndexError, EOFError,FileNotFoundError):
except (IOError, IndexError, EOFError,FileNotFoundError): # pragma: no cover
try:
res = rrdata(csvfile=file+'.gz',rower=rower)
except (IOError, IndexError, EOFError,FileNotFoundError):
@@ -1150,7 +1150,7 @@ def get_my_teams(user):
therower = Rower.objects.get(user=user)
try:
teams1 = therower.team.all()
except AttributeError:
except AttributeError: # pragma: no cover
teams1 = []
teams2 = Team.objects.filter(manager=user)
@@ -1167,7 +1167,7 @@ def get_time(second):
minutes=0
sec=0
microsecond = 0
elif math.isnan(second):
elif math.isnan(second): # pragma: no cover
hours = 0
minutes=0
sec=0
@@ -1182,12 +1182,12 @@ def get_time(second):
# get the workout ID from the SportTracks URI
def getidfromsturi(uri,length=8):
def getidfromsturi(uri,length=8): # pragma: no cover
return uri[len(uri)-length:]
import re
def getidfromuri(uri):
def getidfromuri(uri): # pragma: no cover
m = re.search('/(\w.*)\/(\d+)',uri)
return m.group(2)
@@ -1197,7 +1197,7 @@ from rowers.utils import (
geo_distance,serialize_list,deserialize_list,uniqify,
str2bool,range_to_color_hex,absolute,myqueue,get_call,
calculate_age,rankingdistances,rankingdurations,
is_ranking_piece,my_dict_from_instance,wavg,NoTokenError,
my_dict_from_instance,wavg,NoTokenError,
request_is_ajax
)
@@ -1208,12 +1208,12 @@ def iscoachmember(user):
if not user.is_anonymous:
try:
r = Rower.objects.get(user=user)
except Rower.DoesNotExist:
except Rower.DoesNotExist: # pragma: no cover:
r = Rower(user=user)
r.save()
result = user.is_authenticated and ('coach' in r.rowerplan)
else:
else: # pragma: no cover
result = False
return result
@@ -1256,7 +1256,7 @@ def sendmail(request):
success = response.json().get('success')
form = EmailForm(request.POST)
if form.is_valid() and success:
if form.is_valid() and success: # pragma: no cover
firstname = form.cleaned_data['firstname']
lastname = form.cleaned_data['lastname']
email = form.cleaned_data['email']
@@ -1271,7 +1271,7 @@ def sendmail(request):
else:
if not success:
messages.error(request,'Bots are not welcome')
else:
else: # pragma: no cover
messages.error(request,'Something went wrong. Please try again')
return HttpResponseRedirect('/rowers/email/')
else:
@@ -1285,41 +1285,41 @@ def add_workout_from_strokedata(user,importid,data,strokedata,
workoutsource='concept2'):
try:
workouttype = data['type']
except KeyError:
except KeyError: # pragma: no cover
workouttype = 'rower'
if workouttype not in [x[0] for x in Workout.workouttypes]:
if workouttype not in [x[0] for x in Workout.workouttypes]: # pragma: no cover
workouttype = 'other'
try:
comments = data['comments']
except:
except: # pragma: no cover
comments = ' '
# comments = "Imported data \n %s" % comments
# comments = "Imported data \n"+comments # str(comments)
try:
thetimezone = tz(data['timezone'])
except:
except: # pragma: no cover
thetimezone = 'UTC'
r = getrower(user)
try:
rowdatetime = iso8601.parse_date(data['date_utc'])
except KeyError:
except KeyError: # pragma: no cover
rowdatetime = iso8601.parse_date(data['start_date'])
except ParseError:
except ParseError: # pragma: no cover
rowdatetime = iso8601.parse_date(data['date'])
try:
c2intervaltype = data['workout_type']
except KeyError:
except KeyError: # pragma: no cover
c2intervaltype = ''
try:
title = data['name']
except KeyError:
except KeyError: # pragma: no cover
title = ""
try:
t = data['comments'].split('\n', 1)[0]
@@ -1339,7 +1339,7 @@ def add_workout_from_strokedata(user,importid,data,strokedata,
nr_rows = len(unixtime)
try:
try: # pragma: no cover
latcoord = strokedata.loc[:,'lat']
loncoord = strokedata.loc[:,'lon']
except:
@@ -1356,7 +1356,7 @@ def add_workout_from_strokedata(user,importid,data,strokedata,
try:
spm = strokedata.loc[:,'spm']
except KeyError:
except KeyError: # pragma: no cover
spm = 0*dist2
try:
@@ -1413,10 +1413,10 @@ def add_workout_from_strokedata(user,importid,data,strokedata,
try:
totaldist = data['distance']
totaltime = data['time']/10.
except KeyError:
except KeyError: # pragma: no cover
totaldist = 0
totaltime = 0
else:
else: # pragma: no cover
totaldist = 0
totaltime = 0
@@ -1438,7 +1438,7 @@ def add_workout_from_strokedata(user,importid,data,strokedata,
def keyvalue_get_default(key,options,def_options):
def keyvalue_get_default(key,options,def_options): # pragma: no cover
try:
return options[key]
@@ -1449,14 +1449,14 @@ def keyvalue_get_default(key,options,def_options):
# Creates unix time stamp from a datetime object
def totimestamp(dt, epoch=datetime.datetime(1970,1,1,tzinfo=tz('UTC'))):
def totimestamp(dt, epoch=datetime.datetime(1970,1,1,tzinfo=tz('UTC'))): # pragma: no cover
td = dt - epoch
# return td.total_seconds()
return (td.microseconds + (td.seconds + td.days * 86400) * 10**6) / 10**6
# Check if a column of a dataframe has the required (aantal)
# number of elements. Also checks if the column is a numerical type
# Replaces any faulty columns with zeros
def trydf(df,aantal,column):
def trydf(df,aantal,column): # pragma: no cover
try:
s = df[column]
if len(s) != aantal: