pangea/django-orchestra
7
0
Fork 0
Code Issues 1 Pull requests 2 Packages Projects Releases Wiki Activity

Databases tests passing

Browse source
This commit is contained in:
Marc 2014-10-15 19:29:58 +00:00
parent 3e246f9fe0
commit b93ba235b0
5 changed files with 163 additions and 73 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
orchestra/apps/databases/admin.py
View file

@ -82,6 +82,12 @@ class DatabaseUserAdmin(SelectAccountAdminMixin, ChangePasswordAdminMixin, Exten
self.admin_site.admin_view(useradmin.user_change_password))
) + super(DatabaseUserAdmin, self).get_urls()
def save_model(self, request, obj, form, change):
""" set password """
if not change:
obj.set_password(form.cleaned_data["password1"])
super(DatabaseUserAdmin, self).save_model(request, obj, form, change)
admin.site.register(Database, DatabaseAdmin)
admin.site.register(DatabaseUser, DatabaseUserAdmin)

22
orchestra/apps/databases/backends.py
View file

@ -13,18 +13,23 @@ class MySQLBackend(ServiceController):
model = 'databases.Database'
def save(self, database):
if database.type == database.MYSQL:
context = self.get_context(database)
# Not available on delete()
context['owner'] = database.owner
self.append(
"mysql -e 'CREATE DATABASE `%(database)s`;' || true" % context
)
for user in database.users.all():
context.update({
'username': user.username,
'grant': 'WITH GRANT OPTION' if user == context['owner'] else ''
})
self.append(textwrap.dedent("""\
mysql -e 'GRANT ALL PRIVILEGES ON `%(database)s`.* TO "%(owner)s"@"%(host)s" WITH GRANT OPTION;' \
mysql -e 'GRANT ALL PRIVILEGES ON `%(database)s`.* TO "%(username)s"@"%(host)s" %(grant)s;' \
""" % context
))
def delete(self, database):
if database.type == database.MYSQL:
context = self.get_context(database)
self.append("mysql -e 'DROP DATABASE `%(database)s`;'" % context)
@ -33,7 +38,6 @@ class MySQLBackend(ServiceController):
def get_context(self, database):
return {
'owner': database.owner.username,
'database': database.name,
'host': settings.DATABASES_DEFAULT_HOST,
}
@ -44,7 +48,6 @@ class MySQLUserBackend(ServiceController):
model = 'databases.DatabaseUser'
def save(self, user):
if user.type == user.MYSQL:
context = self.get_context(user)
self.append(textwrap.dedent("""\
mysql -e 'CREATE USER "%(username)s"@"%(host)s";' || true \
@ -56,8 +59,7 @@ class MySQLUserBackend(ServiceController):
))
def delete(self, user):
if user.type == user.MYSQL:
context = self.get_context(database)
context = self.get_context(user)
self.append(textwrap.dedent("""\
mysql -e 'DROP USER "%(username)s"@"%(host)s";' \
""" % context
@ -74,12 +76,6 @@ class MySQLUserBackend(ServiceController):
}
# TODO https://docs.djangoproject.com/en/1.7/ref/signals/#m2m-changed
class MySQLPermissionBackend(ServiceController):
model = 'databases.UserDatabaseRelation'
verbose_name = "MySQL permission"
class MysqlDisk(ServiceMonitor):
model = 'databases.Database'
verbose_name = _("MySQL disk")

28
orchestra/apps/databases/models.py
View file

@ -17,7 +17,6 @@ class Database(models.Model):
validators=[validators.validate_name])
users = models.ManyToManyField('databases.DatabaseUser',
verbose_name=_("users"),related_name='databases')
# through='databases.Role',
type = models.CharField(_("type"), max_length=32,
choices=settings.DATABASES_TYPE_CHOICES,
default=settings.DATABASES_DEFAULT_TYPE)
@ -35,36 +34,11 @@ class Database(models.Model):
""" database owner is the first user related to it """
# Accessing intermediary model to get which is the first user
users = Database.users.through.objects.filter(database_id=self.id)
return users.order_by('-id').first().databaseuser
return users.order_by('id').first().databaseuser
Database.users.through._meta.unique_together = (('database', 'databaseuser'),)
#class Role(models.Model):
# database = models.ForeignKey(Database, verbose_name=_("database"),
# related_name='roles')
# user = models.ForeignKey('databases.DatabaseUser', verbose_name=_("user"),
# related_name='roles')
## is_owner = models.BooleanField(_("owner"), default=False)
#
# class Meta:
# unique_together = ('database', 'user')
#
# def __unicode__(self):
# return "%s@%s" % (self.user, self.database)
#
# @property
# def is_owner(self):
# return datatase.owner == self
#
# def clean(self):
# if self.user.type != self.database.type:
# msg = _("Database and user type doesn't match")
# raise validators.ValidationError(msg)
# roles = self.database.roles.values('id')
# if not roles or (len(roles) == 1 and roles[0].id == self.id):
# self.is_owner = True
class DatabaseUser(models.Model):
MYSQL = 'mysql'

139
orchestra/apps/databases/tests/functional_tests/tests.py
View file

@ -7,11 +7,14 @@ from functools import partial
from django.conf import settings as djsettings
from django.core.management.base import CommandError
from django.core.urlresolvers import reverse
from selenium.webdriver.common.action_chains import ActionChains
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support.select import Select
from orchestra.admin.utils import change_url
from orchestra.apps.accounts.models import Account
from orchestra.apps.orchestration.models import Server, Route
from orchestra.utils.system import run
from orchestra.utils.system import sshrun
from orchestra.utils.tests import (BaseLiveServerTestCase, random_ascii, save_response_on_error,
snapshot_on_error)
@ -59,20 +62,64 @@ class DatabaseTestMixin(object):
self.add(dbname, username, password)
self.validate_create_table(dbname, username, password)
def test_delete(self):
dbname = '%s_database' % random_ascii(5)
username = '%s_dbuser' % random_ascii(5)
password = '@!?%spppP001' % random_ascii(5)
self.add(dbname, username, password)
self.validate_create_table(dbname, username, password)
self.delete(dbname)
self.delete_user(username)
self.validate_delete(dbname, username, password)
self.validate_delete_user(dbname, username)
def test_change_password(self):
dbname = '%s_database' % random_ascii(5)
username = '%s_dbuser' % random_ascii(5)
password = '@!?%spppP001' % random_ascii(5)
self.add(dbname, username, password)
self.addCleanup(self.delete, dbname)
self.addCleanup(self.delete_user, username)
self.validate_create_table(dbname, username, password)
new_password = '@!?%spppP001' % random_ascii(5)
self.change_password(username, new_password)
self.validate_login_error(dbname, username, password)
self.validate_create_table(dbname, username, new_password)
# TODO test add user
# TODO remove user
# TODO remove all users
def test_add_user(self):
dbname = '%s_database' % random_ascii(5)
username = '%s_dbuser' % random_ascii(5)
password = '@!?%spppP001' % random_ascii(5)
self.add(dbname, username, password)
self.addCleanup(self.delete, dbname)
self.addCleanup(self.delete_user, username)
self.validate_create_table(dbname, username, password)
username2 = '%s_dbuser' % random_ascii(5)
password2 = '@!?%spppP001' % random_ascii(5)
self.add_user(username2, password2)
self.addCleanup(self.delete_user, username2)
self.validate_login_error(dbname, username2, password2)
self.add_user_to_db(username2, dbname)
self.validate_create_table(dbname, username, password)
self.validate_create_table(dbname, username2, password2)
def test_delete_user(self):
dbname = '%s_database' % random_ascii(5)
username = '%s_dbuser' % random_ascii(5)
password = '@!?%spppP001' % random_ascii(5)
self.add(dbname, username, password)
self.addCleanup(self.delete, dbname)
self.validate_create_table(dbname, username, password)
username2 = '%s_dbuser' % random_ascii(5)
password2 = '@!?%spppP001' % random_ascii(5)
self.add_user(username2, password2)
self.add_user_to_db(username2, dbname)
self.delete_user(username)
self.validate_login_error(dbname, username, password)
self.validate_create_table(dbname, username2, password2)
self.delete_user(username2)
self.validate_login_error(dbname, username2, password2)
class MySQLBackendMixin(object):
db_type = 'mysql'
@ -97,15 +144,27 @@ class MySQLBackendMixin(object):
def validate_create_table(self, name, username, password):
db = MySQLdb.connect(host=self.MASTER_SERVER, port=3306, user=username, passwd=password, db=name)
cur = db.cursor()
cur.execute('CREATE TABLE %s ( id INT ) ;' % random_ascii(20))
cur.execute('CREATE TABLE table_%s ( id INT ) ;' % random_ascii(10))
def validate_login_error(self, dbname, username, password):
self.assertRaises(MySQLdb.OperationalError,
self.validate_create_table, dbname, username, password)
self.validate_create_table, dbname, username, password
)
def validate_delete(self, name, username, password):
self.asseRaises(MySQLdb.ConnectionError,
self.validate_create_table, name, username, password)
self.assertRaises(MySQLdb.OperationalError,
self.validate_create_table, name, username, password
)
def validate_delete_user(self, name, username):
context = {
'name': name,
'username': username,
}
self.assertEqual('', sshrun(self.MASTER_SERVER,
"""mysql mysql -e 'SELECT * FROM db WHERE db="%(name)s";'""" % context, display=False).stdout)
self.assertEqual('', sshrun(self.MASTER_SERVER,
"""mysql mysql -e 'SELECT * FROM user WHERE user="%(username)s";'""" % context, display=False).stdout)
class RESTDatabaseMixin(DatabaseTestMixin):
@ -121,11 +180,30 @@ class RESTDatabaseMixin(DatabaseTestMixin):
}]
self.rest.databases.create(name=dbname, users=users, type=self.db_type)
@save_response_on_error
def delete(self, dbname):
self.rest.databases.retrieve(name=dbname).delete()
@save_response_on_error
def change_password(self, username, password):
user = self.rest.databaseusers.retrieve(username=username).get()
user.set_password(password)
@save_response_on_error
def add_user(self, username, password):
self.rest.databaseusers.create(username=username, password=password, type=self.db_type)
@save_response_on_error
def add_user_to_db(self, username, dbname):
user = self.rest.databaseusers.retrieve(username=username).get()
db = self.rest.databases.retrieve(name=dbname).get()
db.users.append(user)
db.save()
@save_response_on_error
def delete_user(self, username):
self.rest.databaseusers.retrieve(username=username).delete()
class AdminDatabaseMixin(DatabaseTestMixin):
def setUp(self):
@ -160,16 +238,51 @@ class AdminDatabaseMixin(DatabaseTestMixin):
db = Database.objects.get(name=dbname)
self.admin_delete(db)
@snapshot_on_error
def delete_user(self, username):
user = DatabaseUser.objects.get(username=username)
self.admin_delete(user)
@snapshot_on_error
def change_password(self, username, password):
user = DatabaseUser.objects.get(username=username)
self.admin_change_password(user, password)
@snapshot_on_error
def add_user(self, username, password):
url = self.live_server_url + reverse('admin:databases_databaseuser_add')
self.selenium.get(url)
type_input = self.selenium.find_element_by_id('id_type')
type_select = Select(type_input)
type_select.select_by_value(self.db_type)
username_field = self.selenium.find_element_by_id('id_username')
username_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)
username_field.submit()
self.assertNotEqual(url, self.selenium.current_url)
@snapshot_on_error
def add_user_to_db(self, username, dbname):
database = Database.objects.get(name=dbname, type=self.db_type)
url = self.live_server_url + change_url(database)
self.selenium.get(url)
user = DatabaseUser.objects.get(username=username, type=self.db_type)
users_input = self.selenium.find_element_by_id('id_users')
users_select = Select(users_input)
users_select.select_by_value(str(user.pk))
save = self.selenium.find_element_by_name('_save')
save.submit()
self.assertNotEqual(url, self.selenium.current_url)
@snapshot_on_error
def delete_user(self, username):
user = DatabaseUser.objects.get(username=username)
self.admin_delete(user)
class RESTMysqlDatabaseTest(MySQLBackendMixin, RESTDatabaseMixin, BaseLiveServerTestCase):
pass

1
orchestra/apps/lists/backends.py
View file

@ -30,6 +30,7 @@ class MailmanBackend(ServiceController):
# TODO for list virtual_domains cleaning up we need to know the old domain name when a list changes its address
# domain, but this is not possible with the current design.
# sync the whole file everytime?
# TODO same for mailbox virtual domains
if context['address_domain']:
self.append(textwrap.dedent("""
[[ $(grep "^\s*%(address_domain)s\s*$" %(virtual_alias_domains)s) ]] || {