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 * @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') 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() workoutsbox = Mailbox.objects.create(name='workouts1') workoutsbox.save() failbox = Mailbox.objects.create(name='Failed') failbox.save() m = Message(mailbox=workoutsbox, from_header = u.email, subject = "run", body = """ workout run """) m.save() a2 = 'media/mailbox_attachments/colin3.csv' copy('rowers/tests/testdata/emails/colin.csv',a2) a = MessageAttachment(message=m,document=a2[6:]) a.save() 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', '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) # 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) @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_emailupload( self, mocked_sqlalchemy,mocked_polar_notifications, mock_get, mock_post): out = StringIO() call_command('processemail', stdout=out,testing=True,mailbox='workouts1') self.assertIn('Successfully processed email attachments',out.getvalue()) ws = Workout.objects.filter(name="run") self.assertEqual(len(ws),1) w = ws[0] self.assertEqual(w.workouttype,'Run') @override_settings(TESTING=True) class ZipEmailUpload(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() workoutsbox = Mailbox.objects.create(name='workouts1') workoutsbox.save() failbox = Mailbox.objects.create(name='Failed') failbox.save() m = Message(mailbox=workoutsbox, from_header = u.email, subject = "Sprint", body = """ workout water """) m.save() a2 = 'media/mailbox_attachments/zipfile.zip' copy('rowers/tests/testdata/zipfile.zip',a2) a = MessageAttachment(message=m,document=a2[6:]) a.save() 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_emailupload( self, mocked_sqlalchemy,mocked_polar_notifications, mock_get, mock_post): out = StringIO() call_command('processemail', stdout=out,testing=True,mailbox='workouts1') self.assertIn('Successfully processed email attachments',out.getvalue()) ws = Workout.objects.all() self.assertEqual(len(ws),5) w = ws[4] self.assertEqual(w.name,'Sprint (5)') @override_settings(TESTING=True) class EmailUniCodeUpload(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() workoutsbox = Mailbox.objects.create(name='workouts2') workoutsbox.save() failbox = Mailbox.objects.create(name='Failed') failbox.save() m = Message(mailbox=workoutsbox, from_header = u.email, subject = "Třeboň", body = """ workout water """) m.save() a2 = 'media/mailbox_attachments/colin4a.csv' copy('rowers/tests/testdata/emails/colin.csv',a2) a = MessageAttachment(message=m,document=a2[6:]) a.save() def tearDown(self): filename = 'colin4a.csv' 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_emailupload( self, mocked_sqlalchemy,mocked_polar_notifications, mock_get, mock_post): out = StringIO() call_command('processemail', stdout=out,testing=True,mailbox='workouts2') self.assertIn('Successfully processed email attachments',out.getvalue()) ws = Workout.objects.filter(name="Třeboň") self.assertEqual(len(ws),1) w = ws[0] self.assertEqual(w.workouttype,'water') @override_settings(TESTING=True) class EmailBikeErgUpload(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() workoutsbox = Mailbox.objects.create(name='workouts2') workoutsbox.save() failbox = Mailbox.objects.create(name='Failed') failbox.save() m = Message(mailbox=workoutsbox, from_header = u.email, subject = "bikeerg", body = """ workout bikeerg """) m.save() a2 = 'media/mailbox_attachments/colin3.csv' copy('rowers/tests/testdata/emails/colin.csv',a2) a = MessageAttachment(message=m,document=a2[6:]) a.save() def tearDown(self): filename = 'colin3.csv' 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_emailupload( self, mocked_sqlalchemy,mocked_polar_notifications, mock_get, mock_post): out = StringIO() call_command('processemail', stdout=out,testing=True,mailbox='workouts2') self.assertIn('Successfully processed email attachments',out.getvalue()) ws = Workout.objects.filter(name="bikeerg") self.assertEqual(len(ws),1) w = ws[0] self.assertEqual(w.workouttype,'bikeerg') @override_settings(TESTING=True) class EmailBikeUpload(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() workoutsbox = Mailbox.objects.create(name='workouts3') workoutsbox.save() failbox = Mailbox.objects.create(name='Failed') failbox.save() m = Message(mailbox=workoutsbox, from_header = u.email, subject = "bike", body = """ workout bike """) m.save() a2 = 'media/mailbox_attachments/colin4.csv' copy('rowers/tests/testdata/emails/colin.csv',a2) a = MessageAttachment(message=m,document=a2[6:]) a.save() def tearDown(self): filename = 'colin4.csv' 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_emailupload( self, mocked_sqlalchemy,mocked_polar_notifications, mock_get, mock_post): out = StringIO() call_command('processemail', stdout=out,testing=True,mailbox='workouts3') self.assertIn('Successfully processed email attachments',out.getvalue()) ws = Workout.objects.filter(name="bike") self.assertEqual(len(ws),1) w = ws[0] self.assertEqual(w.workouttype,'Bike') #@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() workoutsbox = Mailbox.objects.create(name='workouts4') workoutsbox.save() failbox = Mailbox.objects.create(name='Failed') failbox.save() for filename in os.listdir(u'rowers/tests/testdata/emails'): m = Message(mailbox=workoutsbox, from_header = u.email, subject = filename, body=""" --- workouttype: water boattype: 4x ... """) m.save() a2 = 'media/mailbox_attachments/'+filename copy(u'rowers/tests/testdata/emails/'+filename,a2) a = MessageAttachment(message=m,document=a2[6:]) a.save() m = Message(mailbox=workoutsbox, from_header = u.email, subject = "3x(5min/2min)/r2 \r2", body = """ workout water """) m.save() a2 = 'media/mailbox_attachments/colin2.csv' copy('rowers/tests/testdata/emails/colin.csv',a2) a = MessageAttachment(message=m,document=a2[6:]) a.save() m = Message(mailbox=workoutsbox, from_header = self.theadmin.email, subject = "johnsworkout", body = """ user john race 1 """) m.save() 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()) @override_settings(TESTING=True) class EmailAdminUpload(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(), birthdate = faker.profile()['birthdate'] ) 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') self.race = RaceFactory(manager = self.theadmin) nu = datetime.datetime.now() workoutsbox = Mailbox.objects.create(name='workouts5') workoutsbox.save() failbox = Mailbox.objects.create(name='Failed') failbox.save() m = Message(mailbox=workoutsbox, from_header = self.theadmin.email, subject = "johnsworkout", body = """ user john race 1 """) m.save() a2 = 'media/mailbox_attachments/minute.csv' copy('rowers/tests/testdata/minute.csv',a2) a = MessageAttachment(message=m,document=a2[6:]) a.save() 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_email_admin_upload( self, mocked_sqlalchemy,mocked_polar_notifications, mock_get, mock_post): out = StringIO() call_command('processemail', stdout=out,testing=True,mailbox='workouts5') self.assertIn('Successfully processed email attachments',out.getvalue()) ws = Workout.objects.filter(name="johnsworkout") if not len(ws): for w in Workout.objects.all(): print(w) self.assertEqual(len(ws),1) w = ws[0] self.assertEqual(w.user.user.username,u'john') results = IndoorVirtualRaceResult.objects.filter( race=self.race) self.assertEqual(len(results),1) result = results[0] self.assertTrue(result.coursecompleted)