Private
Public Access
1
0
Files
rowsandall/rowers/tests/test_uploads2.py

144 lines
5.4 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
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
#@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()
file_list = ['rowers/tests/testdata/testdata.csv',
'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/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',
]
@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)
uploadoptions = form.cleaned_data.copy()
uploadoptions.update(optionsform.cleaned_data)
result = upload_handler(uploadoptions, filename)
self.assertEqual(result["status"], "processing")
@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')
@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")