153 lines
4.6 KiB
Python
153 lines
4.6 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()
|
|
|
|
|
|
class TestDumpDb(TestCase):
|
|
|
|
def test_basic(self):
|
|
c = MagicMock()
|
|
result = mod.dump_db(c, "foo")
|
|
self.assertEqual(result, "foo.sql.gz")
|
|
c.sudo.assert_called_once_with(
|
|
"pg_dump foo | gzip -c > /tmp/foo.sql.gz", user="postgres"
|
|
)
|
|
c.run.assert_called_with("rm /tmp/foo.sql.gz")
|