Jan Dittberner
1649e4592e
This commit adds a set of unit tests for the userdbs app. Some tests will fail because a refactoring to signals comes with the next commit.
296 lines
13 KiB
Python
296 lines
13 KiB
Python
"""
|
|
This module provides tests for :py:mod:`userdbs.models`.
|
|
|
|
"""
|
|
from __future__ import unicode_literals
|
|
|
|
from django.contrib.auth import get_user_model
|
|
from django.test import TestCase
|
|
from django.test.utils import override_settings
|
|
|
|
from osusers.models import User
|
|
from taskresults.models import TaskResult
|
|
from userdbs.models import DB_TYPES, DatabaseUser, UserDatabase
|
|
|
|
Customer = get_user_model()
|
|
|
|
|
|
@override_settings(
|
|
CELERY_ALWAYS_EAGER=True,
|
|
CELERY_CACHE_BACKEND='memory',
|
|
BROKER_BACKEND='memory'
|
|
)
|
|
class TestCaseWithCeleryTasks(TestCase):
|
|
pass
|
|
|
|
|
|
class DatabaseUserManagerTest(TestCaseWithCeleryTasks):
|
|
"""
|
|
Test case for :py:class:`userdbs.models.DatabaseUserManager`.
|
|
|
|
"""
|
|
def setUp(self):
|
|
self.customer = Customer.objects.create_user(username='testcustomer')
|
|
self.osuser = User.objects.create_user(customer=self.customer)
|
|
TaskResult.objects.all().delete()
|
|
|
|
def test_create_database_user_with_name(self):
|
|
dbu = DatabaseUser.objects.create_database_user(
|
|
self.osuser, DB_TYPES.pgsql, 'testname', 'secret')
|
|
self.assertEqual(dbu.name, 'testname')
|
|
self.assertEqual(dbu.osuser, self.osuser)
|
|
self.assertEqual(dbu.db_type, DB_TYPES.pgsql)
|
|
taskres = TaskResult.objects.all()
|
|
self.assertEqual(len(taskres), 1)
|
|
self.assertEqual(taskres[0].creator, 'handle_dbuser_created')
|
|
self.assertEqual(taskres[0].notes, 'pgsql user creation')
|
|
|
|
def test_create_database_user_with_name_no_commit(self):
|
|
dbu = DatabaseUser.objects.create_database_user(
|
|
self.osuser, DB_TYPES.pgsql, 'testname', 'secret', False)
|
|
self.assertEqual(dbu.name, 'testname')
|
|
self.assertEqual(dbu.osuser, self.osuser)
|
|
self.assertEqual(dbu.db_type, DB_TYPES.pgsql)
|
|
self.assertFalse(TaskResult.objects.exists())
|
|
|
|
def test_create_database_user_generate_name(self):
|
|
dbu = DatabaseUser.objects.create_database_user(
|
|
self.osuser, DB_TYPES.pgsql)
|
|
self.assertEqual(dbu.name, '{user}db01'.format(
|
|
user=self.osuser.username))
|
|
self.assertEqual(dbu.osuser, self.osuser)
|
|
self.assertEqual(dbu.db_type, DB_TYPES.pgsql)
|
|
taskres = TaskResult.objects.all()
|
|
self.assertEqual(len(taskres), 1)
|
|
self.assertEqual(taskres[0].creator, 'handle_dbuser_created')
|
|
self.assertEqual(taskres[0].notes, 'pgsql user creation')
|
|
|
|
def test_create_database_user_multiple_generate_name(self):
|
|
dbu = DatabaseUser.objects.create_database_user(
|
|
self.osuser, DB_TYPES.mysql)
|
|
self.assertEqual(dbu.name, '{user}db01'.format(
|
|
user=self.osuser.username))
|
|
self.assertEqual(dbu.osuser, self.osuser)
|
|
self.assertEqual(dbu.db_type, DB_TYPES.mysql)
|
|
dbu = DatabaseUser.objects.create_database_user(
|
|
self.osuser, DB_TYPES.mysql)
|
|
self.assertEqual(dbu.name, '{user}db02'.format(
|
|
user=self.osuser.username))
|
|
self.assertEqual(dbu.osuser, self.osuser)
|
|
self.assertEqual(dbu.db_type, DB_TYPES.mysql)
|
|
taskres = TaskResult.objects.all()
|
|
self.assertEqual(len(taskres), 2)
|
|
self.assertEqual(taskres[0].creator, 'handle_dbuser_created')
|
|
self.assertEqual(taskres[0].notes, 'mysql user creation')
|
|
self.assertEqual(taskres[1].creator, 'handle_dbuser_created')
|
|
self.assertEqual(taskres[1].notes, 'mysql user creation')
|
|
|
|
def test_create_database_user_multiple_gap_generate_name(self):
|
|
dbu = DatabaseUser.objects.create_database_user(
|
|
self.osuser, DB_TYPES.mysql)
|
|
self.assertEqual(dbu.name, '{user}db01'.format(
|
|
user=self.osuser.username))
|
|
self.assertEqual(dbu.osuser, self.osuser)
|
|
self.assertEqual(dbu.db_type, DB_TYPES.mysql)
|
|
dbu = DatabaseUser.objects.create_database_user(
|
|
self.osuser, DB_TYPES.mysql)
|
|
self.assertEqual(dbu.name, '{user}db02'.format(
|
|
user=self.osuser.username))
|
|
self.assertEqual(dbu.osuser, self.osuser)
|
|
self.assertEqual(dbu.db_type, DB_TYPES.mysql)
|
|
DatabaseUser.objects.get(
|
|
name='{user}db01'.format(user=self.osuser.username)).delete()
|
|
dbu = DatabaseUser.objects.create_database_user(
|
|
self.osuser, DB_TYPES.mysql)
|
|
self.assertEqual(dbu.name, '{user}db01'.format(
|
|
user=self.osuser.username))
|
|
self.assertEqual(dbu.osuser, self.osuser)
|
|
self.assertEqual(dbu.db_type, DB_TYPES.mysql)
|
|
taskres = TaskResult.objects.all()
|
|
self.assertEqual(len(taskres), 4)
|
|
self.assertEqual(taskres[0].creator, 'handle_dbuser_created')
|
|
self.assertEqual(taskres[0].notes, 'mysql user creation')
|
|
self.assertEqual(taskres[1].creator, 'handle_dbuser_created')
|
|
self.assertEqual(taskres[1].notes, 'mysql user creation')
|
|
self.assertEqual(taskres[2].creator, 'handle_dbuser_deleted')
|
|
self.assertEqual(taskres[2].notes, 'mysql user deletion')
|
|
self.assertEqual(taskres[3].creator, 'handle_dbuser_created')
|
|
self.assertEqual(taskres[3].notes, 'mysql user creation')
|
|
|
|
|
|
class DatabaseUserTest(TestCaseWithCeleryTasks):
|
|
"""
|
|
Test case for :py:class:`userdbs.models.DatabaseUser`.
|
|
|
|
"""
|
|
def setUp(self):
|
|
self.customer = Customer.objects.create_user(username='testcustomer')
|
|
self.osuser = User.objects.create_user(customer=self.customer)
|
|
self.dbu = DatabaseUser.objects.create_database_user(
|
|
self.osuser, DB_TYPES.pgsql)
|
|
TaskResult.objects.all().delete()
|
|
|
|
def test___str__(self):
|
|
self.assertEqual(
|
|
str(self.dbu),
|
|
'{user}db01 (PostgreSQL for {user})'.format(
|
|
user=self.osuser.username))
|
|
|
|
def test_set_password_pgsql(self):
|
|
self.dbu.set_password('secret')
|
|
taskres = TaskResult.objects.all()
|
|
self.assertEqual(len(taskres), 1)
|
|
self.assertEqual(taskres[0].creator, 'handle_dbuser_password_set')
|
|
self.assertEqual(taskres[0].notes, 'pgsql password change')
|
|
|
|
def test_set_password_mysql(self):
|
|
self.dbu.db_type = DB_TYPES.mysql
|
|
self.dbu.save()
|
|
self.dbu.set_password('secret')
|
|
taskres = TaskResult.objects.all()
|
|
self.assertEqual(len(taskres), 1)
|
|
self.assertEqual(taskres[0].creator, 'handle_dbuser_password_set')
|
|
self.assertEqual(taskres[0].notes, 'mysql password change')
|
|
|
|
def test_delete_no_dbs(self):
|
|
self.dbu.delete()
|
|
taskres = TaskResult.objects.all()
|
|
self.assertEqual(len(taskres), 1)
|
|
self.assertEqual(taskres[0].creator, 'handle_dbuser_deleted')
|
|
self.assertEqual(taskres[0].notes, 'pgsql user deletion')
|
|
|
|
def test_delete_with_dbs(self):
|
|
db = UserDatabase.objects.create_userdatabase(self.dbu)
|
|
dbid = db.id
|
|
self.dbu.delete()
|
|
self.assertFalse(UserDatabase.objects.filter(id=dbid).exists())
|
|
taskres = TaskResult.objects.all()
|
|
self.assertEqual(len(taskres), 3)
|
|
self.assertEqual(taskres[0].creator, 'handle_userdb_created')
|
|
self.assertEqual(taskres[0].notes, 'pgsql database creation')
|
|
self.assertEqual(taskres[1].creator, 'handle_userdb_deleted')
|
|
self.assertEqual(taskres[1].notes, 'pgsql database deletion')
|
|
self.assertEqual(taskres[2].creator, 'handle_dbuser_deleted')
|
|
self.assertEqual(taskres[2].notes, 'pgsql user deletion')
|
|
|
|
|
|
class UserDatabaseManagerTest(TestCaseWithCeleryTasks):
|
|
"""
|
|
Test case for :py:class:`userdbs.models.UserDatabaseManager`.
|
|
|
|
"""
|
|
def setUp(self):
|
|
self.customer = Customer.objects.create_user(username='testcustomer')
|
|
self.osuser = User.objects.create_user(customer=self.customer)
|
|
TaskResult.objects.all().delete()
|
|
|
|
def _create_database_user(self, dbtype):
|
|
self.dbu = DatabaseUser.objects.create_database_user(
|
|
self.osuser, dbtype)
|
|
TaskResult.objects.all().delete()
|
|
|
|
def test_create_userdatabase_with_user_mysql(self):
|
|
db = UserDatabase.objects.create_userdatabase_with_user(
|
|
DB_TYPES.mysql, self.osuser)
|
|
self.assertEqual(db.db_name, '{user}db01'.format(
|
|
user=self.osuser.username))
|
|
self.assertEqual(db.db_user.name, '{user}db01'.format(
|
|
user=self.osuser.username))
|
|
taskres = TaskResult.objects.all()
|
|
self.assertEqual(len(taskres), 2)
|
|
self.assertEqual(taskres[0].creator, 'handle_dbuser_created')
|
|
self.assertEqual(taskres[0].notes, 'mysql user creation')
|
|
self.assertEqual(taskres[1].creator, 'handle_userdb_created')
|
|
self.assertEqual(taskres[1].notes, 'mysql database creation')
|
|
|
|
def test_create_userdatabase_with_user_pgsql(self):
|
|
db = UserDatabase.objects.create_userdatabase_with_user(
|
|
DB_TYPES.pgsql, self.osuser)
|
|
self.assertEqual(db.db_name, '{user}db01'.format(
|
|
user=self.osuser.username))
|
|
self.assertEqual(db.db_user.name, '{user}db01'.format(
|
|
user=self.osuser.username))
|
|
taskres = TaskResult.objects.all()
|
|
self.assertEqual(len(taskres), 2)
|
|
self.assertEqual(taskres[0].creator, 'handle_dbuser_created')
|
|
self.assertEqual(taskres[0].notes, 'pgsql user creation')
|
|
self.assertEqual(taskres[1].creator, 'handle_userdb_created')
|
|
self.assertEqual(taskres[1].notes, 'pgsql database creation')
|
|
|
|
def test_create_userdatabase_given_name_no_new_write(self):
|
|
self._create_database_user(DB_TYPES.pgsql)
|
|
db = UserDatabase.objects.create_userdatabase(self.dbu, db_name='test')
|
|
taskres = TaskResult.objects.all()
|
|
self.assertEqual(len(taskres), 1)
|
|
self.assertEqual(db.db_name, 'test')
|
|
TaskResult.objects.all().delete()
|
|
db.save()
|
|
taskres = TaskResult.objects.all()
|
|
self.assertEqual(len(taskres), 0)
|
|
|
|
def test_create_userdatabase_given_name(self):
|
|
self._create_database_user(DB_TYPES.pgsql)
|
|
db = UserDatabase.objects.create_userdatabase(self.dbu, db_name='test')
|
|
self.assertEqual(db.db_name, 'test')
|
|
taskres = TaskResult.objects.all()
|
|
self.assertEqual(len(taskres), 1)
|
|
self.assertEqual(db.db_name, 'test')
|
|
|
|
def test_create_userdatabase_generate_name_no_commit(self):
|
|
self._create_database_user(DB_TYPES.pgsql)
|
|
db = UserDatabase.objects.create_userdatabase(self.dbu, commit=False)
|
|
self.assertEqual(db.db_name, self.dbu.name)
|
|
self.assertFalse(TaskResult.objects.all().exists())
|
|
|
|
def test_create_userdatabase_generate_name(self):
|
|
self._create_database_user(DB_TYPES.pgsql)
|
|
db = UserDatabase.objects.create_userdatabase(self.dbu)
|
|
self.assertEqual(db.db_name, self.dbu.name)
|
|
|
|
def test_create_userdatabase_multiple_generate_name(self):
|
|
self._create_database_user(DB_TYPES.pgsql)
|
|
db = UserDatabase.objects.create_userdatabase(self.dbu)
|
|
self.assertEqual(db.db_name, self.dbu.name)
|
|
db = UserDatabase.objects.create_userdatabase(self.dbu)
|
|
self.assertEqual(db.db_name, '{user}_02'.format(user=self.dbu.name))
|
|
|
|
def test_create_userdatabase_multiple_gap_generate_name(self):
|
|
self._create_database_user(DB_TYPES.pgsql)
|
|
db = UserDatabase.objects.create_userdatabase(self.dbu)
|
|
self.assertEqual(db.db_name, self.dbu.name)
|
|
dbx = UserDatabase.objects.create_userdatabase(self.dbu)
|
|
self.assertEqual(dbx.db_name, '{user}_02'.format(user=self.dbu.name))
|
|
db = UserDatabase.objects.create_userdatabase(self.dbu)
|
|
self.assertEqual(db.db_name, '{user}_03'.format(user=self.dbu.name))
|
|
dbx.delete()
|
|
db = UserDatabase.objects.create_userdatabase(self.dbu)
|
|
self.assertEqual(db.db_name, '{user}_02'.format(user=self.dbu.name))
|
|
|
|
|
|
class UserDatabaseTest(TestCaseWithCeleryTasks):
|
|
"""
|
|
Test case for :py:class:`userdbs.models.UserDabase`.
|
|
|
|
"""
|
|
def test___str__(self):
|
|
customer = Customer.objects.create_user(username='testcustomer')
|
|
osuser = User.objects.create_user(customer=customer)
|
|
db = UserDatabase.objects.create_userdatabase_with_user(
|
|
DB_TYPES.pgsql, osuser)
|
|
self.assertEqual(
|
|
str(db),
|
|
'{user}db01 ({dbuser})'.format(
|
|
user=osuser.username, dbuser=db.db_user))
|
|
|
|
def test_delete_mysql_db(self):
|
|
customer = Customer.objects.create_user(username='testcustomer')
|
|
osuser = User.objects.create_user(customer=customer)
|
|
TaskResult.objects.all().delete()
|
|
db = UserDatabase.objects.create_userdatabase_with_user(
|
|
DB_TYPES.mysql, osuser)
|
|
db.delete()
|
|
taskres = TaskResult.objects.all()
|
|
self.assertEqual(len(taskres), 3)
|
|
self.assertEqual(taskres[2].creator, 'handle_userdb_deleted')
|
|
self.assertEqual(taskres[2].notes, 'mysql database deletion')
|