Merge lp:~longbow/kewpie/py_pam into lp:kewpie

[Valentine Gostev]

Added basic positive pam test

=== modified file 'lib/sys_mgmt/system_management.py'
--- lib/sys_mgmt/system_management.py	2012-01-26 01:54:17 +0000
+++ lib/sys_mgmt/system_management.py	2012-02-22 16:36:09 +0000
@@ -33,6 +33,7 @@
 # imports
 import os
 import sys
+import pwd
 import shutil
 import getpass
 import commands
@@ -526,6 +527,13 @@
         except OSError:
             self.logging.verbose("PID: %s was not running despite the existence of a pid file.  This may be of note...")
         return
+
+    def user_exists(self, username):
+        try:
+          pwd.getpwnam(username)[0]
+        except KeyError:
+          return False
+        return True
    
     def get_ip_address(self):
         """ We find the ip address of our host.

=== added directory 'percona_tests/percona_pam'
=== added file 'percona_tests/percona_pam/pam_basic_test.py'
--- percona_tests/percona_pam/pam_basic_test.py	1970-01-01 00:00:00 +0000
+++ percona_tests/percona_pam/pam_basic_test.py	2012-02-22 16:36:09 +0000
@@ -0,0 +1,110 @@
+#! /usr/bin/env python
+# -*- mode: python; indent-tabs-mode: nil; -*-
+# vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
+#
+# Copyright (C) 2012 Valentine Gostev
+#
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
+
+import os
+import time
+import shutil
+import signal
+import subprocess
+
+from lib.util.mysqlBaseTestCase import mysqlBaseTestCase
+from lib.util.mysql_methods import execute_cmd
+
+
+server_requirements = [[]]
+servers = []
+server_manager = None
+test_executor = None
+pamcfg = '/etc/pam.d/mysqld'
+
+class basicTest(mysqlBaseTestCase):
+
+    def test_pam_basic(self):
+        opt_matrix_req = ['pam_plugin_dir']
+        self.servers = servers
+        logging = test_executor.logging
+        master_server = servers[0]
+        output_path = os.path.join(master_server.vardir, 'pam.out')
+        # Checking that required options were given
+        test_executor.matrix_manager.matrix_check_req(opt_matrix_req)
+
+        # Create UNIX system account
+        if (test_executor.matrix_manager.option_matrix['pam_user']):
+            pam_user = test_executor.matrix_manager.option_matrix['pam_user']
+        else:
+            pam_user = 'pamuser'
+
+        if not (test_executor.system_manager.user_exists(pam_user)):
+            subprocess.call(["useradd", pam_user])
+
+        # Create PAM config
+        if (os.path.isfile(pamcfg)):
+            os.remove(pamcfg)
+
+        pamcfg_fh = open("/etc/pam.d/mysqld", "wb")
+        pamcfg_fh.write("auth\trequired\tpam_permit.so\n")
+        pamcfg_fh.close();
+
+        # Stop server
+        master_server.stop()
+
+        # Specify mysql plugin dir
+        master_server.server_options.append('--plugin-dir=%s' %(test_executor.matrix_manager.option_matrix['pam_plugin_dir']))
+	# Start server with new options
+        master_server.start()
+        self.assertEqual( master_server.status, 1, msg = 'Server failed to restart')
+	# Install plugin
+        query = "INSTALL PLUGIN auth_pam SONAME \'auth_pam.so\'"
+        expected_result = ''
+        cmd = "%s --protocol=tcp --port=%d -uroot -e \"%s\"" %(master_server.mysql_client
+              , master_server.master_port
+              , query )
+        retcode, output = execute_cmd(cmd, output_path, None, True)
+        self.assertEqual(retcode, 0, msg = cmd)
+        self.assertEqual(output, expected_result, msg = "%s || %s" %(output, expected_result))
+	# Create user
+        query = "CREATE USER \'%s\'@\'%\' IDENTIFIED WITH auth_pam;" %(pam_user)
+        expected_result = ''
+        cmd = "%s --protocol=tcp --port=%d -uroot -e \"%s\"" %(master_server.mysql_client
+              , master_server.master_port
+              , query )
+        retcode, output = execute_cmd(cmd, output_path, None, True)
+        self.assertEqual(retcode, 0, msg = output)
+        self.assertEqual(output, expected_result, msg = "%s || %s" %(output, expected_result))
+	# Grant permissions
+        query = "GRANT ALL ON test.* TO \'%s\'@\'%\';" %(pam_user)
+        expected_result = ''
+        cmd = "%s --protocol=tcp --port=%d --user=root -e \"%s\"" %(master_server.mysql_client
+              , master_server.master_port
+              , query )
+        retcode, output = execute_cmd(cmd, output_path, None, True)
+        self.assertEqual(retcode, 0, msg = output)
+        self.assertEqual(output, expected_result, msg = "%s || %s" %(output, expected_result))
+	# Test user login
+        query = "SHOW TABLES;"
+        expected_result = ''
+        cmd = "%s --plugin-dir=/usr/lib/mysql/plugin/ --protocol=tcp --port=%d --user=%s --password=\'\' -e \"%s\" test" %(master_server.mysql_client
+              , master_server.master_port
+              , pam_user
+              , query )
+        retcode, output = execute_cmd(cmd, output_path, None, True)
+        self.assertEqual(retcode, 0, msg = output)
+        self.assertEqual(output, expected_result, msg = "%s || %s" %(output, expected_result))