# -*- coding: utf-8 -*-

import os
import py
import sys
import signal

from rpython.tool.udir import udir
from pypy.tool.pytest.objspace import gettestobjspace
from pypy.interpreter.gateway import interp2app
from rpython.translator.c.test.test_extfunc import need_sparse_files
from rpython.rlib import rposix

USEMODULES = ['binascii', 'posix', 'signal', 'struct', 'time']
# py3k os.open uses subprocess, requiring the following per platform
if os.name != 'nt':
    USEMODULES += ['fcntl', 'select', '_posixsubprocess', '_socket']
else:
    USEMODULES += ['_rawffi', 'thread']
    USEMODULES += ['_rawffi', 'thread', 'signal', '_cffi_backend']

def setup_module(mod):
    mod.space = gettestobjspace(usemodules=USEMODULES)
    mod.path = udir.join('posixtestfile.txt')
    mod.path.write("this is a test")
    mod.path2 = udir.join('test_posix2-')
    pdir = udir.ensure('posixtestdir', dir=True)
    pdir.join('file1').write("test1")
    os.chmod(str(pdir.join('file1')), 0o600)
    pdir.join('file2').write("test2")
    pdir.join('another_longer_file_name').write("test3")
    mod.pdir = pdir
    if sys.platform == 'darwin':
        # see issue https://bugs.python.org/issue31380
        bytes_dir = udir.ensure('fixc5x9fier.txt', dir=True)
        file_name = 'cafxe9'
        surrogate_name = 'foo'
    else:
        bytes_dir = udir.ensure('fi\xc5\x9fier.txt', dir=True)
        file_name = 'caf\xe9'
        surrogate_name = 'foo\x80'
    bytes_dir.join('somefile').write('who cares?')
    bytes_dir.join(file_name).write('who knows?')
    mod.bytes_dir = bytes_dir
    # an escaped surrogate
    mod.esurrogate_dir = udir.ensure(surrogate_name, dir=True)

    # in applevel tests, os.stat uses the CPython os.stat.
    # Be sure to return times with full precision
    # even when running on top of CPython 2.4.
    os.stat_float_times(True)

    # Initialize sys.filesystemencoding
    # space.call_method(space.getbuiltinmodule('sys'), 'getfilesystemencoding')


GET_POSIX = "(): import %s as m ; return m" % os.name


class AppTestPosix:
    spaceconfig = {'usemodules': USEMODULES}

    def setup_class(cls):
        space = cls.space
        cls.w_runappdirect = space.wrap(cls.runappdirect)
        cls.w_posix = space.appexec([], GET_POSIX)
        cls.w_os = space.appexec([], "(): import os as m ; return m")
        cls.w_path = space.wrap(str(path))
        cls.w_path2 = space.wrap(str(path2))
        cls.w_pdir = space.wrap(str(pdir))
        cls.w_bytes_dir = space.newbytes(str(bytes_dir))
        cls.w_esurrogate_dir = space.newbytes(str(esurrogate_dir))
        if hasattr(os, 'getuid'):
            cls.w_getuid = space.wrap(os.getuid())
            cls.w_geteuid = space.wrap(os.geteuid())
        if hasattr(os, 'getgid'):
            cls.w_getgid = space.wrap(os.getgid())
        if hasattr(os, 'getgroups'):
            cls.w_getgroups = space.newlist([space.wrap(e) for e in os.getgroups()])
        if hasattr(os, 'getpgid'):
            cls.w_getpgid = space.wrap(os.getpgid(os.getpid()))
        if hasattr(os, 'getsid'):
            cls.w_getsid0 = space.wrap(os.getsid(0))
        if hasattr(os, 'sysconf'):
            sysconf_name = os.sysconf_names.keys()[0]
            cls.w_sysconf_name = space.wrap(sysconf_name)
            cls.w_sysconf_value = space.wrap(os.sysconf_names[sysconf_name])
            cls.w_sysconf_result = space.wrap(os.sysconf(sysconf_name))
        if hasattr(os, 'confstr'):
            confstr_name = os.confstr_names.keys()[0]
            cls.w_confstr_name = space.wrap(confstr_name)
            cls.w_confstr_value = space.wrap(os.confstr_names[confstr_name])
            cls.w_confstr_result = space.wrap(os.confstr(confstr_name))
        cls.w_SIGABRT = space.wrap(signal.SIGABRT)
        cls.w_python = space.wrap(sys.executable)
        if hasattr(os, 'major'):
            cls.w_expected_major_12345 = space.wrap(os.major(12345))
            cls.w_expected_minor_12345 = space.wrap(os.minor(12345))
        cls.w_udir = space.wrap(str(udir))

    def setup_method(self, meth):
        if getattr(meth, 'need_sparse_files', False):
            if sys.maxsize < 2**32 and not self.runappdirect:
                # this fails because it uses ll2ctypes to call the posix
                # functions like 'open' and 'lseek', whereas a real compiled
                # C program would macro-define them to their longlong versions
                py.test.skip("emulation of files can't use "
                             "larger-than-long offsets")
            need_sparse_files()

    def test_posix_is_pypy_s(self):
        assert hasattr(self.posix, '_statfields')

    def test_some_posix_basic_operation(self):
        path = self.path
        posix = self.posix
        fd = posix.open(path, posix.O_RDONLY, 0o777)
        fd2 = posix.dup(fd)
        assert posix.get_inheritable(fd2) == False
        assert not posix.isatty(fd2)
        s = posix.read(fd, 1)
        assert s == b't'
        posix.lseek(fd, 5, 0)
        s = posix.read(fd, 1)
        assert s == b'i'
        st = posix.fstat(fd)
        assert st == posix.stat(fd)
        posix.close(fd2)
        posix.close(fd)

        import sys, stat
        assert st[0] == st.st_mode
        assert st[1] == st.st_ino
        assert st[2] == st.st_dev
        assert st[3] == st.st_nlink
        assert st[4] == st.st_uid
        assert st[5] == st.st_gid
        assert st[6] == st.st_size
        assert st[7] == int(st.st_atime)   # in complete corner cases, rounding
        assert st[8] == int(st.st_mtime)   # here could maybe get the wrong
        assert st[9] == int(st.st_ctime)   # integer...

        assert stat.S_IMODE(st.st_mode) & stat.S_IRUSR
        assert stat.S_IMODE(st.st_mode) & stat.S_IWUSR
        if not sys.platform.startswith('win'):
            assert not (stat.S_IMODE(st.st_mode) & stat.S_IXUSR)

        assert st.st_size == 14
        assert st.st_nlink == 1

        if sys.platform.startswith('linux'):
            assert isinstance(st.st_atime, float)
            assert isinstance(st.st_mtime, float)
            assert isinstance(st.st_ctime, float)
            assert hasattr(st, 'st_rdev')

        assert isinstance(st.st_atime_ns, int)
        assert abs(st.st_atime_ns - 1e9*st.st_atime) < 500
        assert abs(st.st_mtime_ns - 1e9*st.st_mtime) < 500
        assert abs(st.st_ctime_ns - 1e9*st.st_ctime) < 500

    def test_stat_float_times(self):
        path = self.path
        posix = self.posix
        current = posix.stat_float_times()
        assert current is True
        try:
            posix.stat_float_times(True)
            st = posix.stat(path)
            assert isinstance(st.st_mtime, float)
            assert st[7] == int(st.st_atime)
            assert posix.stat_float_times(-1) is True

            posix.stat_float_times(False)
            st = posix.stat(path)
            assert isinstance(st.st_mtime, int)
            assert st[7] == st.st_atime
            assert posix.stat_float_times(-1) is False

        finally:
            posix.stat_float_times(current)

    def test_stat_result(self):
        st = self.posix.stat_result((0, 0, 0, 0, 0, 0, 0, 41, 42.1, 43))
        assert st.st_atime == 41
        assert st.st_mtime == 42.1
        assert st.st_ctime == 43
        assert repr(st).startswith('os.stat_result')

    def test_stat_lstat(self):
        import stat
        st = self.posix.stat(".")
        assert stat.S_ISDIR(st.st_mode)
        st = self.posix.stat(b".")
        assert stat.S_ISDIR(st.st_mode)
        st = self.posix.stat(bytearray(b"."))
        assert stat.S_ISDIR(st.st_mode)
        st = self.posix.lstat(".")
        assert stat.S_ISDIR(st.st_mode)

    def test_stat_exception(self):
        import sys
        import errno
        for fn in [self.posix.stat, self.posix.lstat]:
            exc = raises(OSError, fn, "nonexistentdir/nonexistentfile")
            assert exc.value.errno == errno.ENOENT
            assert exc.value.filename == "nonexistentdir/nonexistentfile"

        excinfo = raises(TypeError, self.posix.stat, None)
        assert "can't specify None" in str(excinfo.value)
        excinfo = raises(TypeError, self.posix.stat, 2.)
        assert "should be string, bytes or integer, not float" in str(excinfo.value)
        raises(ValueError, self.posix.stat, -1)
        raises(ValueError, self.posix.stat, b"abc\x00def")
        raises(ValueError, self.posix.stat, u"abc\x00def")

    if hasattr(__import__(os.name), "statvfs"):
        def test_statvfs(self):
            st = self.posix.statvfs(".")
            assert isinstance(st, self.posix.statvfs_result)
            for field in [
                'f_bsize', 'f_frsize', 'f_blocks', 'f_bfree', 'f_bavail',
                'f_files', 'f_ffree', 'f_favail', 'f_flag', 'f_namemax',
            ]:
                assert hasattr(st, field)

    def test_pickle(self):
        import pickle, os
        st = self.posix.stat(os.curdir)
        # print(type(st).__module__)
        s = pickle.dumps(st)
        # print(repr(s))
        new = pickle.loads(s)
        assert new == st
        assert type(new) is type(st)

    def test_open_exception(self):
        posix = self.posix
        try:
            posix.open('qowieuqwoeiu', 0, 0)
        except OSError as e:
            assert e.filename == 'qowieuqwoeiu'
        else:
            assert 0

    def test_filename_exception(self):
        for fname in ['unlink', 'remove',
                      'chdir', 'mkdir', 'rmdir',
                      'listdir', 'readlink',
                      'chroot']:
            if hasattr(self.posix, fname):
                func = getattr(self.posix, fname)
                try:
                    func('qowieuqw/oeiu')
                except OSError as e:
                    assert e.filename == 'qowieuqw/oeiu'
                else:
                    assert 0

    def test_chmod_exception(self):
        try:
            self.posix.chmod('qowieuqw/oeiu', 0)
        except OSError as e:
            assert e.filename == 'qowieuqw/oeiu'
        else:
            assert 0

    def test_chown_exception(self):
        if hasattr(self.posix, 'chown'):
            try:
                self.posix.chown('qowieuqw/oeiu', 0, 0)
            except OSError as e:
                assert e.filename == 'qowieuqw/oeiu'
            else:
                assert 0

    def test_utime_exception(self):
        for arg in [None, (0, 0)]:
            try:
                self.posix.utime('qowieuqw/oeiu', arg)
            except OSError as e:
                pass
            else:
                assert 0

    def test_functions_raise_error(self):
        def ex(func, *args):
            try:
                func(*args)
            except OSError:
                pass
            else:
                raise AssertionError("%s(%s) did not raise" %(
                                     func.__name__,
                                     ", ".join([str(x) for x in args])))
        UNUSEDFD = 123123
        ex(self.posix.open, "qweqwe", 0, 0)
        ex(self.posix.lseek, UNUSEDFD, 123, 0)
        #apparently not posix-required: ex(self.posix.isatty, UNUSEDFD)
        ex(self.posix.read, UNUSEDFD, 123)
        ex(self.posix.write, UNUSEDFD, b"x")
        ex(self.posix.close, UNUSEDFD)
        #UMPF cpython raises IOError ex(self.posix.ftruncate, UNUSEDFD, 123)
        ex(self.posix.fstat, UNUSEDFD)
        ex(self.posix.stat, "qweqwehello")
        # how can getcwd() raise?
        ex(self.posix.dup, UNUSEDFD)

    def test_getcwd(self):
        os, posix = self.os, self.posix
        assert isinstance(posix.getcwd(), str)
        cwdb = posix.getcwdb()
        os.chdir(self.esurrogate_dir)
        try:
            cwd = posix.getcwd()
            assert os.fsencode(cwd) == posix.getcwdb()
        finally:
            os.chdir(cwdb)

    def test_getcwdb(self):
        assert isinstance(self.posix.getcwdb(), bytes)

    def test_listdir(self):
        pdir = self.pdir
        posix = self.posix
        result = posix.listdir(pdir)
        result.sort()
        assert result == ['another_longer_file_name',
                          'file1',
                          'file2']

    def test_listdir_default(self):
        posix = self.posix
        assert posix.listdir() == posix.listdir('.') == posix.listdir(None)

    def test_listdir_bytes(self):
        import sys
        bytes_dir = self.bytes_dir
        posix = self.posix
        result = posix.listdir(bytes_dir)
        assert all(type(x) is bytes for x in result)
        assert b'somefile' in result
        expected = b'caf%E9' if sys.platform == 'darwin' else b'caf\xe9'
        assert expected in result

    def test_listdir_memoryview_returns_unicode(self):
        # XXX unknown why CPython has this behaviour
        bytes_dir = self.bytes_dir
        os, posix = self.os, self.posix
        result1 = posix.listdir(bytes_dir)              # -> list of bytes
        result2 = posix.listdir(memoryview(bytes_dir))  # -> list of unicodes
        assert [os.fsencode(x) for x in result2] == result1

    def test_fdlistdir(self):
        posix = self.posix
        dirfd = posix.open('.', posix.O_RDONLY)
        lst1 = posix.listdir(dirfd)   # does not close dirfd
        lst2 = posix.listdir('.')
        assert lst1 == lst2
        #
        lst3 = posix.listdir(dirfd)   # rewinddir() was used
        assert lst3 == lst1
        posix.close(dirfd)

    def test_undecodable_filename(self):
        import sys
        posix = self.posix
        try:
            'caf\xe9'.encode(sys.getfilesystemencoding(), 'surrogateescape')
        except UnicodeEncodeError:
            pass # probably ascii
        else:
            assert posix.access('caf\xe9', posix.R_OK) is False
        assert posix.access(b'caf\xe9', posix.R_OK) is False
        assert posix.access('caf\udcc0', posix.R_OK) is False
        assert posix.access(b'caf\xc3', posix.R_OK) is False

    def test_access(self):
        pdir = self.pdir + '/file1'
        posix = self.posix

        assert posix.access(pdir, posix.R_OK) is True
        assert posix.access(pdir, posix.W_OK) is True
        import sys
        if sys.platform != "win32":
            assert posix.access(pdir, posix.X_OK) is False

    def test_times(self):
        """
        posix.times() should return a posix.times_result object giving
        float-representations (seconds, effectively) of the four fields from
        the underlying struct tms and the return value.
        """
        result = self.posix.times()
        assert isinstance(self.posix.times(), self.posix.times_result)
        assert isinstance(self.posix.times(), tuple)
        assert len(result) == 5
        for value in result:
            assert isinstance(value, float)
        assert isinstance(result.user, float)
        assert isinstance(result.system, float)
        assert isinstance(result.children_user, float)
        assert isinstance(result.children_system, float)
        assert isinstance(result.elapsed, float)

    def test_strerror(self):
        assert isinstance(self.posix.strerror(0), str)
        assert isinstance(self.posix.strerror(1), str)

    if hasattr(__import__(os.name), "fork"):
        def test_fork(self):
            os = self.posix
            pid = os.fork()
            if pid == 0:   # child
                os._exit(4)
            pid1, status1 = os.waitpid(pid, 0)
            assert pid1 == pid
            assert os.WIFEXITED(status1)
            assert os.WEXITSTATUS(status1) == 4
        pass # <- please, inspect.getsource(), don't crash


    if hasattr(__import__(os.name), "openpty"):
        def test_openpty(self):
            os = self.posix
            master_fd, slave_fd = os.openpty()
            assert isinstance(master_fd, int)
            assert isinstance(slave_fd, int)
            os.write(slave_fd, b'x\n')
            data = os.read(master_fd, 100)
            assert data.startswith(b'x')
            os.close(master_fd)
            os.close(slave_fd)

        def test_openpty_non_inheritable(self):
            os = self.posix
            master_fd, slave_fd = os.openpty()
            assert os.get_inheritable(master_fd) == False
            assert os.get_inheritable(slave_fd) == False
            os.close(master_fd)
            os.close(slave_fd)

    if hasattr(__import__(os.name), "forkpty"):
        def test_forkpty(self):
            import sys
            if 'freebsd' in sys.platform:
                skip("hangs indifinitly on FreeBSD (also on CPython).")
            os = self.posix
            childpid, master_fd = os.forkpty()
            assert isinstance(childpid, int)
            assert isinstance(master_fd, int)
            if childpid == 0:
                data = os.read(0, 100)
                if data.startswith(b'abc'):
                    os._exit(42)
                else:
                    os._exit(43)
            os.write(master_fd, b'abc\n')
            _, status = os.waitpid(childpid, 0)
            assert status >> 8 == 42

    if hasattr(__import__(os.name), "execv"):
        def test_execv(self):
            os = self.posix
            if not hasattr(os, "fork"):
                skip("Need fork() to test execv()")
            pid = os.fork()
            if pid == 0:
                os.execv("/usr/bin/env", ["env", "python", "-c", "open('onefile', 'w').write('1')"])
            os.waitpid(pid, 0)
            assert open("onefile").read() == "1"
            os.unlink("onefile")

        def test_execv_raising(self):
            os = self.posix
            raises(OSError, 'os.execv("saddsadsadsadsa", ["saddsadsasaddsa"])')

        def test_execv_no_args(self):
            os = self.posix
            raises(ValueError, os.execv, "notepad", [])

        def test_execv_raising2(self):
            os = self.posix
            for n in 3, [3, "a"]:
                try:
                    os.execv("xxx", n)
                except TypeError as t:
                    assert str(t) == "execv() arg 2 must be an iterable of strings"
                else:
                    py.test.fail("didn't raise")

        def test_execv_unicode(self):
            os = self.posix
            import sys
            if not hasattr(os, "fork"):
                skip("Need fork() to test execv()")
            try:
                output = "caf\xe9 \u1234\n".encode(sys.getfilesystemencoding())
            except UnicodeEncodeError:
                skip("encoding not good enough")
            pid = os.fork()
            if pid == 0:
                os.execv("/bin/sh", ["sh", "-c",
                                     "echo caf\xe9 \u1234 > onefile"])
            os.waitpid(pid, 0)
            assert open("onefile", "rb").read() == output
            os.unlink("onefile")

        def test_execve(self):
            os = self.posix
            if not hasattr(os, "fork"):
                skip("Need fork() to test execve()")
            pid = os.fork()
            if pid == 0:
                os.execve("/usr/bin/env", ["env", "python", "-c", "import os; open('onefile', 'w').write(os.environ['ddd'])"], {'ddd':'xxx'})
            os.waitpid(pid, 0)
            assert open("onefile").read() == "xxx"
            os.unlink("onefile")

        def test_execve_unicode(self):
            os = self.posix
            import sys
            if not hasattr(os, "fork"):
                skip("Need fork() to test execve()")
            try:
                output = "caf\xe9 \u1234".encode(sys.getfilesystemencoding())
            except UnicodeEncodeError:
                skip("encoding not good enough")
            t = ' abc\uDCFF'
            output += t.encode(sys.getfilesystemencoding(),
                               'surrogateescape')
            pid = os.fork()
            if pid == 0:
                os.execve("/bin/sh", ["sh", "-c",
                                      "echo caf\xe9 \u1234 $t > onefile"],
                          {'ddd': 'xxx', 't': t})
            os.waitpid(pid, 0)
            assert open("onefile", "rb").read() == output + b'\n'
            os.unlink("onefile")
        pass # <- please, inspect.getsource(), don't crash

    if hasattr(__import__(os.name), "spawnv"):
        def test_spawnv(self):
            os = self.posix
            import sys
            print(self.python)
            ret = os.spawnv(os.P_WAIT, self.python,
                            ['python', '-c', 'raise(SystemExit(42))'])
            assert ret == 42

    if hasattr(__import__(os.name), "spawnve"):
        def test_spawnve(self):
            os = self.posix
            env = {'PATH':os.environ['PATH'], 'FOOBAR': '42'}
            ret = os.spawnve(os.P_WAIT, self.python,
                             ['python', '-c',
                              "raise(SystemExit(int(__import__('os').environ['FOOBAR'])))"],
                             env)
            assert ret == 42

    def test_popen(self):
        os = self.os
        for i in range(5):
            stream = os.popen('echo 1')
            res = stream.read()
            assert res == '1\n'
            assert stream.close() is None

    def test_popen_with(self):
        os = self.os
        stream = os.popen('echo 1')
        with stream as fp:
            res = fp.read()
            assert res == '1\n'

    if sys.platform == "win32":
        # using startfile in app_startfile creates global state
        test_popen.dont_track_allocations = True
        test_popen_with.dont_track_allocations = True
        test_popen_child_fds.dont_track_allocations = True

    if hasattr(__import__(os.name), '_getfullpathname'):
        def test__getfullpathname(self):
            # nt specific
            posix = self.posix
            import os
            sysdrv = os.getenv("SystemDrive", "C:")
            # just see if it does anything
            path = sysdrv + 'hubber'
            assert os.sep in posix._getfullpathname(path)
            assert type(posix._getfullpathname(b'C:')) is bytes

    def test_utime(self):
        os = self.posix
        # XXX utimes & float support
        path = self.path2 + "test_utime.txt"
        fh = open(path, "wb")
        fh.write(b"x")
        fh.close()
        from time import time, sleep
        t0 = time()
        sleep(1.1)
        os.utime(path, None)
        assert os.stat(path).st_atime > t0
        os.utime(path, (int(t0), int(t0)))
        assert int(os.stat(path).st_atime) == int(t0)
        t1 = time()
        os.utime(path, (int(t1), int(t1)))
        assert int(os.stat(path).st_atime) == int(t1)

    def test_utime_raises(self):
        os = self.posix
        import errno
        raises(TypeError, "os.utime('xxx', 3)")
        exc = raises(OSError,
                     "os.utime('somefilewhichihopewouldneverappearhere', None)")
        assert exc.value.errno == errno.ENOENT

    for name in rposix.WAIT_MACROS:
        if hasattr(os, name):
            values = [0, 1, 127, 128, 255]
            code = py.code.Source("""
            def test_wstar(self):
                os = self.posix
                %s
            """ % "\n    ".join(["assert os.%s(%d) == %d" % (name, value,
                             getattr(os, name)(value)) for value in values]))
            d = {}
            exec code.compile() in d
            locals()['test_' + name] = d['test_wstar']

    if hasattr(os, 'WIFSIGNALED'):
        def test_wifsignaled(self):
            os = self.posix
            assert os.WIFSIGNALED(0) == False
            assert os.WIFSIGNALED(1) == True

    if hasattr(os, 'uname'):
        def test_os_uname(self):
            os = self.posix
            res = os.uname()
            assert len(res) == 5
            for i in res:
                assert isinstance(i, str)
            assert isinstance(res, tuple)
            assert res == (res.sysname, res.nodename,
                           res.release, res.version, res.machine)

    if hasattr(os, 'getuid'):
        def test_os_getuid(self):
            os = self.posix
            assert os.getuid() == self.getuid
            assert os.geteuid() == self.geteuid

    if hasattr(os, 'setuid'):
        @py.test.mark.skipif("sys.version_info < (2, 7, 4)")
        def test_os_setuid_error(self):
            os = self.posix
            raises(OverflowError, os.setuid, -2)
            raises(OverflowError, os.setuid, 2**32)
            raises(OSError, os.setuid, -1)

    if hasattr(os, 'getgid'):
        def test_os_getgid(self):
            os = self.posix
            assert os.getgid() == self.getgid

    if hasattr(os, 'getgroups'):
        def test_os_getgroups(self):
            os = self.posix
            assert os.getgroups() == self.getgroups

    if hasattr(os, 'setgroups'):
        def test_os_setgroups(self):
            os = self.posix
            raises(TypeError, os.setgroups, [2, 5, "hello"])
            try:
                os.setgroups(os.getgroups())
            except OSError:
                pass

    if hasattr(os, 'initgroups'):
        def test_os_initgroups(self):
            os = self.posix
            raises(OSError, os.initgroups, "crW2hTQC", 100)

    if hasattr(os, 'tcgetpgrp'):
        def test_os_tcgetpgrp(self):
            os = self.posix
            raises(OSError, os.tcgetpgrp, 9999)

    if hasattr(os, 'tcsetpgrp'):
        def test_os_tcsetpgrp(self):
            os = self.posix
            raises(OSError, os.tcsetpgrp, 9999, 1)

    if hasattr(os, 'getpgid'):
        def test_os_getpgid(self):
            os = self.posix
            assert os.getpgid(os.getpid()) == self.getpgid
            raises(OSError, os.getpgid, 1234567)

    if hasattr(os, 'setgid'):
        @py.test.mark.skipif("sys.version_info < (2, 7, 4)")
        def test_os_setgid_error(self):
            os = self.posix
            raises(OverflowError, os.setgid, -2)
            raises(OverflowError, os.setgid, 2**32)
            raises(OSError, os.setgid, -1)
            raises(OSError, os.setgid, 2**32-1)

    if hasattr(os, 'getsid'):
        def test_os_getsid(self):
            os = self.posix
            assert os.getsid(0) == self.getsid0
            raises(OSError, os.getsid, -100000)

    if hasattr(os, 'getresuid'):
        def test_os_getresuid(self):
            os = self.posix
            res = os.getresuid()
            assert len(res) == 3

    if hasattr(os, 'getresgid'):
        def test_os_getresgid(self):
            os = self.posix
            res = os.getresgid()
            assert len(res) == 3

    if hasattr(os, 'setresuid'):
        def test_os_setresuid(self):
            os = self.posix
            a, b, c = os.getresuid()
            os.setresuid(a, b, c)

    if hasattr(os, 'setresgid'):
        def test_os_setresgid(self):
            os = self.posix
            a, b, c = os.getresgid()
            os.setresgid(a, b, c)

    if hasattr(os, 'sysconf'):
        def test_os_sysconf(self):
            os = self.posix
            assert os.sysconf(self.sysconf_value) == self.sysconf_result
            assert os.sysconf(self.sysconf_name) == self.sysconf_result
            assert os.sysconf_names[self.sysconf_name] == self.sysconf_value

        def test_os_sysconf_error(self):
            os = self.posix
            raises(ValueError, os.sysconf, "!@#$%!#$!@#")

    if hasattr(os, 'fpathconf'):
        def test_os_fpathconf(self):
            os = self.posix
            assert os.fpathconf(1, "PC_PIPE_BUF") >= 128
            raises(OSError, os.fpathconf, -1, "PC_PIPE_BUF")
            raises(ValueError, os.fpathconf, 1, "##")

    if hasattr(os, 'pathconf'):
        def test_os_pathconf(self):
            os = self.posix
            assert os.pathconf("/tmp", "PC_NAME_MAX") >= 31
            # Linux: the following gets 'No such file or directory'
            raises(OSError, os.pathconf, "", "PC_PIPE_BUF")
            raises(ValueError, os.pathconf, "/tmp", "##")

    if hasattr(os, 'confstr'):
        def test_os_confstr(self):
            os = self.posix
            assert os.confstr(self.confstr_value) == self.confstr_result
            assert os.confstr(self.confstr_name) == self.confstr_result
            assert os.confstr_names[self.confstr_name] == self.confstr_value

        def test_os_confstr_error(self):
            os = self.posix
            raises(ValueError, os.confstr, "!@#$%!#$!@#")

    if hasattr(os, 'wait'):
        def test_os_wait(self):
            os = self.posix
            exit_status = 0x33

            if not hasattr(os, "fork"):
                skip("Need fork() to test wait()")
            if hasattr(os, "waitpid") and hasattr(os, "WNOHANG"):
                try:
                    while os.waitpid(-1, os.WNOHANG)[0]:
                        pass
                except OSError:  # until we get "No child processes", hopefully
                    pass
            child = os.fork()
            if child == 0: # in child
                os._exit(exit_status)
            else:
                pid, status = os.wait()
                assert child == pid
                assert os.WIFEXITED(status)
                assert os.WEXITSTATUS(status) == exit_status

    if hasattr(os, 'getloadavg'):
        def test_os_getloadavg(self):
            os = self.posix
            l0, l1, l2 = os.getloadavg()
            assert type(l0) is float and l0 >= 0.0
            assert type(l1) is float and l0 >= 0.0
            assert type(l2) is float and l0 >= 0.0

    if hasattr(os, 'major'):
        def test_major_minor(self):
            os = self.posix
            assert os.major(12345) == self.expected_major_12345
            assert os.minor(12345) == self.expected_minor_12345
            assert os.makedev(self.expected_major_12345,
                              self.expected_minor_12345) == 12345
            raises((ValueError, OverflowError), os.major, -1)

    if hasattr(os, 'fsync'):
        def test_fsync(self):
            os = self.posix
            f = open(self.path2, "w")
            try:
                fd = f.fileno()
                os.fsync(fd)
                os.fsync(f)     # <- should also work with a file, or anything
            finally:            #    with a fileno() method
                f.close()
            try:
                # May not raise anything with a buggy libc (or eatmydata)
                os.fsync(fd)
            except OSError:
                pass
            raises(ValueError, os.fsync, -1)

    if hasattr(os, 'fdatasync'):
        def test_fdatasync(self):
            os = self.posix
            f = open(self.path2, "w")
            try:
                fd = f.fileno()
                os.fdatasync(fd)
            finally:
                f.close()
            try:
                # May not raise anything with a buggy libc (or eatmydata)
                os.fdatasync(fd)
            except OSError:
                pass
            raises(ValueError, os.fdatasync, -1)

    if hasattr(os, 'fchdir'):
        def test_fchdir(self):
            os = self.posix
            localdir = os.getcwd()
            os.mkdir(self.path2 + 'fchdir')
            for func in [os.fchdir, os.chdir]:
                try:
                    fd = os.open(self.path2 + 'fchdir', os.O_RDONLY)
                    try:
                        func(fd)
                        mypath = os.getcwd()
                    finally:
                        os.close(fd)
                    assert mypath.endswith('test_posix2-fchdir')
                    raises(OSError, func, fd)
                finally:
                    os.chdir(localdir)
            raises(ValueError, os.fchdir, -1)

    if hasattr(rposix, 'pread'):
        def test_os_pread(self):
            os = self.posix
            fd = os.open(self.path2 + 'test_os_pread', os.O_RDWR | os.O_CREAT)
            try:
                os.write(fd, b'test')
                os.lseek(fd, 0, 0)
                assert os.pread(fd, 2, 1) == b'es'
                assert os.read(fd, 2) == b'te'
            finally:
                os.close(fd)

    if hasattr(rposix, 'pwrite'):
        def test_os_pwrite(self):
            os = self.posix
            fd = os.open(self.path2 + 'test_os_pwrite', os.O_RDWR | os.O_CREAT)
            try:
                os.write(fd, b'test')
                os.lseek(fd, 0, 0)
                os.pwrite(fd, b'xx', 1)
                assert os.read(fd, 4) == b'txxt'
            finally:
                os.close(fd)

    if hasattr(rposix, 'posix_fadvise'):
        def test_os_posix_fadvise(self):
            posix = self.posix
            fd = posix.open(self.path2 + 'test_os_posix_fadvise', posix.O_CREAT | posix.O_RDWR)
            try:
                posix.write(fd, b"foobar")
                assert posix.posix_fadvise(fd, 0, 1, posix.POSIX_FADV_WILLNEED) is None
                assert posix.posix_fadvise(fd, 1, 1, posix.POSIX_FADV_NORMAL) is None
                assert posix.posix_fadvise(fd, 2, 1, posix.POSIX_FADV_SEQUENTIAL) is None
                assert posix.posix_fadvise(fd, 3, 1, posix.POSIX_FADV_RANDOM) is None
                assert posix.posix_fadvise(fd, 4, 1, posix.POSIX_FADV_NOREUSE) is None
                assert posix.posix_fadvise(fd, 5, 1, posix.POSIX_FADV_DONTNEED) is None
                raises(OSError, posix.posix_fadvise, fd, 6, 1, 1234567)
            finally:
                posix.close(fd)

    if hasattr(rposix, 'posix_fallocate'):
        def test_os_posix_posix_fallocate(self):
            posix, os = self.posix, self.os
            fd = os.open(self.path2 + 'test_os_posix_fallocate', os.O_WRONLY | os.O_CREAT)
            try:
                assert posix.posix_fallocate(fd, 0, 10) == 0
            except OSError as inst:
                """ ZFS seems not to support fallocate.
                so skipping solaris-based since it is likely to come with ZFS
                """
                if inst.errno != errno.EINVAL or not sys.platform.startswith("sunos"):
                    raise
            finally:
                os.close(fd)


    def test_largefile(self):
        os = self.posix
        fd = os.open(self.path2 + 'test_largefile',
                     os.O_RDWR | os.O_CREAT, 0o666)
        os.ftruncate(fd, 10000000000)
        res = os.lseek(fd, 9900000000, 0)
        assert res == 9900000000
        res = os.lseek(fd, -5000000000, 1)
        assert res == 4900000000
        res = os.lseek(fd, -5200000000, 2)
        assert res == 4800000000
        os.close(fd)

        st = os.stat(self.path2 + 'test_largefile')
        assert st.st_size == 10000000000
    test_largefile.need_sparse_files = True

    if hasattr(rposix, 'getpriority'):
        def test_os_set_get_priority(self):
            posix, os = self.posix, self.os
            childpid = os.fork()
            if childpid == 0:
                # in the child (avoids changing the priority of the parent
                # process)
                orig_priority = posix.getpriority(posix.PRIO_PROCESS,
                                                  os.getpid())
                orig_grp_priority = posix.getpriority(posix.PRIO_PGRP,
                                                      os.getpgrp())
                posix.setpriority(posix.PRIO_PROCESS, os.getpid(),
                                  orig_priority + 1)
                new_priority = posix.getpriority(posix.PRIO_PROCESS,
                                                 os.getpid())
                assert new_priority == orig_priority + 1
                assert posix.getpriority(posix.PRIO_PGRP, os.getpgrp()) == (
                    orig_grp_priority)
                os._exit(0)    # ok
            #
            pid1, status1 = os.waitpid(childpid, 0)
            assert pid1 == childpid
            assert os.WIFEXITED(status1)
            assert os.WEXITSTATUS(status1) == 0   # else, test failure

    if hasattr(rposix, 'sched_get_priority_max'):
        def test_os_sched_get_priority_max(self):
            import sys
            posix, os = self.posix, self.os
            assert posix.sched_get_priority_max(posix.SCHED_FIFO) != -1
            assert posix.sched_get_priority_max(posix.SCHED_RR) != -1
            assert posix.sched_get_priority_max(posix.SCHED_OTHER) != -1
            if getattr(posix, 'SCHED_BATCH', None):
                assert posix.sched_get_priority_max(posix.SCHED_BATCH) != -1

    if hasattr(rposix, 'sched_get_priority_min'):
        def test_os_sched_get_priority_min(self):
            import sys
            posix, os = self.posix, self.os
            assert posix.sched_get_priority_min(posix.SCHED_FIFO) != -1
            assert posix.sched_get_priority_min(posix.SCHED_RR) != -1
            assert posix.sched_get_priority_min(posix.SCHED_OTHER) != -1
            if getattr(posix, 'SCHED_BATCH', None):
                assert posix.sched_get_priority_min(posix.SCHED_BATCH) != -1

    if hasattr(rposix, 'sched_get_priority_min'):
        def test_os_sched_priority_max_greater_than_min(self):
            posix, os = self.posix, self.os
            policy = posix.SCHED_RR
            low = posix.sched_get_priority_min(policy)
            high = posix.sched_get_priority_max(policy)
            assert isinstance(low, int) == True
            assert isinstance(high, int) == True
            assert  high > low

    if hasattr(rposix, 'sched_yield'):
        def test_sched_yield(self):
            os = self.posix
            #Always suceeds on Linux
            os.sched_yield()

    def test_write_buffer(self):
        os = self.posix
        fd = os.open(self.path2 + 'test_write_buffer',
                     os.O_RDWR | os.O_CREAT, 0o666)
        def writeall(s):
            while s:
                count = os.write(fd, s)
                assert count > 0
                s = s[count:]
        writeall(b'hello, ')
        writeall(memoryview(b'world!\n'))
        res = os.lseek(fd, 0, 0)
        assert res == 0
        data = b''
        while True:
            s = os.read(fd, 100)
            if not s:
                break
            data += s
        assert data == b'hello, world!\n'
        os.close(fd)

    def test_write_unicode(self):
        os = self.posix
        fd = os.open(self.path2 + 'test_write_unicode',
                     os.O_RDWR | os.O_CREAT, 0o666)
        raises(TypeError, os.write, fd, 'X')
        os.close(fd)

    if hasattr(__import__(os.name), "fork"):
        def test_abort(self):
            os = self.posix
            pid = os.fork()
            if pid == 0:
                os.abort()
            pid1, status1 = os.waitpid(pid, 0)
            assert pid1 == pid
            assert os.WIFSIGNALED(status1)
            assert os.WTERMSIG(status1) == self.SIGABRT
        pass # <- please, inspect.getsource(), don't crash

    def test_closerange(self):
        os = self.posix
        if not hasattr(os, 'closerange'):
            skip("missing os.closerange()")
        fds = [os.open(self.path + str(i), os.O_CREAT|os.O_WRONLY, 0o777)
               for i in range(15)]
        fds.sort()
        start = fds.pop()
        stop = start + 1
        while len(fds) > 3 and fds[-1] == start - 1:
            start = fds.pop()
        os.closerange(start, stop)
        for fd in fds:
            os.close(fd)     # should not have been closed
        for fd in range(start, stop):
            raises(OSError, os.fstat, fd)   # should have been closed

    if hasattr(os, 'chown'):
        def test_chown(self):
            my_path = self.path2 + 'test_chown'
            os = self.posix
            raises(OSError, os.chown, my_path, os.getuid(), os.getgid())
            open(my_path, 'w').close()
            os.chown(my_path, os.getuid(), os.getgid())

    if hasattr(os, 'lchown'):
        def test_lchown(self):
            my_path = self.path2 + 'test_lchown'
            os = self.posix
            raises(OSError, os.lchown, my_path, os.getuid(), os.getgid())
            os.symlink('foobar', my_path)
            os.lchown(my_path, os.getuid(), os.getgid())

    if hasattr(os, 'fchown'):
        def test_fchown(self):
            my_path = self.path2 + 'test_fchown'
            os = self.posix
            f = open(my_path, "w")
            os.fchown(f.fileno(), os.getuid(), os.getgid())
            f.close()

    if hasattr(os, 'chmod'):
        def test_chmod(self):
            import sys
            my_path = self.path2 + 'test_chmod'
            os = self.posix
            raises(OSError, os.chmod, my_path, 0o600)
            open(my_path, "w").close()
            if sys.platform == 'win32':
                os.chmod(my_path, 0o400)
                assert (os.stat(my_path).st_mode & 0o600) == 0o400
                os.chmod(self.path, 0o700)
            else:
                os.chmod(my_path, 0o200)
                assert (os.stat(my_path).st_mode & 0o777) == 0o200
                os.chmod(self.path, 0o700)

    if hasattr(os, 'fchmod'):
        def test_fchmod(self):
            my_path = self.path2 + 'test_fchmod'
            os = self.posix
            f = open(my_path, "w")
            os.fchmod(f.fileno(), 0o200)
            assert (os.fstat(f.fileno()).st_mode & 0o777) == 0o200
            f.close()
            assert (os.stat(my_path).st_mode & 0o777) == 0o200

    if hasattr(os, 'mkfifo'):
        def test_mkfifo(self):
            os = self.posix
            os.mkfifo(self.path2 + 'test_mkfifo', 0o666)
            st = os.lstat(self.path2 + 'test_mkfifo')
            import stat
            assert stat.S_ISFIFO(st.st_mode)

    if hasattr(os, 'mknod'):
        def test_mknod(self):
            import stat
            os = self.posix
            # os.mknod() may require root priviledges to work at all
            try:
                # not very useful: os.mknod() without specifying 'mode'
                os.mknod(self.path2 + 'test_mknod-1')
            except OSError as e:
                skip("os.mknod(): got %r" % (e,))
            st = os.lstat(self.path2 + 'test_mknod-1')
            assert stat.S_ISREG(st.st_mode)
            # os.mknod() with S_IFIFO
            os.mknod(self.path2 + 'test_mknod-2', 0o600 | stat.S_IFIFO)
            st = os.lstat(self.path2 + 'test_mknod-2')
            assert stat.S_ISFIFO(st.st_mode)

        def test_mknod_with_ifchr(self):
            # os.mknod() with S_IFCHR
            # -- usually requires root priviledges --
            os = self.posix
            if hasattr(os.lstat('.'), 'st_rdev'):
                import stat
                try:
                    os.mknod(self.path2 + 'test_mknod-3', 0o600 | stat.S_IFCHR,
                             0x105)
                except OSError as e:
                    skip("os.mknod() with S_IFCHR: got %r" % (e,))
                else:
                    st = os.lstat(self.path2 + 'test_mknod-3')
                    assert stat.S_ISCHR(st.st_mode)
                    assert st.st_rdev == 0x105

    if hasattr(os, 'nice') and hasattr(os, 'fork') and hasattr(os, 'waitpid'):
        def test_nice(self):
            os = self.posix
            myprio = os.nice(0)
            #
            pid = os.fork()
            if pid == 0:    # in the child
                res = os.nice(3)
                os._exit(res)
            #
            pid1, status1 = os.waitpid(pid, 0)
            assert pid1 == pid
            assert os.WIFEXITED(status1)
            expected = min(myprio + 3, 19)
            assert os.WEXITSTATUS(status1) == expected

    if sys.platform != 'win32':
        def test_symlink(self):
            posix = self.posix
            bytes_dir = self.bytes_dir
            if bytes_dir is None:
                skip("encoding not good enough")
            dest = bytes_dir + b"/file.txt"
            posix.symlink(bytes_dir + b"/somefile", dest)
            with open(dest) as f:
                data = f.read()
                assert data == "who cares?"
            #
            posix.unlink(dest)
            posix.symlink(memoryview(bytes_dir + b"/somefile"), dest)
            with open(dest) as f:
                data = f.read()
                assert data == "who cares?"

        # XXX skip test if dir_fd is unsupported
        def test_symlink_fd(self):
            posix = self.posix
            bytes_dir = self.bytes_dir
            f = posix.open(bytes_dir, posix.O_RDONLY)
            try:
                posix.symlink('somefile', 'somelink', dir_fd=f)
                assert (posix.readlink(bytes_dir + '/somelink'.encode()) ==
                        'somefile'.encode())
            finally:
                posix.close(f)
                posix.unlink(bytes_dir + '/somelink'.encode())
    else:
        def test_symlink(self):
            posix = self.posix
            raises(NotImplementedError, posix.symlink, 'a', 'b')

    if hasattr(os, 'ftruncate'):
        def test_truncate(self):
            posix = self.posix
            dest = self.path2

            def mkfile(dest, size=4):
                with open(dest, 'wb') as f:
                    f.write(b'd' * size)

            # Check invalid inputs
            mkfile(dest)
            raises(OSError, posix.truncate, dest, -1)
            with open(dest, 'rb') as f:  # f is read-only so cannot be truncated
                raises(OSError, posix.truncate, f.fileno(), 1)
            raises(TypeError, posix.truncate, dest, None)
            raises(TypeError, posix.truncate, None, None)

            # Truncate via file descriptor
            mkfile(dest)
            with open(dest, 'wb') as f:
                posix.truncate(f.fileno(), 1)
            assert 1 == posix.stat(dest).st_size

            # Truncate via filename
            mkfile(dest)
            posix.truncate(dest, 1)
            assert 1 == posix.stat(dest).st_size

            # File does not exist
            e = raises(OSError, posix.truncate, dest + '-DOESNT-EXIST', 0)
            assert e.value.filename == dest + '-DOESNT-EXIST'

    try:
        os.getlogin()
    except (AttributeError, OSError):
        pass
    else:
        def test_getlogin(self):
            assert isinstance(self.posix.getlogin(), str)
            # How else could we test that getlogin is properly
            # working?

    def test_has_kill(self):
        import os
        assert hasattr(os, 'kill')

    def test_pipe_flush(self):
        ffd, gfd = self.posix.pipe()
        f = self.os.fdopen(ffd, 'r')
        g = self.os.fdopen(gfd, 'w')
        g.write('he')
        g.flush()
        x = f.read(1)
        assert x == 'h'
        f.flush()
        x = f.read(1)
        assert x == 'e'

    def test_pipe_inheritable(self):
        fd1, fd2 = self.posix.pipe()
        assert self.posix.get_inheritable(fd1) == False
        assert self.posix.get_inheritable(fd2) == False
        self.posix.close(fd1)
        self.posix.close(fd2)

    def test_pipe2(self):
        if not hasattr(self.posix, 'pipe2'):
            skip("no pipe2")
        fd1, fd2 = self.posix.pipe2(0)
        assert self.posix.get_inheritable(fd1) == True
        assert self.posix.get_inheritable(fd2) == True
        self.posix.close(fd1)
        self.posix.close(fd2)

    def test_O_CLOEXEC(self):
        if not hasattr(self.posix, 'pipe2'):
            skip("no pipe2")
        if not hasattr(self.posix, 'O_CLOEXEC'):
            skip("no O_CLOEXEC")
        fd1, fd2 = self.posix.pipe2(self.posix.O_CLOEXEC)
        assert self.posix.get_inheritable(fd1) == False
        assert self.posix.get_inheritable(fd2) == False
        self.posix.close(fd1)
        self.posix.close(fd2)

    def test_dup2_inheritable(self):
        fd1, fd2 = self.posix.pipe()
        assert self.posix.get_inheritable(fd2) == False
        self.posix.dup2(fd1, fd2)
        assert self.posix.get_inheritable(fd2) == True
        self.posix.dup2(fd1, fd2, False)
        assert self.posix.get_inheritable(fd2) == False
        self.posix.dup2(fd1, fd2, True)
        assert self.posix.get_inheritable(fd2) == True
        self.posix.close(fd1)
        self.posix.close(fd2)

    def test_open_inheritable(self):
        os = self.posix
        fd = os.open(self.path2 + 'test_open_inheritable',
                     os.O_RDWR | os.O_CREAT, 0o666)
        assert os.get_inheritable(fd) == False
        os.close(fd)

    if sys.platform != 'win32':
        def test_sync(self):
            self.posix.sync()   # does not raise

        def test_blocking(self):
            posix = self.posix
            fd = posix.open(self.path, posix.O_RDONLY)
            assert posix.get_blocking(fd) is True
            posix.set_blocking(fd, False)
            assert posix.get_blocking(fd) is False
            posix.set_blocking(fd, True)
            assert posix.get_blocking(fd) is True
            posix.close(fd)

        def test_blocking_error(self):
            posix = self.posix
            raises(OSError, posix.get_blocking, 1234567)
            raises(OSError, posix.set_blocking, 1234567, True)

    if sys.platform != 'win32':
        def test_sendfile(self):
            import _socket, posix
            s1, s2 = _socket.socketpair()
            fd = posix.open(self.path, posix.O_RDONLY)
            res = posix.sendfile(s1.fileno(), fd, 3, 5)
            assert res == 5
            assert posix.lseek(fd, 0, 1) == 0
            data = s2.recv(10)
            expected = b'this is a test'[3:8]
            assert data == expected
            posix.close(fd)
            s2.close()
            s1.close()

        def test_filename_can_be_a_buffer(self):
            import posix, sys
            fsencoding = sys.getfilesystemencoding()
            pdir = (self.pdir + '/file1').encode(fsencoding)
            fd = posix.open(pdir, posix.O_RDONLY)
            posix.close(fd)
            fd = posix.open(memoryview(pdir), posix.O_RDONLY)
            posix.close(fd)

    if sys.platform.startswith('linux'):
        def test_sendfile_no_offset(self):
            import _socket, posix
            s1, s2 = _socket.socketpair()
            fd = posix.open(self.path, posix.O_RDONLY)
            posix.lseek(fd, 3, 0)
            res = posix.sendfile(s1.fileno(), fd, None, 5)
            assert res == 5
            assert posix.lseek(fd, 0, 1) == 8
            data = s2.recv(10)
            expected = b'this is a test'[3:8]
            assert data == expected
            posix.close(fd)
            s2.close()
            s1.close()

        def test_os_lockf(self):
            posix, os = self.posix, self.os
            fd = os.open(self.path2 + 'test_os_lockf', os.O_WRONLY | os.O_CREAT)
            try:
                os.write(fd, b'test')
                os.lseek(fd, 0, 0)
                posix.lockf(fd, posix.F_LOCK, 4)
                posix.lockf(fd, posix.F_ULOCK, 4)
            finally:
                os.close(fd)

    def test_urandom(self):
        os = self.posix
        s = os.urandom(5)
        assert isinstance(s, bytes)
        assert len(s) == 5
        for x in range(50):
            if s != os.urandom(5):
                break
        else:
            assert False, "urandom() always returns the same string"
            # Or very unlucky

    if hasattr(os, 'startfile'):
        def test_startfile(self):
            if not self.runappdirect:
                skip("should not try to import cffi at app-level")
            startfile = self.posix.startfile
            for t1 in [str, unicode]:
                for t2 in [str, unicode]:
                    e = raises(WindowsError, startfile, t1("\\"), t2("close"))
                    assert e.value.args[0] == 1155
                    assert e.value.args[1] == (
                        "No application is associated with the "
                        "specified file for this operation")
                    if len(e.value.args) > 2:
                        assert e.value.args[2] == t1("\\")
            #
            e = raises(WindowsError, startfile, "\\foo\\bar\\baz")
            assert e.value.args[0] == 2
            assert e.value.args[1] == (
                "The system cannot find the file specified")
            if len(e.value.args) > 2:
                assert e.value.args[2] == "\\foo\\bar\\baz"

    @py.test.mark.skipif("sys.platform != 'win32'")
    def test_rename(self):
        os = self.posix
        fname = self.path2 + 'rename.txt'
        with open(fname, "w") as f:
            f.write("this is a rename test")
        unicode_name = str(self.udir) + u'/test\u03be.txt'
        os.rename(fname, unicode_name)
        with open(unicode_name) as f:
            assert f.read() == 'this is a rename test'
        os.rename(unicode_name, fname)
        with open(fname) as f:
            assert f.read() == 'this is a rename test'
        os.unlink(fname)

        
    def test_device_encoding(self):
        import sys
        encoding = self.posix.device_encoding(sys.stdout.fileno())
        # just ensure it returns something reasonable
        assert encoding is None or type(encoding) is str

    if os.name == 'nt':
        def test__getfileinformation(self):
            import os
            path = os.path.join(self.pdir, 'file1')
            with open(path) as fp:
                info = self.posix._getfileinformation(fp.fileno())
            assert len(info) == 3
            assert all(isinstance(obj, int) for obj in info)

        def test__getfinalpathname(self):
            import os
            path = os.path.join(self.pdir, 'file1')
            try:
                result = self.posix._getfinalpathname(path)
            except NotImplementedError:
                skip("_getfinalpathname not supported on this platform")
            assert os.path.exists(result)

    @py.test.mark.skipif("sys.platform == 'win32'")
    def test_rtld_constants(self):
        # check presence of major RTLD_* constants
        self.posix.RTLD_LAZY
        self.posix.RTLD_NOW
        self.posix.RTLD_GLOBAL
        self.posix.RTLD_LOCAL

    def test_error_message(self):
        import sys
        e = raises(OSError, self.posix.open, 'nonexistentfile1', 0)
        assert str(e.value).endswith(": 'nonexistentfile1'")

        e = raises(OSError, self.posix.link, 'nonexistentfile1', 'bok')
        assert str(e.value).endswith(": 'nonexistentfile1' -> 'bok'")
        e = raises(OSError, self.posix.rename, 'nonexistentfile1', 'bok')
        assert str(e.value).endswith(": 'nonexistentfile1' -> 'bok'")
        e = raises(OSError, self.posix.replace, 'nonexistentfile1', 'bok')
        assert str(e.value).endswith(": 'nonexistentfile1' -> 'bok'")

        if sys.platform != 'win32':
            e = raises(OSError, self.posix.symlink, 'bok', '/nonexistentdir/boz')
            assert str(e.value).endswith(": 'bok' -> '/nonexistentdir/boz'")

    if hasattr(rposix, 'getxattr'):
        def test_xattr_simple(self):
            # Minimal testing here, lib-python has better tests.
            os = self.posix
            with open(self.path, 'wb'):
                pass
            init_names = os.listxattr(self.path)
            excinfo = raises(OSError, os.getxattr, self.path, 'user.test')
            assert excinfo.value.filename == self.path
            os.setxattr(self.path, 'user.test', b'', os.XATTR_CREATE, follow_symlinks=False)
            raises(OSError,
                os.setxattr, self.path, 'user.test', b'', os.XATTR_CREATE)
            assert os.getxattr(self.path, 'user.test') == b''
            os.setxattr(self.path, b'user.test', b'foo', os.XATTR_REPLACE)
            assert os.getxattr(self.path, 'user.test', follow_symlinks=False) == b'foo'
            assert set(os.listxattr(self.path)) == set(
                init_names + ['user.test'])
            os.removexattr(self.path, 'user.test', follow_symlinks=False)
            raises(OSError, os.getxattr, self.path, 'user.test')
            assert os.listxattr(self.path, follow_symlinks=False) == init_names

    def test_get_terminal_size(self):
        os = self.posix
        for args in [(), (1,), (0,), (42421,)]:
            try:
                w, h = os.get_terminal_size(*args)
            except (ValueError, OSError):
                continue
            assert isinstance(w, int)
            assert isinstance(h, int)


class AppTestEnvironment(object):
    def setup_class(cls):
        cls.w_path = space.wrap(str(path))

    def test_environ(self):
        import sys, os
        environ = os.environ
        if not environ:
            skip('environ not filled in for untranslated tests')
        for k, v in environ.items():
            assert type(k) is str
            assert type(v) is str
        name = next(iter(environ))
        assert environ[name] is not None
        del environ[name]
        raises(KeyError, lambda: environ[name])

    @py.test.mark.dont_track_allocations('putenv intentionally keeps strings alive')
    def test_environ_nonascii(self):
        import sys, os
        name, value = 'PYPY_TEST_日本', 'foobar日本'
        if not sys.platform == 'win32':
            fsencoding = sys.getfilesystemencoding()
            for s in name, value:
                try:
                    s.encode(fsencoding, 'surrogateescape')
                except UnicodeEncodeError:
                    skip("Requires %s.encode(sys.getfilesystemencoding(), "
                         "'surogateescape') to succeed (or win32)" % ascii(s))

        os.environ[name] = value
        assert os.environ[name] == value
        assert os.getenv(name) == value
        del os.environ[name]
        assert os.environ.get(name) is None
        assert os.getenv(name) is None

    if hasattr(__import__(os.name), "unsetenv"):
        def test_unsetenv_nonexisting(self):
            import os
            os.unsetenv("XYZABC") #does not raise
            try:
                os.environ["ABCABC"]
            except KeyError:
                pass
            else:
                raise AssertionError("did not raise KeyError")
            os.environ["ABCABC"] = "1"
            assert os.environ["ABCABC"] == "1"
            os.unsetenv("ABCABC")
            cmd = '''python -c "import os, sys; sys.exit(int('ABCABC' in os.environ))" '''
            res = os.system(cmd)
            assert res == 0


@py.test.fixture
def check_fsencoding(space, pytestconfig):
    if pytestconfig.getvalue('runappdirect'):
        fsencoding = sys.getfilesystemencoding()
    else:
        fsencoding = space.sys.filesystemencoding
    try:
        u"ą".encode(fsencoding)
    except UnicodeEncodeError:
        py.test.skip("encoding not good enough")

@py.test.mark.usefixtures('check_fsencoding')
class AppTestPosixUnicode:
    def test_stat_unicode(self):
        # test that passing unicode would not raise UnicodeDecodeError
        import os
        try:
            os.stat(u"ą")
        except OSError:
            pass

    def test_open_unicode(self):
        # Ensure passing unicode doesn't raise UnicodeEncodeError
        import os
        try:
            os.open(u"ą", os.O_WRONLY)
        except OSError:
            pass

    def test_remove_unicode(self):
        # See 2 above ;)
        import os
        try:
            os.remove(u"ą")
        except OSError:
            pass


class AppTestUnicodeFilename:
    def setup_class(cls):
        ufilename = (unicode(udir.join('test_unicode_filename_')) +
                     '\u65e5\u672c.txt') # "Japan"
        try:
            f = file(ufilename, 'w')
        except UnicodeEncodeError:
            py.test.skip("encoding not good enough")
        f.write("test")
        f.close()
        cls.space = space
        cls.w_filename = space.wrap(ufilename)
        cls.w_posix = space.appexec([], GET_POSIX)

    def test_open(self):
        fd = self.posix.open(self.filename, self.posix.O_RDONLY)
        try:
            content = self.posix.read(fd, 50)
        finally:
            self.posix.close(fd)
        assert content == b"test"


class AppTestFdVariants:
    # Tests variant functions which also accept file descriptors,
    # dir_fd and follow_symlinks.
    def test_have_functions(self):
        import os
        assert os.stat in os.supports_fd  # fstat() is supported everywhere
        if os.name != 'nt':
            assert os.chdir in os.supports_fd  # fchdir()
        else:
            assert os.chdir not in os.supports_fd
        if os.name == 'posix':
            assert os.open in os.supports_dir_fd  # openat()


class AppTestPep475Retry:
    spaceconfig = {'usemodules': USEMODULES}

    def setup_class(cls):
        if os.name != 'posix':
            skip("xxx tests are posix-only")
        if cls.runappdirect:
            skip("xxx does not work with -A")

        def fd_data_after_delay(space):
            g = os.popen("sleep 5 && echo hello", "r")
            cls._keepalive_g = g
            return space.wrap(g.fileno())

        cls.w_posix = space.appexec([], GET_POSIX)
        cls.w_fd_data_after_delay = cls.space.wrap(
            interp2app(fd_data_after_delay))

    def test_pep475_retry_read(self):
        import _signal as signal
        signalled = []

        def foo(*args):
            signalled.append("ALARM")

        signal.signal(signal.SIGALRM, foo)
        try:
            fd = self.fd_data_after_delay()
            signal.alarm(1)
            got = self.posix.read(fd, 100)
            self.posix.close(fd)
        finally:
            signal.signal(signal.SIGALRM, signal.SIG_DFL)

        assert signalled != []
        assert got.startswith(b'h')