django-orchestra/orchestra/contrib/mailboxes/tests/functional_tests/tests.py

382 lines
14 KiB
Python
Raw Normal View History

2014-10-06 14:57:02 +00:00
import imaplib
import os
import poplib
import smtplib
import time
2014-10-09 17:04:12 +00:00
import textwrap
2014-10-06 14:57:02 +00:00
from email.mime.text import MIMEText
2014-10-04 13:23:04 +00:00
2014-10-09 17:04:12 +00:00
from django.apps import apps
2014-10-06 14:57:02 +00:00
from django.conf import settings as djsettings
from django.contrib.contenttypes.models import ContentType
from django.core.management.base import CommandError
from django.core.urlresolvers import reverse
from selenium.webdriver.support.select import Select
2015-04-05 10:46:24 +00:00
from orchestra.contrib.orchestration.models import Server, Route
from orchestra.contrib.resources.models import Resource
from orchestra.utils.sys import sshrun
2014-10-06 14:57:02 +00:00
from orchestra.utils.tests import BaseLiveServerTestCase, random_ascii, snapshot_on_error, save_response_on_error
from ... import backends, settings
from ...models import Mailbox
2014-10-04 13:23:04 +00:00
2014-10-06 14:57:02 +00:00
class MailboxMixin(object):
MASTER_SERVER = os.environ.get('ORCHESTRA_SLAVE_SERVER', 'localhost')
DEPENDENCIES = (
2015-04-05 10:46:24 +00:00
'orchestra.contrib.orchestration',
'orchestra.contrib.mails',
'orchestra.contrib.resources',
2014-10-06 14:57:02 +00:00
)
def setUp(self):
super(MailboxMixin, self).setUp()
self.add_route()
2014-10-07 13:08:59 +00:00
# clean resource relation from other tests
apps.get_app_config('resources').reload_relations()
2014-10-06 14:57:02 +00:00
djsettings.DEBUG = True
def add_route(self):
server = Server.objects.create(name=self.MASTER_SERVER)
backend = backends.PasswdVirtualUserBackend.get_name()
Route.objects.create(backend=backend, match=True, host=server)
backend = backends.PostfixAddressBackend.get_name()
Route.objects.create(backend=backend, match=True, host=server)
def add_quota_resource(self):
Resource.objects.create(
name='disk',
content_type=ContentType.objects.get_for_model(Mailbox),
period=Resource.LAST,
verbose_name='Mail quota',
unit='MB',
scale=10**6,
on_demand=False,
default_allocation=2000
)
def save(self):
raise NotImplementedError
def add(self):
raise NotImplementedError
def delete(self):
raise NotImplementedError
def update(self):
raise NotImplementedError
def disable(self):
raise NotImplementedError
def add_group(self, username, groupname):
raise NotImplementedError
def login_imap(self, username, password):
mail = imaplib.IMAP4_SSL(self.MASTER_SERVER)
status, msg = mail.login(username, password)
self.assertEqual('OK', status)
self.assertEqual(['Logged in'], msg)
return mail
def login_pop3(self, username, password):
pop = poplib.POP3(self.MASTER_SERVER)
pop.user(username)
pop.pass_(password)
return pop
def send_email(self, to, token):
msg = MIMEText(token)
msg['To'] = to
2014-10-09 17:04:12 +00:00
msg['From'] = 'orchestra@%s' % self.MASTER_SERVER
2014-10-06 14:57:02 +00:00
msg['Subject'] = 'test'
server = smtplib.SMTP(self.MASTER_SERVER, 25)
try:
server.ehlo()
server.starttls()
server.ehlo()
server.sendmail(msg['From'], msg['To'], msg.as_string())
finally:
server.quit()
2014-10-07 13:08:59 +00:00
def validate_mailbox(self, username):
sshrun(self.MASTER_SERVER, "doveadm search -u %s ALL" % username, display=False)
2014-10-06 14:57:02 +00:00
def validate_email(self, username, token):
home = Mailbox.objects.get(name=username).get_home()
sshrun(self.MASTER_SERVER, "grep '%s' %s/Maildir/new/*" % (token, home), display=False)
def test_add(self):
username = '%s_mailbox' % random_ascii(10)
password = '@!?%spppP001' % random_ascii(5)
self.add(username, password)
2014-10-07 13:08:59 +00:00
self.addCleanup(self.delete, username)
2014-10-06 14:57:02 +00:00
imap = self.login_imap(username, password)
2014-10-07 13:08:59 +00:00
self.validate_mailbox(username)
2014-10-06 14:57:02 +00:00
def test_change_password(self):
username = '%s_systemuser' % random_ascii(10)
password = '@!?%spppP001' % random_ascii(5)
self.add(username, password)
2014-10-07 13:08:59 +00:00
self.addCleanup(self.delete, username)
2014-10-06 14:57:02 +00:00
imap = self.login_imap(username, password)
new_password = '@!?%spppP001' % random_ascii(5)
self.change_password(username, new_password)
imap = self.login_imap(username, new_password)
def test_quota(self):
username = '%s_mailbox' % random_ascii(10)
password = '@!?%spppP001' % random_ascii(5)
self.add_quota_resource()
quota = 100
self.add(username, password, quota=quota)
2014-10-07 13:08:59 +00:00
self.addCleanup(self.delete, username)
2014-10-06 14:57:02 +00:00
get_quota = "doveadm quota get -u %s 2>&1|grep STORAGE|awk {'print $5'}" % username
stdout = sshrun(self.MASTER_SERVER, get_quota, display=False).stdout
self.assertEqual(quota*1024, int(stdout))
imap = self.login_imap(username, password)
imap_quota = int(imap.getquotaroot("INBOX")[1][1][0].split(' ')[-1].split(')')[0])
self.assertEqual(quota*1024, imap_quota)
def test_send_email(self):
username = '%s_mailbox' % random_ascii(10)
password = '@!?%spppP001' % random_ascii(5)
self.add(username, password)
2014-10-07 13:08:59 +00:00
self.addCleanup(self.delete, username)
2014-10-06 14:57:02 +00:00
msg = MIMEText("Hola bishuns")
msg['To'] = 'noexists@example.com'
msg['From'] = '%s@%s' % (username, self.MASTER_SERVER)
msg['Subject'] = "test"
server = smtplib.SMTP(self.MASTER_SERVER, 25)
server.login(username, password)
try:
server.sendmail(msg['From'], msg['To'], msg.as_string())
finally:
server.quit()
def test_address(self):
username = '%s_mailbox' % random_ascii(10)
password = '@!?%spppP001' % random_ascii(5)
self.add(username, password)
2014-10-07 13:08:59 +00:00
self.addCleanup(self.delete, username)
domain = '%s_domain.lan' % random_ascii(5)
name = '%s_name' % random_ascii(5)
domain = self.account.domains.create(name=domain)
self.add_address(username, name, domain)
token = random_ascii(100)
self.send_email("%s@%s" % (name, domain), token)
self.validate_email(username, token)
def test_disable(self):
username = '%s_systemuser' % random_ascii(10)
password = '@!?%spppP001' % random_ascii(5)
self.add(username, password)
self.validate_mailbox(username)
2014-10-09 17:04:12 +00:00
# self.addCleanup(self.delete, username)
2014-10-07 13:08:59 +00:00
imap = self.login_imap(username, password)
self.disable(username)
self.assertRaises(imap.error, self.login_imap, username, password)
def test_delete(self):
username = '%s_systemuser' % random_ascii(10)
password = '@!?%sppppP001' % random_ascii(5)
self.add(username, password)
imap = self.login_imap(username, password)
self.validate_mailbox(username)
mailbox = Mailbox.objects.get(name=username)
home = mailbox.get_home()
self.delete(username)
self.assertRaises(Mailbox.DoesNotExist, Mailbox.objects.get, name=username)
self.assertRaises(CommandError, self.validate_mailbox, username)
self.assertRaises(imap.error, self.login_imap, username, password)
self.assertRaises(CommandError,
sshrun, self.MASTER_SERVER, 'ls %s' % home, display=False)
def test_delete_address(self):
username = '%s_mailbox' % random_ascii(10)
password = '@!?%spppP001' % random_ascii(5)
self.add(username, password)
self.addCleanup(self.delete, username)
2014-10-06 14:57:02 +00:00
domain = '%s_domain.lan' % random_ascii(5)
name = '%s_name' % random_ascii(5)
domain = self.account.domains.create(name=domain)
self.add_address(username, name, domain)
token = random_ascii(100)
self.send_email("%s@%s" % (name, domain), token)
self.validate_email(username, token)
2014-10-07 13:08:59 +00:00
self.delete_address(username)
self.send_email("%s@%s" % (name, domain), token)
self.validate_email(username, token)
2014-10-09 17:04:12 +00:00
def test_custom_filtering(self):
username = '%s_mailbox' % random_ascii(10)
password = '@!?%spppP001' % random_ascii(5)
folder = random_ascii(5)
filtering = textwrap.dedent("""
require "fileinto";
if true {
fileinto "%s";
stop;
}""" % folder)
self.add(username, password, filtering=filtering)
self.addCleanup(self.delete, username)
imap = self.login_imap(username, password)
imap.create(folder)
self.validate_mailbox(username)
token = random_ascii(100)
2014-10-17 10:04:47 +00:00
self.send_email("%s@%s" % (username, settings.MAILBOXES_VIRTUAL_MAILBOX_DEFAULT_DOMAIN), token)
2014-10-09 17:04:12 +00:00
home = Mailbox.objects.get(name=username).get_home()
sshrun(self.MASTER_SERVER,
"grep '%s' %s/Maildir/.%s/new/*" % (token, home, folder), display=False)
2014-10-06 14:57:02 +00:00
2014-10-15 12:47:28 +00:00
# TODO test update shit
# TODO test autoreply
2014-10-06 14:57:02 +00:00
class RESTMailboxMixin(MailboxMixin):
def setUp(self):
super(RESTMailboxMixin, self).setUp()
self.rest_login()
@save_response_on_error
2014-10-09 17:04:12 +00:00
def add(self, username, password, quota=None, filtering=None):
2014-10-06 14:57:02 +00:00
extra = {}
if quota:
2014-10-09 17:04:12 +00:00
extra.update({
2014-10-06 14:57:02 +00:00
"resources": [
{
"name": "disk",
"allocated": quota
},
]
2014-10-09 17:04:12 +00:00
})
if filtering:
extra.update({
'filtering': 'CUSTOM',
'custom_filtering': filtering,
})
2014-10-06 14:57:02 +00:00
self.rest.mailboxes.create(name=username, password=password, **extra)
@save_response_on_error
def delete(self, username):
mailbox = self.rest.mailboxes.retrieve(name=username).get()
mailbox.delete()
@save_response_on_error
def change_password(self, username, password):
mailbox = self.rest.mailboxes.retrieve(name=username).get()
mailbox.change_password(password)
@save_response_on_error
def add_address(self, username, name, domain):
mailbox = self.rest.mailboxes.retrieve(name=username).get()
domain = self.rest.domains.retrieve(name=domain.name).get()
self.rest.addresses.create(name=name, domain=domain, mailboxes=[mailbox])
2014-10-07 13:08:59 +00:00
@save_response_on_error
def delete_address(self, username):
mailbox = self.rest.mailboxes.retrieve(name=username).get()
self.rest.addresses.delete()
@save_response_on_error
def change_password(self, username, password):
mailbox = self.rest.mailboxes.retrieve(name=username).get()
mailbox.set_password(password=password)
@save_response_on_error
def disable(self, username):
mailbox = self.rest.mailboxes.retrieve(name=username).get()
mailbox.update(is_active=False)
2014-10-06 14:57:02 +00:00
class AdminMailboxMixin(MailboxMixin):
def setUp(self):
super(AdminMailboxMixin, self).setUp()
self.admin_login()
@snapshot_on_error
2014-10-09 17:04:12 +00:00
def add(self, username, password, quota=None, filtering=None):
2014-10-17 10:04:47 +00:00
url = self.live_server_url + reverse('admin:mailboxes_mailbox_add')
2014-10-06 14:57:02 +00:00
self.selenium.get(url)
2014-10-23 15:38:46 +00:00
# account_input = self.selenium.find_element_by_id('id_account')
# account_select = Select(account_input)
# account_select.select_by_value(str(self.account.pk))
2014-10-06 14:57:02 +00:00
name_field = self.selenium.find_element_by_id('id_name')
name_field.send_keys(username)
password_field = self.selenium.find_element_by_id('id_password1')
password_field.send_keys(password)
password_field = self.selenium.find_element_by_id('id_password2')
password_field.send_keys(password)
2014-10-09 17:04:12 +00:00
2014-10-06 14:57:02 +00:00
if quota is not None:
2014-10-09 17:04:12 +00:00
quota_id = 'id_resources-resourcedata-content_type-object_id-0-allocated'
quota_field = self.selenium.find_element_by_id(quota_id)
2014-10-06 14:57:02 +00:00
quota_field.clear()
quota_field.send_keys(quota)
2014-10-09 17:04:12 +00:00
if filtering is not None:
filtering_input = self.selenium.find_element_by_id('id_filtering')
filtering_select = Select(filtering_input)
filtering_select.select_by_value("CUSTOM")
filtering_inline = self.selenium.find_element_by_id('fieldsetcollapser0')
filtering_inline.click()
time.sleep(0.5)
filtering_field = self.selenium.find_element_by_id('id_custom_filtering')
filtering_field.send_keys(filtering)
2014-10-06 14:57:02 +00:00
name_field.submit()
self.assertNotEqual(url, self.selenium.current_url)
@snapshot_on_error
def delete(self, username):
mailbox = Mailbox.objects.get(name=username)
2014-10-07 13:08:59 +00:00
self.admin_delete(mailbox)
2014-10-06 14:57:02 +00:00
@snapshot_on_error
def change_password(self, username, password):
mailbox = Mailbox.objects.get(name=username)
2014-10-07 13:08:59 +00:00
self.admin_change_password(mailbox, password)
2014-10-06 14:57:02 +00:00
@snapshot_on_error
def add_address(self, username, name, domain):
2014-10-17 10:04:47 +00:00
url = self.live_server_url + reverse('admin:mailboxes_address_add')
2014-10-06 14:57:02 +00:00
self.selenium.get(url)
name_field = self.selenium.find_element_by_id('id_name')
name_field.send_keys(name)
domain_input = self.selenium.find_element_by_id('id_domain')
domain_select = Select(domain_input)
domain_select.select_by_value(str(domain.pk))
mailboxes = self.selenium.find_element_by_id('id_mailboxes_add_all_link')
mailboxes.click()
time.sleep(0.5)
name_field.submit()
self.assertNotEqual(url, self.selenium.current_url)
2014-10-07 13:08:59 +00:00
@snapshot_on_error
def delete_address(self, username):
mailbox = Mailbox.objects.get(name=username)
address = mailbox.addresses.get()
self.admin_delete(address)
@snapshot_on_error
def disable(self, username):
mailbox = Mailbox.objects.get(name=username)
self.admin_disable(mailbox)
2014-10-06 14:57:02 +00:00
class RESTMailboxTest(RESTMailboxMixin, BaseLiveServerTestCase):
pass
class AdminMailboxTest(AdminMailboxMixin, BaseLiveServerTestCase):
pass