Open legaul-cisco opened 2 months ago
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.
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).
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...
[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.
Is your feature request related to a problem? Please describe. No implementation for
os.mkfifo()
(Unix only).Describe the solution you'd like Add implementation so that real FIFOs don't get created in the filesystem.
Describe alternatives you've considered Currently requires additional mocking.
Need to add something like the following to
FakeOsModule
infake_os.py
:However, in using a regular file object in the example above, this doesn't simulate the semantics of a FIFO. Could also take inspiration from
os.pipe()
, exceptos.mkfifo()
doesn't actually open fds, it just creates the entry in the filesystem - maybe create aFakeFIFO
object in the filesystem which would then presumably need to handle open/read/write differently.