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,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,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.getsmallrowdata_db',side_effect=mocked_getsmallrowdata_db) def test_uploadapi(self,mocked_sqlalchemy,mocked_getsmallrowdata_db): 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.getsmallrowdata_db',side_effect=mocked_getsmallrowdata_db) def test_uploadapi2(self,mocked_sqlalchemy,mocked_getsmallrowdata_db): 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.getsmallrowdata_db',side_effect=mocked_getsmallrowdata_db) def test_uploadapi_credentials(self,mocked_sqlalchemy,mocked_getsmallrowdata_db): 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,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,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.polarstuff.get_polar_notifications') @patch('rowers.c2stuff.requests.get', side_effect=mocked_requests) @patch('rowers.c2stuff.requests.post', side_effect=mocked_requests) def test_emailprocessing( self, mocked_sqlalchemy,mocked_polar_notifications, mock_get, mock_post): out = StringIO() call_command('processemail', stdout=out,testing=True,mailbox='workouts4') self.assertIn('Successfully processed email attachments',out.getvalue())