Merge pull request #130 from matyasselmeci/wip/py3

Fix obvious Python 3 incompatibilities

Author: brianhlin

osg-test

@@ -1,5 +1,6 @@
 #!/usr/bin/python
 
+from __future__ import print_function
 from distutils.sysconfig import get_python_lib
 from optparse import OptionParser
 import ConfigParser
@@ -55,7 +56,7 @@ def parse_command_line():
         config_file = ConfigParser.RawConfigParser()
         try:
             config_file.read(opts.config)
-        except ConfigParser.ParsingError, exc:
+        except ConfigParser.ParsingError as exc:
             sys.stderr.write("Error while parsing: %s\n%s\n" % (opts.config, exc))
             sys.stderr.write("Lines with options should not start with a space\n")
             sys.exit(1)
@@ -168,7 +169,7 @@ def get_password():
     if core.options.securepass:
         # If the user also specifies password on the command line, we complain
         if core.options.password != 'vdttest':
-            print 'Only specify one of --password and --securepass'
+            print('Only specify one of --password and --securepass')
             sys.exit(1)
         core.options.password = getpass()
 
@@ -252,6 +253,6 @@ if __name__ == '__main__':
         signal.alarm(0)
         signal.signal(signal.SIGALRM, signal.SIG_DFL)
     else:
-        print 'No tests to run.'
+        print('No tests to run.')
     wrap_up()
     sys.exit(EXIT_CODE)

osgtest/library/core.py

@@ -1,4 +1,5 @@
 """Support and convenience functions for tests."""
+from __future__ import print_function
 
 import errno
 import os
@@ -127,9 +128,9 @@ def end_log():
 def dump_log(outfile=None):
     if outfile is None:
         logfile = open(_log_filename, 'r')
-        print '\n'
+        print('\n')
         for line in logfile:
-            print line.rstrip('\n')
+            print(line.rstrip('\n'))
         logfile.close()
     else:
         shutil.copy(_log_filename, outfile)
@@ -143,7 +144,7 @@ def get_stat(filename):
     '''Return stat for 'filename', None if the file does not exist'''
     try:
         return os.stat(filename)
-    except OSError, exc:
+    except OSError as exc:
         if exc.errno == errno.ENOENT:
             return None
         raise
@@ -493,7 +494,7 @@ def __run_command(command, use_test_user, a_input, a_stdout, a_stderr, log_outpu
         try:
             repr(command)
         except TypeError:
-            print 'Need list or tuple, got %s' % type(command)
+            print('Need list or tuple, got %s' % type(command))
     if use_test_user:
         command = ['runuser', options.username, '-c', ' '.join(map(__prepare_shell_argument, command))]
 
@@ -603,7 +604,7 @@ def el_release():
                 release_file.close()
             match = re.search(r"release (\d)", release_text)
             _el_release = int(match.group(1))
-        except (EnvironmentError, TypeError, ValueError), e:
+        except (EnvironmentError, TypeError, ValueError) as e:
             _log.write("Couldn't determine redhat release: " + str(e) + "\n")
             sys.exit(1)
     return _el_release
@@ -642,7 +643,7 @@ def check_file_and_perms(file_path, owner_name, permissions):
         try:
             file_stat = os.stat(file_path)
             return (file_stat.st_uid == owner_uid and
-                    file_stat.st_mode & 07777 == permissions and
+                    file_stat.st_mode & 0o7777 == permissions and
                     stat.S_ISREG(file_stat.st_mode))
         except OSError:  # file does not exist
             return False
@@ -684,7 +685,7 @@ def install_cert(target_key, source_key, owner_name, permissions):
         os.makedirs(target_dir)
         state[target_key + '-dir'] = target_dir
         os.chown(target_dir, user.pw_uid, user.pw_gid)
-        os.chmod(target_dir, 0755)
+        os.chmod(target_dir, 0o755)
 
     shutil.copy(source_path, target_path)
     state[target_key] = target_path
@@ -697,12 +698,12 @@ def remove_cert(target_key):
     paths associated with the key, as created by the install_cert()
     function.
     """
-    if state.has_key(target_key):
+    if target_key in state:
         os.remove(state[target_key])
-    if state.has_key(target_key + '-backup'):
+    if target_key + '-backup' in state:
         shutil.move(state[target_key + '-backup'],
                     state[target_key])
-    if state.has_key(target_key + '-dir'):
+    if target_key + '-dir' in state:
         target_dir = state[target_key + '-dir']
         if len(os.listdir(target_dir)) == 0:
             os.rmdir(target_dir)

osgtest/library/files.py

@@ -74,7 +74,7 @@ def preserve(path, owner):
     else:
         _backups[backup_id] = None
 
-def write(path, contents, owner=None, backup=True, chown=(0,0), chmod=0600):
+def write(path, contents, owner=None, backup=True, chown=(0,0), chmod=0o600):
     """Write the contents to a file at the path.
 
     The 'owner' argument (default: None) is a string that identifies the owner

osgtest/library/osgunittest.py

@@ -75,41 +75,41 @@ def __init__(self, methodName='runTest'):
 
     def skip_ok(self, message=None):
         "Skip (ok) unconditionally"
-        raise OkSkipException, message
+        raise OkSkipException(message)
 
     def skip_ok_if(self, expr, message=None):
         "Skip (ok) if the expression is true"
         if expr:
-            raise OkSkipException, message
+            raise OkSkipException(message)
 
     def skip_ok_unless(self, expr, message=None):
         "Skip (ok) if the expression is false"
         if not expr:
-            raise OkSkipException, message
+            raise OkSkipException(message)
 
     def skip_bad(self, message=None):
         "Skip (bad) unconditionally"
-        raise BadSkipException, message
+        raise BadSkipException(message)
 
     def skip_bad_if(self, expr, message=None):
         "Skip (bad) if the expression is true"
         if expr:
-            raise BadSkipException, message
+            raise BadSkipException(message)
 
     def skip_bad_unless(self, expr, message=None):
         "Skip (bad) if the expression is false"
         if not expr:
-            raise BadSkipException, message
+            raise BadSkipException(message)
 
     def assertSubsetOf(self, a, b, message=None):
         "Ensure that a is a subset of b "
         if not set(a).issubset(set(b)):
-            raise AssertionError, message
+            raise AssertionError(message)
 
     def failIfSubsetOf(self, a, b, message=None):
         "Ensure that a is not a subset of b"
         if set(a).issubset(set(b)):
-            raise AssertionError, message
+            raise AssertionError(message)
 
     # This is mostly a copy of the method from unittest in python 2.4.
     # There is some code here to test if the 'result' object accepts 'skips',

osgtest/library/voms.py

@@ -74,7 +74,7 @@ def advertise_lsc(vo, hostcert='/etc/grid-security/hostcert.pem'):
     if not os.path.isdir(lsc_dir):
         os.makedirs(lsc_dir)
     vo_lsc_path = os.path.join(lsc_dir, hostname + '.lsc')
-    files.write(vo_lsc_path, (host_dn + '\n', host_issuer + '\n'), backup=False, chmod=0644)
+    files.write(vo_lsc_path, (host_dn + '\n', host_issuer + '\n'), backup=False, chmod=0o644)
 
 
 def advertise_vomses(vo, hostcert='/etc/grid-security/hostcert.pem'):
@@ -86,7 +86,7 @@ def advertise_vomses(vo, hostcert='/etc/grid-security/hostcert.pem'):
     vomses_path = '/etc/vomses'
     contents = ('"%s" "%s" "%d" "%s" "%s"\n' %
                 (vo, hostname, 15151, host_dn, vo))
-    files.write(vomses_path, contents, backup=False, chmod=0644)
+    files.write(vomses_path, contents, backup=False, chmod=0o644)
 
 
 def add_user(vo, usercert, use_voms_admin=False):

osgtest/tests/special_cleanup.py

@@ -134,7 +134,7 @@ def test_06_cleanup_test_certs(self):
                     dest = os.readlink(abs_link_path)
                     if re.match(r'OSG-Test-CA\.', dest):
                         files.remove(abs_link_path)
-                except OSError, e:
+                except OSError as e:
                     if e.errno == errno.EINVAL:
                         continue
 

osgtest/tests/special_user.py

@@ -63,7 +63,7 @@ def test_01_add_user(self):
         # Set up directories
         user = pwd.getpwnam(core.options.username)
         os.chown(user.pw_dir, user.pw_uid, user.pw_gid)
-        os.chmod(user.pw_dir, 0755)
+        os.chmod(user.pw_dir, 0o755)
 
         # Set up certificate
         globus_dir = os.path.join(user.pw_dir, '.globus')
@@ -80,7 +80,7 @@ def test_02_user(self):
             return
         try:
             password_entry = pwd.getpwnam(core.options.username)
-        except KeyError, e:
+        except KeyError as e:
             self.fail("User '%s' should exist but does not" % core.options.username)
         self.assert_(password_entry.pw_dir != '/', "User '%s' has home directory at '/'" % (core.options.username))
         self.assert_(os.path.isdir(password_entry.pw_dir),
@@ -93,4 +93,4 @@ def test_02_user(self):
                      (core.config['user.cert_subject'], password_entry.pw_name),
                      owner='user')
         core.state['system.wrote_mapfile'] = True
-        os.chmod(core.config['system.mapfile'], 0644)
+        os.chmod(core.config['system.mapfile'], 0o644)

osgtest/tests/test_05_mysqld.py

@@ -28,7 +28,7 @@ def test_01_backup_mysql(self):
         self.assert_(not os.path.exists(backup), 'mysql-backup already exists')
         try:
             shutil.move(core.config['mysql.datadir'], backup)
-        except IOError, e:
+        except IOError as e:
             if e.errno == 2:
                 # Folder doesn't exist so we don't have to worry about backups
                 pass

osgtest/tests/test_09_jobs.py

@@ -20,18 +20,18 @@ def test_01_set_job_env(self):
         osg_libdir = os.path.join('/var', 'lib', 'osg')
         try:
             os.makedirs(osg_libdir)
-        except OSError, exc:
+        except OSError as exc:
             if exc.errno != errno.EEXIST:
                 raise
         core.config['osg.job-environment'] = os.path.join(osg_libdir, 'osg-job-environment.conf')
         core.config['osg.local-job-environment'] = os.path.join(osg_libdir, 'osg-local-job-environment.conf')
 
         files.write(core.config['osg.job-environment'],
                     "#!/bin/sh\nJOB_ENV='vdt'\nexport JOB_ENV",
-                    owner='pbs', chmod=0644)
+                    owner='pbs', chmod=0o644)
         files.write(core.config['osg.local-job-environment'],
                     "#!/bin/sh\nLOCAL_JOB_ENV='osg'\nexport LOCAL_JOB_ENV",
-                    owner='pbs', chmod=0644)
+                    owner='pbs', chmod=0o644)
 
         core.state['jobs.env-set'] = True
 

osgtest/tests/test_15_xrootd.py

@@ -42,8 +42,8 @@ def test_01_start_xrootd(self):
         user = pwd.getpwnam("xrootd")
         if core.config['xrootd.gsi'] == "ON":
             core.skip_ok_unless_installed('globus-proxy-utils')
-            core.install_cert('certs.xrootdcert', 'certs.hostcert', 'xrootd', 0644)
-            core.install_cert('certs.xrootdkey', 'certs.hostkey', 'xrootd', 0400)
+            core.install_cert('certs.xrootdcert', 'certs.hostcert', 'xrootd', 0o644)
+            core.install_cert('certs.xrootdkey', 'certs.hostkey', 'xrootd', 0o400)
 
             lcmaps_packages = ('lcmaps', 'lcmaps-db-templates', 'xrootd-lcmaps', 'vo-client', 'vo-client-lcmaps-voms')
             if all([core.rpm_is_installed(x) for x in lcmaps_packages]):