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

removing imports.py and increasing coverage

Browse Source
This commit is contained in:
Sander Roosendaal
2023-09-21 16:09:48 +02:00
parent b1d4534ff2
commit ddac7ef18f
15 changed files with 177 additions and 406 deletions
Download Patch File Download Diff File Expand all files Collapse all files

138
rowers/tests/mocks.py
View File

@@ -57,7 +57,7 @@ from redis import StrictRedis
redis_connection = StrictRedis()
def mocked_grpc(*args, **kwargs):
def mocked_grpc(*args, **kwargs): # pragma: no cover
class insecure_channel:
def __init__(*args,**kwargs):
pass
@@ -84,7 +84,7 @@ def mocked_grpc(*args, **kwargs):
def mocked_send_template_email(*args,**kwargs):
return 1
def mocked_myqueue(*args, **kwargs):
def mocked_myqueue(*args, **kwargs): # pragma: no cover
class Job:
def __init__(self,*args, **kwargs):
self.result = 1
@@ -95,7 +95,7 @@ def mocked_myqueue(*args, **kwargs):
return Job()
def mock_c2open(*args, **kwargs):
def mock_c2open(*args, **kwargs): # pragma: no cover
return('aap')
def mocked_session(*args, **kwargs):
@@ -124,7 +124,7 @@ def mocked_session(*args, **kwargs):
def json(self):
return self.json_data
class MockContentResponse:
class MockContentResponse: # pragma: no cover
def __init__(self,filename,status_code):
with open(filename,'rb') as f:
s = f.read()
@@ -135,7 +135,7 @@ def mocked_session(*args, **kwargs):
return MockEngine()
def mocked_sqlalchemy(*args, **kwargs):
def mocked_sqlalchemy(*args, **kwargs): # pragma: no cover
# return object with method
class MockEngine:
@@ -177,7 +177,7 @@ def mocked_sqlalchemy(*args, **kwargs):
from rowers import courses
def mocked_sqlalchemy_courses(*args, **kwargs):
def mocked_sqlalchemy_courses(*args, **kwargs): # pragma: no cover
# return object with method
cs = courses.kmltocourse('rowers/tests/testdata/thyro.kml')
course = cs[0]
@@ -251,7 +251,7 @@ class DjangoTestCase(TestCase): #, MockTestCase):
#MockTestCase.tearDown(self)
# delete_strokedata(1)
def mocked_tcx_parser(*args, **kwargs):
def mocked_tcx_parser(*args, **kwargs): # pragma: no cover
df = pd.read_csv('rowers/tests/testdata/fake_strokedata.csv')
return rowingdata(df=df)
@@ -272,12 +272,12 @@ def mocked_fetchcperg(*args, **kwargs):
return df
import pandas as pd
def mocked_read_df_sql(id):
def mocked_read_df_sql(id): # pragma: no cover
df = pd.read_csv('rowers/tests/testdata/fake_strokedata.csv')
return df
def mocked_sendmail(*args,**kwargs):
def mocked_sendmail(*args,**kwargs): # pragma: no cover
return HttpResponseRedirect('/rowers/email/')
def mocked_get_video_data(*args, **kwargs):
@@ -299,7 +299,7 @@ def mocked_getrowdata_db(*args, **kwargs):
return df,row
def mocked_getrowdata_uh(*args, **kwargs):
def mocked_getrowdata_uh(*args, **kwargs): # pragma: no cover
df = pd.read_csv('rowers/tests/testdata/uhfull.csv')
id = kwargs['id']
@@ -308,7 +308,7 @@ def mocked_getrowdata_uh(*args, **kwargs):
return df, row
def mocked_getsmallrowdata_uh(*args, **kwargs):
def mocked_getsmallrowdata_uh(*args, **kwargs): # pragma: no cover
df = pd.read_csv('rowers/tests/testdata/uhfull.csv')
return df
@@ -335,33 +335,33 @@ def mocked_getsmallrowdata_db(*args, **kwargs):
return df
def mocked_getsmallrowdata_db_updatecp(*args, **kwargs):
def mocked_getsmallrowdata_db_updatecp(*args, **kwargs): # pragma: no cover
df = pd.read_csv('rowers/tests/testdata/colsfromdb.csv')
return df
def mocked_getsmallrowdata_db_setcp(*args, **kwargs):
def mocked_getsmallrowdata_db_setcp(*args, **kwargs): # pragma: no cover
df = pd.read_csv('rowers/tests/testdata/colsfromdb2.csv')
return df
def mocked_getsmallrowdata_db_water(*args, **kwargs):
def mocked_getsmallrowdata_db_water(*args, **kwargs): # pragma: no cover
df = pd.read_csv('rowers/tests/testdata/colsfromdb3.csv')
return df
def mocked_getsmallrowdata_db_wps(*args, **kwargs):
def mocked_getsmallrowdata_db_wps(*args, **kwargs): # pragma: no cover
df = pd.read_csv('rowers/tests/testdata/driveenergies.csv')
return df
def mocked_getpowerdata_db(*args, **kwargs):
def mocked_getpowerdata_db(*args, **kwargs): # pragma: no cover
df = pd.read_csv('rowers/tests/testdata/fake_powerdata.csv')
return df
def mock_for_interactive_chart(*args, **kwargs):
def mock_for_interactive_chart(*args, **kwargs): # pragma: no cover
df = pd.read_csv('rowers/tests/testdata/interactivechart.csv')
return df
@@ -371,7 +371,7 @@ def mocked_getempowerdata_db(*args, **kwargs):
return df
def mocked_read_df_cols_sql_multistats(ids,columns,convertnewtons=True):
def mocked_read_df_cols_sql_multistats(ids,columns,convertnewtons=True): # pragma: no cover
df = pd.read_csv('rowers/tests/testdata/cumstats.csv')
extracols = []
@@ -387,13 +387,13 @@ def mock_workout_summaries(*args, **kwargs):
df = pd.read_csv('rowers/tests/testdata/workout_summaries.csv')
return df
def mocked_read_df_cols_sql_multi(ids, columns, convertnewtons=True):
def mocked_read_df_cols_sql_multi(ids, columns, convertnewtons=True): # pragma: no cover
df = pd.read_csv('rowers/tests/testdata/fake_strokedata2.csv')
extracols = []
return df, extracols
def mocked_read_df_cols_sql_multiflex(ids, columns, convertnewtons=True):
def mocked_read_df_cols_sql_multiflex(ids, columns, convertnewtons=True): # pragma: no cover
df = pd.read_csv('rowers/tests/testdata/multiflexdata.csv')
extracols = []
@@ -402,11 +402,11 @@ def mocked_read_df_cols_sql_multiflex(ids, columns, convertnewtons=True):
# Mocked Strava
def mocked_stravaexport(f2,workoutname,stravatoken,description='',
activity_type='Rowing'):
activity_type='Rowing'): # pragma: no cover
print("this is mocked strava export")
return 1,'success'
def mocked_fetchcp(*args, **kwargs):
def mocked_fetchcp(*args, **kwargs): # pragma: no cover
df = pd.read_csv('rowers/tests/testdata/otwcpresult.csv')
delta = df['delta']
cpvalue = df['cpvalue']
@@ -426,14 +426,14 @@ def mocked_fetchcp(*args, **kwargs):
return delta, cpvalue, avgpower
def mocked_getcpdata_sql(*args, **kwargs):
def mocked_getcpdata_sql(*args, **kwargs): # pragma: no cover
df = pd.read_csv('rowers/tests/testdata/otwcpresult.csv')
df['cp'] = df['cpvalue']
return df
def mocked_cpraw(*args, **kwargs):
def mocked_cpraw(*args, **kwargs): # pragma: no cover
df = pd.read_csv('rowers/tests/testdata/otwcp_df.csv')
return df
@@ -465,17 +465,17 @@ class MockStravalibClient():
## Higher level - unfortunately didn't succeed in mocking the gateway
def get_client_token(*args, **kwargs):
def get_client_token(*args, **kwargs): # pragma: no cover
return "aap"
def mock_create_customer(*args, **kwargs):
return 121
def mock_make_payment(*args, **kwargs):
def mock_make_payment(*args, **kwargs): # pragma: no cover
return 15,''
def mock_update_subscription(*args, **kwargs):
try:
try: # pragma: no cover
rower = args[0]
data = args[1]
planid = data['plan']
@@ -497,7 +497,7 @@ def mock_update_subscription(*args, **kwargs):
def mock_create_subscription(*args, **kwargs):
return mock_update_subscription(*args, **kwargs)
def mock_cancel_subscription(*args, **kwargs):
def mock_cancel_subscription(*args, **kwargs): # pragma: no cover
themessages = []
errormessages = []
try:
@@ -521,7 +521,7 @@ def mock_mocktest(*args, **kwargs):
## Gateway stuff (not working)
class gatewayresult():
class gatewayresult(): # pragma: no cover
def __init__(self,*args,**kwargs):
self.is_success = kwargs.pop('is_success',True)
self.customer_id = 1
@@ -533,22 +533,22 @@ class gatewayresult():
def __unicode__():
return "mockedgatewayresult"
class credit_card():
class credit_card(): # pragma: no cover
def __init__(self, *args, **kwargs):
self.subscriptions = [vsubscription()]
self.country_of_issuance = 'US'
class paypal_account():
class paypal_account(): # pragma: no cover
def __init__(self, *args, **kwargs):
self.subscriptions = [vsubscription(),vsubscription()]
class customercreateresult:
class customercreateresult: # pragma: no cover
def __init__(self, *args, **kwargs):
self.customer = kwargs.pop('customer',customer())
self.is_success = kwargs.pop('is_success',True)
self.customer_id = 1
class customer():
class customer(): # pragma: no cover
def find(*arg, **kwargs):
return self
@@ -560,17 +560,17 @@ class customer():
self.paypal_accounts = [paypal_account()]
self.id = 1
class client_token():
class client_token(): # pragma: no cover
def generate(*args, **kwargs):
return 'aapnooit'
class plan():
class plan(): # pragma: no cover
def all(*args, **kwargs):
return []
class transaction():
class transaction(): # pragma: no cover
def sale(*args, **kwargs):
return gatewayresult(is_success=True)
@@ -588,7 +588,7 @@ class transaction():
self.created_at = datetime.datetime.now()
self.currency_iso_code = 'EUR'
class vtransaction():
class vtransaction(): # pragma: no cover
def __init__(self,*args, **kwargs):
self.amount = 15
self.credit_card_details = credit_card()
@@ -600,7 +600,7 @@ class vtransaction():
self.created_at = datetime.datetime.now()
self.currency_iso_code = 'EUR'
class vsubscription():
class vsubscription(): # pragma: no cover
def update(*args, **kwargs):
return gatewayresult(is_success=True)
@@ -615,7 +615,7 @@ class vsubscription():
self.price = 15
self.never_expires = True
class subscription():
class subscription(): # pragma: no cover
def create(*args, **kwargs):
return gatewayresult(is_success=True)
@@ -634,7 +634,7 @@ class subscription():
self.price = 15
self.never_expires = True
class vpayment_method():
class vpayment_method(): # pragma: no cover
def create(*args, **kwargs):
return gatewayresult()
@@ -642,25 +642,25 @@ class vpayment_method():
self.token = 'liesjeleerdelotje'
class payment_method():
class payment_method(): # pragma: no cover
def create(*args, **kwargs):
return gatewayresult()
def __init__(self, *args, **kwargs):
self.token = 'liesjeleerdelotje'
class notification():
class notification(): # pragma: no cover
def __init__(self, *args, **kwargs):
print('notifucation')
self.kind = 'subscription_canceled'
class webhook_notification():
class webhook_notification(): # pragma: no cover
def parse(*args, **kwargs):
print(args,kwargs,'parse')
return notification()
# mock braintree gateway
class MockBraintreeGateway:
class MockBraintreeGateway: # pragma: no cover
def __init__(self,*args, **kwargs):
self.customer = customer()
self.client_token = client_token()
@@ -671,13 +671,13 @@ class MockBraintreeGateway:
self.webhook_notification = webhook_notification()
def mocked_gateway(*args, **kwargs):
def mocked_gateway(*args, **kwargs): # pragma: no cover
return MockBraintreeGateway()
# Mocked Rowingdata
class mocked_rowingdata(rowingdata):
class mocked_rowingdata(rowingdata): # pragma: no cover
def __init__(self, *args, **kwargs):
super(mocked_rowingdata).__init__(*args, **kwargs)
@@ -938,19 +938,19 @@ def mocked_requests(*args, **kwargs):
class MockHeaderResponse:
class MockHeaderResponse: # pragma: no cover
def __init__(self, header_data, status_code):
self.headers = header_data
self.status_code = status_code
class MockStreamResponse:
class MockStreamResponse: # pragma: no cover
def __init__(self, file_name, status_code):
self.raw = open(file_name,'rb')
self.status_code = status_code
self.ok = True
self.file_name = file_name
def __enter__(self):
def __enter__(self): # pragma: no cover
return self
def __exit__(self, exc_type, exc_value, traceback):
@@ -977,7 +977,7 @@ def mocked_requests(*args, **kwargs):
class MockSession:
class headers:
def __init__(self,*args,**kwargs):
def __init__(self,*args,**kwargs): # pragma: no cover
pass
def update(self,*args,**kwargs):
@@ -1013,7 +1013,7 @@ def mocked_requests(*args, **kwargs):
args = [kwargs['url']]
if "tofit" in kwargs['url']:
args = [kwargs['url']]
if "tojson" in kwargs['url']:
if "tojson" in kwargs['url']: # pragma: no cover
args = [kwargs['url']]
if not args:
@@ -1156,7 +1156,7 @@ def mocked_requests(*args, **kwargs):
garmintrainingscheduletester = re.compile(garmintrainingscheduleregex)
if garmintester.match(args[0]):
if garmindownloadtester.match(args[0]):
if garmindownloadtester.match(args[0]): # pragma: no cover
return MockStreamResponse('rowers/tests/testdata/3x250m.fit',200)
if garmintrainingtester.match(args[0]):
json_data = {
@@ -1222,10 +1222,10 @@ def mocked_requests(*args, **kwargs):
'refresh_token': 'jHJhFzCfOOKB8oyiayubhLAlxaMkG3ruC1E8YxaR'
}
return MockResponse(json_data,200)
elif tpuploadtester.match(args[0]):
elif tpuploadtester.match(args[0]): # pragma: no cover
return MockResponse(tpuploadresponse,200)
if uaapitester.match(args[0]):
if uaapitester.match(args[0]): # pragma: no cover
if 'access_token' in args[0]:
json_data = {
'access_token': 'TA3n1vrNjuQJWw0TdCDHnjSmrjIPULhTlejMIWqq',
@@ -1244,7 +1244,7 @@ def mocked_requests(*args, **kwargs):
return MockResponse(uauserjson,200)
if uatester.match(args[0]):
if uatester.match(args[0]): # pragma: no cover
if 'access_token' in args[0]:
json_data = {
'access_token': 'TA3n1vrNjuQJWw0TdCDHnjSmrjIPULhTlejMIWqq',
@@ -1254,7 +1254,7 @@ def mocked_requests(*args, **kwargs):
return MockResponse(json_data,200)
if rktester.match(args[0]):
if rktester.match(args[0]): # pragma: no cover
if 'token' in args[0]:
json_data = {
'access_token': 'TA3n1vrNjuQJWw0TdCDHnjSmrjIPULhTlejMIWqq',
@@ -1272,7 +1272,7 @@ def mocked_requests(*args, **kwargs):
else:
json_data = rkworkoutlistjson
return MockResponse(json_data,200)
elif rkusertester.match(args[0]):
elif rkusertester.match(args[0]): # pragma: no cover
json_data = {
"userID": 1234567890,
"profile": "/profile",
@@ -1289,10 +1289,10 @@ def mocked_requests(*args, **kwargs):
"team": "/team"
}
return MockResponse(json_data, 200)
elif rkstrokestester.match(args[0]):
elif rkstrokestester.match(args[0]): # pragma: no cover
return MockResponse(rkstrokesjson,200)
if sttester.match(args[0]):
if sttester.match(args[0]): # pragma: no cover
if 'oauth2/token' in args[0]:
json_data = {
'access_token': 'TA3n1vrNjuQJWw0TdCDHnjSmrjIPULhTlejMIWqq',
@@ -1327,13 +1327,13 @@ def mocked_requests(*args, **kwargs):
if nkstrokestester.match(args[0]):
params = kwargs.pop('params',{})
if 'sessionIds' in params and params['sessionIds'] == '404':
if 'sessionIds' in params and params['sessionIds'] == '404': # pragma: no cover
return MockResponse(nkimpellerstrokedata, 200)
return MockResponse(nkstrokedata,200)
if nkworkoutlisttester.match(args[0]):
params = kwargs.pop('params',{})
if 'after' in params and params['after'] == '1267049972000':
if 'after' in params and params['after'] == '1267049972000': # pragma: no cover
return MockResponse(nkimpellerworkoutlist,200)
return MockResponse(nkworkoutlist,200)
@@ -1357,7 +1357,7 @@ def mocked_requests(*args, **kwargs):
query = kwargs['json']['query']
if 'download' in query:
return MockResponse(rp3linkready,200)
except KeyError:
except KeyError: # pragma: no cover
pass
json_data = rp3workoutlist
return MockResponse(json_data,200)
@@ -1392,13 +1392,13 @@ def mocked_requests(*args, **kwargs):
if c2strokestester.match(args[0]):
return MockResponse(c2strokedata,200)
elif c2importtester.match(args[0]):
if '12' in args[0]:
if '12' in args[0]: # pragma: no cover
return MockResponse(c2workoutdata,200)
elif '31' in args[0]:
return MockResponse(c2timezoneworkoutdata2,200)
elif '32' in args[0]:
elif '32' in args[0]: # pragma: no cover
return MockResponse(c2timezoneworkoutdatabad,200)
else:
else: # pragma: no cover
return MockResponse(c2timezoneworkoutdata,200)
elif c2workoutlisttester.match(args[0]):
return MockResponse(c2workoutlist,200)
@@ -1417,13 +1417,13 @@ def mocked_requests(*args, **kwargs):
}
}
return MockResponse(json_data,200)
elif 'results' in args[0]:
elif 'results' in args[0]: # pragma: no cover
json_data = {
'data': {
'id': 1223,
}
}
else:
else: # pragma: no cover
return MockResponse(c2workoutdata,200)
if fakturoidtester.match(args[0]):
@@ -1451,7 +1451,7 @@ class MockEmailMessage:
def send(self):
return 1
class MockResponse:
class MockResponse: # pragma: no cover
def __init__(self, json_data, status_code):
self.json_data = json_data
self.status_code = status_code
@@ -1465,7 +1465,7 @@ class MockResponse:
class MockOAuth1Session:
class MockOAuth1Session: # pragma: no cover
def __init__(self,*args, **kwargs):
pass