283 lines
11 KiB
Python
283 lines
11 KiB
Python
from __future__ import absolute_import
|
|
from __future__ import division
|
|
from __future__ import print_function
|
|
from __future__ import unicode_literals
|
|
|
|
from .statements import *
|
|
from rowers.mytypes import rowtypes
|
|
import pandas as pd
|
|
|
|
nu = datetime.datetime.now()
|
|
from rowers import tasks
|
|
|
|
# asynchronous tasks
|
|
class AsyncTaskTests(TestCase):
|
|
def setUp(self):
|
|
self.u = UserFactory()
|
|
|
|
self.r = Rower.objects.create(user=self.u,
|
|
birthdate=faker.profile()['birthdate'],
|
|
gdproptin=True,surveydone=True,
|
|
gdproptindate=timezone.now(),
|
|
rowerplan='coach')
|
|
|
|
self.c = Client()
|
|
self.user_workouts = WorkoutFactory.create_batch(5, user=self.r)
|
|
self.factory = RequestFactory()
|
|
self.password = faker.word()
|
|
self.u.set_password(self.password)
|
|
self.u.save()
|
|
|
|
result = get_random_file(filename='rowers/tests/testdata/onwater2.csv')
|
|
|
|
self.wwater = WorkoutFactory(user=self.r,
|
|
csvfilename=result['filename'],
|
|
starttime=result['starttime'],
|
|
startdatetime=result['startdatetime'],
|
|
duration=result['duration'],
|
|
distance=result['totaldist'],
|
|
workouttype = 'water',
|
|
)
|
|
|
|
def test_safetimedelta(self):
|
|
x = 5
|
|
y = tasks.safetimedelta(x)
|
|
self.assertEqual(datetime.timedelta(seconds=5),y)
|
|
|
|
x = np.nan
|
|
y = tasks.safetimedelta(x)
|
|
self.assertEqual(datetime.timedelta(seconds=0),y)
|
|
|
|
@patch('rowers.c2stuff.requests.post', side_effect=mocked_requests)
|
|
@patch('rowers.c2stuff.requests.get', side_effect=mocked_requests)
|
|
def test_c2_sync(self, mock_get, mock_post):
|
|
authorizationstring = str('Bearer aap')
|
|
headers = {'Authorization': authorizationstring,
|
|
'user-agent': 'sanderroosendaal',
|
|
'Content-Type': 'application/json'}
|
|
|
|
data = c2stuff.createc2workoutdata(self.wwater)
|
|
|
|
url = "https://log.concept2.com/api/users/%s/results" % (1)
|
|
res = tasks.handle_c2_sync(1,url,headers,data)
|
|
self.assertEqual(res,1)
|
|
|
|
# to do - mock the grpc channel and metrics_pb2
|
|
@patch('rowers.dataprep.create_engine')
|
|
def test_handle_calctrimp(self, mocked_sqlalchemy):
|
|
result = get_random_file()
|
|
res = tasks.handle_calctrimp(1,result['filename'],200,'male',160,90,52)
|
|
self.assertEqual(res,1)
|
|
|
|
@patch('rowers.tasks.EmailMessage',side_effect=MockEmailMessage)
|
|
def test_handle_updatedps(self,MockEmailMessage):
|
|
result = get_random_file()
|
|
filename = result['filename']
|
|
res = tasks.handle_updatedps('roosendaalsander@gmail.com',[(1,filename)])
|
|
|
|
self.assertEqual(res,1)
|
|
|
|
@patch('rowers.tasks.send_template_email',side_effect=mocked_send_template_email)
|
|
def test_handle_send_email_alert(self,mocked_send_template_email):
|
|
useremail = self.u.email
|
|
userfirstname = self.u.first_name
|
|
userlastname = self.u.last_name,
|
|
rowerfirstname = self.u.first_name,
|
|
alertname = 'Test Alert'
|
|
stats = {
|
|
'percentage': 45,
|
|
'workouts': 3,
|
|
'nr_strokes_qualifying': 23,
|
|
'nr_strokes': 5,
|
|
'median': 33.23,
|
|
'median_q': 32.121,
|
|
'startdate': '2020-01-01',
|
|
'enddate': '2020-23-01',
|
|
}
|
|
|
|
res = tasks.handle_send_email_alert(useremail,userfirstname,userlastname,
|
|
rowerfirstname,alertname,stats)
|
|
|
|
self.assertEqual(res,1)
|
|
|
|
@patch('rowers.tasks.send_template_email',side_effect=mocked_send_template_email)
|
|
def test_handle_send_email_transactions(self,mocked_send_template_email):
|
|
res = tasks.handle_send_email_transaction(self.u.username,self.u.email,23)
|
|
self.assertEqual(res,1)
|
|
|
|
res = tasks.handle_send_email_failed_cancel(self.u.first_name,self.u.email,self.u.username,23)
|
|
self.assertEqual(res,1)
|
|
|
|
res = tasks.handle_send_email_subscription_update(
|
|
self.u.first_name,self.u.email,'testplan',True,23,24,'2020-12-02','down')
|
|
self.assertEqual(res,1)
|
|
|
|
res = tasks.handle_send_email_subscription_update(
|
|
self.u.first_name,self.u.email,'testplan',True,23,24,'2020-12-02','up')
|
|
self.assertEqual(res,1)
|
|
|
|
res = tasks.handle_send_email_subscription_create(
|
|
self.u.username,self.u.email,'pro',True,24,24,'2020-12-02'
|
|
)
|
|
self.assertEqual(res,1)
|
|
|
|
res = tasks.handle_sendemail_expired(self.u.email,self.u.first_name,self.u.last_name,
|
|
'2020-12-12')
|
|
self.assertEqual(res,1)
|
|
|
|
@patch('rowers.tasks.send_template_email',side_effect=mocked_send_template_email)
|
|
def test_handle_raceemails(self,mocked_send_template_email):
|
|
u = self.u
|
|
useremail = u.email
|
|
username = u.first_name
|
|
racename = 'Hop'
|
|
raceid = 12
|
|
registeredname = 'Jaap'
|
|
|
|
res = tasks.handle_sendemail_raceregistration(useremail,username,registeredname,racename,raceid)
|
|
self.assertEqual(res,1)
|
|
|
|
result = get_random_file()
|
|
filename = result['filename']
|
|
|
|
res = tasks.handle_sendemail_coursefail(useremail,username,filename)
|
|
self.assertEqual(res,1)
|
|
|
|
res = tasks.handle_sendemail_optout(useremail,username,registeredname,racename,raceid)
|
|
self.assertEqual(res,1)
|
|
|
|
res = tasks.handle_sendemail_racesubmission(useremail,username,registeredname,racename,raceid)
|
|
self.assertEqual(res,1)
|
|
|
|
res = tasks.handle_send_disqualification_email(useremail,username,'daarom','omdat',racename)
|
|
self.assertEqual(res,1)
|
|
|
|
res = tasks.handle_send_withdraw_email(useremail,username,'daarom','message',racename)
|
|
self.assertEqual(res,1)
|
|
|
|
@patch('rowers.tasks.send_template_email',side_effect=mocked_send_template_email)
|
|
def test_handle_otheremails(self,mocked_send_template_email):
|
|
u = self.u
|
|
useremail = u.email
|
|
username = u.first_name
|
|
userfirstname = u.first_name
|
|
userlastname = u.last_name
|
|
filename = get_random_file()['filename']
|
|
|
|
btvalues = pd.DataFrame({
|
|
'delta':[3,1,3],
|
|
'cpvalues':[100,200,300],
|
|
'pwr':[100,200,300]
|
|
}).to_json()
|
|
|
|
res = tasks.handle_sendemail_breakthrough(1,useremail,userfirstname,userlastname,
|
|
btvalues=btvalues)
|
|
|
|
self.assertEqual(res,1)
|
|
|
|
res = tasks.handle_sendemail_hard(1,useremail,userfirstname,userlastname,
|
|
btvalues=btvalues)
|
|
self.assertEqual(res,1)
|
|
|
|
res = tasks.handle_sendemail_userdeleted(username,useremail)
|
|
self.assertEqual(res,1)
|
|
|
|
res = tasks.handle_sendemail_unrecognized(filename,useremail)
|
|
self.assertEqual(res,1)
|
|
|
|
res = tasks.handle_sendemail_unrecognizedowner(useremail,userfirstname)
|
|
self.assertEqual(res,1)
|
|
|
|
filename = get_random_file()['filename']
|
|
res = tasks.handle_sendemailics(userfirstname,userlastname,useremail,filename)
|
|
self.assertEqual(res,1)
|
|
|
|
filename = get_random_file()['filename']
|
|
res = tasks.handle_sendemailkml(userfirstname,userlastname,useremail,filename)
|
|
self.assertEqual(res,1)
|
|
|
|
filename = get_random_file()['filename']
|
|
res = tasks.handle_sendemailtcx(userfirstname,userlastname,useremail,filename)
|
|
self.assertEqual(res,1)
|
|
|
|
filename = get_random_file()['filename']
|
|
res = tasks.handle_sendemailsummary(userfirstname,userlastname,useremail,filename)
|
|
self.assertEqual(res,1)
|
|
|
|
filename = get_random_file()['filename']
|
|
res = tasks.handle_sendemailcsv(userfirstname,userlastname,useremail,filename)
|
|
self.assertEqual(res,1)
|
|
|
|
filename = get_random_file()['filename']
|
|
res = tasks.handle_sendemail_ical(userfirstname,userlastname,useremail,'url',filename)
|
|
self.assertEqual(res,1)
|
|
|
|
filename = get_random_file()['filename']
|
|
res = tasks.handle_sendemailfile(userfirstname,userlastname,useremail,filename)
|
|
self.assertEqual(res,1)
|
|
|
|
|
|
def test_sigdig(self):
|
|
x = 3.14159
|
|
|
|
y = tasks.sigdig(x,digits=2)
|
|
self.assertEqual(y,'3.1')
|
|
|
|
y = tasks.sigdig(x)
|
|
self.assertEqual(y,'3.14')
|
|
|
|
y = tasks.sigdig(x,digits=0)
|
|
self.assertEqual(y,'0')
|
|
|
|
y = tasks.sigdig(x,digits=1)
|
|
self.assertEqual(y,'3')
|
|
|
|
y = tasks.sigdig(x,digits=4)
|
|
self.assertEqual(y,'3.142')
|
|
|
|
|
|
@patch('rowers.sporttracksstuff.requests.post', side_effect=mocked_requests)
|
|
@patch('rowers.sporttracksstuff.requests.get', side_effect=mocked_requests)
|
|
def test_sporttracks_sync(self, mock_get, mock_post):
|
|
authorizationstring = str('Bearer aap')
|
|
headers = {'Authorization': authorizationstring,
|
|
'user-agent': 'sanderroosendaal',
|
|
'Content-Type': 'application/json'}
|
|
|
|
url = "https://api.sporttracks.mobi/api/v2/fitnessActivities.json"
|
|
|
|
data = sporttracksstuff.createsporttracksworkoutdata(self.wwater)
|
|
|
|
|
|
res = tasks.handle_sporttracks_sync(1,url,headers,json.dumps(data,default=sporttracksstuff.default))
|
|
self.assertEqual(res,1)
|
|
|
|
@patch('rowers.runkeeperstuff.requests.post', side_effect=mocked_requests)
|
|
@patch('rowers.runkeeperstuff.requests.get', side_effect=mocked_requests)
|
|
def test_runkeeper_sync(self, mock_get, mock_post):
|
|
authorizationstring = str('Bearer aap')
|
|
headers = {'Authorization': authorizationstring,
|
|
'user-agent': 'sanderroosendaal',
|
|
'Content-Type': 'application/vnd.com.runkeeper.NewFitnessActivity+json',
|
|
'Content-Length':'nnn'}
|
|
|
|
url = "https://api.runkeeper.com/fitnessActivities"
|
|
|
|
data = runkeeperstuff.createrunkeeperworkoutdata(self.wwater)
|
|
|
|
|
|
res = tasks.handle_runkeeper_sync(1,url,headers,json.dumps(data,default=sporttracksstuff.default))
|
|
self.assertEqual(res,1)
|
|
|
|
@patch('rowers.c2stuff.requests.post',side_effect=mocked_requests)
|
|
@patch('rowers.c2stuff.requests.get',side_effect=mocked_requests)
|
|
def test_import_c2_strokedata(self, mock_get, mock_post):
|
|
c2token = 'aap'
|
|
c2id = 1212
|
|
workoutid = 1
|
|
starttimeunix = 121
|
|
csvfilename = self.wwater.csvfilename
|
|
|
|
res = tasks.handle_c2_import_stroke_data(c2token,c2id,workoutid,starttimeunix,csvfilename)
|
|
self.assertEqual(res,1)
|