Private
Public Access
1
0

adding some tests for garmin training api

This commit is contained in:
Sander Roosendaal
2021-05-17 09:07:04 +02:00
parent e88553a04c
commit d195f03a55
3 changed files with 82 additions and 10 deletions

View File

@@ -133,7 +133,7 @@ def garmin_processcallback(redirect_response,resource_owner_key,resource_owner_s
verifier = oauth_response.get('oauth_verifier')
token = oauth_response.get('oauth_token')
access_token_url = 'https://connectapi.garmin.com/oauth-service/oauth/access_token'
print(token,access_token_url)
# Using OAuth1Session
garmin = OAuth1Session(oauth_data['client_id'],
@@ -193,9 +193,9 @@ def get_garmin_workout_list(user): # pragma: no cover
return result
def garmin_can_export_session(user): # pragma: no cover
def garmin_can_export_session(user):
if user.rower.rowerplan not in ['coach','plan']:
return False
return False # pragma: no cover
result = get_garmin_permissions(user)
if 'WORKOUT_IMPORT' in result:
return True
@@ -215,10 +215,10 @@ def step_to_garmin(step,order=0):
except KeyError:
intensity = None
#durationvaluetype = ''
if durationtype == 'Time': # pragma: no cover
if durationtype == 'Time':
durationtype = 'TIME'
durationvalue = int(durationvalue/1000.)
elif durationtype == 'Distance': # pragma: no cover
elif durationtype == 'Distance':
durationtype = 'DISTANCE'
durationvalue = int(durationvalue/100)
durationvaluetype = 'METER'
@@ -375,7 +375,7 @@ def ps_to_garmin(ps,r):
def get_garmin_permissions(user): # pragma: no cover
def get_garmin_permissions(user):
r = Rower.objects.get(user=user)
if (r.garmintoken == '') or (r.garmintoken is None):
s = "Token doesn't exist. Need to authorize"
@@ -399,7 +399,7 @@ def get_garmin_permissions(user): # pragma: no cover
return []
def garmin_session_create(ps,user): # pragma: no cover
def garmin_session_create(ps,user):
if not ps.steps:
return 0
if not garmin_can_export_session(user):
@@ -427,15 +427,13 @@ def garmin_session_create(ps,user): # pragma: no cover
response = requests.post(url,auth=garminheaders,json=payload)
if response.status_code != 200:
return 0
garmin_workout_id = response.json()['workoutId']
url = 'http://apis.garmin.com/training-api/schedule'
url = 'https://apis.garmin.com/training-api/schedule'
payload = {
'workoutId': garmin_workout_id,
@@ -444,6 +442,7 @@ def garmin_session_create(ps,user): # pragma: no cover
response = requests.post(url,auth=garminheaders,json=payload)
if response.status_code != 200:
return 0

View File

@@ -874,6 +874,10 @@ def mocked_requests(*args, **kwargs):
return MockResponse(json_data,200)
if len(args)==1 and 'userPermissions' in args[0]:
json_data = ['WORKOUT_IMPORT','ACTIVITY_EXPORT']
return MockResponse(json_data,200)
if 'garmin' in args:
return MockOAuth1Session()
@@ -882,6 +886,8 @@ def mocked_requests(*args, **kwargs):
args = [kwargs['url']]
if "tofit" in kwargs['url']:
args = [kwargs['url']]
if "tojson" in kwargs['url']:
args = [kwargs['url']]
if not args:
return MockSession()
@@ -997,10 +1003,23 @@ def mocked_requests(*args, **kwargs):
garmindownloadregex = '.*?garmin\.com\/mockfile?id=1'
garmindownloadtester = re.compile(garmindownloadregex)
garmintrainingregex = '.*?garmin\.com\/training-api\/workout'
garmintrainingtester = re.compile(garmintrainingregex)
garmintrainingscheduleregex = '.*?garmin\.com\/training-api\/schedule'
garmintrainingscheduletester = re.compile(garmintrainingscheduleregex)
if garmintester.match(args[0]):
if garmindownloadtester.match(args[0]):
return MockStreamResponse('rowers/tests/testdata/3x250m.fit',200)
if garmintrainingtester.match(args[0]):
json_data = {
'workoutId':1212,
}
return MockResponse(json_data,200)
if garmintrainingscheduletester.match(args[0]):
json_data = 1234
return MockResponse(json_data,200)
if stravaathletetester.match(args[0]):
json_data = stravaathletejson
@@ -1254,6 +1273,8 @@ class MockResponse:
def json(self):
return self.json_data
class MockOAuth1Session:
def __init__(self,*args, **kwargs):
pass
@@ -1264,5 +1285,14 @@ class MockOAuth1Session:
def post(*args, **kwargs):
return MockResponse({},200)
def fetch_request_token(*args, **kwargs):
return {
'oauth_token':'aap',
'oauth_token_secret':'noot',
}
def authorization_url(*args, **kwargs):
return 'url'
def mocked_invoiceid(*args,**kwargs):
return 1

View File

@@ -18,6 +18,9 @@ from rowers import stravastuff
import urllib
import json
from django.db import transaction
import rowers.garmin_stuff as gs
@pytest.mark.django_db
@override_settings(TESTING=True)
class GarminObjects(DjangoTestCase):
@@ -35,11 +38,31 @@ class GarminObjects(DjangoTestCase):
)
self.r.garmintoken = 'dfdzf'
self.r.garminrefreshtoken = 'fsls'
self.r.rowerplan = 'plan'
self.r.save()
self.c.login(username='john',password='koeinsloot')
self.nu = datetime.datetime.now()
startdate = nu.date()
enddate = (nu+datetime.timedelta(days=3)).date()
preferreddate = startdate
self.ps_trimp = SessionFactory(
startdate=startdate,enddate=enddate,
sessiontype='test',
sessionmode = 'TRIMP',
criterium = 'none',
sessionvalue = 77,
sessionunit='none',
preferreddate=preferreddate,
manager=self.u,
)
self.ps_trimp.interval_string = '10min+4x1000m@200W/20sec+2000m@24spm+10min'
self.ps_trimp.save()
def tearDown(self):
ws = Workout.objects.filter(user=self.r)
for w in ws:
@@ -137,6 +160,26 @@ class GarminObjects(DjangoTestCase):
self.assertEqual(res,1)
@patch('rowers.garmin_stuff.OAuth1Session')
def notest_garmin_callback(self,MockOAuth1Session):
with transaction.atomic():
response = self.c.get('/garmin_callback/?oauth_token=528ea5d9-1163-434d-b172-f428c5d9f522&oauth_verifier=LW33ZMBP8H')
self.assertEqual(response.status_code, 200)
@patch('rowers.garmin_stuff.requests.get',side_effect=mocked_requests)
def test_garmin_can_export_session(self,mock_get):
result = gs.garmin_can_export_session(self.u)
self.assertTrue(result)
def test_ps_to_garmin(self):
res = gs.ps_to_garmin(self.ps_trimp,self.r)
self.assertTrue(len(json.dumps(res))>500)
@patch('rowers.garmin_stuff.requests.get',side_effect=mocked_requests)
@patch('rowers.garmin_stuff.requests.post',side_effect=mocked_requests)
def test_garmin_session_create(self,mock_get,mock_post):
res = gs.garmin_session_create(self.ps_trimp,self.u)
self.assertEqual(res,1212)
@pytest.mark.django_db