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='pro') 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='pro') 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())