feat: add basic postgres module for db setup

tests/test_postgres.py

@@ -0,0 +1,124 @@
+# -*- 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()