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

mocking c2import successfully

Browse Source
This commit is contained in:
Sander Roosendaal
2018-07-01 13:23:24 +02:00
parent e8dd12c8b8
commit cbaada7945
11 changed files with 1804 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
data.txt Normal file
View File

@@ -0,0 +1 @@
{"distance": 13878, "user_id": 457764, "ranked": false, "weight_class": "L", "verified": false, "workout_type": false, "comments": "\n from speedcoach2v2.15 via rowsandall.com", "heart_rate": {"max": 158, "average": 156}, "source": "rowingdata", "date_utc": "2018-06-30 05:31:01", "time_formatted": "1:21:16.7", "time": 48767, "date": "2018-06-30 07:31:01", "timezone": "Europe/Prague", "type": "water", "id": 33991243, "stroke_data": true}

7
rowers/c2stuff.py
View File

@@ -517,10 +517,10 @@ def get_workout(user,c2id):
r = Rower.objects.get(user=user)
if (r.c2token == '') or (r.c2token is None):
s = "Token doesn't exist. Need to authorize"
return custom_exception_handler(401,s)
return custom_exception_handler(401,s) ,0
elif (timezone.now()>r.tokenexpirydate):
s = "Token expired. Needs to refresh."
return custom_exception_handler(401,s)
return custom_exception_handler(401,s),0
else:
# ready to fetch. Hurray
authorizationstring = str('Bearer ' + r.c2token)
@@ -530,6 +530,7 @@ def get_workout(user,c2id):
url = "https://log.concept2.com/api/users/me/results/"+str(c2id)
s = requests.get(url,headers=headers)
data = s.json()['data']
splitdata = None
@@ -544,6 +545,8 @@ def get_workout(user,c2id):
res2 = get_c2_workout_strokes(user,c2id)
if res2.status_code == 200:
strokedata = pd.DataFrame.from_dict(res2.json()['data'])
else:
strokedata = pd.DataFrame()
else:
strokedata = pd.DataFrame()

5
rowers/imports.py
View File

@@ -167,7 +167,8 @@ def imports_get_token(
headers = {'Accept': 'application/json',
'Api-Key': client_id,
'Content-Type': 'application/json'}
'Content-Type': 'application/json',
'user-agent': 'sanderroosendaal'}
if 'json' in oauth_data['content_type']:
@@ -217,7 +218,7 @@ def imports_make_authorization_url(oauth_data):
import urllib
url = oauth_data['authorizaton_uri']+authorization_uri+urllib.urlencode(params)
url = oauth_data['authorizaton_uri']+urllib.urlencode(params)
return HttpResponseRedirect(url)

2
rowers/models.py
View File

@@ -1831,10 +1831,12 @@ class RowerExportForm(ModelForm):
'stravaexportas',
'polar_auto_import',
'c2_auto_export',
'c2_auto_import',
'mapmyfitness_auto_export',
'runkeeper_auto_export',
'sporttracks_auto_export',
'strava_auto_export',
'strava_auto_import',
'trainingpeaks_auto_export',
]

35
rowers/stravastuff.py
View File

@@ -48,40 +48,17 @@ oauth_data = {
# Exchange access code for long-lived access token
def get_token(code):
client_auth = requests.auth.HTTPBasicAuth(STRAVA_CLIENT_ID, STRAVA_CLIENT_SECRET)
post_data = {"grant_type": "authorization_code",
"code": code,
"redirect_uri": STRAVA_REDIRECT_URI,
"client_secret": STRAVA_CLIENT_SECRET,
"client_id":STRAVA_CLIENT_ID,
}
headers = {'user-agent': 'sanderroosendaal'}
response = requests.post("https://www.strava.com/oauth/token",
data=post_data,
headers=headers)
try:
token_json = response.json()
thetoken = token_json['access_token']
except (KeyError,JSONDecodeError):
thetoken = 0
(
thetoken, expires_in, refresh_token
) = imports_get_token(code, oauth_data)
return [thetoken]
# Make authorization URL including random string
def make_authorization_url(request):
# Generate a random string for the state parameter
# Save it for use later to prevent xsrf attacks
state = str(uuid4())
params = {"client_id": STRAVA_CLIENT_ID,
"response_type": "code",
"redirect_uri": STRAVA_REDIRECT_URI,
"scope":"write"}
import urllib
url = "https://www.strava.com/oauth/authorize" +urllib.urlencode(params)
return HttpResponseRedirect(url)
return imports_make_authorization_url(oauth_data)
# Get list of workouts available on Strava
def get_strava_workout_list(user,limit_n=0):

1
rowers/testdata/c2jsonstrokedata.txt vendored Normal file
View File

File diff suppressed because one or more lines are too long

1
rowers/testdata/c2jsonworkoutdata.txt vendored Normal file
View File

@@ -0,0 +1 @@
{"data": {"distance": 13878, "user_id": 457764, "ranked": false, "weight_class": "L", "verified": false, "workout_type": false, "comments": "\n from speedcoach2v2.15 via rowsandall.com", "heart_rate": {"max": 158, "average": 156}, "source": "rowingdata", "date_utc": "2018-06-30 05:31:01", "time_formatted": "1:21:16.7", "time": 48767, "date": "2018-06-30 07:31:01", "timezone": "Europe/Prague", "type": "water", "id": 33991243, "stroke_data": true}}

1720
rowers/testdata/c2strokedata.csv vendored Normal file
View File

File diff suppressed because it is too large Load Diff

84
rowers/tests.py
View File

@@ -1,6 +1,7 @@
from __future__ import print_function
from bs4 import BeautifulSoup
import re
from django.test import TestCase, Client,override_settings
from django.core.management import call_command
from django.utils.six import StringIO
@@ -22,9 +23,11 @@ from rowers.tasks import handle_makeplot
from rowers.utils import serialize_list,deserialize_list
from rowers.utils import NoTokenError
from shutil import copyfile
from nose.tools import assert_true
from mock import Mock, patch
from minimocktest import MockTestCase
import pandas as pd
import rowers.c2stuff as c2stuff
import json
import numpy as np
@@ -51,7 +54,62 @@ class DjangoTestCase(TestCase, MockTestCase):
# Create your tests here.
def mocked_requests_get(*args, **kwargs):
with open('rowers/testdata/c2jsonworkoutdata.txt','r') as infile:
c2workoutdata = json.load(infile)
with open('rowers/testdata/c2jsonstrokedata.txt','r') as infile:
c2strokedata = json.load(infile)
class MockResponse:
def __init__(self, json_data, status_code):
self.json_data = json_data
self.status_code = status_code
def json(self):
return self.json_data
if 'log.concept2.com/' in args[0]:
if 'strokes' in args[0]:
return MockResponse(c2strokedata,200)
else:
return MockResponse(c2workoutdata,200)
return MockResponse(None,404)
class C2Objects(DjangoTestCase):
def setUp(self):
self.c = Client()
self.u = User.objects.create_user('john',
'sander@ds.ds',
'koeinsloot')
self.u.first_name = 'John'
self.u.last_name = 'Sander'
self.u.save()
self.r = Rower.objects.create(user=self.u,gdproptin=True,
gdproptindate=timezone.now()
)
self.r.c2token = '12'
self.r.tokenexpirydate = datetime.datetime.now()+datetime.timedelta(days=1)
self.r.save()
@patch('rowers.c2stuff.requests.get', side_effect=mocked_requests_get)
def test_c2_import(self, mock_get):
self.c.login(username='john',password='koeinsloot')
response = self.c.get('/rowers/workout/c2import/12/',follow=True)
self.assertRedirects(response,
expected_url='/rowers/workout/1/edit',
status_code=302,target_status_code=200)
self.assertEqual(response.status_code, 200)
def test_strokedata(self):
with open('rowers/testdata/c2stroketestdata.txt','r') as infile:
res = json.load(infile)
@@ -65,17 +123,7 @@ class C2Objects(DjangoTestCase):
from rowers.views import add_workout_from_strokedata
u = User.objects.create_user('john',
'sander@ds.ds',
'koeinsloot')
u.first_name = 'John'
u.last_name = 'Sander'
u.save()
r = Rower.objects.create(user=u,gdproptin=True,
gdproptindate=timezone.now()
)
res = add_workout_from_strokedata(u,1,data,strokedata,source='c2')
res = add_workout_from_strokedata(self.u,1,data,strokedata,source='c2')
def test_strokedatanohr(self):
with open('rowers/testdata/c2strokedatanohr.txt','r') as infile:
@@ -90,18 +138,8 @@ class C2Objects(DjangoTestCase):
from rowers.views import add_workout_from_strokedata
u = User.objects.create_user('john',
'sander@ds.ds',
'koeinsloot')
u.first_name = 'John'
u.last_name = 'Sander'
u.save()
r = Rower.objects.create(user=u,gdproptin=True,
gdproptindate=timezone.now()
)
res = add_workout_from_strokedata(u,1,data,strokedata,source='c2')
res = add_workout_from_strokedata(self.u,1,data,strokedata,source='c2')
class StravaObjects(DjangoTestCase):

4
rowers/utils.py
View File

@@ -6,9 +6,11 @@ from django.conf import settings
import uuid
import datetime
import json
import requests
from django.http import HttpResponse
lbstoN = 4.44822
landingpages = (

1
rowers/views.py
View File

@@ -11109,6 +11109,7 @@ def rower_exportsettings_view(request):
setattr(r, attr, value)
r.save()
messages.info(request,'Settings saved')
else:
form = RowerExportForm(instance=r)
return render(request, 'rower_exportsettings.html',