patching sqlalchemy to make tests work
This commit is contained in:
Sander Roosendaal
418
rowers/tests/mocks.py
Normal file
418
rowers/tests/mocks.py
Normal file
@@ -0,0 +1,418 @@
|
||||
import pytest
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
from bs4 import BeautifulSoup
|
||||
import re
|
||||
from nose_parameterized import parameterized
|
||||
from django.test import TestCase, Client,override_settings
|
||||
from django.core.management import call_command
|
||||
from django.utils.six import StringIO
|
||||
from django.test.client import RequestFactory
|
||||
from rowers.views import checkworkoutuser,c2_open
|
||||
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 rowers.c2stuff as c2stuff
|
||||
|
||||
import json
|
||||
import numpy as np
|
||||
|
||||
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()
|
||||
|
||||
from django_mailbox.models import Mailbox,MessageAttachment,Message
|
||||
|
||||
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 MockConnection:
|
||||
def begin(self):
|
||||
return True
|
||||
|
||||
def execute(self):
|
||||
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()
|
||||
|
||||
#@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)
|
||||
|
||||
# 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/c2jsonstrokedata.txt','r') as infile:
|
||||
c2strokedata = json.load(infile)
|
||||
|
||||
polar_json = {
|
||||
'available-user-data': []
|
||||
}
|
||||
|
||||
c2workoutlist = json.load(open('rowers/tests/testdata/c2workoutlist.txt'))
|
||||
|
||||
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
|
||||
}
|
||||
}
|
||||
|
||||
stravaworkoutlist = json.load(open('rowers/tests/testdata/stravaworkoutlist.txt'))
|
||||
sporttracksworkoutlist = json.load(open('rowers/tests/testdata/sporttracksworkouts.txt'))
|
||||
|
||||
rkworkoutlistjson = json.load(open('rowers/tests/testdata/rkworkoutslist.txt','r'))
|
||||
uaworkoutlistjson = json.load(open('rowers/tests/testdata/uaworkoutlist.txt','r'))
|
||||
|
||||
stravasummaryjson = json.load(open('rowers/tests/testdata/stravaworkoutsummary.txt','r'))
|
||||
|
||||
stravatimejson = json.load(open('rowers/tests/testdata/stravatimetestdata.txt','r'))
|
||||
stravavelojson = json.load(open('rowers/tests/testdata/stravavelotestdata.txt','r'))
|
||||
|
||||
stravapowerjson = json.load(open('rowers/tests/testdata/stravapowertestdata.txt','r'))
|
||||
|
||||
stravadistancejson = json.load(open('rowers/tests/testdata/stravadistancetestdata.txt','r'))
|
||||
stravalatlongjson = json.load(open('rowers/tests/testdata/stravalatlongtestdata.txt','r'))
|
||||
stravahrjson = json.load(open('rowers/tests/testdata/stravahrtestdata.txt','r'))
|
||||
stravaspmjson = json.load(open('rowers/tests/testdata/stravaspmtestdata.txt','r'))
|
||||
|
||||
uapostworkoutjson = json.load(open('rowers/tests/testdata/uapostworkoutresponse.txt','r'))
|
||||
|
||||
tpuploadresponse = json.load(open('rowers/tests/testdata/tpuploadresponse.txt','r'))
|
||||
|
||||
stravastreamjson = {
|
||||
'time':stravatimejson,
|
||||
'velocity_smooth':stravavelojson,
|
||||
'distance':stravadistancejson,
|
||||
'latlng':stravalatlongjson,
|
||||
'heartrate':stravahrjson,
|
||||
'cadence':stravaspmjson,
|
||||
'power':stravapowerjson,
|
||||
}
|
||||
|
||||
ststrokesjson = json.load(open('rowers/tests/testdata/sporttracksstrokedata.txt','r'))
|
||||
|
||||
rkstrokesjson = json.load(open('rowers/tests/testdata/rkstrokes.txt','r'))
|
||||
|
||||
uastrokesjson = json.load(open('rowers/tests/testdata/uastrokes.txt','r'))
|
||||
uauserjson = json.load(open('rowers/tests/testdata/uauser.txt','r'))
|
||||
|
||||
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
|
||||
|
||||
|
||||
class MockHeaderResponse:
|
||||
def __init__(self, header_data, status_code):
|
||||
self.headers = header_data
|
||||
self.status_code = status_code
|
||||
|
||||
class MockSession:
|
||||
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 not args:
|
||||
return MockSession()
|
||||
|
||||
polartester = re.compile('.*?polaraccesslink\.com')
|
||||
c2tester = re.compile('.*?log\.concept2\.com')
|
||||
stravatester = re.compile('.*?strava\.com')
|
||||
sttester = re.compile('.*?sporttracks\.mobi')
|
||||
rktester = re.compile('.*?runkeeper\.com')
|
||||
uatester = re.compile('.*?mapmyfitness\.com')
|
||||
tptester = re.compile('.*?trainingpeaks\.com')
|
||||
|
||||
c2importregex = '.*?concept2.com\/api\/users\/me\/results\/\d+'
|
||||
c2importtester = re.compile(c2importregex)
|
||||
|
||||
c2uploadregex = '.*?concept2.com\/api\/users\/\d+\/results$'
|
||||
c2uploadtester = re.compile(c2uploadregex)
|
||||
|
||||
c2strokesregex = '.*?concept2.com\/api\/users\/me\/results\/\d+\/strokes'
|
||||
c2strokestester = re.compile(c2strokesregex)
|
||||
|
||||
c2workoutlistregex = '.*?concept2\.com\/api\/users\/me\/results\?page=\d'
|
||||
c2workoutlisttester = re.compile(c2workoutlistregex)
|
||||
|
||||
stravaworkoutlistregex = '.*?strava\.com\/api\/v3\/athlete\/activities'
|
||||
stravaworkoutlisttester = re.compile(stravaworkoutlistregex)
|
||||
|
||||
stravastreamregex = '.*?strava\.com\/api\/v3\/activities\/\d+\/streams\/(\w.*?)\?'
|
||||
stravastreamtester = re.compile(stravastreamregex)
|
||||
|
||||
stravasummaryregex = '.*?strava\.com\/api\/v3\/activities\/\d+$'
|
||||
stravasummarytester = re.compile(stravasummaryregex)
|
||||
|
||||
stuploadregex = '.*?sporttracks\.mobi\/api\/v2\/fitnessActivities.json$'
|
||||
stuploadtester = re.compile(stuploadregex)
|
||||
|
||||
stworkoutlistregex = '.*?sporttracks\.mobi\/api\/v2\/fitnessActivities$'
|
||||
stworkoutlisttester = re.compile(stworkoutlistregex)
|
||||
|
||||
ststrokesregex = '.*?sporttracks\.mobi\/api\/v2\/fitnessActivities/\d+$'
|
||||
ststrokestester = re.compile(ststrokesregex)
|
||||
|
||||
rkuploadregex = '.*?api\.runkeeper\.com\/fitnessActivities$'
|
||||
rkuploadtester = re.compile(rkuploadregex)
|
||||
rkuserregex = '.*?api\.runkeeper\.com\/user$'
|
||||
rkusertester = re.compile(rkuserregex)
|
||||
rkstrokesregex = '.*?api\.runkeeper\.com\/fitnessActivities/\d+$'
|
||||
rkstrokestester = re.compile(rkstrokesregex)
|
||||
|
||||
uaapiregex = '.*?api\.ua\.com'
|
||||
uaapitester = re.compile(uaapiregex)
|
||||
|
||||
uauploadregex = '.*?api\.ua\.com\/v7.1\/workout\/$'
|
||||
uauploadtester = re.compile(uauploadregex)
|
||||
|
||||
uastrokesregex = '.*?api\.ua\.com\/v7.1\/workout\/\d+'
|
||||
uastrokestester = re.compile(uastrokesregex)
|
||||
|
||||
ualistregex = '.*?api\.ua\.com\/v7.1\/workout\/\?user'
|
||||
ualisttester = re.compile(ualistregex)
|
||||
|
||||
uauserregex = '.*?api\.ua\.com\/v7.1\/user\/self\/'
|
||||
uausertester = re.compile(uauserregex)
|
||||
|
||||
tpuploadregex = '.*?trainingpeaks\.com\/v1\/file'
|
||||
tpuploadtester = re.compile(tpuploadregex)
|
||||
|
||||
if polartester.match(args[0]):
|
||||
json_data = polar_json
|
||||
return MockResponse(json_data,200)
|
||||
|
||||
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]):
|
||||
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 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": 1531385304
|
||||
}
|
||||
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]):
|
||||
return MockResponse(c2workoutdata,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)
|
||||
|
||||
return MockResponse(None,404)
|
||||
@@ -37,7 +37,9 @@ import json
|
||||
import numpy as np
|
||||
|
||||
from rowers import urls
|
||||
from rowers.views import error500_view,error404_view,error400_view,error403_view
|
||||
from rowers.views import (
|
||||
error500_view,error404_view,error400_view,error403_view
|
||||
)
|
||||
|
||||
from rowers.dataprep import delete_strokedata
|
||||
|
||||
@@ -46,343 +48,9 @@ redis_connection = StrictRedis()
|
||||
|
||||
from django_mailbox.models import Mailbox,MessageAttachment,Message
|
||||
|
||||
#@pytest.mark.django_db
|
||||
class DjangoTestCase(TestCase, MockTestCase):
|
||||
def _pre_setup(self):
|
||||
MockTestCase.setUp(self)
|
||||
TestCase._pre_setup(self)
|
||||
self.client = Client()
|
||||
from rowers.tests.mocks import *
|
||||
|
||||
def _post_teardown(self):
|
||||
TestCase._post_teardown(self)
|
||||
MockTestCase.tearDown(self)
|
||||
delete_strokedata(1)
|
||||
|
||||
# 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/c2jsonstrokedata.txt','r') as infile:
|
||||
c2strokedata = json.load(infile)
|
||||
|
||||
polar_json = {
|
||||
'available-user-data': []
|
||||
}
|
||||
|
||||
c2workoutlist = json.load(open('rowers/tests/testdata/c2workoutlist.txt'))
|
||||
|
||||
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
|
||||
}
|
||||
}
|
||||
|
||||
stravaworkoutlist = json.load(open('rowers/tests/testdata/stravaworkoutlist.txt'))
|
||||
sporttracksworkoutlist = json.load(open('rowers/tests/testdata/sporttracksworkouts.txt'))
|
||||
|
||||
rkworkoutlistjson = json.load(open('rowers/tests/testdata/rkworkoutslist.txt','r'))
|
||||
uaworkoutlistjson = json.load(open('rowers/tests/testdata/uaworkoutlist.txt','r'))
|
||||
|
||||
stravasummaryjson = json.load(open('rowers/tests/testdata/stravaworkoutsummary.txt','r'))
|
||||
|
||||
stravatimejson = json.load(open('rowers/tests/testdata/stravatimetestdata.txt','r'))
|
||||
stravavelojson = json.load(open('rowers/tests/testdata/stravavelotestdata.txt','r'))
|
||||
|
||||
stravapowerjson = json.load(open('rowers/tests/testdata/stravapowertestdata.txt','r'))
|
||||
|
||||
stravadistancejson = json.load(open('rowers/tests/testdata/stravadistancetestdata.txt','r'))
|
||||
stravalatlongjson = json.load(open('rowers/tests/testdata/stravalatlongtestdata.txt','r'))
|
||||
stravahrjson = json.load(open('rowers/tests/testdata/stravahrtestdata.txt','r'))
|
||||
stravaspmjson = json.load(open('rowers/tests/testdata/stravaspmtestdata.txt','r'))
|
||||
|
||||
uapostworkoutjson = json.load(open('rowers/tests/testdata/uapostworkoutresponse.txt','r'))
|
||||
|
||||
tpuploadresponse = json.load(open('rowers/tests/testdata/tpuploadresponse.txt','r'))
|
||||
|
||||
stravastreamjson = {
|
||||
'time':stravatimejson,
|
||||
'velocity_smooth':stravavelojson,
|
||||
'distance':stravadistancejson,
|
||||
'latlng':stravalatlongjson,
|
||||
'heartrate':stravahrjson,
|
||||
'cadence':stravaspmjson,
|
||||
'power':stravapowerjson,
|
||||
}
|
||||
|
||||
ststrokesjson = json.load(open('rowers/tests/testdata/sporttracksstrokedata.txt','r'))
|
||||
|
||||
rkstrokesjson = json.load(open('rowers/tests/testdata/rkstrokes.txt','r'))
|
||||
|
||||
uastrokesjson = json.load(open('rowers/tests/testdata/uastrokes.txt','r'))
|
||||
uauserjson = json.load(open('rowers/tests/testdata/uauser.txt','r'))
|
||||
|
||||
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
|
||||
|
||||
|
||||
class MockHeaderResponse:
|
||||
def __init__(self, header_data, status_code):
|
||||
self.headers = header_data
|
||||
self.status_code = status_code
|
||||
|
||||
class MockSession:
|
||||
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 not args:
|
||||
return MockSession()
|
||||
|
||||
polartester = re.compile('.*?polaraccesslink\.com')
|
||||
c2tester = re.compile('.*?log\.concept2\.com')
|
||||
stravatester = re.compile('.*?strava\.com')
|
||||
sttester = re.compile('.*?sporttracks\.mobi')
|
||||
rktester = re.compile('.*?runkeeper\.com')
|
||||
uatester = re.compile('.*?mapmyfitness\.com')
|
||||
tptester = re.compile('.*?trainingpeaks\.com')
|
||||
|
||||
c2importregex = '.*?concept2.com\/api\/users\/me\/results\/\d+'
|
||||
c2importtester = re.compile(c2importregex)
|
||||
|
||||
c2uploadregex = '.*?concept2.com\/api\/users\/\d+\/results$'
|
||||
c2uploadtester = re.compile(c2uploadregex)
|
||||
|
||||
c2strokesregex = '.*?concept2.com\/api\/users\/me\/results\/\d+\/strokes'
|
||||
c2strokestester = re.compile(c2strokesregex)
|
||||
|
||||
c2workoutlistregex = '.*?concept2\.com\/api\/users\/me\/results\?page=\d'
|
||||
c2workoutlisttester = re.compile(c2workoutlistregex)
|
||||
|
||||
stravaworkoutlistregex = '.*?strava\.com\/api\/v3\/athlete\/activities'
|
||||
stravaworkoutlisttester = re.compile(stravaworkoutlistregex)
|
||||
|
||||
stravastreamregex = '.*?strava\.com\/api\/v3\/activities\/\d+\/streams\/(\w.*?)\?'
|
||||
stravastreamtester = re.compile(stravastreamregex)
|
||||
|
||||
stravasummaryregex = '.*?strava\.com\/api\/v3\/activities\/\d+$'
|
||||
stravasummarytester = re.compile(stravasummaryregex)
|
||||
|
||||
stuploadregex = '.*?sporttracks\.mobi\/api\/v2\/fitnessActivities.json$'
|
||||
stuploadtester = re.compile(stuploadregex)
|
||||
|
||||
stworkoutlistregex = '.*?sporttracks\.mobi\/api\/v2\/fitnessActivities$'
|
||||
stworkoutlisttester = re.compile(stworkoutlistregex)
|
||||
|
||||
ststrokesregex = '.*?sporttracks\.mobi\/api\/v2\/fitnessActivities/\d+$'
|
||||
ststrokestester = re.compile(ststrokesregex)
|
||||
|
||||
rkuploadregex = '.*?api\.runkeeper\.com\/fitnessActivities$'
|
||||
rkuploadtester = re.compile(rkuploadregex)
|
||||
rkuserregex = '.*?api\.runkeeper\.com\/user$'
|
||||
rkusertester = re.compile(rkuserregex)
|
||||
rkstrokesregex = '.*?api\.runkeeper\.com\/fitnessActivities/\d+$'
|
||||
rkstrokestester = re.compile(rkstrokesregex)
|
||||
|
||||
uaapiregex = '.*?api\.ua\.com'
|
||||
uaapitester = re.compile(uaapiregex)
|
||||
|
||||
uauploadregex = '.*?api\.ua\.com\/v7.1\/workout\/$'
|
||||
uauploadtester = re.compile(uauploadregex)
|
||||
|
||||
uastrokesregex = '.*?api\.ua\.com\/v7.1\/workout\/\d+'
|
||||
uastrokestester = re.compile(uastrokesregex)
|
||||
|
||||
ualistregex = '.*?api\.ua\.com\/v7.1\/workout\/\?user'
|
||||
ualisttester = re.compile(ualistregex)
|
||||
|
||||
uauserregex = '.*?api\.ua\.com\/v7.1\/user\/self\/'
|
||||
uausertester = re.compile(uauserregex)
|
||||
|
||||
tpuploadregex = '.*?trainingpeaks\.com\/v1\/file'
|
||||
tpuploadtester = re.compile(tpuploadregex)
|
||||
|
||||
if polartester.match(args[0]):
|
||||
json_data = polar_json
|
||||
return MockResponse(json_data,200)
|
||||
|
||||
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]):
|
||||
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 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": 1531385304
|
||||
}
|
||||
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]):
|
||||
return MockResponse(c2workoutdata,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)
|
||||
|
||||
return MockResponse(None,404)
|
||||
|
||||
#@pytest.mark.django_db
|
||||
@pytest.mark.django_db
|
||||
class C2Objects(DjangoTestCase):
|
||||
def setUp(self):
|
||||
self.c = Client()
|
||||
@@ -472,7 +140,8 @@ class C2Objects(DjangoTestCase):
|
||||
self.assertEqual(response.status_code,200)
|
||||
|
||||
@patch('rowers.c2stuff.requests.get', side_effect=mocked_requests)
|
||||
def test_c2_import(self, mock_get):
|
||||
@patch('rowers.dataprep.create_engine')
|
||||
def test_c2_import(self, mock_get, mocked_sqlalchemy):
|
||||
|
||||
response = self.c.get('/rowers/workout/c2import/12/',follow=True)
|
||||
|
||||
@@ -578,7 +247,8 @@ class C2ObjectsTokenExpired(DjangoTestCase):
|
||||
self.assertEqual(response.status_code,200)
|
||||
|
||||
@patch('rowers.c2stuff.requests.get', side_effect=mocked_requests)
|
||||
def test_c2_import(self, mock_get):
|
||||
@patch('rowers.dataprep.create_engine')
|
||||
def test_c2_import(self, mock_get, mocked_sqlalchemy):
|
||||
|
||||
response = self.c.get('/rowers/workout/c2import/12/',follow=True)
|
||||
|
||||
@@ -839,7 +509,7 @@ class RunKeeperObjects(DjangoTestCase):
|
||||
|
||||
|
||||
|
||||
#@pytest.mark.django_db
|
||||
@pytest.mark.django_db
|
||||
class UAObjects(DjangoTestCase):
|
||||
def setUp(self):
|
||||
self.c = Client()
|
||||
@@ -929,7 +599,8 @@ class UAObjects(DjangoTestCase):
|
||||
self.assertEqual(response.status_code,200)
|
||||
|
||||
@patch('rowers.imports.requests.get', side_effect=mocked_requests)
|
||||
def test_underarmour_import(self, mock_get):
|
||||
@patch('rowers.dataprep.create_engine')
|
||||
def test_underarmour_import(self, mock_get, mocked_sqlalchemy):
|
||||
|
||||
response = self.c.get('/rowers/workout/underarmourimport/12/',follow=True)
|
||||
|
||||
|
||||
@@ -49,8 +49,10 @@ from rowers.tests.test_imports import mocked_requests
|
||||
|
||||
from django_mailbox.models import Mailbox,MessageAttachment,Message
|
||||
|
||||
from rowers.tests.mocks import mocked_sqlalchemy
|
||||
|
||||
#@pytest.mark.django_db
|
||||
|
||||
@pytest.mark.django_db
|
||||
class UploadTests(TestCase):
|
||||
def setUp(self):
|
||||
redis_connection.publish('tasks','KILL')
|
||||
@@ -151,7 +153,8 @@ workout water
|
||||
pass
|
||||
|
||||
@patch('requests.get', side_effect=mocked_requests)
|
||||
def test_emailprocessing(self, mock_get):
|
||||
@patch('rowers.dataprep.create_engine')
|
||||
def test_emailprocessing(self, mock_get, mocked_sqlalchemy):
|
||||
out = StringIO()
|
||||
call_command('processemail', stdout=out,testing=True)
|
||||
self.assertIn('Successfully processed email attachments',out.getvalue())
|
||||
|
||||
Reference in New Issue
Block a user