[Jython-checkins] jython: from:

frank.wierzbicki jython-checkins at python.org
Thu Jun 28 19:37:41 CEST 2012


http://hg.python.org/jython/rev/a28bdeb695df
changeset:   6767:a28bdeb695df
user:        Frank Wierzbicki <fwierzbicki at gmail.com>
date:        Wed Jun 27 19:47:23 2012 -0700
summary:
  from:
http://hg.python.org/cpython/Lib/test/test_posix.py@22db03646d9b

files:
  Lib/test/test_posix.py |  467 +++++++++++++++++++++++++++++
  1 files changed, 467 insertions(+), 0 deletions(-)


diff --git a/Lib/test/test_posix.py b/Lib/test/test_posix.py
new file mode 100644
--- /dev/null
+++ b/Lib/test/test_posix.py
@@ -0,0 +1,467 @@
+"Test posix functions"
+
+from test import test_support
+
+# Skip these tests if there is no posix module.
+posix = test_support.import_module('posix')
+
+import errno
+import sys
+import time
+import os
+import pwd
+import shutil
+import stat
+import sys
+import tempfile
+import unittest
+import warnings
+
+_DUMMY_SYMLINK = os.path.join(tempfile.gettempdir(),
+                              test_support.TESTFN + '-dummy-symlink')
+
+warnings.filterwarnings('ignore', '.* potential security risk .*',
+                        RuntimeWarning)
+
+class PosixTester(unittest.TestCase):
+
+    def setUp(self):
+        # create empty file
+        fp = open(test_support.TESTFN, 'w+')
+        fp.close()
+        self.teardown_files = [ test_support.TESTFN ]
+
+    def tearDown(self):
+        for teardown_file in self.teardown_files:
+            os.unlink(teardown_file)
+
+    def testNoArgFunctions(self):
+        # test posix functions which take no arguments and have
+        # no side-effects which we need to cleanup (e.g., fork, wait, abort)
+        NO_ARG_FUNCTIONS = [ "ctermid", "getcwd", "getcwdu", "uname",
+                             "times", "getloadavg", "tmpnam",
+                             "getegid", "geteuid", "getgid", "getgroups",
+                             "getpid", "getpgrp", "getppid", "getuid",
+                           ]
+
+        with warnings.catch_warnings():
+            warnings.filterwarnings("ignore", "", DeprecationWarning)
+            for name in NO_ARG_FUNCTIONS:
+                posix_func = getattr(posix, name, None)
+                if posix_func is not None:
+                    posix_func()
+                    self.assertRaises(TypeError, posix_func, 1)
+
+    if hasattr(posix, 'getresuid'):
+        def test_getresuid(self):
+            user_ids = posix.getresuid()
+            self.assertEqual(len(user_ids), 3)
+            for val in user_ids:
+                self.assertGreaterEqual(val, 0)
+
+    if hasattr(posix, 'getresgid'):
+        def test_getresgid(self):
+            group_ids = posix.getresgid()
+            self.assertEqual(len(group_ids), 3)
+            for val in group_ids:
+                self.assertGreaterEqual(val, 0)
+
+    if hasattr(posix, 'setresuid'):
+        def test_setresuid(self):
+            current_user_ids = posix.getresuid()
+            self.assertIsNone(posix.setresuid(*current_user_ids))
+            # -1 means don't change that value.
+            self.assertIsNone(posix.setresuid(-1, -1, -1))
+
+        def test_setresuid_exception(self):
+            # Don't do this test if someone is silly enough to run us as root.
+            current_user_ids = posix.getresuid()
+            if 0 not in current_user_ids:
+                new_user_ids = (current_user_ids[0]+1, -1, -1)
+                self.assertRaises(OSError, posix.setresuid, *new_user_ids)
+
+    if hasattr(posix, 'setresgid'):
+        def test_setresgid(self):
+            current_group_ids = posix.getresgid()
+            self.assertIsNone(posix.setresgid(*current_group_ids))
+            # -1 means don't change that value.
+            self.assertIsNone(posix.setresgid(-1, -1, -1))
+
+        def test_setresgid_exception(self):
+            # Don't do this test if someone is silly enough to run us as root.
+            current_group_ids = posix.getresgid()
+            if 0 not in current_group_ids:
+                new_group_ids = (current_group_ids[0]+1, -1, -1)
+                self.assertRaises(OSError, posix.setresgid, *new_group_ids)
+
+    @unittest.skipUnless(hasattr(posix, 'initgroups'),
+                         "test needs os.initgroups()")
+    def test_initgroups(self):
+        # It takes a string and an integer; check that it raises a TypeError
+        # for other argument lists.
+        self.assertRaises(TypeError, posix.initgroups)
+        self.assertRaises(TypeError, posix.initgroups, None)
+        self.assertRaises(TypeError, posix.initgroups, 3, "foo")
+        self.assertRaises(TypeError, posix.initgroups, "foo", 3, object())
+
+        # If a non-privileged user invokes it, it should fail with OSError
+        # EPERM.
+        if os.getuid() != 0:
+            name = pwd.getpwuid(posix.getuid()).pw_name
+            try:
+                posix.initgroups(name, 13)
+            except OSError as e:
+                self.assertEqual(e.errno, errno.EPERM)
+            else:
+                self.fail("Expected OSError to be raised by initgroups")
+
+    def test_statvfs(self):
+        if hasattr(posix, 'statvfs'):
+            self.assertTrue(posix.statvfs(os.curdir))
+
+    def test_fstatvfs(self):
+        if hasattr(posix, 'fstatvfs'):
+            fp = open(test_support.TESTFN)
+            try:
+                self.assertTrue(posix.fstatvfs(fp.fileno()))
+            finally:
+                fp.close()
+
+    def test_ftruncate(self):
+        if hasattr(posix, 'ftruncate'):
+            fp = open(test_support.TESTFN, 'w+')
+            try:
+                # we need to have some data to truncate
+                fp.write('test')
+                fp.flush()
+                posix.ftruncate(fp.fileno(), 0)
+            finally:
+                fp.close()
+
+    def test_dup(self):
+        if hasattr(posix, 'dup'):
+            fp = open(test_support.TESTFN)
+            try:
+                fd = posix.dup(fp.fileno())
+                self.assertIsInstance(fd, int)
+                os.close(fd)
+            finally:
+                fp.close()
+
+    def test_confstr(self):
+        if hasattr(posix, 'confstr'):
+            self.assertRaises(ValueError, posix.confstr, "CS_garbage")
+            self.assertEqual(len(posix.confstr("CS_PATH")) > 0, True)
+
+    def test_dup2(self):
+        if hasattr(posix, 'dup2'):
+            fp1 = open(test_support.TESTFN)
+            fp2 = open(test_support.TESTFN)
+            try:
+                posix.dup2(fp1.fileno(), fp2.fileno())
+            finally:
+                fp1.close()
+                fp2.close()
+
+    def fdopen_helper(self, *args):
+        fd = os.open(test_support.TESTFN, os.O_RDONLY)
+        fp2 = posix.fdopen(fd, *args)
+        fp2.close()
+
+    def test_fdopen(self):
+        if hasattr(posix, 'fdopen'):
+            self.fdopen_helper()
+            self.fdopen_helper('r')
+            self.fdopen_helper('r', 100)
+
+    def test_osexlock(self):
+        if hasattr(posix, "O_EXLOCK"):
+            fd = os.open(test_support.TESTFN,
+                         os.O_WRONLY|os.O_EXLOCK|os.O_CREAT)
+            self.assertRaises(OSError, os.open, test_support.TESTFN,
+                              os.O_WRONLY|os.O_EXLOCK|os.O_NONBLOCK)
+            os.close(fd)
+
+            if hasattr(posix, "O_SHLOCK"):
+                fd = os.open(test_support.TESTFN,
+                             os.O_WRONLY|os.O_SHLOCK|os.O_CREAT)
+                self.assertRaises(OSError, os.open, test_support.TESTFN,
+                                  os.O_WRONLY|os.O_EXLOCK|os.O_NONBLOCK)
+                os.close(fd)
+
+    def test_osshlock(self):
+        if hasattr(posix, "O_SHLOCK"):
+            fd1 = os.open(test_support.TESTFN,
+                         os.O_WRONLY|os.O_SHLOCK|os.O_CREAT)
+            fd2 = os.open(test_support.TESTFN,
+                          os.O_WRONLY|os.O_SHLOCK|os.O_CREAT)
+            os.close(fd2)
+            os.close(fd1)
+
+            if hasattr(posix, "O_EXLOCK"):
+                fd = os.open(test_support.TESTFN,
+                             os.O_WRONLY|os.O_SHLOCK|os.O_CREAT)
+                self.assertRaises(OSError, os.open, test_support.TESTFN,
+                                  os.O_RDONLY|os.O_EXLOCK|os.O_NONBLOCK)
+                os.close(fd)
+
+    def test_fstat(self):
+        if hasattr(posix, 'fstat'):
+            fp = open(test_support.TESTFN)
+            try:
+                self.assertTrue(posix.fstat(fp.fileno()))
+            finally:
+                fp.close()
+
+    def test_stat(self):
+        if hasattr(posix, 'stat'):
+            self.assertTrue(posix.stat(test_support.TESTFN))
+
+    def _test_all_chown_common(self, chown_func, first_param):
+        """Common code for chown, fchown and lchown tests."""
+        if os.getuid() == 0:
+            try:
+                # Many linux distros have a nfsnobody user as MAX_UID-2
+                # that makes a good test case for signedness issues.
+                #   http://bugs.python.org/issue1747858
+                # This part of the test only runs when run as root.
+                # Only scary people run their tests as root.
+                ent = pwd.getpwnam('nfsnobody')
+                chown_func(first_param, ent.pw_uid, ent.pw_gid)
+            except KeyError:
+                pass
+        else:
+            # non-root cannot chown to root, raises OSError
+            self.assertRaises(OSError, chown_func,
+                              first_param, 0, 0)
+
+        # test a successful chown call
+        chown_func(first_param, os.getuid(), os.getgid())
+
+    @unittest.skipUnless(hasattr(posix, 'chown'), "test needs os.chown()")
+    def test_chown(self):
+        # raise an OSError if the file does not exist
+        os.unlink(test_support.TESTFN)
+        self.assertRaises(OSError, posix.chown, test_support.TESTFN, -1, -1)
+
+        # re-create the file
+        open(test_support.TESTFN, 'w').close()
+        self._test_all_chown_common(posix.chown, test_support.TESTFN)
+
+    @unittest.skipUnless(hasattr(posix, 'fchown'), "test needs os.fchown()")
+    def test_fchown(self):
+        os.unlink(test_support.TESTFN)
+
+        # re-create the file
+        test_file = open(test_support.TESTFN, 'w')
+        try:
+            fd = test_file.fileno()
+            self._test_all_chown_common(posix.fchown, fd)
+        finally:
+            test_file.close()
+
+    @unittest.skipUnless(hasattr(posix, 'lchown'), "test needs os.lchown()")
+    def test_lchown(self):
+        os.unlink(test_support.TESTFN)
+        # create a symlink
+        os.symlink(_DUMMY_SYMLINK, test_support.TESTFN)
+        self._test_all_chown_common(posix.lchown, test_support.TESTFN)
+
+    def test_chdir(self):
+        if hasattr(posix, 'chdir'):
+            posix.chdir(os.curdir)
+            self.assertRaises(OSError, posix.chdir, test_support.TESTFN)
+
+    def test_lsdir(self):
+        if hasattr(posix, 'lsdir'):
+            self.assertIn(test_support.TESTFN, posix.lsdir(os.curdir))
+
+    def test_access(self):
+        if hasattr(posix, 'access'):
+            self.assertTrue(posix.access(test_support.TESTFN, os.R_OK))
+
+    def test_umask(self):
+        if hasattr(posix, 'umask'):
+            old_mask = posix.umask(0)
+            self.assertIsInstance(old_mask, int)
+            posix.umask(old_mask)
+
+    def test_strerror(self):
+        if hasattr(posix, 'strerror'):
+            self.assertTrue(posix.strerror(0))
+
+    def test_pipe(self):
+        if hasattr(posix, 'pipe'):
+            reader, writer = posix.pipe()
+            os.close(reader)
+            os.close(writer)
+
+    def test_tempnam(self):
+        if hasattr(posix, 'tempnam'):
+            with warnings.catch_warnings():
+                warnings.filterwarnings("ignore", "tempnam", DeprecationWarning)
+                self.assertTrue(posix.tempnam())
+                self.assertTrue(posix.tempnam(os.curdir))
+                self.assertTrue(posix.tempnam(os.curdir, 'blah'))
+
+    def test_tmpfile(self):
+        if hasattr(posix, 'tmpfile'):
+            with warnings.catch_warnings():
+                warnings.filterwarnings("ignore", "tmpfile", DeprecationWarning)
+                fp = posix.tmpfile()
+                fp.close()
+
+    def test_utime(self):
+        if hasattr(posix, 'utime'):
+            now = time.time()
+            posix.utime(test_support.TESTFN, None)
+            self.assertRaises(TypeError, posix.utime, test_support.TESTFN, (None, None))
+            self.assertRaises(TypeError, posix.utime, test_support.TESTFN, (now, None))
+            self.assertRaises(TypeError, posix.utime, test_support.TESTFN, (None, now))
+            posix.utime(test_support.TESTFN, (int(now), int(now)))
+            posix.utime(test_support.TESTFN, (now, now))
+
+    def _test_chflags_regular_file(self, chflags_func, target_file):
+        st = os.stat(target_file)
+        self.assertTrue(hasattr(st, 'st_flags'))
+        chflags_func(target_file, st.st_flags | stat.UF_IMMUTABLE)
+        try:
+            new_st = os.stat(target_file)
+            self.assertEqual(st.st_flags | stat.UF_IMMUTABLE, new_st.st_flags)
+            try:
+                fd = open(target_file, 'w+')
+            except IOError as e:
+                self.assertEqual(e.errno, errno.EPERM)
+        finally:
+            posix.chflags(target_file, st.st_flags)
+
+    @unittest.skipUnless(hasattr(posix, 'chflags'), 'test needs os.chflags()')
+    def test_chflags(self):
+        self._test_chflags_regular_file(posix.chflags, test_support.TESTFN)
+
+    @unittest.skipUnless(hasattr(posix, 'lchflags'), 'test needs os.lchflags()')
+    def test_lchflags_regular_file(self):
+        self._test_chflags_regular_file(posix.lchflags, test_support.TESTFN)
+
+    @unittest.skipUnless(hasattr(posix, 'lchflags'), 'test needs os.lchflags()')
+    def test_lchflags_symlink(self):
+        testfn_st = os.stat(test_support.TESTFN)
+
+        self.assertTrue(hasattr(testfn_st, 'st_flags'))
+
+        os.symlink(test_support.TESTFN, _DUMMY_SYMLINK)
+        self.teardown_files.append(_DUMMY_SYMLINK)
+        dummy_symlink_st = os.lstat(_DUMMY_SYMLINK)
+
+        posix.lchflags(_DUMMY_SYMLINK,
+                       dummy_symlink_st.st_flags | stat.UF_IMMUTABLE)
+        try:
+            new_testfn_st = os.stat(test_support.TESTFN)
+            new_dummy_symlink_st = os.lstat(_DUMMY_SYMLINK)
+
+            self.assertEqual(testfn_st.st_flags, new_testfn_st.st_flags)
+            self.assertEqual(dummy_symlink_st.st_flags | stat.UF_IMMUTABLE,
+                             new_dummy_symlink_st.st_flags)
+        finally:
+            posix.lchflags(_DUMMY_SYMLINK, dummy_symlink_st.st_flags)
+
+    def test_getcwd_long_pathnames(self):
+        if hasattr(posix, 'getcwd'):
+            dirname = 'getcwd-test-directory-0123456789abcdef-01234567890abcdef'
+            curdir = os.getcwd()
+            base_path = os.path.abspath(test_support.TESTFN) + '.getcwd'
+
+            try:
+                os.mkdir(base_path)
+                os.chdir(base_path)
+            except:
+#               Just returning nothing instead of the SkipTest exception,
+#               because the test results in Error in that case.
+#               Is that ok?
+#                raise unittest.SkipTest, "cannot create directory for testing"
+                return
+
+            try:
+                def _create_and_do_getcwd(dirname, current_path_length = 0):
+                    try:
+                        os.mkdir(dirname)
+                    except:
+                        raise unittest.SkipTest, "mkdir cannot create directory sufficiently deep for getcwd test"
+
+                    os.chdir(dirname)
+                    try:
+                        os.getcwd()
+                        if current_path_length < 4099:
+                            _create_and_do_getcwd(dirname, current_path_length + len(dirname) + 1)
+                    except OSError as e:
+                        expected_errno = errno.ENAMETOOLONG
+                        if 'sunos' in sys.platform or 'openbsd' in sys.platform:
+                            expected_errno = errno.ERANGE # Issue 9185
+                        self.assertEqual(e.errno, expected_errno)
+                    finally:
+                        os.chdir('..')
+                        os.rmdir(dirname)
+
+                _create_and_do_getcwd(dirname)
+
+            finally:
+                os.chdir(curdir)
+                shutil.rmtree(base_path)
+
+    @unittest.skipUnless(hasattr(os, 'getegid'), "test needs os.getegid()")
+    def test_getgroups(self):
+        with os.popen('id -G') as idg:
+            groups = idg.read().strip()
+
+        if not groups:
+            raise unittest.SkipTest("need working 'id -G'")
+
+        # 'id -G' and 'os.getgroups()' should return the same
+        # groups, ignoring order and duplicates.
+        # #10822 - it is implementation defined whether posix.getgroups()
+        # includes the effective gid so we include it anyway, since id -G does
+        self.assertEqual(
+                set([int(x) for x in groups.split()]),
+                set(posix.getgroups() + [posix.getegid()]))
+
+class PosixGroupsTester(unittest.TestCase):
+
+    def setUp(self):
+        if posix.getuid() != 0:
+            raise unittest.SkipTest("not enough privileges")
+        if not hasattr(posix, 'getgroups'):
+            raise unittest.SkipTest("need posix.getgroups")
+        if sys.platform == 'darwin':
+            raise unittest.SkipTest("getgroups(2) is broken on OSX")
+        self.saved_groups = posix.getgroups()
+
+    def tearDown(self):
+        if hasattr(posix, 'setgroups'):
+            posix.setgroups(self.saved_groups)
+        elif hasattr(posix, 'initgroups'):
+            name = pwd.getpwuid(posix.getuid()).pw_name
+            posix.initgroups(name, self.saved_groups[0])
+
+    @unittest.skipUnless(hasattr(posix, 'initgroups'),
+                         "test needs posix.initgroups()")
+    def test_initgroups(self):
+        # find missing group
+
+        g = max(self.saved_groups) + 1
+        name = pwd.getpwuid(posix.getuid()).pw_name
+        posix.initgroups(name, g)
+        self.assertIn(g, posix.getgroups())
+
+    @unittest.skipUnless(hasattr(posix, 'setgroups'),
+                         "test needs posix.setgroups()")
+    def test_setgroups(self):
+        for groups in [[0], range(16)]:
+            posix.setgroups(groups)
+            self.assertListEqual(groups, posix.getgroups())
+
+
+def test_main():
+    test_support.run_unittest(PosixTester, PosixGroupsTester)
+
+if __name__ == '__main__':
+    test_main()

-- 
Repository URL: http://hg.python.org/jython


More information about the Jython-checkins mailing list