rowsandall/rowers/tests/test_emails.py

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()
        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)
        a3 = 'media/mailbox_attachments/colin4.csv'
        copy('rowers/tests/testdata/emails/colin.csv',a3)
        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',
            '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')



    @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)


    @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)