at the point where the test is there and fails
No changes to the code yet, so we expected it to fail
This commit is contained in:
Sander Roosendaal
213
rowers/tests/test_emails.py
Normal file
213
rowers/tests/test_emails.py
Normal file
@@ -0,0 +1,213 @@
|
||||
#from __future__ import print_function
|
||||
from statements import *
|
||||
|
||||
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'
|
||||
copyfile('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,WindowsError):
|
||||
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
|
||||
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
|
||||
copyfile(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'
|
||||
copyfile('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,WindowsError):
|
||||
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())
|
||||
|
||||
|
||||
|
||||
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()
|
||||
)
|
||||
|
||||
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 = self.theadmin.email,
|
||||
subject = "johnsworkout",
|
||||
body = """
|
||||
user john
|
||||
race 1
|
||||
""")
|
||||
m.save()
|
||||
a2 = 'media/mailbox_attachments/minute.csv'
|
||||
copyfile('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,WindowsError):
|
||||
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')
|
||||
|
||||
|
||||
Reference in New Issue
Block a user