import itertools
import os
import stat
import sys
import time

from testutils import assert_raises

assert os.name == "posix" or os.name == "nt"


def windows_stat_preserves_time64():
    import tempfile

    with tempfile.TemporaryDirectory() as directory:
        filename = os.path.join(directory, "time64")
        with open(filename, "wb"):
            pass
        timestamp = 2_208_988_800  # 2040-01-01, beyond signed 32-bit seconds.
        os.utime(filename, (timestamp, timestamp))
        with open(filename, "rb") as file:
            results = [os.stat(filename), os.fstat(file.fileno())]
        with os.scandir(directory) as entries:
            results.append(next(entries).stat())
        for result in results:
            assert result.st_mtime == timestamp
            assert result.st_mtime_ns == timestamp * 1_000_000_000


if sys.platform == "win32":
    windows_stat_preserves_time64()


def stat_rejects_negative_file_descriptors():
    import errno
    import os

    for stat in (os.stat, os.fstat):
        for fd in (-1, -5, -9, -100, -(2**31)):
            try:
                stat(fd)
            except OSError as exc:
                assert exc.errno == errno.EBADF, (stat, fd, exc)
            else:
                raise AssertionError((stat, fd, "accepted a negative file descriptor"))


# Keep the original RustPython-only coverage: CPython treats -1 as a path sentinel.
if os.name == "posix" and sys.implementation.name == "rustpython":
    stat_rejects_negative_file_descriptors()


def chmod_supports_directory_descriptors():
    import errno
    import tempfile

    assert os.chmod in os.supports_dir_fd
    assert os.chmod in os.supports_fd
    with tempfile.TemporaryDirectory() as directory:
        filename = os.path.join(directory, "file")
        with open(filename, "wb"):
            pass
        dir_fd = os.open(directory, os.O_RDONLY)
        try:
            os.chmod("file", 0o640, dir_fd=dir_fd)
            assert stat.S_IMODE(os.stat(filename).st_mode) == 0o640

            # Absolute paths ignore even an invalid directory descriptor.
            os.chmod(path=filename, mode=0o600, dir_fd=-1)
            assert stat.S_IMODE(os.stat(filename).st_mode) == 0o600

            class DirectoryDescriptor:
                def __index__(self):
                    return dir_fd

            os.chmod(b"file", 0o640, dir_fd=DirectoryDescriptor())
            assert stat.S_IMODE(os.stat(filename).st_mode) == 0o640
            with assert_raises(OSError) as error:
                os.chmod("file", 0o600, dir_fd=-1)
            assert error.exception.errno == errno.EBADF
            assert error.exception.filename == "file"
            for invalid, message in (
                (-(2**31) - 1, "fd is less than minimum"),
                (2**31, "fd is greater than maximum"),
            ):
                with assert_raises(OverflowError) as error:
                    os.chmod(filename, 0o600, dir_fd=invalid)
                assert str(error.exception) == message

            link = os.path.join(directory, "link")
            os.symlink("file", link)
            os.chmod("link", 0o600, dir_fd=dir_fd)
            assert stat.S_IMODE(os.stat(filename).st_mode) == 0o600
            for path, options, error_type, message in (
                (link, {}, NotImplementedError, "follow_symlinks unavailable"),
                (
                    "link",
                    {"dir_fd": dir_fd},
                    ValueError,
                    "cannot use dir_fd and follow_symlinks together",
                ),
            ):
                try:
                    os.chmod(path, 0o640, follow_symlinks=False, **options)
                except error_type as error:
                    assert message in str(error)
                assert stat.S_IMODE(os.stat(filename).st_mode) == 0o600
        finally:
            os.close(dir_fd)


if os.name == "posix" and hasattr(os, "fchmod"):
    chmod_supports_directory_descriptors()


fd = os.open("README.md", os.O_RDONLY)
assert fd > 0

os.close(fd)
assert_raises(OSError, lambda: os.read(fd, 10))
assert_raises(FileNotFoundError, lambda: os.open("DOES_NOT_EXIST", os.O_RDONLY))
assert_raises(FileNotFoundError, lambda: os.open("DOES_NOT_EXIST", os.O_WRONLY))
assert_raises(
    FileNotFoundError, lambda: os.rename("DOES_NOT_EXIST", "DOES_NOT_EXIST 2")
)

# sendfile only supports in_fd as non-socket on linux and solaris
if hasattr(os, "sendfile") and sys.platform.startswith("linux"):
    src_fd = os.open("README.md", os.O_RDONLY)
    dest_fd = os.open("destination.md", os.O_RDWR | os.O_CREAT)
    src_len = os.stat("README.md").st_size

    bytes_sent = os.sendfile(dest_fd, src_fd, 0, src_len)
    assert src_len == bytes_sent

    os.lseek(dest_fd, 0, 0)
    assert os.read(src_fd, src_len) == os.read(dest_fd, bytes_sent)
    os.close(src_fd)
    os.close(dest_fd)
    os.remove("destination.md")

try:
    os.open("DOES_NOT_EXIST", 0)
except OSError as err:
    assert err.errno == 2

assert os.O_RDONLY == 0
assert os.O_WRONLY == 1
assert os.O_RDWR == 2

ENV_KEY = "TEST_ENV_VAR"
ENV_VALUE = "value"

assert os.getenv(ENV_KEY) is None
assert ENV_KEY not in os.environ
assert os.getenv(ENV_KEY, 5) == 5
os.environ[ENV_KEY] = ENV_VALUE
assert ENV_KEY in os.environ
assert os.getenv(ENV_KEY) == ENV_VALUE
del os.environ[ENV_KEY]
assert ENV_KEY not in os.environ
assert os.getenv(ENV_KEY) is None

if os.name == "posix":
    os.putenv(ENV_KEY, ENV_VALUE)
    os.unsetenv(ENV_KEY)
    assert os.getenv(ENV_KEY) is None

assert os.curdir == "."
assert os.pardir == ".."
assert os.extsep == "."

if os.name == "nt":
    assert os.sep == "\\"
    assert os.linesep == "\r\n"
    assert os.altsep == "/"
    assert os.pathsep == ";"
else:
    assert os.sep == "/"
    assert os.linesep == "\n"
    assert os.altsep is None
    assert os.pathsep == ":"

assert os.fspath("Testing") == "Testing"
assert os.fspath(b"Testing") == b"Testing"
assert_raises(TypeError, lambda: os.fspath([1, 2, 3]))


class TestWithTempDir:
    def __enter__(self):
        if os.name == "nt":
            base_folder = os.environ["TEMP"]
        else:
            base_folder = "/tmp"

        name = os.path.join(base_folder, "rustpython_test_os_" + str(int(time.time())))

        while os.path.isdir(name):
            name = name + "_"

        os.mkdir(name)
        self.name = name
        return name

    def __exit__(self, exc_type, exc_val, exc_tb):
        pass


class TestWithTempCurrentDir:
    def __enter__(self):
        self.prev_cwd = os.getcwd()

    def __exit__(self, exc_type, exc_val, exc_tb):
        os.chdir(self.prev_cwd)


FILE_NAME = "test1"
FILE_NAME2 = "test2"
FILE_NAME3 = "test3"
SYMLINK_FILE = "symlink"
SYMLINK_FOLDER = "symlink1"
FOLDER = "dir1"
CONTENT = b"testing"
CONTENT2 = b"rustpython"
CONTENT3 = b"BOYA"

with TestWithTempDir() as tmpdir:
    fname = os.path.join(tmpdir, FILE_NAME)
    fd = os.open(fname, os.O_WRONLY | os.O_CREAT | os.O_EXCL)
    assert os.write(fd, CONTENT2) == len(CONTENT2)
    os.close(fd)

    # A valid negative access time must reach the invalid modification time.
    for atime in (-1, -0.25):
        assert_raises(TypeError, os.utime, fname, (atime, object()))
    assert_raises(TypeError, os.utime, fname, ns=(-1, object()))

    fd = os.open(fname, os.O_WRONLY | os.O_APPEND)
    assert os.write(fd, CONTENT3) == len(CONTENT3)
    os.close(fd)

    assert_raises(
        FileExistsError, lambda: os.open(fname, os.O_WRONLY | os.O_CREAT | os.O_EXCL)
    )

    fd = os.open(fname, os.O_RDONLY)
    assert os.read(fd, len(CONTENT2)) == CONTENT2
    assert os.read(fd, len(CONTENT3)) == CONTENT3
    os.close(fd)

    fname3 = os.path.join(tmpdir, FILE_NAME3)
    os.rename(fname, fname3)
    assert os.path.exists(fname) is False
    assert os.path.exists(fname3) is True

    fd = os.open(fname3, 0)
    assert os.read(fd, len(CONTENT2) + len(CONTENT3)) == CONTENT2 + CONTENT3
    os.close(fd)

    assert not os.isatty(fd)

    # TODO: get os.lseek working on windows
    if os.name != "nt":
        fd = os.open(fname3, 0)
        assert os.read(fd, len(CONTENT2)) == CONTENT2
        assert os.read(fd, len(CONTENT3)) == CONTENT3
        os.lseek(fd, len(CONTENT2), os.SEEK_SET)
        assert os.read(fd, len(CONTENT3)) == CONTENT3
        os.close(fd)

    os.rename(fname3, fname)
    assert os.path.exists(fname3) is False
    assert os.path.exists(fname) is True

    # wait a little bit to ensure that the file times aren't the same
    time.sleep(0.1)

    fname2 = os.path.join(tmpdir, FILE_NAME2)
    with open(fname2, "wb"):
        pass
    folder = os.path.join(tmpdir, FOLDER)
    os.mkdir(folder)

    symlink_file = os.path.join(tmpdir, SYMLINK_FILE)
    os.symlink(fname, symlink_file)
    symlink_folder = os.path.join(tmpdir, SYMLINK_FOLDER)
    os.symlink(folder, symlink_folder)

    names = set()
    paths = set()
    dirs = set()
    dirs_no_symlink = set()
    files = set()
    files_no_symlink = set()
    symlinks = set()
    for dir_entry in os.scandir(tmpdir):
        names.add(dir_entry.name)
        paths.add(dir_entry.path)
        if dir_entry.is_dir():
            assert stat.S_ISDIR(dir_entry.stat().st_mode) is True
            dirs.add(dir_entry.name)
        if dir_entry.is_dir(follow_symlinks=False):
            assert stat.S_ISDIR(dir_entry.stat().st_mode) is True
            dirs_no_symlink.add(dir_entry.name)
        if dir_entry.is_file():
            files.add(dir_entry.name)
            assert stat.S_ISREG(dir_entry.stat().st_mode) is True
        if dir_entry.is_file(follow_symlinks=False):
            files_no_symlink.add(dir_entry.name)
            assert stat.S_ISREG(dir_entry.stat().st_mode) is True
        if dir_entry.is_symlink():
            symlinks.add(dir_entry.name)

    assert names == set([FILE_NAME, FILE_NAME2, FOLDER, SYMLINK_FILE, SYMLINK_FOLDER])
    assert paths == set([fname, fname2, folder, symlink_file, symlink_folder])
    assert dirs == set([FOLDER, SYMLINK_FOLDER])
    assert dirs_no_symlink == set([FOLDER])
    assert files == set([FILE_NAME, FILE_NAME2, SYMLINK_FILE])
    assert files_no_symlink == set([FILE_NAME, FILE_NAME2])
    assert symlinks == set([SYMLINK_FILE, SYMLINK_FOLDER])

    # Stat
    stat_res = os.stat(fname)
    print(stat_res.st_mode)
    assert stat.S_ISREG(stat_res.st_mode) is True
    print(stat_res.st_ino)
    print(stat_res.st_dev)
    print(stat_res.st_nlink)
    print(stat_res.st_uid)
    print(stat_res.st_gid)
    print(stat_res.st_size)
    assert stat_res.st_size == len(CONTENT2) + len(CONTENT3)
    print(stat_res.st_atime)
    print(stat_res.st_ctime)
    print(stat_res.st_mtime)
    # test that it all of these times are greater than the 10 May 2019,
    # when this test was written
    assert stat_res.st_atime > 1557500000
    assert stat_res.st_ctime > 1557500000
    assert stat_res.st_mtime > 1557500000

    bytes_stats_res = os.stat(fname.encode())

    stat_file2 = os.stat(fname2)
    print(stat_file2.st_ctime)
    assert stat_file2.st_ctime > stat_res.st_ctime

    # wait a little bit to ensures that the access/modify time will change
    time.sleep(0.1)

    old_atime = stat_res.st_atime
    old_mtime = stat_res.st_mtime

    fd = os.open(fname, os.O_RDWR)
    os.write(fd, CONTENT)
    os.fsync(fd)

    # wait a little bit to ensures that the access/modify time is different
    time.sleep(0.1)

    os.read(fd, 1)
    os.fsync(fd)
    os.close(fd)

    # retrieve update file stats
    stat_res = os.stat(fname)
    print(stat_res.st_atime)
    print(stat_res.st_ctime)
    print(stat_res.st_mtime)
    if os.name != "nt":
        # access time on windows has a resolution ranging from 1 hour to 1 day
        # https://docs.microsoft.com/en-gb/windows/desktop/api/minwinbase/ns-minwinbase-filetime
        assert stat_res.st_atime > old_atime, "Access time should be update"
        assert stat_res.st_atime > stat_res.st_mtime
    assert stat_res.st_mtime > old_mtime, "Modified time should be update"

    # stat default is follow_symlink=True
    os.stat(fname).st_ino == os.stat(symlink_file).st_ino
    os.stat(fname).st_mode == os.stat(symlink_file).st_mode

    os.stat(fname, follow_symlinks=False).st_ino == os.stat(
        symlink_file, follow_symlinks=False
    ).st_ino
    os.stat(fname, follow_symlinks=False).st_mode == os.stat(
        symlink_file, follow_symlinks=False
    ).st_mode

    # os.chmod
    if os.name != "nt":
        os.chmod(fname, 0o666)
        assert oct(os.stat(fname).st_mode) == "0o100666"

    # os.chown
    if os.name != "nt":
        # setup
        root_in_posix = False
        if hasattr(os, "geteuid"):
            root_in_posix = os.geteuid() == 0
        try:
            import pwd

            all_users = [u.pw_uid for u in pwd.getpwall()]
        except (ImportError, AttributeError):
            all_users = []

        fname1 = os.path.join(tmpdir, FILE_NAME)
        fname2 = os.path.join(tmpdir, FILE_NAME2)
        fd = os.open(fname2, os.O_RDONLY)

        # test chown without root permissions
        if not root_in_posix and len(all_users) > 1:
            uid_1, uid_2 = all_users[:2]
            gid = os.stat(fname1).st_gid
            assert_raises(PermissionError, lambda: os.chown(fname1, uid_1, gid))
            assert_raises(PermissionError, lambda: os.chown(fname1, uid_2, gid))

        # test chown with root perm and file name
        if root_in_posix and len(all_users) > 1:
            uid_1, uid_2 = all_users[:2]
            gid = os.stat(fname1).st_gid
            os.chown(fname1, uid_1, gid)
            uid = os.stat(fname1).st_uid
            assert uid == uid_1
            os.chown(fname1, uid_2, gid)
            uid = os.stat(fname1).st_uid
            assert uid == uid_2

        # test chown with root perm and file descriptor
        if root_in_posix and len(all_users) > 1:
            uid_1, uid_2 = all_users[:2]
            gid = os.stat(fd).st_gid
            os.chown(fd, uid_1, gid)
            uid = os.stat(fd).st_uid
            assert uid == uid_1
            os.chown(fd, uid_2, gid)
            uid = os.stat(fd).st_uid
            assert uid == uid_2

        # test gid change
        if hasattr(os, "getgroups"):
            groups = os.getgroups()
            if len(groups) > 1:
                gid_1, gid_2 = groups[:2]
                uid = os.stat(fname1).st_uid

                os.chown(fname1, uid, gid_1)
                gid = os.stat(fname1).st_gid
                assert gid == gid_1

                os.chown(fname1, uid, gid_2)
                gid = os.stat(fname1).st_gid
                assert gid == gid_2

        # teardown
        os.close(fd)

    # os.path
    assert os.path.exists(fname) is True
    assert os.path.exists("NO_SUCH_FILE") is False
    assert os.path.isfile(fname) is True
    assert os.path.isdir(folder) is True
    assert os.path.isfile(folder) is False
    assert os.path.isdir(fname) is False

    assert os.path.basename(fname) == FILE_NAME
    assert os.path.dirname(fname) == tmpdir

    with TestWithTempCurrentDir():
        os.chdir(tmpdir)
        assert os.path.realpath(os.getcwd()) == os.path.realpath(tmpdir)
        assert os.path.exists(FILE_NAME)

# supports
assert isinstance(os.supports_fd, set)
assert isinstance(os.supports_dir_fd, set)
assert isinstance(os.supports_follow_symlinks, set)

# get pid
assert isinstance(os.getpid(), int)

# unix
if "win" not in sys.platform:
    assert isinstance(os.getegid(), int)
    assert isinstance(os.getgid(), int)
    assert isinstance(os.getsid(os.getpid()), int)
    assert isinstance(os.getuid(), int)
    assert isinstance(os.geteuid(), int)
    assert isinstance(os.getppid(), int)
    assert isinstance(os.getpgid(os.getpid()), int)

    if os.getuid() != 0:
        assert_raises(PermissionError, lambda: os.setgid(42))
        assert_raises(PermissionError, lambda: os.setegid(42))
        assert_raises(PermissionError, lambda: os.setpgid(os.getpid(), 42))
        assert_raises(PermissionError, lambda: os.setuid(42))
        assert_raises(PermissionError, lambda: os.seteuid(42))
        assert_raises(PermissionError, lambda: os.setreuid(42, 42))
        assert_raises(PermissionError, lambda: os.setresuid(42, 42, 42))

    # pty
    a, b = os.openpty()
    assert isinstance(a, int)
    assert isinstance(b, int)
    assert isinstance(os.ttyname(b), str)
    assert_raises(OSError, lambda: os.ttyname(9999))
    os.close(b)
    os.close(a)

    # os.get_blocking, os.set_blocking
    # TODO: windows support should be added for below functions
    # os.pipe,
    # os.set_inheritable, os.get_inheritable,
    rfd, wfd = os.pipe()
    try:
        os.write(wfd, CONTENT2)
        assert os.read(rfd, len(CONTENT2)) == CONTENT2
        assert not os.get_inheritable(rfd)
        assert not os.get_inheritable(wfd)
        os.set_inheritable(rfd, True)
        os.set_inheritable(wfd, True)
        assert os.get_inheritable(rfd)
        assert os.get_inheritable(wfd)
        os.set_inheritable(rfd, True)
        os.set_inheritable(wfd, True)
        os.set_inheritable(rfd, True)
        os.set_inheritable(wfd, True)
        assert os.get_inheritable(rfd)
        assert os.get_inheritable(wfd)

        assert os.get_blocking(rfd)
        assert os.get_blocking(wfd)
        os.set_blocking(rfd, False)
        os.set_blocking(wfd, False)
        assert not os.get_blocking(rfd)
        assert not os.get_blocking(wfd)
        os.set_blocking(rfd, True)
        os.set_blocking(wfd, True)
        os.set_blocking(rfd, True)
        os.set_blocking(wfd, True)
        assert os.get_blocking(rfd)
        assert os.get_blocking(wfd)
    finally:
        os.close(rfd)
        os.close(wfd)

# os.pipe2
if sys.platform.startswith("linux") or sys.platform.startswith("freebsd"):
    rfd, wfd = os.pipe2(0)
    try:
        os.write(wfd, CONTENT2)
        assert os.read(rfd, len(CONTENT2)) == CONTENT2
        assert os.get_inheritable(rfd)
        assert os.get_inheritable(wfd)
        assert os.get_blocking(rfd)
        assert os.get_blocking(wfd)
    finally:
        os.close(rfd)
        os.close(wfd)
    rfd, wfd = os.pipe2(os.O_CLOEXEC | os.O_NONBLOCK)
    try:
        os.write(wfd, CONTENT2)
        assert os.read(rfd, len(CONTENT2)) == CONTENT2
        assert not os.get_inheritable(rfd)
        assert not os.get_inheritable(wfd)
        assert not os.get_blocking(rfd)
        assert not os.get_blocking(wfd)
    finally:
        os.close(rfd)
        os.close(wfd)

with TestWithTempDir() as tmpdir:
    for i in range(0, 4):
        file_name = os.path.join(tmpdir, "file" + str(i))
        with open(file_name, "w") as f:
            f.write("test")

    expected_files = ["file0", "file1", "file2", "file3"]

    dir_iter = os.scandir(tmpdir)
    collected_files = [dir_entry.name for dir_entry in dir_iter]

    assert set(collected_files) == set(expected_files)

    with assert_raises(StopIteration):
        next(dir_iter)

    dir_iter.close()

    expected_files_bytes = [
        (file.encode(), os.path.join(tmpdir, file).encode()) for file in expected_files
    ]

    dir_iter_bytes = os.scandir(tmpdir.encode())
    collected_files_bytes = [
        (dir_entry.name, dir_entry.path) for dir_entry in dir_iter_bytes
    ]

    assert set(collected_files_bytes) == set(expected_files_bytes)

    dir_iter_bytes.close()

    collected_files = os.listdir(tmpdir)
    assert set(collected_files) == set(expected_files)

    collected_files = os.listdir(tmpdir.encode())
    assert set(collected_files) == set([file.encode() for file in expected_files])

    with TestWithTempCurrentDir():
        os.chdir(tmpdir)
        with os.scandir() as dir_iter:
            collected_files = [dir_entry.name for dir_entry in dir_iter]
            assert set(collected_files) == set(expected_files)

# system()
if os.name in ("posix", "nt"):
    assert os.system("echo test") == 0
    assert os.system("&") != 0

    for arg in [None, 1, 1.0, TabError]:
        assert_raises(TypeError, os.system, arg)

# Testing for os.pathconf_names
if not sys.platform.startswith("win"):
    assert len(os.pathconf_names) > 0
    assert "PC_NAME_MAX" in os.pathconf_names
    for option, index in os.pathconf_names.items():
        if sys.platform == "darwin":
            # TODO: check why it fails
            if option in ["PC_MAX_CANON", "PC_MAX_INPUT", "PC_VDISABLE"]:
                continue
        assert os.pathconf("/", index) == os.pathconf("/", option)

# os.access - test with empty path and nonexistent files
assert os.access("", os.F_OK) is False
assert os.access("", os.R_OK) is False
assert os.access("", os.W_OK) is False
assert os.access("", os.X_OK) is False
assert os.access("nonexistent_file_12345", os.F_OK) is False
assert os.access("nonexistent_file_12345", os.W_OK) is False
assert os.access("README.md", os.F_OK) is True
assert os.access("README.md", os.R_OK) is True

# argv and the group list are sequences; an arbitrary iterable must be rejected
# rather than drained.
if hasattr(os, "posix_spawn"):
    with assert_raises(TypeError):
        os.posix_spawn("/bin/true", map(str, itertools.count()), os.environ)
if hasattr(os, "setgroups"):
    with assert_raises(TypeError):
        os.setgroups(itertools.count())

# The optional second argument fills the fields past the visible ones, and the
# getters must not index past what __new__ stored.
assert os.stat_result(tuple(range(10))).st_atime == 7
assert os.stat_result(tuple(range(10)), {"st_atime": 1.5}).st_atime == 1.5
with assert_raises(TypeError):
    os.stat_result(tuple(range(10)), ["st_atime"])


def mkfifo_paths_and_directory_descriptors():
    import errno
    import inspect
    import tempfile
    import warnings

    assert str(inspect.signature(os.mkfifo)) == "(path, mode=438, *, dir_fd=None)"
    assert os.mkfifo not in os.supports_fd
    assert os.mkfifo not in os.supports_follow_symlinks
    if sys.platform.startswith("linux"):
        assert os.mkfifo in os.supports_dir_fd

    class Path:
        def __init__(self, value):
            self.value = value

        def __fspath__(self):
            return self.value

    class Index:
        def __init__(self, value):
            self.value = value

        def __index__(self):
            return self.value

    class BadIndex:
        def __index__(self):
            raise RuntimeError("mkfifo index marker")

    # Never open a FIFO: creating, inspecting and unlinking it cannot wait for
    # another process to connect. All paths belong to this temporary directory.
    with tempfile.TemporaryDirectory(prefix="rustpython-mkfifo-") as directory:
        path = os.path.join(directory, "fifo")
        for value in (path, os.fsencode(path), Path(path), Path(os.fsencode(path))):
            assert os.mkfifo(path=value, mode=0) is None
            info = os.stat(path)
            assert stat.S_ISFIFO(info.st_mode)
            assert stat.S_IMODE(info.st_mode) == 0
            with assert_raises(FileExistsError) as error:
                os.mkfifo(value)
            assert error.exception.errno == errno.EEXIST
            assert error.exception.filename is None
            os.unlink(path)

        assert os.mkfifo(path, dir_fd=None) is None
        assert stat.S_ISFIFO(os.stat(path).st_mode)
        assert stat.S_IMODE(os.stat(path).st_mode) & ~0o666 == 0
        os.unlink(path)

        for value in (None, 1.5):
            assert_raises(TypeError, os.mkfifo, path, value)
        for value in (2**31, -(2**31) - 1, 2**100, -(2**100)):
            with assert_raises(OverflowError) as error:
                os.mkfifo(path, value)
            assert str(error.exception) == "Python int too large to convert to C int"
            with assert_raises(OverflowError) as error:
                os.mkfifo(path, dir_fd=value)
            expected = (
                "fd is greater than maximum" if value > 0 else "fd is less than minimum"
            )
            assert str(error.exception) == expected
        for value in (1.5, "fd"):
            assert_raises(TypeError, os.mkfifo, path, dir_fd=value)
        for value in (None, 1):
            with assert_raises(TypeError) as error:
                os.mkfifo(value)
            expected = "NoneType" if value is None else "int"
            assert str(error.exception) == (
                "mkfifo: path should be string, bytes or os.PathLike, not " + expected
            )
        assert_raises(TypeError, os.mkfifo, Path(None))
        for value in (path + "\0", os.fsencode(path) + b"\0"):
            with assert_raises(ValueError) as error:
                os.mkfifo(value)
            assert str(error.exception) == "mkfifo: embedded null character in path"
        assert_raises(TypeError, os.mkfifo, path, 0o600, None)
        for options in ({"mode": BadIndex()}, {"dir_fd": BadIndex()}):
            with assert_raises(RuntimeError) as error:
                os.mkfifo(path, **options)
            assert str(error.exception) == "mkfifo index marker"
        with assert_raises(FileNotFoundError) as error:
            os.mkfifo(os.path.join(directory, "missing", "fifo"))
        assert error.exception.errno == errno.ENOENT
        assert error.exception.filename is None

        directory_fd = os.open(directory, os.O_RDONLY)
        try:
            if os.mkfifo in os.supports_dir_fd:
                assert (
                    os.mkfifo(Path(b"fifo"), Index(0o600), dir_fd=Index(directory_fd))
                    is None
                )
                assert stat.S_ISFIFO(os.stat(path).st_mode)
                assert stat.S_IMODE(os.stat(path).st_mode) & ~0o600 == 0
                os.unlink(path)
                for value in (-1, -(2**31)):
                    assert os.mkfifo(path, dir_fd=value) is None
                    assert stat.S_ISFIFO(os.stat(path).st_mode)
                    os.unlink(path)
                with assert_raises(OSError) as error:
                    os.mkfifo("fifo", dir_fd=-1)
                assert error.exception.errno == errno.EBADF
                assert error.exception.filename is None
                with warnings.catch_warnings(record=True) as records:
                    warnings.simplefilter("always")
                    assert os.mkfifo(path, dir_fd=True) is None
                assert len(records) == 1
                assert records[0].category is RuntimeWarning
                assert str(records[0].message) == "bool is used as a file descriptor"
                os.unlink(path)
            else:
                with assert_raises(NotImplementedError) as error:
                    os.mkfifo(path, dir_fd=directory_fd)
                assert str(error.exception) == "dir_fd unavailable on this platform"
        finally:
            os.close(directory_fd)


if sys.platform.startswith("linux"):
    assert hasattr(os, "mkfifo")
if hasattr(os, "mkfifo"):
    mkfifo_paths_and_directory_descriptors()


import errno


def nonblocking_windows_pipe():
    reader, writer = os.pipe()
    try:
        for fd in (reader, writer):
            assert os.get_blocking(fd)
            os.set_blocking(fd, False)
            assert not os.get_blocking(fd)
            os.set_blocking(fd, True)
            assert os.get_blocking(fd)
            os.set_blocking(fd, False)

        assert os.read(reader, 0) == b""
        assert os.write(writer, b"") == 0
        try:
            os.read(reader, 1)
        except BlockingIOError as error:
            assert error.errno == errno.EAGAIN
        else:
            raise AssertionError("empty nonblocking pipe did not raise")
    finally:
        os.close(reader)
        os.close(writer)


if os.name == "nt":
    nonblocking_windows_pipe()