300 lines
10 KiB
Python
300 lines
10 KiB
Python
from __future__ import absolute_import
|
|
from __future__ import division
|
|
from __future__ import print_function
|
|
from __future__ import unicode_literals
|
|
|
|
#from __future__ import print_function
|
|
from .statements import *
|
|
nu = datetime.datetime.now()
|
|
from django.db import transaction
|
|
import shutil
|
|
|
|
from rowers.views import add_defaultfavorites
|
|
from rowers.dataflow import process_single_file, upload_handler, unzip_and_process
|
|
from django.core.files.uploadedfile import SimpleUploadedFile
|
|
from django.conf import settings
|
|
from rowingdata import get_file_type
|
|
|
|
file_list = [
|
|
'rowers/tests/testdata/testdata.csv',
|
|
'rowers/tests/testdata/testdata.csv.gz',
|
|
'rowers/tests/testdata/tim.csv',
|
|
'rowers/tests/testdata/crewnerddata.tcx',
|
|
'rowers/tests/testdata/Speedcoach2example.csv',
|
|
'rowers/tests/testdata/Impeller.csv',
|
|
'rowers/tests/testdata/speedcoach3test3.csv',
|
|
'rowers/tests/testdata/SpeedCoach2Linkv1.27.csv',
|
|
'rowers/tests/testdata/SpeedCoach2Link_interval.csv',
|
|
'rowers/tests/testdata/NoHR.tcx',
|
|
'rowers/tests/testdata/rowinginmotionexample.tcx',
|
|
'rowers/tests/testdata/RP_testdata.csv',
|
|
'rowers/tests/testdata/mystery.csv',
|
|
'rowers/tests/testdata/RP_interval.csv',
|
|
'rowers/tests/testdata/3x250m.fit',
|
|
'rowers/tests/testdata/painsled_desktop_example.csv',
|
|
'rowers/tests/testdata/ergdata_example.csv',
|
|
'rowers/tests/testdata/boatcoach_2021-09-09__18-15-53.csv',
|
|
'rowers/tests/testdata/colinforce.csv',
|
|
'rowers/tests/testdata/PainsledForce.csv',
|
|
'rowers/tests/testdata/EmpowerSpeedCoachForce.csv',
|
|
'rowers/tests/testdata/boatcoach.csv',
|
|
'rowers/tests/testdata/ergstick.csv',
|
|
]
|
|
|
|
fail_list = [
|
|
'rowers/tests/testdata/lofoten.jpg',
|
|
'rowers/tests/testdata/c2records.json',
|
|
'rowers/tests/testdata/alphen.kml',
|
|
'rowers/tests/testdata/testdata.gpx'
|
|
]
|
|
|
|
|
|
#@pytest.mark.django_db
|
|
@override_settings(TESTING=True)
|
|
class ViewTest(TestCase):
|
|
def setUp(self):
|
|
redis_connection.publish('tasks','KILL')
|
|
self.c = Client()
|
|
self.u = User.objects.create_user('john',
|
|
'sander@ds.ds',
|
|
'koeinsloot')
|
|
self.r = Rower.objects.create(user=self.u,gdproptin=True, ftpset=True,surveydone=True,
|
|
gdproptindate=timezone.now(),
|
|
rowerplan='pro',
|
|
)
|
|
|
|
add_defaultfavorites(self.r)
|
|
|
|
self.nu = datetime.datetime.now()
|
|
|
|
# copy every file in fail_list to rowers/tests/testdata/backup folder
|
|
# Zorg ervoor dat de backup-map bestaat
|
|
backup_dir = 'rowers/tests/testdata/backup'
|
|
os.makedirs(backup_dir, exist_ok=True)
|
|
|
|
# Kopieer elk bestand in fail_list naar de backup-map
|
|
for file_path in fail_list:
|
|
if os.path.exists(file_path):
|
|
shutil.copy(file_path, backup_dir)
|
|
else:
|
|
print(f"Bestand niet gevonden: {file_path}")
|
|
|
|
def tearDown(self):
|
|
backup_dir = 'rowers/tests/testdata/backup'
|
|
for file_path in fail_list:
|
|
backup_file = os.path.join(backup_dir, os.path.basename(file_path))
|
|
if os.path.exists(backup_file):
|
|
shutil.copy(backup_file, os.path.dirname(file_path))
|
|
else:
|
|
print(f"Backup-bestand niet gevonden: {backup_file}")
|
|
|
|
|
|
|
|
@parameterized.expand(file_list)
|
|
@patch('rowers.dataflow.myqueue')
|
|
def test_upload_view(self, filename, mocked_myqueue):
|
|
# simple test to see if upload view works. Submits a DocumentsForm to /rowers/workout/upload/
|
|
login = self.c.login(username='john',password='koeinsloot')
|
|
self.assertTrue(login)
|
|
|
|
with open(filename, 'rb') as f:
|
|
file_content = f.read()
|
|
uploaded_file = SimpleUploadedFile(
|
|
"testdata.csv",
|
|
file_content,
|
|
content_type="text/csv"
|
|
)
|
|
form_data = {
|
|
'title':'test',
|
|
'workouttype':'rower',
|
|
'boattype':'1x',
|
|
'notes':'aap noot mies',
|
|
'make_plot':False,
|
|
'rpe':6,
|
|
'upload_to_c2':False,
|
|
'plottype':'timeplot',
|
|
'landingpage':'workout_edit_view',
|
|
'raceid':0,
|
|
'file': filename,
|
|
}
|
|
|
|
|
|
request = RequestFactory()
|
|
request.user = self.u
|
|
form = DocumentsForm(data = form_data,files={'file': uploaded_file})
|
|
self.assertTrue(form.is_valid())
|
|
|
|
optionsform = UploadOptionsForm(form_data,request=request)
|
|
self.assertTrue(optionsform.is_valid())
|
|
|
|
response = self.c.post('/rowers/workout/upload/', data = form_data,
|
|
files = {'file': uploaded_file}, follow=True)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
uploadoptions = form.cleaned_data.copy()
|
|
uploadoptions.update(optionsform.cleaned_data)
|
|
result = upload_handler(uploadoptions, filename)
|
|
|
|
self.assertEqual(result["status"], "processing")
|
|
|
|
@parameterized.expand(fail_list)
|
|
@patch('rowers.dataflow.myqueue')
|
|
def test_upload_view(self, filename, mocked_myqueue):
|
|
# simple test to see if upload view works. Submits a DocumentsForm to /rowers/workout/upload/
|
|
login = self.c.login(username='john',password='koeinsloot')
|
|
self.assertTrue(login)
|
|
|
|
with open(filename, 'rb') as f:
|
|
file_content = f.read()
|
|
uploaded_file = SimpleUploadedFile(
|
|
"testdata.csv",
|
|
file_content,
|
|
content_type="text/csv"
|
|
)
|
|
form_data = {
|
|
'title':'test',
|
|
'workouttype':'rower',
|
|
'boattype':'1x',
|
|
'notes':'aap noot mies',
|
|
'make_plot':False,
|
|
'rpe':6,
|
|
'upload_to_c2':False,
|
|
'plottype':'timeplot',
|
|
'landingpage':'workout_edit_view',
|
|
'raceid':0,
|
|
'file': filename,
|
|
}
|
|
|
|
|
|
request = RequestFactory()
|
|
request.user = self.u
|
|
form = DocumentsForm(data = form_data,files={'file': uploaded_file})
|
|
self.assertTrue(form.is_valid())
|
|
|
|
optionsform = UploadOptionsForm(form_data,request=request)
|
|
self.assertTrue(optionsform.is_valid())
|
|
|
|
response = self.c.post('/rowers/workout/upload/', data = form_data,
|
|
files = {'file': uploaded_file}, follow=True)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
uploadoptions = form.cleaned_data.copy()
|
|
uploadoptions.update(optionsform.cleaned_data)
|
|
result = upload_handler(uploadoptions, filename)
|
|
|
|
self.assertEqual(result["status"], "error")
|
|
|
|
@parameterized.expand(file_list)
|
|
@patch('rowers.dataflow.myqueue')
|
|
def test_process_single_file(self, filename, mocked_myqueue):
|
|
uploadoptions = {
|
|
'title':'test',
|
|
'workouttype':'rower',
|
|
'boattype':'1x',
|
|
'notes':'aap noot mies',
|
|
'make_plot':False,
|
|
'rpe':6,
|
|
'upload_to_c2':False,
|
|
'plottype':'timeplot',
|
|
'landingpage':'workout_edit_view',
|
|
'raceid':0,
|
|
'user': self.u,
|
|
'file': filename,
|
|
}
|
|
result, f2 = process_single_file(filename, uploadoptions, 1)
|
|
self.assertEqual(result, True)
|
|
os.remove(f2+'.gz')
|
|
|
|
# process a single file without 'user'
|
|
@patch('rowers.dataflow.myqueue')
|
|
def test_process_single_file_nouser(self, mocked_myqueue):
|
|
filename = 'rowers/tests/testdata/testdata.csv'
|
|
uploadoptions = {
|
|
'title':'test',
|
|
'workouttype':'rower',
|
|
'boattype':'1x',
|
|
'notes':'aap noot mies',
|
|
'make_plot':False,
|
|
'rpe':6,
|
|
'upload_to_c2':False,
|
|
'plottype':'timeplot',
|
|
'landingpage':'workout_edit_view',
|
|
'raceid':0,
|
|
'useremail': self.u.email,
|
|
'file': filename,
|
|
}
|
|
result, f2 = process_single_file(filename, uploadoptions, 1)
|
|
self.assertEqual(result, True)
|
|
os.remove(f2+'.gz')
|
|
|
|
# process a zip file
|
|
@patch('rowers.dataflow.myqueue')
|
|
def test_process_single_zipfile(self, mocked_myqueue):
|
|
filename = 'rowers/tests/testdata/zipfile.zip'
|
|
uploadoptions = {
|
|
'title':'test',
|
|
'workouttype':'rower',
|
|
'boattype':'1x',
|
|
'notes':'aap noot mies',
|
|
'make_plot':False,
|
|
'rpe':6,
|
|
'upload_to_c2':False,
|
|
'plottype':'timeplot',
|
|
'landingpage':'workout_edit_view',
|
|
'raceid':0,
|
|
'user': self.u,
|
|
'file': filename,
|
|
}
|
|
result = process_single_file(filename, uploadoptions, 1)
|
|
|
|
self.assertEqual(result["status"], "error")
|
|
|
|
result = upload_handler(uploadoptions, filename)
|
|
|
|
self.assertEqual(result["status"], "processing")
|
|
|
|
# process a single file without 'title'
|
|
@patch('rowers.dataflow.myqueue')
|
|
def test_process_single_file_nouser(self, mocked_myqueue):
|
|
filename = 'rowers/tests/testdata/testdata.csv'
|
|
uploadoptions = {
|
|
'workouttype':'rower',
|
|
'boattype':'1x',
|
|
'notes':'aap noot mies',
|
|
'make_plot':False,
|
|
'rpe':6,
|
|
'upload_to_c2':False,
|
|
'plottype':'timeplot',
|
|
'landingpage':'workout_edit_view',
|
|
'raceid':0,
|
|
'user': self.u,
|
|
'file': filename,
|
|
}
|
|
result, f2 = process_single_file(filename, uploadoptions, 1)
|
|
self.assertEqual(result, True)
|
|
os.remove(f2+'.gz')
|
|
|
|
@patch('rowers.dataflow.myqueue')
|
|
def test_process_zip_file(self, mocked_myqueue):
|
|
filename = 'rowers/tests/testdata/zipfile.zip'
|
|
uploadoptions = {
|
|
'title':'test',
|
|
'workouttype':'rower',
|
|
'boattype':'1x',
|
|
'notes':'aap noot mies',
|
|
'make_plot':False,
|
|
'rpe':6,
|
|
'upload_to_c2':False,
|
|
'plottype':'timeplot',
|
|
'landingpage':'workout_edit_view',
|
|
'raceid':0,
|
|
'user': self.u,
|
|
'file': filename,
|
|
}
|
|
result = unzip_and_process(filename, uploadoptions, 1)
|
|
self.assertEqual(result['status'], "completed")
|
|
|
|
|