125 lines
4.1 KiB
Python
125 lines
4.1 KiB
Python
# -*- 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()
|