gva/gnuviechadmin/userdbs/tests/test_models.py

"""
This module provides tests for :py:mod:`userdbs.models`.

"""
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.order_by("created").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.order_by("created").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.order_by("created").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.order_by("created").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")
        os_user = User.objects.create_user(customer=customer)
        db = UserDatabase.objects.create_userdatabase_with_user(DB_TYPES.pgsql, os_user)
        self.assertEqual(
            str(db),
            "{user}db01 ({dbuser})".format(user=os_user.username, dbuser=db.db_user),
        )

    def test_delete_mysql_db(self):
        customer = Customer.objects.create_user(username="testcustomer")
        os_user = User.objects.create_user(customer=customer)
        TaskResult.objects.all().delete()
        db = UserDatabase.objects.create_userdatabase_with_user(DB_TYPES.mysql, os_user)
        db.delete()
        task_result = TaskResult.objects.order_by("created").all()
        self.assertEqual(len(task_result), 3)
        self.assertEqual(task_result[2].creator, "handle_userdb_deleted")
        self.assertEqual(task_result[2].notes, "mysql database deletion")