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)) 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 arrow 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() from django_mailbox.models import Mailbox,MessageAttachment,Message 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 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) 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_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 def mocked_getsmallrowdata_db(*args, **kwargs): df = pd.read_csv('rowers/tests/testdata/colsfromdb.csv') return df def mocked_getsmallrowdata_db_water(*args, **kwargs): df = pd.read_csv('rowers/tests/testdata/colsfromdb.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 def StravaActivity(): def wait(*args, **kwargs): return 1 class MockStravalibClient(): def upload_activity(*args, **kwargs): return StravaActivity() def update_activity(*args, **kwargs): return StravaActivity() # 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 = 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 customer(): def find(*arg, **kwargs): return self def create(*args, **kwargs): return gatewayresult(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' # 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() 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/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')) ststrokesjson_nogps = json.load(open('rowers/tests/testdata/sporttracksstrokedatanolocation.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]): if ststrokestester.match(args[0]).group(1) == '13': return MockResponse(ststrokesjson_nogps,200) else: return MockResponse(strokesjson,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": arrow.now().timestamp+3600 } 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)