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 class fakejob: def __init__(self): self.id = 1 class fakerequest: def __init__(self): self.id = 1 self.job = fakejob() # 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) for w in self.user_workouts: res = dataprep.setcp(w) 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 tearDown(self): for workout in self.user_workouts: try: os.remove(workout.csvfilename) except (IOError,OSError,FileNotFoundError): pass for filename in os.listdir('rowers/tests/testdata/temp'): path = os.path.join('rowers/tests/testdata/temp/',filename) if not os.path.isdir(path): try: os.remove(path) except: pass def test_summaryfromsplitdata(self): splitdata = [ { "type": "distance", "time": 415, "rest_time": 600, "stroke_rate": 35, "distance": 220, "heart_rate": { "ending": 160, "rest": 60 } }, { "type": "distance", "time": 347, "rest_time": 600, "stroke_rate": 45, "distance": 220, "heart_rate": { "ending": 170, "rest": 70 } } ] data = { "date": "2015-08-30 14:24:00", "timezone": "Europe/London", "distance": 440, "time": 762, "type": "rower", "weight_class": "H", "heart_rate": { "average": 140 }, "workout_type": "FixedDistanceInterval", "rest_distance": 43, "rest_time": 1200, "workout": { "targets": { "stroke_rate": 30, "heart_rate_zone": 4, "pace": 1050 }, "intervals": [ { "type": "distance", "time": 415, "rest_time": 600, "stroke_rate": 35, "distance": 220, "heart_rate": { "ending": 160, "rest": 60 } }, { "type": "distance", "time": 347, "rest_time": 600, "stroke_rate": 45, "distance": 220, "heart_rate": { "ending": 170, "rest": 70 } } ] } } res = tasks.summaryfromsplitdata(splitdata,data,'test.csv') self.assertEqual(len(res[0]),478) @patch('rowers.tasks.stravalib.Client', side_effect=MockStravalibClient) def test_strava_sync(self,MockStravalibClient): res = tasks.handle_strava_sync('aap',self.wwater.id,self.wwater.csvfilename,'aap','activitytype','aa') self.assertEqual(res,1) 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.tasks.requests.get',side_effect=mocked_requests) @patch('rowers.tasks.requests.post',side_effect=mocked_requests) @patch('rowers.tasks.requests.session',side_effect=mocked_requests) def test_fetch_strava_workout(self, mock_get, mock_post, mock_Session): res = tasks.fetch_strava_workout('aap',None,12,'rowers/tests/testdata/temp/tesmp.csv', self.u.id) self.assertEqual(res,1) @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) res = tasks.handle_sendemail_coachrequest(useremail,username,'sdsd','Fred') self.assertEqual(res,1) res = tasks.handle_sendemail_coachoffer_rejected(useremail,username,username) self.assertEqual(res,1) res = tasks.handle_sendemail_coachoffer_accepted(useremail,username,username) self.assertEqual(res,1) res = tasks.handle_sendemail_coachrequest_rejected(useremail,username,username) self.assertEqual(res,1) res = tasks.handle_sendemail_coachrequest_accepted(useremail,username,username) self.assertEqual(res,1) res = tasks.handle_sendemail_coacheerequest(useremail,username,'aaee',username) self.assertEqual(res,1) res = tasks.handle_sendemail_invite(useremail,username,'ss','team','Fred Hachee') self.assertEqual(res,1) res = tasks.handle_sendemailnewresponse( userfirstname, userlastname, useremail, userfirstname,userlastname, 'er staat een paard in de gang', self.wwater.name, self.wwater.id, 1 ) self.assertEqual(res,1) res = tasks.handle_sendemailnewcomment( userfirstname, userlastname, useremail, userfirstname,userlastname, 'er staat een paard in de gang', self.wwater.name, self.wwater.id, 1 ) self.assertEqual(res,1) res = tasks.handle_send_template_email('aa.html',useremail,useremail,userfirstname,'aa','bb') self.assertEqual(res,1) res = tasks.handle_sendemail_message(useremail,useremail,userfirstname,'aap', 'noot',userlastname) self.assertEqual(res,1) res = tasks.handle_sendemail_request(useremail,username,'asas','asas',userfirstname,12) self.assertEqual(res,1) res = tasks.handle_sendemail_request_accept(useremail,username,'asas',userfirstname,12) self.assertEqual(res,1) res = tasks.handle_sendemail_request_reject(useremail,username,'asas',userfirstname,12) self.assertEqual(res,1) res = tasks.handle_sendemail_member_dropped(useremail,username,'asas',userfirstname,12) self.assertEqual(res,1) res = tasks.handle_sendemail_team_removed(useremail,username,'asas',userfirstname,12) self.assertEqual(res,1) res = tasks.handle_sendemail_invite_reject(useremail,username,'asas',userfirstname,12) self.assertEqual(res,1) res = tasks.handle_sendemail_invite_accept(useremail,username,'asas',userfirstname,12) 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) @patch('rowers.tasks.grpc',side_effect=mocked_grpc) @patch('rowers.tasks.send_template_email',side_effect=mocked_send_template_email) def test_handle_otwsetpower(self,mocked_send_template_email,mocked_grpc): f1 = get_random_file(filename='rowers/tests/testdata/sprintervals.csv')['filename'] boattype = '1x' boatclass = 'water' coastalbrand = 'other' weightvalue = 80. first_name = self.u.first_name last_name = self.u.last_name email = self.u.email workoutid = self.wwater.id job = fakerequest() res = tasks.handle_otwsetpower(f1,boattype,boatclass,coastalbrand, weightvalue,first_name,last_name,email,workoutid, jobkey='23') self.assertEqual(res,1) @patch('rowers.dataprepnodjango.create_engine') def test_handle_updateergcp(self,mocked_sqlalchemy): f1 = get_random_file()['filename'] res = tasks.handle_updateergcp(1,[f1]) self.assertEqual(res,1) @patch('rowers.dataprepnodjango.getsmallrowdata_db') def test_cp_from_workoutids(self,mocked_getsmallrowdata_db): ids = [1] powerfourmin,power2k,powerhour = tasks.cp_from_workoutids(ids) self.assertFalse(powerfourmin==0) self.assertFalse(power2k==0) self.assertFalse(powerhour==0) @patch('rowers.dataprepnodjango.getsmallrowdata_db') def test_handle_updatecp(self,mocked_getsmallrowdata_db): rower_id = 1 workoutids = [1] res = tasks.handle_updatecp(rower_id,workoutids) self.assertEqual(res,1) @patch('rowers.dataprepnodjango.getsmallrowdata_db') def test_handle_setcp(self,mocked_getsmallrowdata_db): strokesdf = pd.read_csv('rowers/tests/testdata/uhfull.csv') filename = 'rowers/tests/testdata/temp/pq.gz' workoutids = 1 res = tasks.handle_setcp(strokesdf,filename,1) self.assertEqual(res,1) try: os.remove(filename) except FileNotFoundError: pass