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

fixing failing tests

Browse Source
This commit is contained in:
Sander Roosendaal
2023-02-10 17:44:23 +01:00
parent 2981c59a5d
commit 792faf414e
5 changed files with 82 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
rowers/integrations/c2.py
View File

@@ -114,7 +114,7 @@ class C2Integration(SyncIntegration):
return super(C2Integration, self).open(*args, **kwargs)
def token_refresh(self, *args, **kwargs):
return super(C2Integration, self).open(*args, **kwargs)
return super(C2Integration, self).token_refresh(*args, **kwargs)
def make_authorization_url(self, *args, **kwargs):
scope = "user:read,results:write"

16
rowers/integrations/integrations.py
View File

@@ -162,6 +162,8 @@ class SyncIntegration(metaclass=ABCMeta):
return token
def do_refresh_token(self, *args, **kwargs) -> (str, int, str):
refreshtoken = getattr(self.rower, self.oauth_data['refreshtokenname'])
access_token = kwargs.get('access_token','')
post_data = {"grant_type": "refresh_token",
"client_secret": self.oauth_data['client_secret'],
"client_id": self.oauth_data['client_id'],
@@ -229,19 +231,19 @@ class SyncIntegration(metaclass=ABCMeta):
@abstractmethod
def token_refresh(self, *args, **kwargs) -> str:
refreshtoken = getattr(self.rower, oauth['refreshtokenname'])
refreshtoken = getattr(self.rower, self.oauth_data['refreshtokenname'])
if not refreshtoken:
refreshtoken = getattr(self.rower, oauth['tokenname'])
refreshtoken = getattr(self.rower, self.oauth_data['tokenname'])
access_token, expires_in, refresh_token = self.do_refresh_token()
expirydatetime = timezone.now()+timedelta(seconds=expires_in)
setattr(self.rower, tokenname, access_token)
if expirydatename is not None:
setattr(self.rower, expirydatename, expirydatetime)
if refreshtokenname is not None:
setattr(self.rower, refreshtokenname, refresh_token)
setattr(self.rower, self.oauth_data['tokenname'], access_token)
if self.oauth_data['expirydatename'] is not None:
setattr(self.rower, self.oauth_data['expirydatename'], expirydatetime)
if self.oauth_data['refreshtokenname'] is not None:
setattr(self.rower, self.oauth_data['refreshtokenname'], refresh_token)
self.rower.save()

2
rowers/tests/statements.py
View File

@@ -37,7 +37,7 @@ from django.core.files.uploadedfile import SimpleUploadedFile
from io import StringIO
from django.test.client import RequestFactory
from rowers.integrations import *
from rowers.forms import (
DocumentsForm,CNsummaryForm,RegistrationFormUniqueEmail,

11
rowers/tests/test_async_tasks.py
View File

@@ -186,15 +186,16 @@ class AsyncTaskTests(TestCase):
self.u.id)
self.assertEqual(res,1)
@patch('rowers.c2stuff.requests.post', side_effect=mocked_requests)
@patch('rowers.c2stuff.requests.get', side_effect=mocked_requests)
@patch('rowers.integrations.c2.requests.post', side_effect=mocked_requests)
@patch('rowers.integrations.c2.requests.get', side_effect=mocked_requests)
def test_c2_sync(self, mock_get, mock_post):
authorizationstring = str('Bearer aap')
headers = {'Authorization': authorizationstring,
'user-agent': 'sanderroosendaal',
'Content-Type': 'application/json'}
data = c2stuff.createc2workoutdata(self.wwater)
integration = C2Integration(self.u)
data = integration.createworkoutdata(self.wwater)
url = "https://log.concept2.com/api/users/%s/results" % (1)
res = tasks.handle_c2_sync(1,url,headers,data)
@@ -469,8 +470,8 @@ class AsyncTaskTests(TestCase):
self.assertEqual(res,1)
@patch('rowers.c2stuff.requests.post',side_effect=mocked_requests)
@patch('rowers.c2stuff.requests.get',side_effect=mocked_requests)
@patch('rowers.integrations.c2.requests.post',side_effect=mocked_requests)
@patch('rowers.integrations.c2.requests.get',side_effect=mocked_requests)
def test_import_c2_strokedata(self, mock_get, mock_post):
c2token = 'aap'
c2id = 1212

67
rowers/views/importviews.py
View File

@@ -1832,14 +1832,77 @@ def workout_getrp3importview(request, externalid):
url = reverse('workout_rp3import_view')
return HttpResponseRedirect(url)
@login_required()
def workout_getimportview(request, externalid, source='c2', do_async=True):
def workout_getimportview_old(request, externalid, source='c2', do_async=True):
if 'startdate' in request.session and source == 'nk': # pragma: no cover
startdate = request.session.get('startdate')
enddate = request.session.get('enddate')
try:
result = importsources[source].get_workout(request.user, externalid, do_async=do_async,
startdate=startdate, enddate=enddate)
except NoTokenError:
return HttpResponseRedirect(reverse(importauthorizeviews[source]))
url = reverse(importlistviews[source])
return HttpResponseRedirect(url)
try:
result = importsources[source].get_workout(request.user, externalid,
do_async=do_async)
except NoTokenError:
return HttpResponseRedirect(reverse(importauthorizeviews[source]))
if result: # pragma: no cover
messages.info(
request, "Your workout will be imported in the background")
# this should return to the respective import list page
else: # pragma: no cover
messages.error(request, 'Error getting the workout')
url = reverse(importlistviews[source])
return HttpResponseRedirect(url)
# Imports all new workouts from SportTracks
@login_required()
def workout_getsporttracksworkout_all(request):
res = sporttracksstuff.get_sporttracks_workout_list(request.user)
if (res.status_code == 200):
r = getrower(request.user)
stids = [int(getidfromuri(item['uri']))
for item in res.json()['items']]
knownstids = uniqify([
w.uploadedtosporttracks for w in Workout.objects.filter(user=r)
])
newids = [stid for stid in stids if stid not in knownstids]
for sporttracksid in newids:
id = sporttracksstuff.get_workout(
request.user, sporttracksid)
if id == 0: # pragma: no cover
messages.error(
request, "Something went wrong with workout {id}".format(id=sporttracksid))
else:
w = Workout.objects.get(id=id)
w.uploadedtosporttracks = sporttracksid
w.save()
url = reverse('workouts_view')
return HttpResponseRedirect(url)
@login_required()
def workout_getimportview(request, externalid, source='c2', do_async=True):
try:
integration = importsources[source](request.user)
except TypeError:
return workout_getimportview_old(request, externalid, source=source, do_async=True)
if 'startdate' in request.session and source == 'nk': # pragma: no cover
startdate = request.session.get('startdate')
enddate = request.session.get('enddate')
try:
result = integration.get_workout(externalid, startdate=startdate, enddate=enddate)