1490 lines
44 KiB
Python
1490 lines
44 KiB
Python
from __future__ import absolute_import
|
|
from __future__ import division
|
|
from __future__ import print_function
|
|
from __future__ import unicode_literals
|
|
|
|
import pytest
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
from rowers.opaque import encoder
|
|
encoded1 = str(encoder.encode_hex(1))
|
|
encoded2 = str(encoder.encode_hex(2))
|
|
encoded13 = str(encoder.encode_hex(13))
|
|
|
|
from bs4 import BeautifulSoup
|
|
import re
|
|
from parameterized import parameterized
|
|
from django.test import TestCase, Client,override_settings
|
|
from django.core.management import call_command
|
|
|
|
from django.test.client import RequestFactory
|
|
from rowers.models import Workout, User, Rower, WorkoutForm,RowerForm,GraphImage
|
|
from rowers.forms import DocumentsForm,CNsummaryForm,RegistrationFormUniqueEmail
|
|
import rowers.plots as plots
|
|
import rowers.interactiveplots as iplots
|
|
import datetime
|
|
from rowingdata import rowingdata as rdata
|
|
from rowingdata import rower as rrower
|
|
from django.utils import timezone
|
|
from rowers.rows import handle_uploaded_file
|
|
from django.core.files.uploadedfile import SimpleUploadedFile
|
|
from time import strftime,strptime,mktime,time,daylight
|
|
import os
|
|
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 arrow
|
|
from django.http import HttpResponseRedirect
|
|
|
|
import json
|
|
import numpy as np
|
|
from rowingdata import rowingdata, figure
|
|
|
|
from rowers import urls
|
|
from rowers.views import (
|
|
error500_view,error404_view,error400_view,error403_view
|
|
)
|
|
|
|
from rowers.dataprep import delete_strokedata
|
|
|
|
from redis import StrictRedis
|
|
redis_connection = StrictRedis()
|
|
|
|
|
|
def mocked_grpc(*args, **kwargs):
|
|
class insecure_channel:
|
|
def __init__(*args,**kwargs):
|
|
pass
|
|
|
|
class channel_ready_future:
|
|
def __init__(*args,**kwargs):
|
|
pass
|
|
|
|
def result(self,*args,**kwargs):
|
|
pass
|
|
|
|
class Result:
|
|
def __init__(*args,**kwargs):
|
|
self.result = 1
|
|
|
|
class calculator_pb2_grpc:
|
|
def PowerStub(*args,**kwargs):
|
|
def __init__(*args,**kwargs):
|
|
pass
|
|
|
|
def CalcPower(*args,**kwargs):
|
|
return Result()
|
|
|
|
def mocked_send_template_email(*args,**kwargs):
|
|
return 1
|
|
|
|
def mocked_myqueue(*args, **kwargs):
|
|
class Job:
|
|
def __init__(self,*args, **kwargs):
|
|
self.result = 1
|
|
self.id = 1
|
|
|
|
def revoke(self):
|
|
return 1
|
|
|
|
return Job()
|
|
|
|
def mock_c2open(*args, **kwargs):
|
|
return('aap')
|
|
|
|
def mocked_session(*args, **kwargs):
|
|
|
|
class MockEngine:
|
|
def __init__(self, *args, **kwargs):
|
|
self.headers = MockHeaders()
|
|
|
|
def post(self, *args, **kwargs):
|
|
return MockResponse({},200)
|
|
|
|
class MockHeaders:
|
|
def update(*args, **kwargs):
|
|
return None
|
|
|
|
class MockResponse:
|
|
def __init__(self, json_data, status_code):
|
|
self.json_data = json_data
|
|
self.status_code = status_code
|
|
self.reason = 'mock reason'
|
|
self.headers = {
|
|
'Location':'MockLocation',
|
|
}
|
|
self.ok = True
|
|
|
|
def json(self):
|
|
return self.json_data
|
|
|
|
class MockContentResponse:
|
|
def __init__(self,filename,status_code):
|
|
with open(filename,'rb') as f:
|
|
s = f.read()
|
|
self.content = s
|
|
self.status_code = status_code
|
|
self.ok = True
|
|
|
|
|
|
|
|
return MockEngine()
|
|
|
|
def mocked_sqlalchemy(*args, **kwargs):
|
|
# return object with method
|
|
|
|
class MockEngine:
|
|
def connect(self):
|
|
return MockConnection()
|
|
|
|
def dispose(self):
|
|
return True
|
|
|
|
def raw_connection(self):
|
|
return True
|
|
|
|
class QueryResult:
|
|
def fetchall(self):
|
|
return []
|
|
|
|
class MockConnection:
|
|
def begin(self):
|
|
return True
|
|
|
|
def execute(self,query):
|
|
if 'polygon' in query:
|
|
return QueryResult()
|
|
return True
|
|
|
|
def execute(self):
|
|
return True
|
|
|
|
def close(self):
|
|
return True
|
|
|
|
def __exit__(self, *args, **kwargs):
|
|
return True
|
|
|
|
def __enter__(self, *args, **kwargs):
|
|
return True
|
|
|
|
return MockEngine()
|
|
|
|
from rowers import courses
|
|
|
|
def mocked_sqlalchemy_courses(*args, **kwargs):
|
|
# return object with method
|
|
cs = courses.kmltocourse('rowers/tests/testdata/thyro.kml')
|
|
course = cs[0]
|
|
cname = course['name']
|
|
cnotes = course['description']
|
|
polygons = course['polygons']
|
|
|
|
lijst = []
|
|
i=0
|
|
for polygon in polygons:
|
|
lijst.append((i,polygon['name']))
|
|
i = i+1
|
|
|
|
puntenlijst = []
|
|
i = 0
|
|
for p in polygons[0]['points']:
|
|
puntenlijst.append((i,p['latitude'],p['longitude']))
|
|
i = i+1
|
|
|
|
|
|
class MockEngine:
|
|
def connect(self):
|
|
return MockConnection()
|
|
|
|
def dispose(self):
|
|
return True
|
|
|
|
def raw_connection(self):
|
|
return True
|
|
|
|
class QueryResultPolygons:
|
|
def fetchall(self):
|
|
return lijst
|
|
|
|
class QueryResultPoints:
|
|
def fetchall(self):
|
|
return puntenlijst
|
|
|
|
class MockConnection:
|
|
def begin(self):
|
|
return True
|
|
|
|
def execute(self,query):
|
|
if 'latitude' in query:
|
|
return QueryResultPoints()
|
|
if 'polygon' in query:
|
|
return QueryResultPolygons()
|
|
return True
|
|
|
|
def close(self):
|
|
return True
|
|
|
|
def __exit__(self, *args, **kwargs):
|
|
return True
|
|
|
|
def __enter__(self, *args, **kwargs):
|
|
return True
|
|
|
|
return MockEngine()
|
|
|
|
|
|
#@pytest.mark.django_db
|
|
class DjangoTestCase(TestCase): #, MockTestCase):
|
|
def _pre_setup(self):
|
|
# MockTestCase.setUp(self)
|
|
TestCase._pre_setup(self)
|
|
self.client = Client()
|
|
|
|
def _post_teardown(self):
|
|
TestCase._post_teardown(self)
|
|
#MockTestCase.tearDown(self)
|
|
# delete_strokedata(1)
|
|
|
|
def mocked_tcx_parser(*args, **kwargs):
|
|
df = pd.read_csv('rowers/tests/testdata/fake_strokedata.csv')
|
|
|
|
return rowingdata(df=df)
|
|
|
|
def mocked_polar_notifications(*args, **kwargs):
|
|
return []
|
|
|
|
|
|
class mocked_canvas():
|
|
def print_figure(*args, **kwargs):
|
|
return True
|
|
|
|
# Mocked Dataprep
|
|
|
|
def mocked_fetchcperg(*args, **kwargs):
|
|
df = pd.read_csv('rowers/tests/testdata/cpdataerg.csv')
|
|
|
|
return df
|
|
|
|
import pandas as pd
|
|
def mocked_read_df_sql(id):
|
|
df = pd.read_csv('rowers/tests/testdata/fake_strokedata.csv')
|
|
|
|
return df
|
|
|
|
def mocked_sendmail(*args,**kwargs):
|
|
return HttpResponseRedirect('/rowers/email/')
|
|
|
|
def mocked_get_video_data(*args, **kwargs):
|
|
with open('rowers/tests/testdata/videodata.json','r') as infile:
|
|
data = json.load(infile)
|
|
|
|
maxtime = 4135.
|
|
with open('rowers/tests/testdata/videometrics.json','r') as infile:
|
|
metrics = json.load(infile)
|
|
|
|
return data,metrics,maxtime
|
|
|
|
def mocked_getrowdata_db(*args, **kwargs):
|
|
df = pd.read_csv('rowers/tests/testdata/getrowdata_mock.csv')
|
|
|
|
id = kwargs['id']
|
|
|
|
row = Workout.objects.get(id=id)
|
|
|
|
return df,row
|
|
|
|
def mocked_getrowdata_uh(*args, **kwargs):
|
|
df = pd.read_csv('rowers/tests/testdata/uhfull.csv')
|
|
|
|
id = kwargs['id']
|
|
|
|
row = Workout.objects.get(id=id)
|
|
|
|
return df, row
|
|
|
|
def mocked_getsmallrowdata_uh(*args, **kwargs):
|
|
df = pd.read_csv('rowers/tests/testdata/uhfull.csv')
|
|
|
|
return df
|
|
|
|
def mocked_getsmallrowdata_forfusion(*args, **kwargs):
|
|
df = pd.read_csv('rowers/tests/testdata/getrowdata_mock.csv')
|
|
|
|
return df
|
|
|
|
from collections import OrderedDict
|
|
|
|
def mocked_videodata(*args, **kwargs):
|
|
df = pd.read_csv('rowers/tests/testdata/testvideodata.csv')
|
|
data = df .to_dict()
|
|
df = pd.read_csv('rowers/tests/testdata/testvideodata_metrics.csv')
|
|
metrics = OrderedDict(df.to_dict())
|
|
|
|
maxtime = 3282.4
|
|
|
|
return data, metrics, maxtime
|
|
|
|
def mocked_getsmallrowdata_db(*args, **kwargs):
|
|
df = pd.read_csv('rowers/tests/testdata/colsfromdb.csv')
|
|
|
|
return df
|
|
|
|
def mocked_getsmallrowdata_db_updatecp(*args, **kwargs):
|
|
df = pd.read_csv('rowers/tests/testdata/colsfromdb.csv')
|
|
|
|
return df
|
|
|
|
def mocked_getsmallrowdata_db_setcp(*args, **kwargs):
|
|
df = pd.read_csv('rowers/tests/testdata/colsfromdb2.csv')
|
|
|
|
return df
|
|
|
|
|
|
def mocked_getsmallrowdata_db_water(*args, **kwargs):
|
|
df = pd.read_csv('rowers/tests/testdata/colsfromdb3.csv')
|
|
|
|
return df
|
|
|
|
def mocked_getsmallrowdata_db_wps(*args, **kwargs):
|
|
df = pd.read_csv('rowers/tests/testdata/driveenergies.csv')
|
|
|
|
return df
|
|
|
|
def mocked_getpowerdata_db(*args, **kwargs):
|
|
df = pd.read_csv('rowers/tests/testdata/fake_powerdata.csv')
|
|
|
|
return df
|
|
|
|
def mock_for_interactive_chart(*args, **kwargs):
|
|
df = pd.read_csv('rowers/tests/testdata/interactivechart.csv')
|
|
|
|
return df
|
|
|
|
def mocked_getempowerdata_db(*args, **kwargs):
|
|
df = pd.read_csv('rowers/tests/testdata/fake_empowerdata.csv')
|
|
|
|
return df
|
|
|
|
def mocked_read_df_cols_sql_multistats(ids,columns,convertnewtons=True):
|
|
df = pd.read_csv('rowers/tests/testdata/cumstats.csv')
|
|
extracols = []
|
|
|
|
return df, extracols
|
|
|
|
def mocked_read_cols_df_sql(*args, **kwargs):
|
|
df = pd.read_csv('rowers/tests/testdata/readcols.csv')
|
|
|
|
return df,[]
|
|
|
|
|
|
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):
|
|
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):
|
|
df = pd.read_csv('rowers/tests/testdata/multiflexdata.csv')
|
|
extracols = []
|
|
|
|
return df, extracols
|
|
|
|
# Mocked Strava
|
|
|
|
def mocked_stravaexport(f2,workoutname,stravatoken,description='',
|
|
activity_type='Rowing'):
|
|
print("this is mocked strava export")
|
|
return 1,'success'
|
|
|
|
def mocked_fetchcp(*args, **kwargs):
|
|
df = pd.read_csv('rowers/tests/testdata/otwcpresult.csv')
|
|
delta = df['delta']
|
|
cpvalue = df['cpvalue']
|
|
|
|
try:
|
|
theworkouts = args[1]
|
|
except:
|
|
theworkouts = []
|
|
|
|
theids = [int(w.id) for w in theworkouts]
|
|
|
|
avgpower = {}
|
|
thepowers = [203,153,143,152,196,211,223]
|
|
|
|
for i in range(len(theids)):
|
|
avgpower[id[i]] = thepowers[i]
|
|
|
|
return delta, cpvalue, avgpower
|
|
|
|
def mocked_getcpdata_sql(*args, **kwargs):
|
|
df = pd.read_csv('rowers/tests/testdata/otwcpresult.csv')
|
|
|
|
df['cp'] = df['cpvalue']
|
|
|
|
return df
|
|
|
|
def mocked_cpraw(*args, **kwargs):
|
|
df = pd.read_csv('rowers/tests/testdata/otwcp_df.csv')
|
|
|
|
return df
|
|
|
|
class StravalibResponse():
|
|
def __init__(self,*args, **kwargs):
|
|
self.id = 1
|
|
|
|
class StravaActivity():
|
|
def wait(*args, **kwargs):
|
|
return StravalibResponse()
|
|
|
|
def __init__(*args, **kwargs):
|
|
pass
|
|
|
|
|
|
|
|
class MockStravalibClient():
|
|
def upload_activity(*args, **kwargs):
|
|
return StravaActivity()
|
|
|
|
def update_activity(*args, **kwargs):
|
|
return StravaActivity()
|
|
|
|
def __init__(*args,**kwargs):
|
|
pass
|
|
|
|
# Mocked Braintree
|
|
|
|
## Higher level - unfortunately didn't succeed in mocking the gateway
|
|
|
|
def get_client_token(*args, **kwargs):
|
|
return "aap"
|
|
|
|
def mock_create_customer(*args, **kwargs):
|
|
return 121
|
|
|
|
def mock_make_payment(*args, **kwargs):
|
|
return 15,''
|
|
|
|
def mock_update_subscription(*args, **kwargs):
|
|
try:
|
|
rower = args[0]
|
|
data = args[1]
|
|
planid = data['plan']
|
|
plan = PaidPlan.objects.get(id=planid)
|
|
amount = data['amount']
|
|
|
|
rower.paidplan = plan
|
|
rower.planexpires = (datetime.datetime.now()+datetime.timedelta(days=365)).date()
|
|
rower.clubsize = plan.clubsize
|
|
rower.paymenttype = plan.paymenttype
|
|
rower.rowerplan = plan.shortname
|
|
rower.subscription_id = 12
|
|
rower.save()
|
|
except:
|
|
amount = 20
|
|
|
|
return True,amount
|
|
|
|
def mock_create_subscription(*args, **kwargs):
|
|
return mock_update_subscription(*args, **kwargs)
|
|
|
|
def mock_cancel_subscription(*args, **kwargs):
|
|
themessages = []
|
|
errormessages = []
|
|
try:
|
|
basicplans = PaidPlan.objects.filter(price=0,paymentprocessor='braintree')
|
|
rower.paidplan = basicplans[0]
|
|
rower.teamplanexpires = timezone.now()
|
|
rower.planexpires = timezone.now()
|
|
rower.clubsize = 0
|
|
rower.rowerplan = 'basic'
|
|
rower.subscription_id = None
|
|
rower.save()
|
|
themessages.append("Your plan was reset to basic")
|
|
except:
|
|
pass
|
|
|
|
return True, themessages,errormessages
|
|
|
|
def mock_mocktest(*args, **kwargs):
|
|
return '121'
|
|
|
|
|
|
## Gateway stuff (not working)
|
|
|
|
class gatewayresult():
|
|
def __init__(self,*args,**kwargs):
|
|
self.is_success = kwargs.pop('is_success',True)
|
|
self.customer_id = 1
|
|
self.transaction = vtransaction()
|
|
self.payment_method = vpayment_method()
|
|
self.subscription = vsubscription()
|
|
self.customer = kwargs.pop('customer',customer())
|
|
|
|
def __unicode__():
|
|
return "mockedgatewayresult"
|
|
|
|
class credit_card():
|
|
def __init__(self, *args, **kwargs):
|
|
self.subscriptions = [vsubscription()]
|
|
self.country_of_issuance = 'US'
|
|
|
|
class paypal_account():
|
|
def __init__(self, *args, **kwargs):
|
|
self.subscriptions = [vsubscription(),vsubscription()]
|
|
|
|
class customercreateresult:
|
|
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():
|
|
def find(*arg, **kwargs):
|
|
return self
|
|
|
|
def create(*args, **kwargs):
|
|
return customercreateresult(is_success=True)
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
self.credit_cards = [credit_card(),credit_card()]
|
|
self.paypal_accounts = [paypal_account()]
|
|
self.id = 1
|
|
|
|
class client_token():
|
|
def generate(*args, **kwargs):
|
|
return 'aapnooit'
|
|
|
|
class plan():
|
|
def all(*args, **kwargs):
|
|
return []
|
|
|
|
|
|
|
|
class transaction():
|
|
def sale(*args, **kwargs):
|
|
return gatewayresult(is_success=True)
|
|
|
|
def search(*args, **kwargs):
|
|
return [gatewayresult(is_success=True),gatewayresult(is_success=True)]
|
|
|
|
def __init__(self,*args, **kwargs):
|
|
self.amount = 15
|
|
self.credit_card_details = credit_card()
|
|
self.customer = {
|
|
'first_name': 'John',
|
|
'last_name': 'Doe',
|
|
'id': 12
|
|
}
|
|
self.created_at = datetime.datetime.now()
|
|
self.currency_iso_code = 'EUR'
|
|
|
|
class vtransaction():
|
|
def __init__(self,*args, **kwargs):
|
|
self.amount = 15
|
|
self.credit_card_details = credit_card()
|
|
self.customer = {
|
|
'first_name': 'John',
|
|
'last_name': 'Doe',
|
|
'id': 12
|
|
}
|
|
self.created_at = datetime.datetime.now()
|
|
self.currency_iso_code = 'EUR'
|
|
|
|
class vsubscription():
|
|
def update(*args, **kwargs):
|
|
return gatewayresult(is_success=True)
|
|
|
|
def cancel(*args, **kwargs):
|
|
return gatewayresult(is_success=True)
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
self.id = '121'
|
|
self.billing_period_end_date = (datetime.datetime.now()+datetime.timedelta(days=365)).date()
|
|
self.status = 'Active'
|
|
self.plan_id = 12
|
|
self.price = 15
|
|
self.never_expires = True
|
|
|
|
class subscription():
|
|
def create(*args, **kwargs):
|
|
return gatewayresult(is_success=True)
|
|
|
|
def update(*args, **kwargs):
|
|
return gatewayresult(is_success=True)
|
|
|
|
def cancel(*args, **kwargs):
|
|
return gatewayresult(is_success=True)
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
self.id = '121'
|
|
self.billing_period_end_date = (datetime.datetime.now()+datetime.timedelta(days=365)).date()
|
|
self.transactions = [vtransaction()]
|
|
self.status = 'Active'
|
|
self.plan_id = 12
|
|
self.price = 15
|
|
self.never_expires = True
|
|
|
|
class vpayment_method():
|
|
def create(*args, **kwargs):
|
|
return gatewayresult()
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
self.token = 'liesjeleerdelotje'
|
|
|
|
|
|
class payment_method():
|
|
def create(*args, **kwargs):
|
|
return gatewayresult()
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
self.token = 'liesjeleerdelotje'
|
|
|
|
class notification():
|
|
def __init__(self, *args, **kwargs):
|
|
print('notifucation')
|
|
self.kind = 'subscription_canceled'
|
|
|
|
class webhook_notification():
|
|
def parse(*args, **kwargs):
|
|
print(args,kwargs,'parse')
|
|
return notification()
|
|
|
|
# mock braintree gateway
|
|
class MockBraintreeGateway:
|
|
def __init__(self,*args, **kwargs):
|
|
self.customer = customer()
|
|
self.client_token = client_token()
|
|
self.plan = plan()
|
|
self.transaction = transaction()
|
|
self.subscription = subscription()
|
|
self.payment_method = payment_method()
|
|
self.webhook_notification = webhook_notification()
|
|
|
|
|
|
def mocked_gateway(*args, **kwargs):
|
|
return MockBraintreeGateway()
|
|
|
|
|
|
# Mocked Rowingdata
|
|
|
|
class mocked_rowingdata(rowingdata):
|
|
def __init__(self, *args, **kwargs):
|
|
super(mocked_rowingdata).__init__(*args, **kwargs)
|
|
|
|
def get_metersplot_erg(self, title):
|
|
fig1 = figure.Figure(figsize(12,10))
|
|
return fig1
|
|
|
|
def get_timeplot_erg(self, title):
|
|
fig1 = figure.Figure(figsize(12,10))
|
|
return fig1
|
|
|
|
def get_piechart(self, title):
|
|
fig1 = figure.Figure(figsize(12,10))
|
|
return fig1
|
|
|
|
def get_timeplot_otwempower(self, title):
|
|
fig1 = figure.Figure(figsize(12,10))
|
|
return fig1
|
|
|
|
def get_timeplot_otw(self, title):
|
|
fig1 = figure.Figure(figsize(12,10))
|
|
return fig1
|
|
|
|
def get_metersplot_otwempower(self, title):
|
|
fig1 = figure.Figure(figsize(12,10))
|
|
return fig1
|
|
|
|
def get_metersplot_erg2(self, title):
|
|
fig1 = figure.Figure(figsize(12,10))
|
|
return fig1
|
|
|
|
def get_timeplot_erg2(self, title):
|
|
fig1 = figure.Figure(figsize(12,10))
|
|
return fig1
|
|
|
|
def get_time_otwpower(self, title):
|
|
fig1 = figure.Figure(figsize(12,10))
|
|
return fig1
|
|
|
|
def get_power_piechart(self, title):
|
|
fig1 = figure.Figure(figsize(12,10))
|
|
return fig1
|
|
|
|
|
|
# Mocked Requests
|
|
|
|
# to be done add polar mocks (for email processing)
|
|
def mocked_requests(*args, **kwargs):
|
|
with open('rowers/tests/testdata/c2jsonworkoutdata.txt','r') as infile:
|
|
c2workoutdata = json.load(infile)
|
|
|
|
with open('rowers/tests/testdata/c2_timezone.json','r') as infile:
|
|
c2timezoneworkoutdata = json.load(infile)
|
|
|
|
with open('rowers/tests/testdata/rojabolist.json','r') as infile:
|
|
rojabodata = json.load(infile)
|
|
|
|
with open('rowers/tests/testdata/c2_timezone2.json','r') as infile:
|
|
c2timezoneworkoutdata2 = json.load(infile)
|
|
|
|
with open('rowers/tests/testdata/c2_timezonebad.json','r') as infile:
|
|
c2timezoneworkoutdatabad = json.load(infile)
|
|
|
|
with open('rowers/tests/testdata/c2jsonstrokedata.txt','r') as infile:
|
|
c2strokedata = json.load(infile)
|
|
|
|
with open('rowers/tests/testdata/nk_strokes.json','r') as infile:
|
|
nkstrokedata = json.load(infile)
|
|
|
|
with open('rowers/tests/testdata/nk_list.json','r') as infile:
|
|
nkworkoutlist = json.load(infile)
|
|
|
|
with open('rowers/tests/testdata/rp3_list.json','r') as infile:
|
|
rp3workoutlist = json.load(infile)
|
|
|
|
rp3linkready = {'data': {'download': {'id': 591621, 'status': 'ready', 'link': 'https://rp3rowing-app.com/api/workouts/591621/download?type=csv'}}}
|
|
|
|
with open('rowers/tests/testdata/example-session-strokes-with-impeller-data.json','r') as infile:
|
|
nkimpellerstrokedata = json.load(infile)
|
|
|
|
with open('rowers/tests/testdata/example-session-with-impeller-data.json','r') as infile:
|
|
nkimpellerworkoutlist = json.load(infile)
|
|
|
|
polar_json = {
|
|
"transaction-id":240522162,
|
|
"resource-uri":"https://polaraccesslink.com/v3/users/40273947/exercise-transactions/240522162"
|
|
}
|
|
|
|
polar_exercise_list = {
|
|
"exercises": [
|
|
"https://www.polaraccesslink.com/v3/users/12/exercise-transactions/34/exercises/56",
|
|
"https://www.polaraccesslink.com/v3/users/12/exercise-transactions/34/exercises/120"
|
|
]
|
|
}
|
|
|
|
polar_user_data = {
|
|
"polar-user-id": 475,
|
|
"member-id": "i09u9ujj",
|
|
"registration-date": "2011-10-14T12:50:37.000Z",
|
|
"first-name": "Eka",
|
|
"last-name": "Toka",
|
|
"birthdate": "1985-09-06",
|
|
"gender": "MALE",
|
|
"weight": 66,
|
|
"height": 170,
|
|
"field": [
|
|
{
|
|
"value": "2",
|
|
"index": 0,
|
|
"name": "number-of-children"
|
|
}
|
|
]
|
|
}
|
|
|
|
polar_exercise_dict = {
|
|
"id": 1937529874,
|
|
"upload-time": "2008-10-13T10:40:02Z",
|
|
"polar-user": "https://www.polaraccesslink/v3/users/1",
|
|
"transaction-id": 179879,
|
|
"device": "Polar M400",
|
|
"device-id": "1111AAAA",
|
|
"start-time": "2008-10-13T10:40:02Z",
|
|
"start-time-utc-offset": 180,
|
|
"duration": "PT2H44M",
|
|
"calories": 530,
|
|
"distance": 1600,
|
|
"heart-rate": {
|
|
"average": 129,
|
|
"maximum": 147
|
|
},
|
|
"training-load": 143.22,
|
|
"sport": "OTHER",
|
|
"has-route": True,
|
|
"club-id": 999,
|
|
"club-name": "Polar Club",
|
|
"detailed-sport-info": "WALKING",
|
|
"fat-percentage": 60,
|
|
"carbohydrate-percentage": 38,
|
|
"protein-percentage": 2
|
|
}
|
|
|
|
with open('rowers/tests/testdata/c2workoutlist.txt') as f:
|
|
c2workoutlist = json.load(f)
|
|
|
|
c2uploadjson = {
|
|
"data": {
|
|
"id": 339,
|
|
"user_id": 1,
|
|
"date": "2015-08-05 13:15:41",
|
|
"timezone": "Europe/London",
|
|
"date_utc": "2015-08-05 12:15:41",
|
|
"distance": 5649,
|
|
"type": "rower",
|
|
"time": 8649,
|
|
"time_formatted": "14:24.9",
|
|
"workout_type": "JustRow",
|
|
"source": "ErgData",
|
|
"weight_class": "H",
|
|
"verified": True,
|
|
"ranked": False
|
|
}
|
|
}
|
|
|
|
with open('rowers/tests/testdata/stravaworkoutlist.txt') as f:
|
|
stravaworkoutlist = json.load(f)
|
|
|
|
with open('rowers/tests/testdata/sporttracksworkouts.txt') as f:
|
|
sporttracksworkoutlist = json.load(f)
|
|
|
|
#nkworkoutlist = json.load(open('rowers/tests/testdata/nkworkouts.txt'))
|
|
|
|
|
|
with open('rowers/tests/testdata/rkworkoutslist.txt','r') as f:
|
|
rkworkoutlistjson = json.load(f)
|
|
|
|
with open('rowers/tests/testdata/uaworkoutlist.txt','r') as f:
|
|
uaworkoutlistjson = json.load(f)
|
|
|
|
with open('rowers/tests/testdata/stravaworkoutsummary.txt','r') as f:
|
|
stravasummaryjson = json.load(f)
|
|
|
|
with open('rowers/tests/testdata/stravatimetestdata.txt','r') as f:
|
|
stravatimejson = json.load(f)
|
|
|
|
with open('rowers/tests/testdata/stravavelotestdata.txt','r') as f:
|
|
stravavelojson = json.load(f)
|
|
|
|
with open('rowers/tests/testdata/stravapowertestdata.txt','r') as f:
|
|
stravapowerjson = json.load(f)
|
|
|
|
with open('rowers/tests/testdata/stravadistancetestdata.txt','r') as f:
|
|
stravadistancejson = json.load(f)
|
|
|
|
with open('rowers/tests/testdata/stravalatlongtestdata.txt','r') as f:
|
|
stravalatlongjson = json.load(f)
|
|
|
|
with open('rowers/tests/testdata/stravahrtestdata.txt','r') as f:
|
|
stravahrjson = json.load(f)
|
|
|
|
with open('rowers/tests/testdata/stravaspmtestdata.txt','r') as f:
|
|
stravaspmjson = json.load(f)
|
|
|
|
|
|
with open('rowers/tests/testdata/uapostworkoutresponse.txt','r') as f:
|
|
uapostworkoutjson = json.load(f)
|
|
|
|
|
|
with open('rowers/tests/testdata/tpuploadresponse.txt','r') as f:
|
|
tpuploadresponse = json.load(f)
|
|
|
|
stravastreamjson = {
|
|
'time':stravatimejson,
|
|
'velocity_smooth':stravavelojson,
|
|
'distance':stravadistancejson,
|
|
'latlng':stravalatlongjson,
|
|
'heartrate':stravahrjson,
|
|
'cadence':stravaspmjson,
|
|
'watts':stravapowerjson,
|
|
}
|
|
|
|
with open('rowers/tests/testdata/sporttracksstrokedata.txt','r') as f:
|
|
ststrokesjson = json.load(f)
|
|
|
|
with open('rowers/tests/testdata/sporttracksstrokedatanolocation.txt','r') as f:
|
|
ststrokesjson_nogps = json.load(f)
|
|
|
|
with open('rowers/tests/testdata/rkstrokes.txt','r') as f:
|
|
rkstrokesjson = json.load(f)
|
|
|
|
with open('rowers/tests/testdata/uastrokes.txt','r') as f:
|
|
uastrokesjson = json.load(f)
|
|
|
|
with open('rowers/tests/testdata/uauser.txt','r') as f:
|
|
uauserjson = json.load(f)
|
|
|
|
with open('rowers/tests/testdata/strava_athlete.txt') as f:
|
|
stravaathletejson = json.load(f)
|
|
|
|
class MockResponse:
|
|
def __init__(self, json_data, status_code):
|
|
self.json_data = json_data
|
|
self.status_code = status_code
|
|
self.ok = True
|
|
self.headers = {
|
|
'Location':'MockLocation',
|
|
}
|
|
|
|
def json(self):
|
|
return self.json_data
|
|
|
|
class MockContentResponse:
|
|
def __init__(self,filename,status_code):
|
|
with open(filename,'rb') as f:
|
|
s = f.read()
|
|
self.content = s
|
|
self.status_code = status_code
|
|
self.ok = True
|
|
|
|
|
|
|
|
class MockHeaderResponse:
|
|
def __init__(self, header_data, status_code):
|
|
self.headers = header_data
|
|
self.status_code = status_code
|
|
|
|
class MockStreamResponse:
|
|
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):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_value, traceback):
|
|
self.raw.close()
|
|
|
|
|
|
def __delete__(self):
|
|
self.raw.close()
|
|
|
|
|
|
def __del__(self):
|
|
try:
|
|
self.raw.close()
|
|
except TypeError:
|
|
pass
|
|
|
|
class MockOAuth1Session:
|
|
def __init__(self,*args, **kwargs):
|
|
pass
|
|
|
|
def get(*args,**kwargs):
|
|
return MockStreamResponse('rowers/tests/testdata/3x250m.fit',200)
|
|
|
|
|
|
class MockSession:
|
|
class headers:
|
|
def __init__(self,*args,**kwargs):
|
|
pass
|
|
|
|
def update(self,*args,**kwargs):
|
|
pass
|
|
|
|
def post(self, *args, **kwargs):
|
|
json_data = {'id':1}
|
|
return MockResponse(json_data,200)
|
|
|
|
def send(self,prepped):
|
|
# prepped.url
|
|
# prepped.body (post data)
|
|
# prepped.headers
|
|
# prepped.method
|
|
|
|
json_data = {
|
|
'access_token': 'TA3n1vrNjuQJWw0TdCDHnjSmrjIPULhTlejMIWqq',
|
|
'expires_in': 604800,
|
|
'refresh_token': 'jHJhFzCfOOKB8oyiayubhLAlxaMkG3ruC1E8YxaR',
|
|
}
|
|
|
|
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()
|
|
|
|
if 'url' in kwargs:
|
|
if 'rp3' in kwargs['url']:
|
|
args = [kwargs['url']]
|
|
if "tofit" in kwargs['url']:
|
|
args = [kwargs['url']]
|
|
if "tojson" in kwargs['url']:
|
|
args = [kwargs['url']]
|
|
|
|
if not args:
|
|
return MockSession()
|
|
|
|
|
|
if "tofit" in args[0]:
|
|
jsonresponse = {
|
|
'name': '',
|
|
'sport': 'rowing',
|
|
'filename': '/home/sander/python/rowsandall/media/630a9e78-6d34-4eb3-8d53-4c02b2e95fff.fit',
|
|
'steps': [
|
|
{
|
|
'wkt_step_name': '0',
|
|
'stepId': 0,
|
|
'durationType': 'Distance',
|
|
'durationValue': 100000,
|
|
'intensity': 'Active'
|
|
},
|
|
{
|
|
'wkt_step_name': '1',
|
|
'stepId': 1,
|
|
'durationType': 'RepeatUntilStepsCmplt',
|
|
'targetValue': 4,
|
|
'durationValue': 0
|
|
}
|
|
]
|
|
}
|
|
|
|
|
|
return MockResponse(jsonresponse,200)
|
|
|
|
if 'rojabo' in args[0]:
|
|
return MockResponse(rojabodata, 200)
|
|
|
|
|
|
polartester = re.compile(r'.*?polaraccesslink\.com')
|
|
polarremotetester = re.compile(r'.*?polarremote\.com')
|
|
c2tester = re.compile(r'.*?log\.concept2\.com')
|
|
stravatester = re.compile(r'.*?strava\.com')
|
|
sttester = re.compile(r'.*?sporttracks\.mobi')
|
|
rktester = re.compile(r'.*?runkeeper\.com')
|
|
uatester = re.compile(r'.*?mapmyfitness\.com')
|
|
tptester = re.compile(r'.*?trainingpeaks\.com')
|
|
nktester = re.compile(r'.*?nkrowlink\.com')
|
|
rp3tester = re.compile(r'.*?rp3rowing-app\.com')
|
|
garmintester = re.compile(r'.*?garmin\.com')
|
|
fakturoidtester = re.compile(r'.*?fakturoid\.cz')
|
|
|
|
polarlistregex = r'.*?polaraccesslink\.com\/.*\/(\d+)$'
|
|
polarlisttester = re.compile(polarlistregex)
|
|
|
|
polarexerciseregex = r'.*?polaraccesslink\.com\/.*\/(\d+)\/exercises\/(\d+)$'
|
|
polarexercisetester = re.compile(polarexerciseregex)
|
|
|
|
polartcxregex = r'.*?polaraccesslink\.com\/.*\/(\d+)\/tcx'
|
|
polartcxtester = re.compile(polartcxregex)
|
|
|
|
polaruserregex = r'.*?polaraccesslink\.com\/.*\/users$'
|
|
polarusertester = re.compile(polaruserregex)
|
|
|
|
polarnotificationregex = r'.*polaraccesslink\.com\/.*\/notifications'
|
|
polarnotificationtester = re.compile(polarnotificationregex)
|
|
|
|
c2importregex = r'.*?concept2.com\/api\/users\/me\/results\/\d+'
|
|
c2importtester = re.compile(c2importregex)
|
|
|
|
c2uploadregex = r'.*?concept2.com\/api\/users\/\d+\/results$'
|
|
c2uploadtester = re.compile(c2uploadregex)
|
|
|
|
c2strokesregex = r'.*?concept2.com\/api\/users\/me\/results\/\d+\/strokes'
|
|
c2strokestester = re.compile(c2strokesregex)
|
|
|
|
c2workoutlistregex = r'.*?concept2\.com\/api\/users\/me\/results\?page=\d'
|
|
c2workoutlisttester = re.compile(c2workoutlistregex)
|
|
|
|
nkworkoutlistregex = r'.*?nkrowlink\.com\/api\/v1\/sessions'
|
|
nkworkoutlisttester = re.compile(nkworkoutlistregex)
|
|
|
|
nkstrokesregex = r'.*?nkrowlink\.com\/api\/v1\/sessions\/strokes'
|
|
nkstrokestester = re.compile(nkstrokesregex)
|
|
|
|
rp3graphregex = r'.*?rp3rowing-app\.com\/graphql'
|
|
rp3graphtester = re.compile(rp3graphregex)
|
|
|
|
stravaathleteregex = r'.*?strava\.com\/api\/v3\/athlete$'
|
|
stravaathletetester = re.compile(stravaathleteregex)
|
|
|
|
|
|
stravaworkoutlistregex = r'.*?strava\.com\/api\/v3\/athlete\/activities'
|
|
stravaworkoutlisttester = re.compile(stravaworkoutlistregex)
|
|
|
|
stravastreamregex = r'.*?strava\.com\/api\/v3\/activities\/\d+\/streams\/(\w.*?)\?'
|
|
stravastreamtester = re.compile(stravastreamregex)
|
|
|
|
stravasummaryregex = r'.*?strava\.com\/api\/v3\/activities\/\d+$'
|
|
stravasummarytester = re.compile(stravasummaryregex)
|
|
|
|
stuploadregex = r'.*?sporttracks\.mobi\/api\/v2\/fitnessActivities.json$'
|
|
stuploadtester = re.compile(stuploadregex)
|
|
|
|
stworkoutlistregex = r'.*?sporttracks\.mobi\/api\/v2\/fitnessActivities$'
|
|
stworkoutlisttester = re.compile(stworkoutlistregex)
|
|
|
|
ststrokesregex = r'.*?sporttracks\.mobi\/api\/v2\/fitnessActivities/(\d+)$'
|
|
ststrokestester = re.compile(ststrokesregex)
|
|
|
|
rkuploadregex = r'.*?api\.runkeeper\.com\/fitnessActivities$'
|
|
rkuploadtester = re.compile(rkuploadregex)
|
|
rkuserregex = r'.*?api\.runkeeper\.com\/user$'
|
|
rkusertester = re.compile(rkuserregex)
|
|
rkstrokesregex = r'.*?api\.runkeeper\.com\/fitnessActivities/\d+$'
|
|
rkstrokestester = re.compile(rkstrokesregex)
|
|
|
|
uaapiregex = r'.*?api\.ua\.com'
|
|
uaapitester = re.compile(uaapiregex)
|
|
|
|
uauploadregex = r'.*?api\.ua\.com\/v7.1\/workout\/$'
|
|
uauploadtester = re.compile(uauploadregex)
|
|
|
|
uastrokesregex = r'.*?api\.ua\.com\/v7.1\/workout\/\d+'
|
|
uastrokestester = re.compile(uastrokesregex)
|
|
|
|
ualistregex = r'.*?api\.ua\.com\/v7.1\/workout\/\?user'
|
|
ualisttester = re.compile(ualistregex)
|
|
|
|
uauserregex = r'.*?api\.ua\.com\/v7.1\/user\/self\/'
|
|
uausertester = re.compile(uauserregex)
|
|
|
|
tpuploadregex = r'.*?trainingpeaks\.com\/v3\/file'
|
|
tpuploadtester = re.compile(tpuploadregex)
|
|
|
|
garmindownloadregex = r'.*?garmin\.com\/mockfile?id=1'
|
|
garmindownloadtester = re.compile(garmindownloadregex)
|
|
|
|
garmintrainingregex = r'.*?garmin\.com\/training-api\/workout'
|
|
garmintrainingtester = re.compile(garmintrainingregex)
|
|
|
|
garmintrainingscheduleregex = r'.*?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
|
|
return MockResponse(json_data,200)
|
|
|
|
if polarremotetester.match(args[0]):
|
|
json_data = {
|
|
'access_token':'aap',
|
|
'expires_in':31535999,
|
|
'x_user_id':475,
|
|
}
|
|
return MockResponse(json_data,200)
|
|
|
|
if polartester.match(args[0]):
|
|
if polarnotificationtester.match(args[0]):
|
|
json_data = {
|
|
"available-user-data": [
|
|
{
|
|
"user-id": 475,
|
|
"data-type": "ACTIVITY_SUMMARY",
|
|
"url": "https://www.polaraccesslink.com/v3/users/475/activity-transactions"
|
|
},
|
|
{
|
|
'user-id': 475,
|
|
'data-type': 'EXERCISE',
|
|
'url': 'https://polaraccesslink.com/v3/users/40273947/exercise-transactions'
|
|
}
|
|
]
|
|
}
|
|
return MockResponse(json_data,200)
|
|
elif polarusertester.match(args[0]):
|
|
json_data = polar_user_data
|
|
return MockResponse(json_data,200)
|
|
elif polartcxtester.match(args[0]):
|
|
filename = 'rowers/tests/testdata/polar_response.tcx'
|
|
return MockContentResponse(filename,200)
|
|
elif polarexercisetester.match(args[0]):
|
|
json_data = polar_exercise_dict
|
|
return MockResponse(json_data,200)
|
|
elif polarlisttester.match(args[0]):
|
|
json_data = polar_exercise_list
|
|
return MockResponse(json_data,200)
|
|
|
|
# getting user data
|
|
json_data = polar_json
|
|
|
|
return MockResponse(json_data,201)
|
|
|
|
if tptester.match(args[0]):
|
|
if 'token' in args[0]:
|
|
json_data = {
|
|
'access_token': 'TA3n1vrNjuQJWw0TdCDHnjSmrjIPULhTlejMIWqq',
|
|
'expires_in': 604800,
|
|
'refresh_token': 'jHJhFzCfOOKB8oyiayubhLAlxaMkG3ruC1E8YxaR'
|
|
}
|
|
return MockResponse(json_data,200)
|
|
elif tpuploadtester.match(args[0]):
|
|
return MockResponse(tpuploadresponse,200)
|
|
|
|
if uaapitester.match(args[0]):
|
|
if 'access_token' in args[0]:
|
|
json_data = {
|
|
'access_token': 'TA3n1vrNjuQJWw0TdCDHnjSmrjIPULhTlejMIWqq',
|
|
'expires_in': 604800,
|
|
'refresh_token': 'jHJhFzCfOOKB8oyiayubhLAlxaMkG3ruC1E8YxaR'
|
|
}
|
|
return MockResponse(json_data,200)
|
|
elif uauploadtester.match(args[0]):
|
|
if 'data' in kwargs:
|
|
return MockResponse(uapostworkoutjson,200)
|
|
elif uastrokestester.match(args[0]):
|
|
return MockResponse(uastrokesjson,200)
|
|
elif ualisttester.match(args[0]):
|
|
return MockResponse(uaworkoutlistjson,200)
|
|
elif uausertester.match(args[0]):
|
|
return MockResponse(uauserjson,200)
|
|
|
|
|
|
if uatester.match(args[0]):
|
|
if 'access_token' in args[0]:
|
|
json_data = {
|
|
'access_token': 'TA3n1vrNjuQJWw0TdCDHnjSmrjIPULhTlejMIWqq',
|
|
'expires_in': 604800,
|
|
'refresh_token': 'jHJhFzCfOOKB8oyiayubhLAlxaMkG3ruC1E8YxaR'
|
|
}
|
|
return MockResponse(json_data,200)
|
|
|
|
|
|
if rktester.match(args[0]):
|
|
if 'token' in args[0]:
|
|
json_data = {
|
|
'access_token': 'TA3n1vrNjuQJWw0TdCDHnjSmrjIPULhTlejMIWqq',
|
|
'expires_in': 604800,
|
|
'refresh_token': 'jHJhFzCfOOKB8oyiayubhLAlxaMkG3ruC1E8YxaR'
|
|
}
|
|
return MockResponse(json_data,200)
|
|
elif rkuploadtester.match(args[0]):
|
|
if 'data' in kwargs:
|
|
# post
|
|
header_data = {
|
|
'Location': '/fitnessActivities/12'
|
|
}
|
|
return MockHeaderResponse(header_data,200)
|
|
else:
|
|
json_data = rkworkoutlistjson
|
|
return MockResponse(json_data,200)
|
|
elif rkusertester.match(args[0]):
|
|
json_data = {
|
|
"userID": 1234567890,
|
|
"profile": "/profile",
|
|
"settings": "/settings",
|
|
"fitness_activities": "/fitnessActivities",
|
|
"strength_training_activities": "/strengthTrainingActivities",
|
|
"background_activities": "/backgroundActivities",
|
|
"sleep": "/sleep",
|
|
"nutrition": "/nutrition",
|
|
"weight": "/weight",
|
|
"general_measurements": "/generalMeasurements",
|
|
"diabetes": "/diabetes",
|
|
"records": "/records",
|
|
"team": "/team"
|
|
}
|
|
return MockResponse(json_data, 200)
|
|
elif rkstrokestester.match(args[0]):
|
|
return MockResponse(rkstrokesjson,200)
|
|
|
|
if sttester.match(args[0]):
|
|
if 'oauth2/token' in args[0]:
|
|
json_data = {
|
|
'access_token': 'TA3n1vrNjuQJWw0TdCDHnjSmrjIPULhTlejMIWqq',
|
|
'expires_in': 604800,
|
|
'refresh_token': 'jHJhFzCfOOKB8oyiayubhLAlxaMkG3ruC1E8YxaR'
|
|
}
|
|
return MockResponse(json_data,200)
|
|
if ststrokestester.match(args[0]):
|
|
if ststrokestester.match(args[0]).group(1) == '13':
|
|
return MockResponse(ststrokesjson_nogps,200)
|
|
else:
|
|
return MockResponse(ststrokesjson,200)
|
|
if stuploadtester.match(args[0]):
|
|
json_data = {
|
|
"uris": [
|
|
"https://api.sporttracks.mobi/api/v2/fitnessActivities/123456.json"
|
|
]
|
|
}
|
|
return MockResponse(json_data, 200)
|
|
if stworkoutlisttester.match(args[0]):
|
|
return MockResponse(sporttracksworkoutlist,200)
|
|
|
|
|
|
if nktester.match(args[0]):
|
|
if 'token' in args[0]:
|
|
json_data = {
|
|
'access_token': 'TA3n1vrNjuQJWw0TdCDHnjSmrjIPULhTlejMIWqq',
|
|
'expires_in': 604800,
|
|
'refresh_token': 'jHJhFzCfOOKB8oyiayubhLAlxaMkG3ruC1E8YxaR'
|
|
}
|
|
return MockResponse(json_data,200)
|
|
if nkstrokestester.match(args[0]):
|
|
params = kwargs.pop('params',{})
|
|
|
|
if 'sessionIds' in params and params['sessionIds'] == '404':
|
|
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':
|
|
return MockResponse(nkimpellerworkoutlist,200)
|
|
return MockResponse(nkworkoutlist,200)
|
|
|
|
if rp3tester.match(args[0]):
|
|
if 'token' in args[0]:
|
|
json_data = {
|
|
'access_token': 'TA3n1vrNjuQJWw0TdCDHnjSmrjIPULhTlejMIWqq',
|
|
'expires_in': 604800,
|
|
'refresh_token': 'jHJhFzCfOOKB8oyiayubhLAlxaMkG3ruC1E8YxaR'
|
|
}
|
|
return MockResponse(json_data,200)
|
|
|
|
if 'api' in args[0]:
|
|
if '591621' in args[0]:
|
|
return MockContentResponse('rowers/tests/testdata/rp3content.csv',200)
|
|
|
|
|
|
if rp3graphtester.match(args[0]):
|
|
if 'json' in kwargs:
|
|
try:
|
|
query = kwargs['json']['query']
|
|
if 'download' in query:
|
|
return MockResponse(rp3linkready,200)
|
|
except KeyError:
|
|
pass
|
|
json_data = rp3workoutlist
|
|
return MockResponse(json_data,200)
|
|
|
|
|
|
|
|
|
|
|
|
if stravatester.match(args[0]):
|
|
if stravaworkoutlisttester.match(args[0]):
|
|
return MockResponse(stravaworkoutlist,200)
|
|
if stravastreamtester.match(args[0]):
|
|
metric = stravastreamtester.match(args[0]).group(1)
|
|
json_data = stravastreamjson[metric]
|
|
return MockResponse(json_data,200)
|
|
elif stravasummarytester.match(args[0]):
|
|
return MockResponse(stravasummaryjson,200)
|
|
elif 'token' in args[0]:
|
|
json_data = {
|
|
"token_type": "Bearer",
|
|
"access_token": "987654321234567898765432123456789",
|
|
"refresh_token": "1234567898765432112345678987654321",
|
|
"expires_at": arrow.now().timestamp()+7200
|
|
}
|
|
|
|
return MockResponse(json_data,200)
|
|
|
|
|
|
if c2tester.match(args[0]):
|
|
if c2uploadtester.match(args[0]):
|
|
return MockResponse(c2uploadjson,201)
|
|
if c2strokestester.match(args[0]):
|
|
return MockResponse(c2strokedata,200)
|
|
elif c2importtester.match(args[0]):
|
|
if '12' in args[0]:
|
|
return MockResponse(c2workoutdata,200)
|
|
elif '31' in args[0]:
|
|
return MockResponse(c2timezoneworkoutdata2,200)
|
|
elif '32' in args[0]:
|
|
return MockResponse(c2timezoneworkoutdatabad,200)
|
|
else:
|
|
return MockResponse(c2timezoneworkoutdata,200)
|
|
elif c2workoutlisttester.match(args[0]):
|
|
return MockResponse(c2workoutlist,200)
|
|
elif 'access_token' in args[0]:
|
|
json_data = {
|
|
'access_token': 'TA3n1vrNjuQJWw0TdCDHnjSmrjIPULhTlejMIWqq',
|
|
'expires_in': 604800,
|
|
'refresh_token': 'jHJhFzCfOOKB8oyiayubhLAlxaMkG3ruC1E8YxaR'
|
|
}
|
|
return MockResponse(json_data,200)
|
|
elif 'users/me' in args[0]:
|
|
json_data = {
|
|
'data': {
|
|
'username': 'john',
|
|
'id': 1234,
|
|
}
|
|
}
|
|
return MockResponse(json_data,200)
|
|
elif 'results' in args[0]:
|
|
json_data = {
|
|
'data': {
|
|
'id': 1223,
|
|
}
|
|
}
|
|
else:
|
|
return MockResponse(c2workoutdata,200)
|
|
|
|
if fakturoidtester.match(args[0]):
|
|
if 'invoices' in args[0]:
|
|
response = {
|
|
'url':'aap',
|
|
'id':1,
|
|
}
|
|
return MockResponse(response,200)
|
|
|
|
response = [
|
|
{
|
|
'id':1,
|
|
'url':'aap',
|
|
}
|
|
]
|
|
return MockResponse(response,200)
|
|
|
|
return MockResponse(None,404)
|
|
|
|
class MockEmailMessage:
|
|
def __init__(*args,**kwargs):
|
|
pass
|
|
|
|
def send(self):
|
|
return 1
|
|
|
|
class MockResponse:
|
|
def __init__(self, json_data, status_code):
|
|
self.json_data = json_data
|
|
self.status_code = status_code
|
|
self.ok = True
|
|
self.headers = {
|
|
'Location':'MockLocation',
|
|
}
|
|
|
|
def json(self):
|
|
return self.json_data
|
|
|
|
|
|
|
|
class MockOAuth1Session:
|
|
def __init__(self,*args, **kwargs):
|
|
pass
|
|
|
|
def get(*args,**kwargs):
|
|
return MockStreamResponse('rowers/tests/testdata/3x250m.fit',200)
|
|
|
|
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
|