From b93ba235b0e991006bb78bd907c4153e54e28d38 Mon Sep 17 00:00:00 2001 From: Marc Date: Wed, 15 Oct 2014 19:29:58 +0000 Subject: [PATCH] Databases tests passing --- orchestra/apps/databases/admin.py | 6 + orchestra/apps/databases/backends.py | 60 ++++---- orchestra/apps/databases/models.py | 28 +--- .../databases/tests/functional_tests/tests.py | 141 ++++++++++++++++-- orchestra/apps/lists/backends.py | 1 + 5 files changed, 163 insertions(+), 73 deletions(-) diff --git a/orchestra/apps/databases/admin.py b/orchestra/apps/databases/admin.py index 0d1eb187..bfcc81a3 100644 --- a/orchestra/apps/databases/admin.py +++ b/orchestra/apps/databases/admin.py @@ -81,6 +81,12 @@ class DatabaseUserAdmin(SelectAccountAdminMixin, ChangePasswordAdminMixin, Exten (r'^(\d+)/password/$', 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) diff --git a/orchestra/apps/databases/backends.py b/orchestra/apps/databases/backends.py index 3b0c611f..c5da2108 100644 --- a/orchestra/apps/databases/backends.py +++ b/orchestra/apps/databases/backends.py @@ -13,27 +13,31 @@ class MySQLBackend(ServiceController): model = 'databases.Database' def save(self, database): - if database.type == database.MYSQL: - context = self.get_context(database) - self.append( - "mysql -e 'CREATE DATABASE `%(database)s`;' || true" % context - ) + 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) + context = self.get_context(database) + self.append("mysql -e 'DROP DATABASE `%(database)s`;'" % context) def commit(self): self.append("mysql -e 'FLUSH PRIVILEGES;'") def get_context(self, database): return { - 'owner': database.owner.username, 'database': database.name, 'host': settings.DATABASES_DEFAULT_HOST, } @@ -44,24 +48,22 @@ 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 \ - """ % context - )) - self.append(textwrap.dedent("""\ - mysql -e 'UPDATE mysql.user SET Password="%(password)s" WHERE User="%(username)s";' \ - """ % context - )) + context = self.get_context(user) + self.append(textwrap.dedent("""\ + mysql -e 'CREATE USER "%(username)s"@"%(host)s";' || true \ + """ % context + )) + self.append(textwrap.dedent("""\ + mysql -e 'UPDATE mysql.user SET Password="%(password)s" WHERE User="%(username)s";' \ + """ % context + )) def delete(self, user): - if user.type == user.MYSQL: - context = self.get_context(database) - self.append(textwrap.dedent("""\ - mysql -e 'DROP USER "%(username)s"@"%(host)s";' \ - """ % context - )) + context = self.get_context(user) + self.append(textwrap.dedent("""\ + mysql -e 'DROP USER "%(username)s"@"%(host)s";' \ + """ % context + )) def commit(self): self.append("mysql -e 'FLUSH PRIVILEGES;'") @@ -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") diff --git a/orchestra/apps/databases/models.py b/orchestra/apps/databases/models.py index 87795102..7aec82c4 100644 --- a/orchestra/apps/databases/models.py +++ b/orchestra/apps/databases/models.py @@ -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' diff --git a/orchestra/apps/databases/tests/functional_tests/tests.py b/orchestra/apps/databases/tests/functional_tests/tests.py index 765f01e8..5b4fd725 100644 --- a/orchestra/apps/databases/tests/functional_tests/tests.py +++ b/orchestra/apps/databases/tests/functional_tests/tests.py @@ -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_change_password(self): + 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) + + 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) - # TODO test add user - # TODO remove user - # TODO remove all users 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,10 +180,29 @@ 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): @@ -160,15 +238,50 @@ 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): diff --git a/orchestra/apps/lists/backends.py b/orchestra/apps/lists/backends.py index 509a9e84..b8f490d4 100644 --- a/orchestra/apps/lists/backends.py +++ b/orchestra/apps/lists/backends.py @@ -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) ]] || {