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

377 lines
14 KiB
Python

import imaplib
import os
import poplib
import smtplib
import time
import textwrap
from email.mime.text import MIMEText
from django.apps import apps
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
from orchestra.contrib.orchestration.models import Server, Route
from orchestra.contrib.resources.models import Resource
from orchestra.utils.sys import sshrun
from orchestra.utils.tests import BaseLiveServerTestCase, random_ascii, snapshot_on_error, save_response_on_error
from ... import backends, settings
from ...models import Mailbox
class MailboxMixin(object):
MASTER_SERVER = os.environ.get('ORCHESTRA_SLAVE_SERVER', 'localhost')
DEPENDENCIES = (
'orchestra.contrib.orchestration',
'orchestra.contrib.mails',
'orchestra.contrib.resources',
)
def setUp(self):
super(MailboxMixin, self).setUp()
self.add_route()
# clean resource relation from other tests
apps.get_app_config('resources').reload_relations()
djsettings.DEBUG = True
def add_route(self):
server = Server.objects.create(name=self.MASTER_SERVER)
backend = backends.RoundcubeIdentityController.get_name()
Route.objects.create(backend=backend, match=True, host=server)
backend = backends.PostfixAddressController.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
msg['From'] = 'orchestra@%s' % self.MASTER_SERVER
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()
def validate_mailbox(self, username):
sshrun(self.MASTER_SERVER, "doveadm search -u %s ALL" % username, display=False)
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)
self.addCleanup(self.delete, username)
imap = self.login_imap(username, password)
self.validate_mailbox(username)
def test_change_password(self):
username = '%s_systemuser' % random_ascii(10)
password = '@!?%spppP001' % random_ascii(5)
self.add(username, password)
self.addCleanup(self.delete, username)
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)
self.addCleanup(self.delete, username)
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)
self.addCleanup(self.delete, username)
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)
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)
# self.addCleanup(self.delete, username)
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)
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)
self.delete_address(username)
self.send_email("%s@%s" % (name, domain), token)
self.validate_email(username, token)
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)
self.send_email("%s@%s" % (username, settings.MAILBOXES_VIRTUAL_MAILBOX_DEFAULT_DOMAIN), token)
home = Mailbox.objects.get(name=username).get_home()
sshrun(self.MASTER_SERVER,
"grep '%s' %s/Maildir/.%s/new/*" % (token, home, folder), display=False)
# TODO test update shit
# TODO test autoreply
class RESTMailboxMixin(MailboxMixin):
def setUp(self):
super(RESTMailboxMixin, self).setUp()
self.rest_login()
@save_response_on_error
def add(self, username, password, quota=None, filtering=None):
extra = {}
if quota:
extra.update({
"resources": [
{
"name": "disk",
"allocated": quota
},
]
})
if filtering:
extra.update({
'filtering': 'CUSTOM',
'custom_filtering': filtering,
})
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])
@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 disable(self, username):
mailbox = self.rest.mailboxes.retrieve(name=username).get()
mailbox.update(is_active=False)
class AdminMailboxMixin(MailboxMixin):
def setUp(self):
super(AdminMailboxMixin, self).setUp()
self.admin_login()
@snapshot_on_error
def add(self, username, password, quota=None, filtering=None):
url = self.live_server_url + reverse('admin:mailboxes_mailbox_add')
self.selenium.get(url)
# account_input = self.selenium.find_element_by_id('id_account')
# account_select = Select(account_input)
# account_select.select_by_value(str(self.account.pk))
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)
if quota is not None:
quota_id = 'id_resources-resourcedata-content_type-object_id-0-allocated'
quota_field = self.selenium.find_element_by_id(quota_id)
quota_field.clear()
quota_field.send_keys(quota)
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)
name_field.submit()
self.assertNotEqual(url, self.selenium.current_url)
@snapshot_on_error
def delete(self, username):
mailbox = Mailbox.objects.get(name=username)
self.admin_delete(mailbox)
@snapshot_on_error
def change_password(self, username, password):
mailbox = Mailbox.objects.get(name=username)
self.admin_change_password(mailbox, password)
@snapshot_on_error
def add_address(self, username, name, domain):
url = self.live_server_url + reverse('admin:mailboxes_address_add')
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)
@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)
class RESTMailboxTest(RESTMailboxMixin, BaseLiveServerTestCase):
pass
class AdminMailboxTest(AdminMailboxMixin, BaseLiveServerTestCase):
pass