165 lines
5.6 KiB
Python
165 lines
5.6 KiB
Python
#from __future__ import print_function
|
|
import pytest
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
from bs4 import BeautifulSoup
|
|
import re
|
|
from nose_parameterized import parameterized
|
|
from django.test import TestCase, Client,override_settings
|
|
from django.core.management import call_command
|
|
from django.utils.six import StringIO
|
|
from django.test.client import RequestFactory
|
|
from rowers.views import checkworkoutuser,c2_open
|
|
from rowers.models import Workout, User, Rower, WorkoutForm,RowerForm,GraphImage
|
|
from rowers.forms import DocumentsForm,CNsummaryForm,RegistrationFormUniqueEmail
|
|
import rowers.plots as plots
|
|
import rowers.interactiveplots as iplots
|
|
import datetime
|
|
from rowingdata import rowingdata as rdata
|
|
from rowingdata import rower as rrower
|
|
from django.utils import timezone
|
|
from rowers.rows import handle_uploaded_file
|
|
from django.core.files.uploadedfile import SimpleUploadedFile
|
|
from time import strftime,strptime,mktime,time,daylight
|
|
import os
|
|
from rowers.tasks import handle_makeplot
|
|
from rowers.utils import serialize_list,deserialize_list
|
|
from rowers.utils import NoTokenError
|
|
from shutil import copyfile
|
|
from nose.tools import assert_true
|
|
from mock import Mock, patch
|
|
from minimocktest import MockTestCase
|
|
import pandas as pd
|
|
import rowers.c2stuff as c2stuff
|
|
|
|
import json
|
|
import numpy as np
|
|
|
|
from rowers import urls
|
|
from rowers.views import error500_view,error404_view,error400_view,error403_view
|
|
|
|
from rowers.dataprep import delete_strokedata
|
|
|
|
from redis import StrictRedis
|
|
redis_connection = StrictRedis()
|
|
|
|
from rowers.tests.test_imports import mocked_requests
|
|
|
|
|
|
from django_mailbox.models import Mailbox,MessageAttachment,Message
|
|
|
|
from rowers.tests.mocks import mocked_sqlalchemy
|
|
|
|
|
|
@pytest.mark.django_db
|
|
class UploadTests(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(),
|
|
getemailnotifications = False,
|
|
)
|
|
|
|
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 = "3x(5min/2min)/r2 \r2",
|
|
body = """
|
|
workout run
|
|
""")
|
|
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()
|
|
|
|
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('requests.get', side_effect=mocked_requests)
|
|
def test_email_workouttype(self, mock_get):
|
|
out = StringIO()
|
|
call_command('processemail', stdout=out, testing=True)
|
|
w = Workout.objects.get(id=1)
|
|
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()
|
|
)
|
|
|
|
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()
|
|
|
|
|
|
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())
|
|
|