243 lines
9.9 KiB
Python
243 lines
9.9 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 *
|
|
from django.db import transaction
|
|
|
|
@override_settings(TESTING=True)
|
|
class EmailUpload(TestCase):
|
|
def setUp(self):
|
|
redis_connection.publish('tasks','KILL')
|
|
self.c = Client()
|
|
u = User.objects.create_user('john',
|
|
'sander@ds.ds',
|
|
'koeinsloot',
|
|
first_name='John',
|
|
last_name='Sloot')
|
|
r = Rower.objects.create(user=u,gdproptin=True, ftpset=True,surveydone=True,
|
|
gdproptindate=timezone.now(),
|
|
emailalternatives=['sander2@ds.nl']
|
|
)
|
|
|
|
self.theadmin = UserFactory(is_staff=True)
|
|
self.rtheadmin = Rower.objects.create(user=self.theadmin,
|
|
birthdate = faker.profile()['birthdate'],
|
|
gdproptin=True, ftpset=True,surveydone=True,
|
|
gdproptindate=timezone.now(),
|
|
rowerplan='coach')
|
|
|
|
nu = datetime.datetime.now()
|
|
|
|
a2 = 'media/mailbox_attachments/colin3.csv'
|
|
copy('rowers/tests/testdata/emails/colin.csv',a2)
|
|
a3 = 'media/mailbox_attachments/colin4.csv'
|
|
copy('rowers/tests/testdata/emails/colin.csv',a3)
|
|
a4 = 'media/mailbox_attachments/colin5.csv'
|
|
copy('rowers/tests/testdata/emails/colin.csv',a4)
|
|
a5 = 'media/mailbox_attachments/colin6.csv'
|
|
copy('rowers/tests/testdata/emails/colin.csv',a5)
|
|
a6 = 'media/mailbox_attachments/colin7.csv'
|
|
copy('rowers/tests/testdata/emails/colin.csv',a6)
|
|
|
|
|
|
def tearDown(self):
|
|
for filename in os.listdir('media/mailbox_attachments'):
|
|
path = os.path.join('media/mailbox_attachments/',filename)
|
|
if not os.path.isdir(path):
|
|
try:
|
|
os.remove(path)
|
|
except (IOError,FileNotFoundError,OSError):
|
|
pass
|
|
|
|
|
|
|
|
|
|
@patch('rowers.dataprep.create_engine')
|
|
@patch('rowers.dataprep.read_data',side_effect=mocked_read_data)
|
|
def test_uploadapi(self,mocked_sqlalchemy,mocked_read_data):
|
|
form_data = {
|
|
'title': 'test',
|
|
'workouttype':'rower',
|
|
'boattype': '1x',
|
|
'notes': 'aap noot mies',
|
|
'rpe':1,
|
|
'make_plot': False,
|
|
'upload_to_C2': False,
|
|
'plottype': 'timeplot',
|
|
'file': 'media/mailbox_attachments/colin3.csv',
|
|
'secret': settings.UPLOAD_SERVICE_SECRET,
|
|
'user': 1,
|
|
}
|
|
|
|
url = reverse('workout_upload_api')
|
|
response = self.c.post(url,form_data,HTTP_HOST='127.0.0.1:4533')
|
|
self.assertEqual(response.status_code,200)
|
|
|
|
form_data['file'] = 'media/mailbox_attachments/colin4.csv'
|
|
response = self.c.post(url,json.dumps(form_data),HTTP_HOST='127.0.0.1:4533',
|
|
content_type='application/json')
|
|
self.assertEqual(response.status_code,200)
|
|
|
|
# should also test if workout is created
|
|
w = Workout.objects.get(id=1)
|
|
self.assertEqual(w.name,'test')
|
|
self.assertEqual(w.notes,'aap noot mies')
|
|
|
|
# test exceptions
|
|
secret = form_data.pop('secret')
|
|
form_data['file'] = 'media/mailbox_attachments/colin5.csv'
|
|
response = self.c.post(url,form_data,HTTP_HOST='127.0.0.1:4533')
|
|
self.assertEqual(response.status_code,400)
|
|
|
|
form_data['secret'] = 'wrong'
|
|
form_data['file'] = 'media/mailbox_attachments/colin6.csv'
|
|
response = self.c.post(url,form_data,HTTP_HOST='127.0.0.1:4533')
|
|
self.assertEqual(response.status_code,403)
|
|
|
|
form_data['secret'] = secret
|
|
filename = form_data.pop('file')
|
|
response = self.c.post(url,form_data,HTTP_HOST='127.0.0.1:4533')
|
|
self.assertEqual(response.status_code,400)
|
|
|
|
form_data['file'] = filename
|
|
os.remove(filename)
|
|
response = self.c.post(url,form_data,HTTP_HOST='127.0.0.1:4533')
|
|
self.assertEqual(response.status_code,400)
|
|
|
|
|
|
@patch('rowers.dataprep.create_engine')
|
|
@patch('rowers.dataprep.read_data',side_effect=mocked_read_data)
|
|
def test_uploadapi2(self,mocked_sqlalchemy,mocked_read_data):
|
|
form_data = {"secret": settings.UPLOAD_SERVICE_SECRET,
|
|
"user": 1,
|
|
"file": 'media/mailbox_attachments/colin3.csv',
|
|
"title": "JustGo-17256M",
|
|
"workouttype": "water",
|
|
"boattype": "1x",
|
|
"nkid": 1206081,
|
|
"inboard": 0.86,
|
|
"oarlength": 2.84,
|
|
"summary": "Workout Summary - JustGo-17256M\n--|Total|--Total---|--Avg--|-Avg-|-Avg--|-Avg-|-Max-|-Avg\n--|Dist-|--Time----|-Pace--|-Pwr-|-SPM--|-HR--|-HR--|-DPS\n--|17256|01:29:09.0|02:35.2|100.9| 16.5 |125.0|138.0|12.4\nWorkout Details\n#-|SDist|-Split-|-SPace-|-Pwr-|-SPM--|-AvgHR-|DPS-\n01|17256|29:09.0|02:35.2| 100 | 16.5 | 125.0 |12.4\n",
|
|
"oarlockfirmware": "001.01",
|
|
"elapsedTime": 5349.0,
|
|
"totalDistance": 17256.13,
|
|
"useImpeller": False,
|
|
"seatNumber": 1
|
|
}
|
|
|
|
|
|
url = reverse('workout_upload_api')
|
|
response = self.c.post(url,json.dumps(form_data),HTTP_HOST='127.0.0.1:4533',
|
|
content_type='application/json')
|
|
self.assertEqual(response.status_code,200)
|
|
|
|
# should also test if workout is created
|
|
w = Workout.objects.get(id=1)
|
|
self.assertEqual(len(w.summary),324)
|
|
|
|
|
|
|
|
@patch('rowers.dataprep.create_engine')
|
|
@patch('rowers.dataprep.read_data',side_effect=mocked_read_data)
|
|
def test_uploadapi3(self,mocked_sqlalchemy,mocked_read_data):
|
|
with transaction.atomic():
|
|
form_data = {
|
|
'title': 'test',
|
|
'workouttype':'rower',
|
|
'boattype': '1x',
|
|
'notes': 'aap noot mies',
|
|
'make_plot': False,
|
|
'upload_to_C2': False,
|
|
'plottype': 'timeplot',
|
|
'rpe':4,
|
|
'file': 'media/mailbox_attachments/colin3.csv',
|
|
'secret': settings.UPLOAD_SERVICE_SECRET,
|
|
'useremail': 'sander2@ds.nl',
|
|
}
|
|
|
|
url = reverse('workout_upload_api')
|
|
response = self.c.post(url,form_data,HTTP_HOST='127.0.0.1:4533')
|
|
self.assertEqual(response.status_code,200)
|
|
|
|
# should also test if workout is created
|
|
w = Workout.objects.get(id=1)
|
|
self.assertEqual(w.name,'test')
|
|
self.assertEqual(w.notes,'aap noot mies')
|
|
|
|
@patch('rowers.dataprep.create_engine')
|
|
@patch('rowers.dataprep.read_data',side_effect=mocked_read_data)
|
|
def test_uploadapi_credentials(self,mocked_sqlalchemy,mocked_read_data):
|
|
form_data = {
|
|
'title': 'test',
|
|
'workouttype':'rower',
|
|
'boattype': '1x',
|
|
'notes': 'aap noot mies',
|
|
'make_plot': False,
|
|
'upload_to_C2': False,
|
|
'plottype': 'timeplot',
|
|
'file': 'media/mailbox_attachments/colin3.csv',
|
|
'secret': 'potjandorie2',
|
|
'user': 1,
|
|
}
|
|
|
|
url = reverse('workout_upload_api')
|
|
response = self.c.post(url,form_data)
|
|
self.assertEqual(response.status_code,403)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
#@pytest.mark.django_db
|
|
@override_settings(TESTING=True)
|
|
class EmailTests(TestCase):
|
|
def setUp(self):
|
|
redis_connection.publish('tasks','KILL')
|
|
u = User.objects.create_user('john',
|
|
'sander@ds.ds',
|
|
'koeinsloot')
|
|
r = Rower.objects.create(user=u,gdproptin=True, ftpset=True,surveydone=True,
|
|
gdproptindate=timezone.now()
|
|
)
|
|
|
|
self.theadmin = UserFactory(is_staff=True)
|
|
self.rtheadmin = Rower.objects.create(user=self.theadmin,
|
|
birthdate = faker.profile()['birthdate'],
|
|
gdproptin=True, ftpset=True,surveydone=True,
|
|
gdproptindate=timezone.now(),
|
|
rowerplan='coach')
|
|
|
|
nu = datetime.datetime.now()
|
|
a2 = 'media/mailbox_attachments/colin3.csv'
|
|
copy('rowers/tests/testdata/emails/colin.csv',a2)
|
|
a3 = 'media/mailbox_attachments/colin4.csv'
|
|
copy('rowers/tests/testdata/emails/colin.csv',a3)
|
|
a4 = 'media/mailbox_attachments/colin5.csv'
|
|
copy('rowers/tests/testdata/emails/colin.csv',a4)
|
|
a5 = 'media/mailbox_attachments/colin6.csv'
|
|
copy('rowers/tests/testdata/emails/colin.csv',a5)
|
|
a6 = 'media/mailbox_attachments/colin7.csv'
|
|
copy('rowers/tests/testdata/emails/colin.csv',a6)
|
|
|
|
|
|
def tearDown(self):
|
|
for filename in os.listdir('media/mailbox_attachments'):
|
|
path = os.path.join('media/mailbox_attachments/',filename)
|
|
if not os.path.isdir(path):
|
|
try:
|
|
os.remove(path)
|
|
except (IOError,FileNotFoundError,OSError):
|
|
pass
|
|
|
|
@patch('rowers.dataprep.create_engine')
|
|
@patch('rowers.integrations.c2.requests.get', side_effect=mocked_requests)
|
|
@patch('rowers.integrations.c2.requests.post', side_effect=mocked_requests)
|
|
def test_emailprocessing(
|
|
self, mocked_sqlalchemy,mocked_get, mock_post):
|
|
out = StringIO()
|
|
call_command('processemail', stdout=out,testing=True,mailbox='workouts4')
|
|
self.assertIn('Successfully processed email attachments',out.getvalue())
|