Merge branch 'develop' into feature/idoklad

rowers/tests/test_api.py

@@ -22,12 +22,454 @@ from rowers.opaque import encoder
 
 from rest_framework.test import APIRequestFactory, force_authenticate
 
+UPLOAD_SERVICE_URL = '/rowers/workout/api/upload/'
+UPLOAD_SERVICE_SECRET = "FoYezZWLSyfAVimumpHEeYsJjsNCerxV"
+
 import json
 
+# import BeautifulSoup
+from bs4 import BeautifulSoup
+
 from rowers.ownapistuff import *
 from rowers.views.apiviews import *
 from rowers.models import APIKey
 
+from rowers.teams import add_member, add_coach
+from rowers.views.analysisviews import histodata
+
+class TeamFactory(factory.DjangoModelFactory):
+    class Meta:
+        model = Team
+
+    name = factory.LazyAttribute(lambda _: faker.word())
+    notes = faker.text()
+    private = 'open'
+    viewing = 'allmembers'
+
+class StravaPrivacy(TestCase):
+    def setUp(self):
+        self.u = UserFactory()
+        self.u2 = UserFactory()
+        self.u3 = UserFactory()
+
+        self.r = Rower.objects.create(user=self.u,
+                                      birthdate=faker.profile()['birthdate'],
+                                      gdproptin=True, ftpset=True,surveydone=True,
+                                      gdproptindate=timezone.now(),
+                                      rowerplan='coach',subscription_id=1)
+        
+        self.r.stravatoken = '12'
+        self.r.stravarefreshtoken = '123'
+        self.r.stravatokenexpirydate = arrow.get(datetime.datetime.now()-datetime.timedelta(days=1)).datetime
+        self.r.strava_owner_id = 4
+
+        self.r.save()
+
+        self.c = Client()
+
+        self.factory = RequestFactory()
+        self.password = faker.word()
+        self.u.set_password(self.password)
+        self.u.save()
+        self.factory = APIRequestFactory()
+        
+        self.r2 = Rower.objects.create(user=self.u2,
+                                       birthdate=faker.profile()['birthdate'],
+                                       gdproptin=True, ftpset=True,surveydone=True,
+                                       gdproptindate=timezone.now(),
+                                       rowerplan='coach',clubsize=3)
+
+        self.r3 = Rower.objects.create(user=self.u3,
+                                       birthdate=faker.profile()['birthdate'],
+                                       gdproptin=True, ftpset=True,surveydone=True,
+                                       gdproptindate=timezone.now(),
+                                       rowerplan='basic')
+        
+        self.c = Client()
+
+        self.password2 = faker.word()
+        self.u2.set_password(self.password2)
+        self.u2.save()
+
+        self.password3 = faker.word()
+        self.u3.set_password(self.password3)
+        self.u3.save()
+
+        self.team = TeamFactory(manager=self.u2)
+
+        # all are team members
+        add_member(self.team.id, self.r)
+        add_member(self.team.id, self.r2)
+        add_member(self.team.id, self.r3)
+
+        self.user_workouts = WorkoutFactory.create_batch(5, user=self.r)
+        for w in self.user_workouts:
+            if w.id <= 2:
+                w.workoutsource = 'strava'
+                w.privacy = 'hidden'
+            elif w.id == 3: # user can change privacy but cannot change workoutsource
+                w.workoutsource = 'strava'
+                w.privacy = 'visible'
+            else:
+                w.workoutsource = 'concept2'
+                w.privacy = 'visible'
+            w.team.add(self.team)
+            w.csvfilename = get_random_file(filename='rowers/tests/testdata/thyro.csv')['filename']
+            w.save()
+
+        # r2 coaches r
+        add_coach(self.r2, self.r)
+
+        self.factory = APIRequestFactory()
+
+    def tearDown(self):
+        for workout in self.user_workouts:
+            try:
+                os.remove(workout.csvfilename)
+            except (OSError, FileNotFoundError, IOError):
+                pass
+
+    # Test if workout with workoutsource strava and privacy hidden can be seen by coach
+    def test_privacy_coach(self):
+        login = self.c.login(username=self.u2.username, password=self.password2)
+        self.assertTrue(login)
+
+        w = self.user_workouts[0]
+        url = reverse('workout_view',kwargs={'id':encoder.encode_hex(w.id)})
+        response = self.c.get(url)
+        self.assertEqual(response.status_code,403)
+
+    # Same test as above but for 'workout_edit_view'
+    def test_privacy_coach_edit(self):
+        login = self.c.login(username=self.u2.username, password=self.password2)
+        self.assertTrue(login)
+
+        w = self.user_workouts[0]
+        url = reverse('workout_edit_view',kwargs={'id':encoder.encode_hex(w.id)})
+        response = self.c.get(url)
+        self.assertEqual(response.status_code,403)
+
+    # Test if workout with workoutsource strava and privacy hidden can be seen by team member
+    def test_privacy_member(self):
+        login = self.c.login(username=self.u3.username, password=self.password3)
+        self.assertTrue(login)
+
+        w = self.user_workouts[0]
+        url = reverse('workout_view',kwargs={'id':encoder.encode_hex(w.id)})
+        response = self.c.get(url)
+        self.assertEqual(response.status_code,403)
+
+    # Same test as above but for 'workout_edit_view'
+    def test_privacy_member_edit(self):
+        login = self.c.login(username=self.u3.username, password=self.password3)
+        self.assertTrue(login)
+
+        w = self.user_workouts[0]
+        url = reverse('workout_edit_view',kwargs={'id':encoder.encode_hex(w.id)})
+        response = self.c.get(url)
+        self.assertEqual(response.status_code,403)
+
+    # same test as above but with user r and the response code should be 200
+    def test_privacy_owner(self):
+        login = self.c.login(username=self.u.username, password=self.password)
+        self.assertTrue(login)
+        
+        w = self.user_workouts[0]
+        url = reverse('workout_view',kwargs={'id':encoder.encode_hex(w.id)})
+        response = self.c.get(url)
+        self.assertEqual(response.status_code,200)
+
+    # same test as above but for 'workout_edit_view'
+    def test_privacy_owner_edit(self):
+        login = self.c.login(username=self.u.username, password=self.password)
+        self.assertTrue(login)
+        
+        w = self.user_workouts[0]
+        url = reverse('workout_edit_view',kwargs={'id':encoder.encode_hex(w.id)})
+        response = self.c.get(url)
+        self.assertEqual(response.status_code,200)
+
+                
+
+    # test if list_workouts returns all workouts for user r
+    def test_list_workouts(self):
+        login = self.c.login(username=self.u.username, password=self.password)
+        self.assertTrue(login)
+
+        url = reverse('workouts_view')
+        response = self.c.get(url)
+        self.assertEqual(response.status_code,200)
+
+        # the response.content is html, so we need to parse it
+        soup = BeautifulSoup(response.content, 'html.parser')
+        # the workouts look like <a href="/rowers/workout/{id}/...">...</a> and there should be 5 unique ids
+        # the id is a hex string
+        workouts = set([a['href'].split('/')[3] for a in soup.find_all('a') if a['href'].startswith('/rowers/workout/')])
+
+        # throw out "c2import", "nkimport", "stravaimport", "concept2import", "sporttracksimport" from the set
+        workouts = set([w for w in workouts if w not in [
+            'upload', 'addmanual', 'c2import', 'polarimport', 'rp3import', 'nkimport', 'stravaimport', 'concept2import', 'sporttracksimport',
+            'intervalsimport']])
+        
+        self.assertEqual(len(workouts),5)
+
+        
+    # same test as above but list_workouts with team id = self.team.id
+    def test_list_workouts_team(self):
+        login = self.c.login(username=self.u.username, password=self.password)
+        self.assertTrue(login)
+
+        url = reverse('workouts_view',kwargs={'teamid':self.team.id})
+        response = self.c.get(url)
+        self.assertEqual(response.status_code,200)
+
+        # the response.content is html, so we need to parse it
+        soup = BeautifulSoup(response.content, 'html.parser')
+        # the workouts look like <a href="/rowers/workout/{id}/...">...</a> and there should be 5 unique ids
+        # the id is a hex string
+        workouts = set([a['href'].split('/')[3] for a in soup.find_all('a') if a['href'].startswith('/rowers/workout/')])
+
+        # throw out "c2import", "nkimport", "stravaimport", "concept2import", "sporttracksimport" from the set
+        workouts = set([w for w in workouts if w not in [
+            'upload', 'addmanual', 'c2import', 'polarimport', 'rp3import', 'nkimport', 'stravaimport', 'concept2import', 'sporttracksimport',
+            'intervalsimport']])
+        
+        self.assertEqual(len(workouts),2)
+
+    # same test as the previous one but with self.r2 and the number of workouts found should 0
+    def test_list_workouts_team_coach(self):
+        login = self.c.login(username=self.u2.username, password=self.password2)
+        self.assertTrue(login)
+
+        url = reverse('workouts_view',kwargs={'teamid':self.team.id})
+        response = self.c.get(url)
+        self.assertEqual(response.status_code,200)
+
+        # the response.content is html, so we need to parse it
+        soup = BeautifulSoup(response.content, 'html.parser')
+        # the workouts look like <a href="/rowers/workout/{id}/...">...</a> and there should be 5 unique ids
+        # the id is a hex string
+        workouts = set([a['href'].split('/')[3] for a in soup.find_all('a') if a['href'].startswith('/rowers/workout/')])
+
+        # throw out "c2import", "nkimport", "stravaimport", "concept2import", "sporttracksimport" from the set
+        workouts = set([w for w in workouts if w not in [
+            'upload', 'addmanual', 'c2import', 'polarimport', 'rp3import', 'nkimport', 'stravaimport', 'concept2import', 'sporttracksimport',
+            'intervalsimport']])
+        
+        self.assertEqual(len(workouts),2)
+
+    # same test as above but with without the teamid kwarg but with a rowerid=self.r.id
+    def test_list_workouts_team_coach2(self):
+        login = self.c.login(username=self.u2.username, password=self.password2)
+        self.assertTrue(login)
+
+        url = reverse('workouts_view',kwargs={'rowerid':self.r.id})
+        response = self.c.get(url)
+        self.assertEqual(response.status_code,200)
+
+        # the response.content is html, so we need to parse it
+        soup = BeautifulSoup(response.content, 'html.parser')
+        # the workouts look like <a href="/rowers/workout/{id}/...">...</a> and there should be 5 unique ids
+        # the id is a hex string
+        workouts = set([a['href'].split('/')[3] for a in soup.find_all('a') if a['href'].startswith('/rowers/workout/')])
+
+        # throw out "c2import", "nkimport", "stravaimport", "concept2import", "sporttracksimport" from the set
+        workouts = set([w for w in workouts if w not in [
+            'upload', 'addmanual', 'c2import', 'polarimport', 'rp3import', 'nkimport', 'stravaimport', 'concept2import', 'sporttracksimport',
+            'intervalsimport']])
+        
+        self.assertEqual(len(workouts),2)
+    
+    # same test as the previous one but with self.r3 and the number of workouts found should 0
+    def test_list_workouts_team_member(self):
+        login = self.c.login(username=self.u3.username, password=self.password3)
+        self.assertTrue(login)
+
+        url = reverse('workouts_view',kwargs={'teamid':self.team.id})
+        response = self.c.get(url)
+        self.assertEqual(response.status_code,200)
+
+        # the response.content is html, so we need to parse it
+        soup = BeautifulSoup(response.content, 'html.parser')
+        # the workouts look like <a href="/rowers/workout/{id}/...">...</a> and there should be 5 unique ids
+        # the id is a hex string
+        workouts = set([a['href'].split('/')[3] for a in soup.find_all('a') if a['href'].startswith('/rowers/workout/')])
+
+        # throw out "c2import", "nkimport", "stravaimport", "concept2import", "sporttracksimport" from the set
+        workouts = set([w for w in workouts if w not in [
+            'upload', 'addmanual', 'c2import', 'polarimport', 'rp3import', 'nkimport', 'stravaimport', 'concept2import', 'sporttracksimport',
+            'intervalsimport']])
+        
+        self.assertEqual(len(workouts),2)
+
+    # now test strava import and test if the created workout has workoutsource strava and privacy hidden
+    @patch('rowers.utils.requests.get', side_effect=mocked_requests)
+    @patch('rowers.integrations.strava.requests.post', side_effect=mocked_requests)
+    @patch('rowers.dataprep.read_data')
+    def test_stravaimport(self, mock_get, mock_post, mocked_read_data):
+        login = self.c.login(username=self.u.username, password=self.password)
+        self.assertTrue(login)
+
+        # remove all self.workouts
+        Workout.objects.filter(user=self.r).delete()
+
+        # create a workout using dataprep.new_workout_from_file with workoutsource = strava
+        result = get_random_file(filename='rowers/tests/testdata/thyro.csv')
+        workout_id, message, filename = dataprep.new_workout_from_file(self.r, result['filename'],
+                                                                       workoutsource='strava', makeprivate=True)
+
+        # check if the workout was created
+        ws = Workout.objects.filter(user=self.r)
+        self.assertEqual(len(ws),1)
+        w = ws[0]
+        self.assertEqual(w.workoutsource,'strava')
+        self.assertEqual(w.privacy,'hidden')
+
+    # same as test above but makeprivate = False
+    @patch('rowers.utils.requests.get', side_effect=mocked_requests)
+    @patch('rowers.integrations.strava.requests.post', side_effect=mocked_requests)
+    @patch('rowers.dataprep.read_data')
+    def test_stravaimport_public(self, mock_get, mock_post, mocked_read_data):
+        login = self.c.login(username=self.u.username, password=self.password)
+        self.assertTrue(login)
+
+        # remove all self.workouts
+        Workout.objects.filter(user=self.r).delete()
+
+        # create a workout using dataprep.new_workout_from_file with workoutsource = strava
+        result = get_random_file(filename='rowers/tests/testdata/thyro.csv')
+        workout_id, message, filename = dataprep.new_workout_from_file(self.r, result['filename'],
+                                                                       workoutsource='strava', makeprivate=False)
+
+        # check if the workout was created
+        ws = Workout.objects.filter(user=self.r)
+        self.assertEqual(len(ws),1)
+        w = ws[0]
+        self.assertEqual(w.workoutsource,'strava')
+        self.assertEqual(w.privacy,'hidden')
+
+
+    # test ownapi with stravaid = '122'
+    def test_ownapi(self):
+        # remove all self.workouts
+        Workout.objects.filter(user=self.r).delete()
+        
+        result = get_random_file(filename='rowers/tests/testdata/thyro.csv')
+        uploadoptions = {
+            'workouttype': 'water',
+            'boattype': '1x',
+            'notes': 'A test file upload',
+            'stravaid': '122',
+            'secret': UPLOAD_SERVICE_SECRET,
+            'user': self.u.id,
+            'file': result['filename'],
+        }
+        url = reverse('workout_upload_api')
+        response = self.c.post(url, uploadoptions)
+        self.assertEqual(response.status_code,200)
+
+        # check if the workout was created
+        ws = Workout.objects.filter(user=self.r)
+        self.assertEqual(len(ws),1)
+        w = ws[0]
+        self.assertEqual(w.workoutsource,'strava')
+        self.assertEqual(w.privacy,'hidden')
+
+
+    # test some analysis, should only use the workouts with workoutsource != strava
+    #@patch('rowers.dataprep.read_data', side_effect=mocked_read_data)
+    #def test_workouts_analysis(self, mocked_read_data):
+    def test_workouts_analysis(self):
+        login = self.c.login(username=self.u.username, password=self.password)
+        self.assertTrue(login)
+
+        url = '/rowers/history/'
+        response = self.c.get(url)
+        self.assertEqual(response.status_code,200)
+
+        url = '/rowers/history/data/'
+        response = self.c.get(url)
+        self.assertEqual(response.status_code,200)
+        # response.json() has a key "script" with a javascript script
+        # check if this is correct
+        self.assertTrue('script' in response.json())
+
+        # now check histogram
+        startdate = (self.user_workouts[0].startdatetime-datetime.timedelta(days=3)).date()
+        enddate = (self.user_workouts[0].startdatetime+datetime.timedelta(days=3)).date()
+
+        # make sure the dates are not naive
+        try:
+            startdate = pytz.utc.localize(startdate)
+        except (ValueError, AttributeError):
+            pass
+        try:
+            enddate = pytz.utc.localize(enddate)
+        except (ValueError, AttributeError):
+            pass
+
+        form_data = {
+            'function':'histo',
+            'xparam':'hr',
+            'plotfield':'spm',
+            'yparam':'pace',
+            'groupby':'spm',
+            'palette':'monochrome_blue',
+            'xaxis':'time',
+            'yaxis1':'power',
+            'yaxis2':'hr',
+            'startdate':startdate,
+            'enddate':enddate,
+            'plottype':'scatter',
+            'spmmin':15,
+            'spmmax':55,
+            'workmin':0,
+            'workmax':1500,
+            'includereststrokes':False,
+            'modality':'all',
+            'waterboattype':['1x','2x','4x'],
+            'userid':self.u.id,
+            'workouts':[w.id for w in Workout.objects.filter(user=self.r)],
+            }
+
+        form = AnalysisChoiceForm(form_data)
+        optionsform = AnalysisOptionsForm(form_data)
+        dateform = DateRangeForm(form_data)
+
+        result = form.is_valid()
+        if not result:
+            print(form.errors)
+
+        self.assertTrue(form.is_valid())
+        self.assertTrue(optionsform.is_valid())
+        self.assertTrue(dateform.is_valid())
+
+        response = self.c.post('/rowers/user-analysis-select/',form_data)
+
+        self.assertEqual(response.status_code,200)
+
+        # count number of workouts by counting the number of occurences of '<label for="id_workouts_xx">' in response.content where xx is a number
+        # print all lines of response.content that contain '<label for="id_workouts_'
+        #print([line for line in response.content.decode('utf-8').split('\n') if '<label for="id_workouts_' in line])
+        #print(form_data['workouts'])
+        #self.assertEqual(response.content.count(b'<label for="id_workouts_'),2) <-- if we forbid the user to use strava workouts
+        self.assertEqual(response.content.count(b'<label for="id_workouts_'),5)
+
+        # get data from histodata function
+        ws = Workout.objects.filter(user=self.r)
+
+        script, div = histodata(ws,form_data)
+        # script has a line starting with 'data = [ ... ]'
+        # we need to get that line
+        data = [line for line in script.split('\n') if line.startswith('data = [')][0]
+        # the line should be a list of float values
+        self.assertTrue(data.startswith('data = ['))
+        self.assertTrue(data.endswith(']'))
+        # count the number of commas between the brackets
+        #self.assertEqual(data.count(','),2062) <-- if we forbid the user to use strava workouts
+        self.assertEqual(data.count(','),5155)
+        
+
 class OwnApi(TestCase):
     def setUp(self):
         self.u = UserFactory()
@@ -36,9 +478,7 @@ class OwnApi(TestCase):
                                       birthdate=faker.profile()['birthdate'],
                                       gdproptin=True, ftpset=True,surveydone=True,
                                       gdproptindate=timezone.now(),
-                                      rowerplan='coach',subscription_id=1)
-
-
+                                      rowerplan='pro',subscription_id=1)
         self.c = Client()
         self.user_workouts = WorkoutFactory.create_batch(5, user=self.r)
         self.factory = RequestFactory()

rowers/tests/test_races.py

@@ -106,10 +106,26 @@ class ChallengesTest(TestCase):
                                      workouttype = 'water',
                                      )
 
+
         self.wthyro.startdatetime = arrow.get(nu).datetime
         self.wthyro.date = nu.date()
         self.wthyro.save()
 
+        result = get_random_file(filename='rowers/tests/testdata/thyro.csv')
+        self.w_strava = WorkoutFactory(user=self.r,
+                                       csvfilename=result['filename'],
+                                       starttime=result['starttime'],
+                                       startdatetime=result['startdatetime'],
+                                       duration=result['duration'],
+                                       distance=result['totaldist'],
+                                       workouttype = 'water',
+                                       workoutsource = 'strava',
+                                       privacy = 'hidden',
+                                       )
+        self.w_strava.startdatetime = arrow.get(nu).datetime
+        self.w_strava.date = nu.date()
+        self.w_strava.save()
+
         result = get_random_file(filename='rowers/tests/testdata/thyro.csv')
         self.wthyro2 = WorkoutFactory(user=self.r2,
                                      csvfilename=result['filename'],
@@ -591,6 +607,78 @@ class ChallengesTest(TestCase):
 
         self.assertEqual(response.status_code, 200)
 
+    # repeat previous test for self.w_strava, but the response status of virtualevent_submit_result_view should be 403 and len(records) should be 0
+    @patch('django.contrib.gis.geoip2.GeoIP2.city', side_effect=mocked_requests)
+    def test_fastestrace_view_strava(self, mock_get):
+        login = self.c.login(username=self.u.username, password=self.password)
+        self.assertTrue(login)
+        
+        race = self.FastestRace
+
+        if self.r.birthdate:
+            age = calculate_age(self.r.birthdate)
+        else:
+            age = 25
+
+        # look at event
+        url = reverse('virtualevent_view',kwargs={'id':race.id})
+        response = self.c.get(url)
+        self.assertEqual(response.status_code,200)
+
+        # register
+        url = reverse('virtualevent_register_view',kwargs={'id':race.id})
+        response = self.c.get(url)
+        self.assertEqual(response.status_code,200)
+
+
+        form_data = {
+            'teamname': 'ApeTeam',
+            'boatclass': 'water',
+            'boattype': '1x',
+            'weightcategory': 'hwt',
+            'adaptiveclass': 'None',
+            'age': age,
+            'mix': False,
+            'acceptsocialmedia': True,
+        }
+        form = VirtualRaceResultForm(form_data)
+        self.assertTrue(form.is_valid())
+
+
+        response = self.c.post(url,form_data,follow=True)
+        expected_url = reverse('virtualevent_view',kwargs={'id':race.id})
+        self.assertRedirects(response, expected_url=expected_url,
+                             status_code=302,target_status_code=200)
+
+        self.assertEqual(response.status_code, 200)
+
+        # submit workout
+        url = reverse('virtualevent_submit_result_view',kwargs={'id':race.id,'workoutid':self.w_strava.id})
+        response = self.c.get(url)
+        self.assertEqual(response.status_code, 200)
+
+        # response.content should have a form with only one instance of <label for="id_workouts_0">
+        self.assertEqual(response.content.count(b'<label for="id_workouts_0">'),1)
+
+
+
+        records = IndoorVirtualRaceResult.objects.filter(userid=self.u.id)
+        self.assertEqual(len(records),1)
+
+        record = records[0]
+
+
+        form_data = {
+            'workouts':[self.w_strava.id],
+            'record':record.id,
+        }
+
+        response = self.c.post(url,form_data,follow=True)
+
+        self.assertEqual(response.status_code, 200)
+
+        # in response.content, there should be a p with class errormessage and the text "Error in form"
+        self.assertTrue(b'Error in form' in response.content)
 
     @patch('django.contrib.gis.geoip2.GeoIP2.city', side_effect=mocked_requests)
     def test_virtualevents_view(self, mock_get):