from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
from __future__ import unicode_literals
from django.db import transaction
#from __future__ import print_function
from .statements import *
nu = datetime.datetime.now()
import rowers
from rowers import dataprep
from rowers import tasks
from rowers import c2stuff
import urllib
import json
import pandas as pd
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 ... 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']])
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 ... 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']])
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 ... 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']])
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 ... 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']])
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 ... 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']])
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 '