`os.mkfifo` is not supported

[legaul-cisco]

`os.mkfifo` is not supported

[LewisGaul]

[Excuse the switch to personal account]

Yes, I included the full details of how I happen to be using select to shed some light here - I agree that having to mock that explicitly wouldn't be too bad, it was more that something was being created in the real filesystem while pyfakefs was active that was most off-putting. Of course if there's a way to emulate the real-world behaviour of FIFOs in pyfakefs that'd be a bonus :)

The use of subprocess I'd agree is probably going to be common, because FIFOs can be used for inter-process communication as illustrated in the example.

Feel free to add me as reviewer if/when you get around to a PR if that would help.

[mrbean-bremen]

Thanks! This is something to consider. I'm a bit unsure about the usage, as I have never used FIFOs, but I will have a look. As far as I understand, this would be a normal (fake) file object, with a special handling for reading and writing. I'm not sure how well it would play with real code though - if I don't back it with a real pipe (as done in FakePipe), operations like select on the descriptor will fail. Also, as far as I understand, it is mostly used in the context of several processes, which would not work with pyfakefs (the file system being in memory).

Can you give a use case / example of how the code to be faked would look like? I guess you have a use case where you need this to be faked.

[legaul-cisco]

Here's a runnable example:

#!/usr/bin/env python3

import contextlib
import os
import os.path
import select
import subprocess
import tempfile
import textwrap
import time
from pathlib import Path

def read_line_from_fifo(fd: int, *, timeout: float, encoding: str = "utf-8") -> str:
    """
    Read a line from a FIFO.

    Blocks until a newline or EOF are received, or until the timeout is hit.

    :param fd:
        The FIFO file descriptor, as returned from os.open().
    :param timeout:
        The total time in seconds to attempt to read a line.
    :param encoding:
        The encoding to use when converting from bytes.
    :return:
        The line read from the FIFO.
    :raises TimeoutError:
        If the timeout is hit before a line is read.
    """
    # Read one character at a time until we reach a newline or EOF. Use
    # select to check whether any data is available to read from the FIFO,
    # which allows use of a timeout.
    line = b""
    start_time = time.monotonic()
    while True:
        remaining_time = timeout - (time.monotonic() - start_time)
        if remaining_time <= 0:
            raise TimeoutError(f"Timeout waiting for data on FIFO after {timeout}s")
        readable_fds, _, _ = select.select([fd], [], [], remaining_time)
        if readable_fds:
            char = os.read(fd, 1)
            if char in (b"\n", b""):
                break
            line += char
    return line.decode(encoding)

with contextlib.ExitStack() as stack:
    tmpdir = Path(stack.enter_context(tempfile.TemporaryDirectory()))
    fifo_path = tmpdir / "fifo"
    os.mkfifo(fifo_path)
    script_path = tmpdir / "script.sh"
    script_path.write_text(
        textwrap.dedent(
            f"""\
            #!/bin/bash
            echo "ready" >"{fifo_path}"
            sleep 100  # Do something...
            """
        )
    )
    script_path.chmod(0o755)
    child = subprocess.Popen(script_path)
    fifo_fd = os.open(fifo_path, os.O_RDONLY | os.O_NONBLOCK)
    stack.callback(os.close, fifo_fd)
    line = read_line_from_fifo(fifo_fd, timeout=10)
    if line != "ready":
        raise Exception(f"Unexpected line: {line}")

print("Success")

The child process signals to the parent when it is ready using the FIFO provided to it.

If I wanted to test code like this (perhaps structured a bit differently), it would be nice to be able to mock out the creation of the child process (e.g. with pytest-subprocess) and either be able to simulate writing to the FIFO or simply mock out read_line_from_fifo(). Either way, I'd prefer files not be created in the real filesystem when pyfakefs is active (i.e. via the os.mkfifo() call).

[mrbean-bremen]

Thank you! I see the same things I mentioned before as problematic: subprocess, where you already have a solution (I actually didn't know about pytest-subprocess, I should probably mention it in the documentation), and select, which will not work with a fake descriptor. I'm not sure about that one: you could just mock it away, but that may change the functionality of the code, or I could try to add it to the patched modules. This may or may not be a good idea, I have to check this more closely... Anyway, I won't do this in a first version.

I will handle another issue first, and will have a closer look at this one some time later, so this may take some time...