477 lines
18 KiB
Python
477 lines
18 KiB
Python
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='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.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,
|
|
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='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,
|
|
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='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,
|
|
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='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,
|
|
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='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,
|
|
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='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,
|
|
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='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)
|
|
|