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') u = User.objects.create_user('john', 'sander@ds.ds', 'koeinsloot') r = Rower.objects.create(user=u,gdproptin=True, gdproptindate=timezone.now() ) self.theadmin = UserFactory(is_staff=True) self.rtheadmin = Rower.objects.create(user=self.theadmin, birthdate = faker.profile()['birthdate'], gdproptin=True, gdproptindate=timezone.now(), rowerplan='coach') nu = datetime.datetime.now() workoutsbox = Mailbox.objects.create(name='workouts') 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.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) 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') #@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, gdproptindate=timezone.now() ) self.theadmin = UserFactory(is_staff=True) self.rtheadmin = Rower.objects.create(user=self.theadmin, birthdate = faker.profile()['birthdate'], gdproptin=True, gdproptindate=timezone.now(), rowerplan='coach') nu = datetime.datetime.now() workoutsbox = Mailbox.objects.create(name='workouts') 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) 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, 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, gdproptindate=timezone.now(), rowerplan='coach') self.race = RaceFactory(manager = self.theadmin) nu = datetime.datetime.now() workoutsbox = Mailbox.objects.create(name='workouts') 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) 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)