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 import json from rowers.ownapistuff import * from rowers.views.apiviews import * from rowers.models import APIKey class OwnApi(TestCase): def setUp(self): self.u = 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.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() self.factory = APIRequestFactory() self.apikey = APIKey.objects.create(user=self.u) def test_strokedataform(self): login = self.c.login(username=self.u.username, password=self.password) self.assertTrue(login) w = self.user_workouts[0] url = reverse('strokedataform',kwargs={'id':encoder.encode_hex(w.id)}) response = self.c.get(url) self.assertEqual(response.status_code,200) with patch('rowers.dataprep.read_data') as mock_read_data: mock_read_data.return_value = pl.read_csv('rowers/tests/testdata/colsfromdb.csv') url = reverse('strokedatajson',kwargs={'id':w.id}) request = self.factory.get(url) request.user = self.u force_authenticate(request, user=self.u) response = strokedatajson(request,id=w.id) self.assertEqual(response.status_code,200) # response must be json strokedata = json.loads(response.content) df = pl.from_dict(strokedata) self.assertFalse(df.is_empty()) form_data = { "distance": [23, 46, 48], "time": [3200, 6700, 10099], "spm": [16.4, 21.2, 19.8], "pace": [155068, 144402, 138830], "power": [84.6, 117.2, 141.3], "hr": [85, 91, 95] } result = get_random_file(filename='rowers/tests/testdata/thyro.csv') w2 = Workout.objects.create( user=self.r, csvfilename=result['filename'], duration=result['duration'], startdatetime=result['startdatetime'], workouttype='water', starttime=result['starttime'], ) url = reverse('strokedatajson',kwargs={'id':w2.id}) request = self.factory.post(url,{'strokedata':form_data},format='json') request.user = self.u request.data = json.dumps({'strokedata':form_data}) strokedata = json.loads(request.data)['strokedata'] force_authenticate(request, user=self.u) with patch('rowers.dataprep.getrowdata_db') as mock_getrowdata: mock_getrowdata.return_value = (pd.DataFrame(),None) response = strokedatajson(request,id=w.id) self.assertEqual(response.status_code,201) def test_strokedataform_v2(self): login = self.c.login(username=self.u.username, password=self.password) self.assertTrue(login) w = self.user_workouts[1] url = reverse('strokedataform_v2',kwargs={'id':encoder.encode_hex(w.id)}) response = self.c.get(url) self.assertEqual(response.status_code,200) url = reverse('strokedatajson_v2',kwargs={'id':w.id}) request = self.factory.get(url) request.user = self.u force_authenticate(request, user=self.u) with patch('rowers.dataprep.read_data') as mock_read_data: mock_read_data.return_value = pl.read_csv('rowers/tests/testdata/colsfromdb.csv') response = strokedatajson_v2(request,id=w.id) self.assertEqual(response.status_code,200) # response must be json strokedata = json.loads(response.content) df = pl.from_dicts(strokedata['data']) self.assertFalse(df.is_empty()) form_data = { "data": [ { "time": 3200.0000476837, "pace": 155068.4885951763, "hr": 85.7857142857, "power": 84.6531131591, "distance": 23, "spm": 16.380952381 }, { "time": 6700.0000476837, "pace" : 144402.6407586741, "hr": 91.2142857143, "power": 117.458827834, "distance": 36, "spm": 21.1666666667 }, { "time": 10099.9999046326, "pace": 138830.8712654931, "hr": 95.7142857143, "power": 141.31057207, "distance": 48, "spm": 19.8095238095 } ] } result = get_random_file(filename='rowers/tests/testdata/thyro.csv') w2 = Workout.objects.create( user=self.r, csvfilename=result['filename'], duration=result['duration'], startdatetime=result['startdatetime'], workouttype='water', starttime=result['starttime'], ) url = reverse('strokedatajson_v2',kwargs={'id':w2.id}) request = self.factory.post(url,form_data,format='json') request.user = self.u request.data = json.dumps(form_data) force_authenticate(request, user=self.u) with patch('rowers.dataprep.getrowdata_db') as mock_getrowdata: mock_getrowdata.return_value = (pd.DataFrame(),None) response = strokedatajson_v2(request,id=w.id) self.assertEqual(response.status_code,200) def test_strokedataform_tcx(self): login = self.c.login(username=self.u.username, password=self.password) self.assertTrue(login) w = self.user_workouts[1] url = reverse('strokedata_tcx') with open('rowers/tests/testdata/crewnerddata.tcx') as f: tcxdata_str = f.read() result = get_random_file(filename='rowers/tests/testdata/thyro.csv') request = self.factory.post(url, data = tcxdata_str, content_type='application/xml') request.user = self.u request.content_type = 'application/xml' force_authenticate(request, user=self.u) with patch('rowers.dataprep.getrowdata_db') as mock_getrowdata: mock_getrowdata.return_value = (pd.DataFrame(),None) response = strokedata_tcx(request) self.assertEqual(response.status_code,200) def test_strokedataform_rowingdata(self): url = reverse('strokedata_rowingdata') filename = 'rowers/tests/testdata/testdata.csv' f = open(filename, 'rb') # Use Basic Auth header (alternative if using a custom view) credentials = f"{self.u.username}:{self.password}" base64_credentials = base64.b64encode(credentials.encode()).decode() # Use Basic Auth header headers = { "HTTP_AUTHORIZATION": f"Basic {base64_credentials}" } form_data = { "workouttype": "rower", "boattype": "1x", "notes": "A test file upload", } # Send POST request response = self.client.post( url, {"file": f, **form_data}, format="multipart", # Ensure multipart/form-data is used **headers, # Optional if login doesn't suffice ) f.close() # Assertions self.assertEqual(response.status_code, 201) def test_strokedataform_rowingdata_apikey(self): url = reverse('strokedata_rowingdata_apikey') filename = 'rowers/tests/testdata/testdata.csv' f = open(filename, 'rb') # Use API Key header headers = { "HTTP_AUTHORIZATION": self.apikey.key, } form_data = { "workouttype": "rower", "boattype": "1x", "notes": "A test file upload", } # Send POST request response = self.client.post( url, {"file": f, **form_data}, format="multipart", # Ensure multipart/form-data is used **headers, # Optional if login doesn't suffice ) f.close() # Assertions self.assertEqual(response.status_code, 201) def test_strokedataform_empty(self): login = self.c.login(username=self.u.username, password=self.password) self.assertTrue(login) w = self.user_workouts[1] url = reverse('strokedata_tcx') request = self.factory.post(url, data = '', content_type='application/xml') request.user = self.u request.content_type = 'application/xml' force_authenticate(request, user=self.u) with patch('rowers.dataprep.getrowdata_db') as mock_getrowdata: mock_getrowdata.return_value = (pd.DataFrame(),None) response = strokedata_tcx(request) self.assertEqual(response.status_code,405) def test_strokedataform_v3(self): login = self.c.login(username=self.u.username, password=self.password) self.assertTrue(login) w = self.user_workouts[1] url = reverse('strokedatajson_v3') request = self.factory.get(url) request.user = self.u force_authenticate(request, user=self.u) response = strokedatajson_v3(request) self.assertEqual(response.status_code,405) strokedata = { "data": [ { "time": 3200.0000476837, "pace": 155068.4885951763, "hr": 85.7857142857, "power": 84.6531131591, "distance": 23, "spm": 16.380952381 }, { "time": 6700.0000476837, "pace" : 144402.6407586741, "hr": 91.2142857143, "power": 117.458827834, "distance": 36, "spm": 21.1666666667 }, { "time": 10099.9999046326, "pace": 138830.8712654931, "hr": 95.7142857143, "power": 141.31057207, "distance": 48, "spm": 19.8095238095 } ] } form_data = { "distance": 2100, "elapsedTime": 592, "duration": "0:09:52", "name": "Test Workout (GO)", "startdatetime": "2023-01-16 17:54:35.588838+00:00", "workouttype": "water", "boattype": "1x", "notes": "some\nnotes", "strokes": strokedata, } url = reverse('strokedatajson_v3') request = self.factory.post(url,form_data,format='json') request.user = self.u request.data = json.dumps(form_data) force_authenticate(request, user=self.u) with patch('rowers.dataprep.getrowdata_db') as mock_getrowdata: mock_getrowdata.return_value = (pd.DataFrame(),None) response = strokedatajson_v3(request) self.assertEqual(response.status_code,200) response = json.loads(response.content) x = response['workout id']