wuttamess/tests/test_postgres.py

# -*- coding: utf-8; -*-

from unittest import TestCase
from unittest.mock import MagicMock, patch

from wuttamess import postgres as mod


class TestSql(TestCase):

    def test_basic(self):
        c = MagicMock()
        mod.sql(c, "select @@version")
        c.sudo.assert_called_once_with('psql --tuples-only --no-align --command="select @@version" ',
                                       user='postgres')


class TestUserExists(TestCase):

    def test_user_exists(self):
        c = MagicMock()
        with patch.object(mod, 'sql') as sql:
            sql.return_value.stdout = 'foo'
            self.assertTrue(mod.user_exists(c, 'foo'))
            sql.assert_called_once_with(c, "SELECT rolname FROM pg_roles WHERE rolname = 'foo'", port=None)

    def test_user_does_not_exist(self):
        c = MagicMock()
        with patch.object(mod, 'sql') as sql:
            sql.return_value.stdout = ''
            self.assertFalse(mod.user_exists(c, 'foo'))
            sql.assert_called_once_with(c, "SELECT rolname FROM pg_roles WHERE rolname = 'foo'", port=None)


class TestCreateUser(TestCase):

    def test_basic(self):
        c = MagicMock()
        with patch.object(mod, 'set_user_password') as set_user_password:
            mod.create_user(c, 'foo', checkfirst=False)
            c.sudo.assert_called_once_with('createuser --no-createrole --no-superuser foo',
                                           user='postgres')
            set_user_password.assert_not_called()

    def test_user_exists(self):
        c = MagicMock()

        with patch.object(mod, 'user_exists') as user_exists:
            user_exists.return_value = True

            mod.create_user(c, 'foo')
            user_exists.assert_called_once_with(c, 'foo', port=None)
            c.sudo.assert_not_called()

    def test_with_password(self):
        c = MagicMock()
        with patch.object(mod, 'set_user_password') as set_user_password:
            mod.create_user(c, 'foo', 'foopass', checkfirst=False)
            c.sudo.assert_called_once_with('createuser --no-createrole --no-superuser foo',
                                           user='postgres')
            set_user_password.assert_called_once_with(c, 'foo', 'foopass', port=None)


class TestSetUserPassword(TestCase):

    def test_basic(self):
        c = MagicMock()
        with patch.object(mod, 'sql') as sql:
            mod.set_user_password(c, 'foo', 'foopass')
            sql.assert_called_once_with(c, "ALTER USER \\\"foo\\\" PASSWORD 'foopass';",
                                        port=None, hide=True, echo=False)


class TestDbExists(TestCase):

    def test_db_exists(self):
        c = MagicMock()
        with patch.object(mod, 'sql') as sql:
            sql.return_value.stdout = 'foo'
            self.assertTrue(mod.db_exists(c, 'foo'))
            sql.assert_called_once_with(c, "SELECT datname FROM pg_database WHERE datname = 'foo'", port=None)

    def test_db_does_not_exist(self):
        c = MagicMock()
        with patch.object(mod, 'sql') as sql:
            sql.return_value.stdout = ''
            self.assertFalse(mod.db_exists(c, 'foo'))
            sql.assert_called_once_with(c, "SELECT datname FROM pg_database WHERE datname = 'foo'", port=None)


class TestCreateDb(TestCase):

    def test_basic(self):
        c = MagicMock()
        mod.create_db(c, 'foo', checkfirst=False)
        c.sudo.assert_called_once_with('createdb foo', user='postgres')

    def test_db_exists(self):
        c = MagicMock()

        with patch.object(mod, 'db_exists') as db_exists:
            db_exists.return_value = True

            mod.create_db(c, 'foo')
            db_exists.assert_called_once_with(c, 'foo', port=None)
            c.sudo.assert_not_called()


class TestDropDb(TestCase):

    def test_basic(self):
        c = MagicMock()
        mod.drop_db(c, 'foo', checkfirst=False)
        c.sudo.assert_called_once_with('dropdb foo', user='postgres')

    def test_db_does_not_exist(self):
        c = MagicMock()

        with patch.object(mod, 'db_exists') as db_exists:
            db_exists.return_value = False

            mod.drop_db(c, 'foo')
            db_exists.assert_called_once_with(c, 'foo')
            c.sudo.assert_not_called()