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

lots a small stuff

Browse Source
This commit is contained in:
Sander Roosendaal
2021-04-26 19:35:11 +02:00
parent 594ee6239a
commit 62d06a2439
33 changed files with 144 additions and 149 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
context_processors.py
View File

@@ -1,6 +1,6 @@
from django.conf import settings
def google_analytics(request):
def google_analytics(request): # pragma: no cover
"""
Use the variables returned in this function to
render your Google Analytics tracking code template.

4
rowers/alerts.py
View File

@@ -201,6 +201,6 @@ def alert_get_stats(alert,nperiod=0): # pragma: no cover
def checkalertowner(alert,user):
if alert.manager == user:
return True
if alert.rower.user == user:
if alert.rower.user == user: # pragma: no cover
return True
return False
return False # pragma: no cover

2
rowers/context_processors.py
View File

@@ -5,7 +5,7 @@ from __future__ import unicode_literals
from django.conf import settings # import the settings file
def braintree_merchant(request):
def braintree_merchant(request): # pragma: no cover
# return the value you want as a dictionnary. you may add multiple values in there.
# return {'BRAINTREE_MERCHANT_ID': settings.BRAINTREE_MERCHANT_ID}
return {'BRAINTREE_MERCHANT_ID': 'jytq7yxsm66qqdzb' }

12
rowers/courses.py
View File

@@ -83,7 +83,7 @@ def get_polygons(polygonpms):
coordinates = pm.findall('.//opengis:coordinates',ns)
if coordinates:
cc = coordinates[0].text
else:
else: # pragma: no cover
cc = ''
pointstring = cc.split()
@@ -157,7 +157,7 @@ def kmltocourse(f):
doc = et.parse(f)
courses = doc.findall('.//opengis:Folder[opengis:Placemark]',ns)
if not courses:
if not courses: # pragma: no cover
courses = doc.findall('.//opengis:Document[opengis:Placemark]',ns)
if not courses:
courses = doc.findall('.//opengis:Placemark',ns)
@@ -165,9 +165,9 @@ def kmltocourse(f):
if courses:
return crewnerdcourse(courses)
polygonpms = doc.findall('.//opengis:Placemark[opengis:Polygon]',ns)
polygonpms = doc.findall('.//opengis:Placemark[opengis:Polygon]',ns) # pragma: no cover
return get_polygons(polygonpms)
return get_polygons(polygonpms) # pragma: no cover
@@ -190,7 +190,7 @@ def createcourse(
g = geocoder.osm([latitude,longitude],method='reverse')
if g.ok:
country = g.json['country']
else:
else: # pragma: no cover
country = 'unknown'
c.country = country
c.save()
@@ -210,7 +210,7 @@ def createcourse(
return c
def get_time_course(ws,course):
def get_time_course(ws,course): # pragma: no cover
coursetimeseconds = 0.0
coursecompleted = False

4
rowers/decorators.py
View File

@@ -13,7 +13,7 @@ from django.contrib import messages
try:
from functools import wraps
except ImportError:
except ImportError: # pragma: no cover
from django.utils.functional import wraps
@@ -40,7 +40,7 @@ def user_passes_test(test_func, message=default_message,login_url=None,redirect_
return _wrapped_view
return decorator
def login_required_message(function=None, message=default_message):
def login_required_message(function=None, message=default_message): # pragma: no cover
"""
Decorator for views that checks that the user is logged in, redirecting
to the log-in page if necessary.

12
rowers/emails.py
View File

@@ -81,19 +81,19 @@ def send_template_email(from_email,to_email,subject,
# html_content = newlinetobr(html_content)
if 'bcc' in kwargs and 'cc' in kwargs:
if 'bcc' in kwargs and 'cc' in kwargs: # pragma: no cover
msg = EmailMultiAlternatives(subject, text_content, from_email, to_email,cc=kwargs['cc'],
bcc=kwargs['bcc'])
elif 'bcc' in kwargs:
elif 'bcc' in kwargs: # pragma: no cover
msg = EmailMultiAlternatives(subject, text_content, from_email, to_email,bcc=kwargs['bcc'])
elif 'cc' in kwargs:
elif 'cc' in kwargs: # pragma: no cover
msg = EmailMultiAlternatives(subject, text_content, from_email, to_email,cc=kwargs['cc'])
else:
msg = EmailMultiAlternatives(subject, text_content, from_email, to_email)
msg.attach_alternative(html_content, "text/html")
if 'attach_file' in kwargs:
if 'attach_file' in kwargs: # pragma: no cover
fileobj = kwargs['attach_file']
if os.path.isfile(fileobj):
msg.attach_file(fileobj)
@@ -107,7 +107,7 @@ def send_template_email(from_email,to_email,subject,
except IOError:
pass
if 'emailbounced' in kwargs:
if 'emailbounced' in kwargs: # pragma: no cover
emailbounced = kwargs['emailbounced']
else:
emailbounced = False
@@ -116,7 +116,7 @@ def send_template_email(from_email,to_email,subject,
if not emailbounced:
res = msg.send()
else:
else: # pragma: no cover
return 0
return res

10
rowers/fakturoid.py
View File

@@ -32,7 +32,7 @@ def get_contacts(rower):
with open('braintreewebhooks.log','a') as f:
f.write('Searching Contact Status code '+str(res.status_code)+'\n')
if res.status_code != 200:
if res.status_code != 200: # pragma: no cover
return None
with open('braintreewebhooks.log','a') as f:
@@ -74,7 +74,7 @@ def create_contact(rower):
with open('braintreewebhooks.log','a') as f:
f.write('Status Code '+str(res.status_code)+'\n')
if res.status_code not in [200,201]:
if res.status_code not in [200,201]: # pragma: no cover
return 0
with open('braintreewebhooks.log','a') as f:
@@ -96,7 +96,7 @@ def create_invoice(rower,amount,braintreeid,dosend=True,
with open('braintreewebhooks.log','a') as f:
f.write('Creating invoice for contact iD '+str(contact_id)+'\n')
if not contact_id:
if not contact_id: # pragma: no cover
return 0
post_data = {
@@ -121,7 +121,7 @@ def create_invoice(rower,amount,braintreeid,dosend=True,
with open('braintreewebhooks.log','a') as f:
f.write('Invoice Created - status code '+str(res.status_code)+'\n')
if res.status_code not in [200,201]:
if res.status_code not in [200,201]: # pragma: no cover
return 0
url = res.json()['url']
@@ -140,7 +140,7 @@ def create_invoice(rower,amount,braintreeid,dosend=True,
with open('braintreewebhooks.log','a') as f:
f.write('Invoice Set to paid - status code '+str(res.status_code)+'\n')
if res.status_code not in [200,201]:
if res.status_code not in [200,201]: # pragma: no cover
return 0
if dosend:

6
rowers/garmin_stuff.py
View File

@@ -133,7 +133,7 @@ def garmin_open(user): # pragma: no cover
return token
def get_garmin_file(r,callbackURL,starttime,fileType):
def get_garmin_file(r,callbackURL,starttime,fileType): # pragma: no cover
job = myqueue(
queue,
handle_get_garmin_file,
@@ -462,9 +462,9 @@ def garmin_workouts_from_details(data):
for activity in activities:
try: # pragma: no cover
garmintoken = activity['userAccessToken']
except KeyError:
except KeyError: # pragma: no cover
return 0
except TypeError:
except TypeError: # pragma: no cover
return 0
try:
r = Rower.objects.get(garmintoken=garmintoken)

32
rowers/imports.py
View File

@@ -109,7 +109,7 @@ def imports_open(user,oauth_data):
expirydatename,
oauth_data,
)
elif tokenexpirydate is None and expirydatename is not None and 'strava' in expirydatename:
elif tokenexpirydate is None and expirydatename is not None and 'strava' in expirydatename: # pragma: no cover
token = imports_token_refresh(
user,
tokenname,
@@ -142,7 +142,7 @@ def imports_do_refresh_token(refreshtoken,oauth_data,access_token=''):
if 'grant_type' in oauth_data:
if oauth_data['grant_type']:
post_data['grant_type'] = oauth_data['grant_type']
else:
else: # pragma: no cover
grant_type = post_data.pop('grant_type',None)
if oauth_data['bearer_auth']:
@@ -155,7 +155,7 @@ def imports_do_refresh_token(refreshtoken,oauth_data,access_token=''):
response = requests.post(baseurl,
data=json.dumps(post_data),
headers=headers,verify=False)
except:
except: # pragma: no cover
raise NoTokenError("Failed to get token")
else:
try:
@@ -163,19 +163,19 @@ def imports_do_refresh_token(refreshtoken,oauth_data,access_token=''):
data=post_data,
headers=headers,verify=False,
)
except:
except: # pragma: no cover
raise NoTokenError("Failed to get token")
if response.status_code == 200 or response.status_code == 201:
token_json = response.json()
else:
else: # pragma: no cover
raise NoTokenError("User has no token")
try:
thetoken = token_json['access_token']
except KeyError:
except KeyError: # pragma: no cover
raise NoTokenError("User has no token")
try:
@@ -184,7 +184,7 @@ def imports_do_refresh_token(refreshtoken,oauth_data,access_token=''):
try:
expires_at = arrow.get(token_json['expires_at']).timestamp()
expires_in = expires_at - arrow.now().timestamp()
except KeyError:
except KeyError: # pragma: no cover
expires_in = 0
try:
refresh_token = token_json['refresh_token']
@@ -192,7 +192,7 @@ def imports_do_refresh_token(refreshtoken,oauth_data,access_token=''):
refresh_token = refreshtoken
try:
expires_in = int(expires_in)
except (TypeError,ValueError):
except (TypeError,ValueError): # pragma: no cover
expires_in = 0
return [thetoken,expires_in,refresh_token]
@@ -234,7 +234,7 @@ def imports_get_token(
post_data['grant_type'] = oauth_data['grant_type']
if 'strava' in oauth_data['autorization_uri']:
post_data['grant_type'] = "authorization_code"
else:
else: # pragma: no cover
grant_type = post_data.pop('grant_type',None)
@@ -253,28 +253,28 @@ def imports_get_token(
token_json = response.json()
try:
thetoken = token_json['access_token']
except KeyError:
except KeyError: # pragma: no cover
return [0,0,0]
try:
refresh_token = token_json['refresh_token']
except KeyError:
except KeyError: # pragma: no cover
refresh_token = ''
try:
expires_in = token_json['expires_in']
except KeyError:
except KeyError: # pragma: no cover
expires_in = 0
try:
expires_in = int(expires_in)
except (ValueError,TypeError):
except (ValueError,TypeError): # pragma: no cover
expires_in = 0
else:
else: # pragma: no cover
return [0,response.text,0]
return [thetoken,expires_in,refresh_token]
# Make authorization URL including random string
def imports_make_authorization_url(oauth_data):
def imports_make_authorization_url(oauth_data): # pragma: no cover
# Generate a random string for the state parameter
# Save it for use later to prevent xsrf attacks
@@ -299,7 +299,7 @@ def imports_token_refresh(user,tokenname,refreshtokenname,expirydatename,oauth_d
refreshtoken = getattr(r,refreshtokenname)
# for Strava transition
if not refreshtoken:
if not refreshtoken: # pragma: no cover
refreshtoken = getattr(r,tokenname)

20
rowers/mailprocessing.py
View File

@@ -40,7 +40,7 @@ queuehigh = django_rq.get_queue('default')
# Sends a confirmation with a link to the workout
from rowers.emails import send_template_email
def send_confirm(user, name, link, options):
def send_confirm(user, name, link, options): # pragma: no cover
d = {
'first_name':user.first_name,
'name':name,
@@ -65,12 +65,12 @@ def rdata(file, rower=rrower()):
""" Reads rowingdata data or returns 0 on Error """
try:
result = rrdata(csvfile=file, rower=rower)
except IOError:
except IOError: # pragma: no cover
try:
result = rrdata(csvfile=file + '.gz', rower=rower)
except IOError:
result = 0
except TypeError:
except TypeError: # pragma: no cover
try:
result = rrdata(csvfile=file)
except IOError:
@@ -88,18 +88,18 @@ def make_new_workout_from_email(rower, datafile, name, cntr=0,testing=False):
workouttype = 'rower'
impeller = False
try:
try: # pragma: no cover
datafilename = datafile.name
fileformat = get_file_type('media/' + datafilename)
raise ValueError
except IOError:
except IOError: # pragma: no cover
datafilename = datafile.name + '.gz'
fileformat = get_file_type('media/' + datafilename)
except AttributeError:
datafilename = datafile
fileformat = get_file_type('media/' + datafile)
if len(fileformat) == 3 and fileformat[0] == 'zip':
if len(fileformat) == 3 and fileformat[0] == 'zip': # pragma: no cover
with zipfile.ZipFile('media/' + datafilename) as zip_file:
datafilename = zip_file.extract(
zip_file.namelist()[0],
@@ -112,7 +112,7 @@ def make_new_workout_from_email(rower, datafile, name, cntr=0,testing=False):
f,e = os.path.splitext(datafilename)
if fileformat == 'unknown' and 'txt' not in e:
fcopy = "media/"+datafilename
if not testing:
if not testing: # pragma: no cover
if settings.CELERY:
res = handle_sendemail_unrecognized.delay(
fcopy,
@@ -141,7 +141,7 @@ def make_new_workout_from_email(rower, datafile, name, cntr=0,testing=False):
if fileformat != 'csv':
filename_mediadir, summary, oarlength, inboard,fileformat,impeller = dataprep.handle_nonpainsled(
'media/' + datafilename, fileformat, summary)
if not filename_mediadir:
if not filename_mediadir: # pragma: no cover
return 0
else:
filename_mediadir = 'media/' + datafilename
@@ -149,7 +149,7 @@ def make_new_workout_from_email(rower, datafile, name, cntr=0,testing=False):
oarlength = 2.89
row = rdata(filename_mediadir)
if row == 0:
if row == 0: # pragma: no cover
return 0
# change filename
@@ -169,7 +169,7 @@ def make_new_workout_from_email(rower, datafile, name, cntr=0,testing=False):
dosummary = (fileformat != 'fit' and 'speedcoach2' not in fileformat)
dosummary = dosummary or summary == ''
if name == '':
if name == '': # pragma: no cover
name = 'Workout from Background Queue'
id, message = dataprep.save_workout_database(

4
rowers/management/commands/processemail.py
View File

@@ -150,7 +150,7 @@ def get_from_address(message):
except AttributeError: # pragma: no cover
pass
if "quiske" in first_line:
if "quiske" in first_line: # pragma: no cover
match = re.search(r'[\w\.-]+@[\w\.-]+', first_line)
return match.group(0)
@@ -186,7 +186,7 @@ class Command(BaseCommand):
else: # pragma: no cover
workoutmailbox = Mailbox.objects.get(name='workouts')
if 'failedmailbox' in options:
if 'failedmailbox' in options: # pragma: no cover
failedmailbox = Mailbox.objects.get(name=options['failedmailbox'])
else: # pragma: no cover
failedmailbox = Mailbox.objects.get(name='Failed')

4
rowers/metrics.py
View File

@@ -377,7 +377,7 @@ dtypes = {}
for name,d in rowingmetrics:
if d['numtype'] == 'float':
dtypes[name] = float
elif d['numtype'] == 'int':
elif d['numtype'] == 'int': # pragma: no cover
dtypes[name] = int
axesnew = [
@@ -479,7 +479,7 @@ This value should be fairly constant across all stroke rates.""",
def calc_trimp(df,sex,hrmax,hrmin,hrftp):
def calc_trimp(df,sex,hrmax,hrmin,hrftp): # pragma: no cover
if sex == 'male':
f = 1.92
else:

8
rowers/middleware.py
View File

@@ -44,10 +44,10 @@ class SurveyMiddleWare(object):
if request.user.is_authenticated and request.path not in allowed_paths:
r = getrower(request.user)
nexturl = request.path
if 'survey' in nexturl:
if 'survey' in nexturl: # pragma: no cover
nexturl = '/rowers/list-workouts'
mustseesurvey = request.user.date_joined <= timezone.now()-datetime.timedelta(days=14) and not r.surveydone
if mustseesurvey:
if mustseesurvey: # pragma: no cover
return redirect(
'/rowers/survey/?next=%s' % nexturl
)
@@ -64,7 +64,7 @@ class GDPRMiddleWare(object):
if request.user.is_authenticated and request.path not in allowed_paths:
r = getrower(request.user)
nexturl = request.path
if 'optin' in nexturl:
if 'optin' in nexturl: # pragma: no cover
nexturl = '/rowers/list-workouts'
if not r.gdproptin:
return redirect(
@@ -82,7 +82,7 @@ class RowerPlanMiddleWare(object):
def __call__(self, request):
if request.user.is_authenticated and request.user.rower.rowerplan not in ['basic','freecoach']:
if request.user.rower.paymenttype == 'single':
if request.user.rower.planexpires < timezone.now().date():
if request.user.rower.planexpires < timezone.now().date(): # pragma: no cover
messg = 'Your paid plan has expired. We have reset you to a free basic plan.'
messages.error(request,messg)
r = getrower(request.user)

12
rowers/nkimportutils.py
View File

@@ -12,7 +12,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)
@@ -51,7 +51,7 @@ def add_workout_from_data(userid,nkid,data,strokedata,source='nk',splitdata=None
totalDistance = totalDistanceGps
useImpeller = False
if speedInput:
if speedInput: # pragma: no cover
totdalDistance = totalDistanceImp
useImpeller = True
@@ -70,7 +70,7 @@ def add_workout_from_data(userid,nkid,data,strokedata,source='nk',splitdata=None
oarlockfirmware = oarlocksession["firmwareVersion"]
except KeyError:
oarlockfirmware = ''
else:
else: # pragma: no cover
boatName = ''
oarLength = 289
oarInboardLength = 88
@@ -103,7 +103,7 @@ def add_workout_from_data(userid,nkid,data,strokedata,source='nk',splitdata=None
response = session.post(UPLOAD_SERVICE_URL,json=uploadoptions)
if response.status_code != 200:
if response.status_code != 200: # pragma: no cover
return 0,response.text
try:
@@ -219,10 +219,6 @@ def get_nk_summary(workoutdata,strokedata):
return stri1
return stri1
def get_nk_allstats(data,workoutdata):
stri = get_nk_summary(data, workoutdata) + \
get_nk_intervalstats(data, workoutdata)

10
rowers/otw_power_calculator_pb2_grpc.py
View File

@@ -25,12 +25,12 @@ class PowerServicer(object):
"""Power service definition
"""
def CalcPower(self, request, context):
def CalcPower(self, request, context): # pragma: no cover
# missing associated documentation comment in .proto file
pass
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
context.set_code(grpc.StatusCode.UNIMPLEMENTED) # pragma: no cover
context.set_details('Method not implemented!') # pragma: no cover
raise NotImplementedError('Method not implemented!') # pragma: no cover
def add_PowerServicer_to_server(servicer, server):
@@ -40,7 +40,7 @@ def add_PowerServicer_to_server(servicer, server):
request_deserializer=otw__power__calculator__pb2.WorkoutPowerRequest.FromString,
response_serializer=otw__power__calculator__pb2.CalculationResult.SerializeToString,
),
}
} # pragma: no cover
generic_handler = grpc.method_handlers_generic_handler(
'otw_power_calculator.Power', rpc_method_handlers)
server.add_generic_rpc_handlers((generic_handler,))

14
rowers/permissions.py
View File

@@ -15,35 +15,35 @@ class IsOwnerOrReadOnly(permissions.BasePermission):
def has_object_permission(self, request, view, obj):
# Read permissions are allowed to any request,
# so we'll always allow GET, HEAD or OPTIONS requests.
if request.method in permissions.SAFE_METHODS:
if request.method in permissions.SAFE_METHODS: # pragma: no cover
return True
# Write permissions are only allowed to the owner of the snippet.
return obj.user == request.user
return obj.user == request.user # pragma: no cover
class IsOwnerOrNot(permissions.BasePermission):
def has_object_permission(self, request, view, obj):
def has_object_permission(self, request, view, obj): # pragma: no cover
r = Rower.objects.get(user=request.user)
return (obj.user == r)
class IsRowerOrNot(permissions.BasePermission):
def has_object_permission(self, request, view, obj):
def has_object_permission(self, request, view, obj): # pragma: no cover
r = Rower.objects.get(user=request.user)
return (r in obj.rower.all())
class IsPlanOrHigher(permissions.BasePermission):
def has_object_permission(self, request, view, obj):
def has_object_permission(self, request, view, obj): # pragma: no cover
r = Rower.objects.get(user=request.user)
return r not in ['basic','pro','freecoach']
class IsCompetitorOrNot(permissions.BasePermission):
def has_object_permission(self, request, view, obj):
def has_object_permission(self, request, view, obj): # pragma: no cover
return (obj.userid == request.user.id)
class IsManagerOrReadOnly(permissions.BasePermission):
def has_object_permission(self, request, view, obj):
def has_object_permission(self, request, view, obj): # pragma: no cover
if request.method in permissions.SAFE_METHODS:
return True

5
rowers/plots.py
View File

@@ -36,7 +36,7 @@ def y_axis_range(ydata,miny=0,padding=.1,ultimate=[-1e9,1e9]):
if (yrange == 0):
if (yrange == 0): # pragma: no cover
if ymin == 0:
yrangemin = -padding
else:
@@ -49,7 +49,7 @@ def y_axis_range(ydata,miny=0,padding=.1,ultimate=[-1e9,1e9]):
yrangemin = ymin-padding*yrange
yrangemax = ymax+padding*yrange
if (yrangemin < ultimate[0]):
if (yrangemin < ultimate[0]): # pragma: no cover
yrangemin = ultimate[0]
if (yrangemax > ultimate[1]):
@@ -104,4 +104,3 @@ def mkplot(row,title):
plt.subplots_adjust(hspace=0)
return fig

8
rowers/polarstuff.py
View File

@@ -99,7 +99,7 @@ def get_token(code): # pragma: no cover
return [thetoken,expires_in,user_id]
# Make authorization URL including random string
def make_authorization_url():
def make_authorization_url(): # pragma: no cover
# Generate a random string for the state parameter
# Save it for use later to prevent xsrf attacks
state = str(uuid4())
@@ -145,7 +145,7 @@ def get_polar_notifications():
from rowers.rower_rules import ispromember
def get_all_new_workouts(available_data,testing=False):
for record in available_data:
for record in available_data: # pragma: no cover
if testing:
print(record)
if record['data-type'] == 'EXERCISE':
@@ -159,7 +159,7 @@ def get_all_new_workouts(available_data,testing=False):
except Rower.DoesNotExist:
pass
return 1
return 1 # pragma: no cover
def get_polar_workouts(user):
@@ -245,7 +245,7 @@ def get_polar_workouts(user):
# commit transaction
requests.put(url, headers=headers)
return exercise_list
return exercise_list # pragma: no cover
def get_polar_user_info(user,physical=False): # pragma: no cover
r = Rower.objects.get(user=user)

8
rowers/rowing_workout_metrics_pb2_grpc.py
View File

@@ -28,12 +28,12 @@ class MetricsServicer(object):
def CalcMetrics(self, request, context):
# missing associated documentation comment in .proto file
pass
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
context.set_code(grpc.StatusCode.UNIMPLEMENTED) # pragma: no cover
context.set_details('Method not implemented!') # pragma: no cover
raise NotImplementedError('Method not implemented!') # pragma: no cover
def add_MetricsServicer_to_server(servicer, server):
def add_MetricsServicer_to_server(servicer, server): # pragma: no cover
rpc_method_handlers = {
'CalcMetrics': grpc.unary_unary_rpc_method_handler(
servicer.CalcMetrics,

4
rowers/rp3stuff.py
View File

@@ -97,7 +97,7 @@ def get_token(code): # pragma: no cover
return thetoken,expires_in,refresh_token
# Make authorization URL including random string
def make_authorization_url(request):
def make_authorization_url(request): # pragma: no cover
return imports_make_authorization_url(oauth_data)
@@ -217,7 +217,7 @@ def get_rp3_workout_link(user,workout_id,waittime=3,max_attempts=20): # pragma:
return get_rp3_workout_token(workout_id,auth_token,waittime=waittime,max_attempts=max_attempts)
def get_rp3_workout(user,workout_id,startdatetime=None):
def get_rp3_workout(user,workout_id,startdatetime=None): # pragma: no cover
url = get_rp3_workout_link(user,workout_id)
filename = 'media/RP3Import_'+str(workout_id)+'.csv'

4
rowers/runkeeperstuff.py
View File

@@ -108,7 +108,7 @@ def get_workout(user,runkeeperid,do_async=False):
try:
data = s.json()
except ValueError:
except ValueError: # pragma: no cover
data = {}
return data,"Something went wrong with the workout import"
@@ -429,7 +429,7 @@ def add_workout_from_data(user,importid,data,strokedata,source='runkeeper',
latseries = pd.Series(latcoord,index=times_location)
try:
latseries = latseries.groupby(latseries.index).first()
except TypeError:
except TypeError: # pragma: no cover
latseries = 0.0*distseries
lonseries = pd.Series(loncoord,index=times_location)

6
rowers/scoring.py
View File

@@ -102,7 +102,7 @@ def save_scoring(name,user,filename,id=0,notes=""):
try:
boattype = row['BoatType']
except KeyError:
except KeyError: # pragma: no cover
boattype = '1x'
try:
@@ -113,7 +113,7 @@ def save_scoring(name,user,filename,id=0,notes=""):
sex = 'mixed'
else:
sex = 'female'
except KeyError:
except KeyError: # pragma: no cover
sex = 'female'
try:
@@ -135,7 +135,7 @@ def save_scoring(name,user,filename,id=0,notes=""):
try:
skillclass = row['SkillClass']
except KeyError:
except KeyError: # pragma: no cover
skillclass = 'Open'
# finding existing standard

2
rowers/serializers.py
View File

@@ -226,7 +226,7 @@ class WorkoutSerializer(serializers.ModelSerializer):
return Workout.objects.create(**validated_data)
def update(self, instance, validated_data):
def update(self, instance, validated_data): # pragma: no cover
d = validated_data['date']
t = validated_data['starttime']
rowdatetime = datetime.datetime(d.year,

6
rowers/underarmourstuff.py
View File

@@ -214,7 +214,7 @@ def createunderarmourworkoutdata(w):
}
if haslatlon:
if haslatlon: # pragma: no cover
timeseries["position"] = locdata
data = {
@@ -268,7 +268,7 @@ def refresh_ua_actlist(user): # pragma: no cover
try:
activities = pd.read_csv('static/rigging/ua2.csv',index_col='id')
actdict = activities.to_dict()['Name']
except:
except: # pragma: no cover
actdict = {}
@@ -276,7 +276,7 @@ def get_typefromid(typeid,user):
r = Rower.objects.get(user=user)
try:
res = actdict[int(typeid)]
except KeyError:
except KeyError: # pragma: no cover
authorizationstring = str('Bearer ' + r.underarmourtoken)
headers = {'Authorization': authorizationstring,
'Api-Key': UNDERARMOUR_CLIENT_KEY,

2
rowers/urls.py
View File

@@ -130,7 +130,7 @@ class EntryViewSet(viewsets.ModelViewSet):
model = VirtualRaceResult
serializer_class = EntrySerializer
def get_queryset(self):
def get_queryset(self): # pragma: no cover
try:
return VirtualRaceResult.objects.filter(userid=self.request.user.id)
except TypeError:

6
rowers/views/apiviews.py
View File

@@ -71,7 +71,7 @@ def strokedataform_v2(request,id=0):
'teams':get_my_teams(request.user),
'id':id,
'workout':w,
})
}) # pragma: no cover
# Process the POSTed stroke data according to the API definition
@@ -96,7 +96,7 @@ def strokedatajson_v2(request,id):
try:
id = int(id)
except ValueError:
except ValueError: # pragma: no cover
return HttpResponse("Not a valid workout number",status=404)
if request.method == 'GET':
@@ -127,7 +127,7 @@ def strokedatajson_v2(request,id):
logfile.write("\n")
except KeyError:
logfile.write("No data in request.data\n")
except (AttributeError,TypeError):
except (AttributeError,TypeError): # pragma: no cover
logfile.write("No data in request\n")
checkdata, r = dataprep.getrowdata_db(id=row.id)
if not checkdata.empty: # pragma: no cover

4
rowers/views/errorviews.py
View File

@@ -11,7 +11,7 @@ from django.test import SimpleTestCase, override_settings
from django.urls import path
def servererror_view(request):
def servererror_view(request): # pragma: no cover
raise ValueError
# Custom error pages with Rowsandall headers
@@ -36,7 +36,7 @@ def error400_view(request, exception):
response.status_code = 400
return response
def error403_view(request,*args, **kwargs):
def error403_view(request,*args, **kwargs): # pragma: no cover
response = render(request,'403.html', {},status=403)
# context_instance = RequestContext(request))

12
rowers/views/exportviews.py
View File

@@ -74,16 +74,16 @@ def plannedsessions_coach_icsemail_view(request,userid=0):
sps = get_sessions_manager(request.user,teamid=0,
enddate=enddate,
startdate=startdate)
else:
else: # pragma: no cover
rteams = therower.team.filter(viewing='allmembers')
sps = get_sessions(therower,startdate=startdate,enddate=enddate)
if therower.rowerplan != 'freecoach':
rowers = [therower]
else:
else: # pragma: no cover
rowers = []
for ps in sps:
for ps in sps: # pragma: no cover
if 'coach' in request.user.rower.rowerplan:
rowers += ps.rower.all().exclude(rowerplan='freecoach')
else:
@@ -138,7 +138,7 @@ def plannedsessions_coach_icsemail_view(request,userid=0):
@login_required()
def course_kmldownload_view(request,id=0):
r = getrower(request.user)
if r.emailbounced:
if r.emailbounced: # pragma: no cover
message = "Please check your email address first. Email to this address bounced."
messages.error(request,message)
return HttpResponseRedirect(
@@ -190,7 +190,7 @@ def workout_gpxemail_view(request,id=0):
@login_required()
def workouts_summaries_email_view(request):
r = getrower(request.user)
if r.emailbounced:
if r.emailbounced: # pragma: no cover
message = "Please check your email address first. Email to this address bounced."
messages.error(request, message)
return HttpResponseRedirect(
@@ -259,7 +259,7 @@ def workout_csvemail_view(request,id=0):
# Get Workout CSV file and send it to user's email address
@login_required()
@permission_required('rower.is_staff',fn=get_user_by_userid,raise_exception=True)
def workout_csvtoadmin_view(request,id=0):
def workout_csvtoadmin_view(request,id=0): # pragma: no cover
message = ""
r = getrower(request.user)
w = get_workout(id)

18
rowers/views/otherviews.py
View File

@@ -16,7 +16,7 @@ def download_fit(request,filename=''):
pss = PlannedSession.objects.filter(fitfile=filename)
if len(pss) != 1:
if len(pss) != 1: # pragma: no cover
raise Http404("Could not find the required file")
ps = pss[0]
@@ -24,7 +24,7 @@ def download_fit(request,filename=''):
if ps.manager == request.user or request.user.rower in ps.rower.all():
owns = True
if not owns:
if not owns: # pragma: no cover
raise PermissionDenied("You are not allowed to download this file")
fitfile = ps.fitfile
@@ -33,21 +33,21 @@ def download_fit(request,filename=''):
except FileNotFoundError:
raise Http404("File not found")
response['Content-Disposition'] = 'attachment; filename="%s"' % filename
response['Content-Type'] = 'application/octet-stream'
response['Content-Disposition'] = 'attachment; filename="%s"' % filename # pragma: no cover
response['Content-Type'] = 'application/octet-stream' # pragma: no cover
return response
return response # pragma: no cover
@login_required()
def failed_queue_view(request):
if not request.user.is_staff:
if not request.user.is_staff: # pragma: no cover
raise PermissionDenied("Not Allowed")
q = Queue('failed', connection=Redis())
resultslist = []
for job in q.jobs:
for job in q.jobs: # pragma: no cover
traceback = str(job.exc_info)
@@ -82,7 +82,7 @@ def failed_queue_empty(request):
@login_required()
def failed_job_view(request,id=0):
if not request.user.is_staff:
if not request.user.is_staff: # pragma: no cover
raise PermissionDenied("Not Allowed")
q = Queue('failed', connection=Redis())
@@ -94,7 +94,7 @@ def failed_job_view(request,id=0):
@login_required()
def errormessage_view(request,errormessage='aap'):
def errormessage_view(request,errormessage='aap'): # pragma: no cover
if (errormessage=='3dsecure'):
errormessage = '3D Secure Card Verification Error. Please check your card details.'
messages.error(request,errormessage)

4
rowers/weather.py
View File

@@ -99,8 +99,8 @@ def get_metar_data(airportcode,unixtime):
return [wind_ms,windbearing,message,rawtext,timestamp]
message = 'Failed to download METAR data'
return [0,0,message,'',timestamp]
message = 'Failed to download METAR data' # pragma: no cover
return [0,0,message,'',timestamp] # pragma: no cover
# Get wind data (and translate from knots to m/s)

34
rowsandall_app/settings.py
View File

@@ -265,11 +265,11 @@ LOGOUT_REDIRECT_URL = '/'
PROGRESS_CACHE_SECRET = CFG['progress_cache_secret']
try:
UPLOAD_SERVICE_URL = CFG['upload_service_url']
except KeyError:
except KeyError: # pragma: no cover
UPLOAD_SERVICE_URL = "http://localhost:8000/rowers/workout/api/upload/"
try:
UPLOAD_SERVICE_SECRET = CFG['upload_service_secret']
except KeyError:
except KeyError: # pragma: no cover
UPLOAD_SERVICE_SECRET = "FoYezZWLSyfAVimumpHEeYsJjsNCerxV"
# Concept 2
@@ -507,75 +507,75 @@ except KeyError:
try:
BRAINTREE_MERCHANT_ID = CFG['braintree_merchant_id']
except KeyError:
except KeyError: # pragma: no cover
BRAINTREE_MERCHANT_ID = ''
try:
BRAINTREE_MERCHANT_ACCOUNT_ID = CFG['braintree_merchant_account_id']
except KeyError:
except KeyError: # pragma: no cover
BRAINTREE_MERCHANT_ACCOUNT_ID = 'rowsandallEUR'
try:
BRAINTREE_PUBLIC_KEY = CFG['braintree_public_key']
except KeyError:
except KeyError: # pragma: no cover
BRAINTREE_PUBLIC_KEY = ''
try:
BRAINTREE_PRIVATE_KEY = CFG['braintree_private_key']
except KeyError:
except KeyError: # pragma: no cover
BRAINTREE_PRIVATE_KEY = ''
try:
BRAINTREE_SANDBOX_MERCHANT_ID = CFG['braintree_sandbox_merchant_id']
except KeyError:
except KeyError: # pragma: no cover
BRAINTREE_SANDBOX_MERCHANT_ID = ''
try:
BRAINTREE_SANDBOX_PUBLIC_KEY = CFG['braintree_sandbox_public_key']
except KeyError:
except KeyError: # pragma: no cover
BRAINTREE_SANDBOX_PUBLIC_KEY = ''
try:
BRAINTREE_SANDBOX_PRIVATE_KEY = CFG['braintree_sandbox_private_key']
except KeyError:
except KeyError: # pragma: no cover
BRAINTREE_SANDBOX_PRIVATE_KEY = ''
try:
PAYMENT_PROCESSING_ON = CFG['payment_processing_on']
except KeyError:
except KeyError: # pragma: no cover
PAYMENT_PROCESSING_ON = False
try:
FAKTUROID_API_KEY = CFG['fakturoid_api_key']
except KeyError:
except KeyError: # pragma: no cover
FAKTUROID_API_KEY = ''
try:
FAKTUROID_EMAIL = CFG['fakturoid_email']
except KeyError:
except KeyError: # pragma: no cover
FAKTUROID_EMAIL = ''
try:
FAKTUROID_SLUG = CFG['fakturoid_slug']
except KeyError:
except KeyError: # pragma: no cover
FAKTUROID_SLUG = ''
# ID obfuscation
try:
OPAQUE_SECRET_KEY = CFG['opaque_secret_key']
except KeyError:
except KeyError: # pragma: no cover
OPAQUE_SECRET_KEY = 0xa193443a
# Celery or RQ
try:
CELERY = CFG['use_celery']
except KeyError:
except KeyError: # pragma: no cover
CELERY = False
try:
WORKOUTS_FIT_TOKEN = CFG['workouts_fit_token']
WORKOUTS_FIT_URL = CFG['workouts_fit_url']
except KeyError:
except KeyError: # pragma: no cover
WORKOUTS_FIT_TOKEN = 'aapnootmies'
WORKOUTS_FIT_URL = 'http://localhost:50053/tojson'
@@ -584,7 +584,7 @@ except KeyError:
try:
RECAPTCHA_SITE_KEY = CFG['recaptcha_site_key']
RECAPTCHA_SITE_SECRET = CFG['recaptcha_site_secret']
except KeyError:
except KeyError: # pragma: no cover
RECAPTCHA_SITE_KEY = ''
RECAPTCHA_SITE_SECRET = ''

2
rowsandall_app/urls.py
View File

@@ -100,7 +100,7 @@ urlpatterns += [
# monkey patch workaround for bug in recurrence library
django.views.i18n.javascript_catalog = None
if settings.DEBUG:
if settings.DEBUG: # pragma: no cover
import debug_toolbar
import django
urlpatterns += [

4
rowsandall_app/views.py
View File

@@ -16,12 +16,12 @@ def landingview(request):
'landingpage.html',
)
def logoview(request):
def logoview(request): # pragma: no cover
image_data = open(settings.STATIC_ROOT+"/img/apple-icon-144x144.png", "rb").read()
return HttpResponse(image_data, content_type="image/png")
def rootview(request):
def rootview(request): # pragma: no cover
magicsentence = rmain()
loginform = LoginForm()