Commits

Mikhail Korobov committed 4deefbd

execute_sql selects DB_NAME by default for non-superusers; Postgres.execute_sql is fixed for non-superuser (thanks Andrey Rahmatullin for help); dedicated tests for postgres; faster mysql tests.

  • Participants
  • Parent commits 43adc9d

Comments (0)

Files changed (7)

fab_deploy/db/base.py

         raise NotImplementedError()
 
     @task
-    def execute_sql(self, sql, user=None, password=None):
+    def execute_sql(self, sql, user=None, password=None, db_name=None):
         """ Executes passed sql command. """
         raise NotImplementedError()
 

fab_deploy/db/mysql.py

         return user, password
 
     @task
-    def execute_sql(self, sql, user=None, password=None):
+    def execute_sql(self, sql, user=None, password=None, db_name=None):
         """ Executes passed sql command using mysql shell. """
         user, password = self._credentials_for_sql(user, password)
         sql = sql.replace('"', r'\"')
-        return run('echo "%s" | mysql --user="%s" --password="%s"' % (sql, user , password))
+
+        if user != self.SUPERUSER:
+            db_name = db_name or env.conf.DB_NAME
+        else:
+            db_name = '' # do not select database for superuser
+
+        return run('echo "%s" | mysql --user="%s" --password="%s" %s' % (sql, user , password, db_name))
 
 instance = Mysql()
 __all__ = instance.expose_to_current_module()

fab_deploy/db/postgres.py

         system.aptitude_install(packages)
 
     @task
-    def execute_sql(self, sql, user=None, password=None):
+    def execute_sql(self, sql, user=None, password=None, db_name=None):
         db_user = user or env.conf.DB_USER
         #db_host = env.conf.DB_HOST
         if password is None and db_user != self.SUPERUSER:
             password = env.conf.DB_PASSWORD
 
-        pwd_cmd = 'set PGPASSWORD="%s";' % password if password is not None else ''
         sql = sql.replace('"', r'\"')
-        cmd = 'psql --user="%s" --command "%s"' % (db_user, sql,)
 
         # TODO: superuser password auth?
-        if user == self.SUPERUSER:
+        if db_user == self.SUPERUSER:
+            cmd = 'psql --command "%s"' % sql
             @utils.run_as_sudo
             def sudo_cmd():
                 return sudo(cmd, user=db_user)
             return sudo_cmd()
 
-        return run(pwd_cmd + cmd, pty=False, combine_stderr=False)
+        pwd_cmd = "export PGPASSWORD='%s';" % password if password is not None else ''
+
+        # TODO: configurable DB_HOST
+        # we have to set host because password auth is off for sockets by default
+        db_name = db_name or env.conf.DB_NAME
+        cmd = 'psql %s --user="%s" --host=127.0.0.1 --command "%s"' % (db_name, db_user, sql,)
+        return run(pwd_cmd + cmd, pty=False, combine_stderr=False, shell=False)
 
     @task
     def _user_exists(self, db_user):

fab_deploy_tests/runtests.py

     atexit.register(beep)
     FabDeployTest.vm_name = sys.argv[1]
 
-    common = [SshTest, MysqlTest, MysqlNonRootTest, BasicTest]
+    common = [SshTest, MysqlTest, MysqlNonRootTest, BasicTest,
+              PostgresInstallTest, PostgresSetupTest]
 
     suites = {
         'misc': TestSuite(load(common + [

fab_deploy_tests/tests/__init__.py

 
 from .base import *
 from .mysql import *
+from .postgres import *
 from .system_tests import *
 from .utils_tests import *
 from .crontab import *

fab_deploy_tests/tests/mysql.py

         env.conf['DB_PASSWORD'] = '123'
         env.conf['DB_NAME'] = 'new_database'
 
-    def test_mysql(self):
+    def test_install(self):
         setup_sudo()
 
         self.assertFalse(mysql_is_installed())
 
 class MysqlNonRootTest(FabDeployTest):
     host = 'root@127.0.0.1:2222'
+    snapshot = 'fabtest-prepared-server'
 
     def setup_conf(self):
         super(MysqlNonRootTest, self).setup_conf()
         env.conf['DB_USER'] = 'foo'
 
     def test_mysql(self):
-        setup_sudo()
-
-        self.assertFalse(mysql_is_installed())
-
-        fab(mysql.install)
         self.assertTrue(mysql_is_installed())
 
         self.assertFalse(database_exists('new_database'))
         fab(mysql.create_user)
 
         # this will fail if permissions are not set correctly
-        output = fab(mysql.execute_sql, 'use new_database; create table baz (id int); show tables;')
+        output = fab(mysql.execute_sql, 'create table baz (id int); show tables;')
         tables = output.splitlines()[1:]
         self.assertEqual(tables[0], 'baz')
 

fab_deploy_tests/tests/postgres.py

+# -*- coding: utf-8 -*-
+from __future__ import absolute_import
+from fabric.api import env
+from fabtest import fab
+from fab_deploy.db import postgres
+from ..utils import setup_sudo
+from .base import FabDeployTest
+
+def postgres_is_installed():
+    return fab(postgres.is_installed)
+
+def database_exists(db_name):
+    user_name = postgres.Postgres.SUPERUSER
+    databases = fab(postgres.execute_sql, 'select datname from pg_database;', user_name)
+    return db_name in databases
+
+class PostgresInstallTest(FabDeployTest):
+    host = 'root@127.0.0.1:2222'
+
+    def setup_conf(self):
+        super(PostgresInstallTest, self).setup_conf()
+        env.conf['DB_PASSWORD'] = '123'
+        env.conf['DB_NAME'] = 'new_database'
+        env.conf['DB_USER'] = 'vasia'
+
+    def test_install(self):
+        setup_sudo()
+        self.assertFalse(postgres_is_installed())
+        fab(postgres.install)
+        self.assertTrue(postgres_is_installed())
+
+
+class PostgresSetupTest(FabDeployTest):
+    host = 'root@127.0.0.1:2222'
+    snapshot = 'fabtest-prepared-server'
+
+    def setup_conf(self):
+        super(PostgresSetupTest, self).setup_conf()
+        env.conf['DB_PASSWORD'] = '123'
+        env.conf['DB_NAME'] = 'new_database'
+        env.conf['DB_USER'] = 'vasia'
+
+    def test_postgres(self):
+        self.assertTrue(postgres_is_installed())
+
+        self.assertFalse(database_exists('new_database'))
+        fab(postgres.create_user)
+        fab(postgres.create_db)
+        self.assertTrue(database_exists('new_database'))
+
+        # create table
+        fab(postgres.execute_sql, "create table baz (id int);")
+
+        # check if it is created
+        test_sql = "SELECT table_name FROM information_schema.tables WHERE table_schema = 'public';"
+        output = fab(postgres.execute_sql, test_sql)
+        tables = output.splitlines()[1:]
+        self.assertEqual(tables[1].strip(), 'baz')