Repository: pytest-dev/pyfakefs (python)

Issue:
Missing mock of `os.dup`
**Describe the bug**
The `os.dup` function is not implemented by `pyfakefs`, and defers to the underlying `dup`. This can cause the pyfakefs file numbers to be given to the real `dup` syscall, leading to unexpected behavior.

**How To Reproduce**
```python
def read_with_dup(s: Path):
    s.write_text("heythere")
    fd1 = os.open(s, os.O_RDONLY)
    fd2 = os.dup(fd1)
    dat1 = os.read(fd1, 3)
    dat2 = os.read(fd2, 10)
    os.close(fd1)
    os.close(fd2)
    assert dat1 == b"hey"
    assert dat2 == b"there"

def test_realfs():
    read_with_dup(Path("file.txt"))

def test_fakefs(fs):
    read_with_dup(Path("file.txt"))
```

The fileno returned by the patched `os.open` will be given to the built-in `os.dup`, which may or may not generate an error depending on whether `fd1` is coincidentally equal to a real file descriptor. The call to `os.read(fd2)` will likely fail since the patched `os.read` will be given a fileno that wasn't created by `pyfakefs`.

**Your environment**
Please run the following in the environment where the problem happened and
paste the output.
```console
$ python -c "import platform; print(platform.platform())"
Linux-6.7.6-arch1-1-x86_64-with-glibc2.39
$ python -c "import sys; print('Python', sys.version)"
Python 3.11.7 (main, Jan 29 2024, 16:03:57) [GCC 13.2.1 20230801]
$ python -c "from pyfakefs import __version__; print('pyfakefs', __version__)"
pyfakefs 5.4.dev0
$ python -c "import pytest; print('pytest', pytest.__version__)"
pytest 7.2.0
```

---

This one may be trickier to solve since `os.dup` is also useful with non-filesystem file descriptors. i.e. it would be bad if `pyfakefs` broke `os.dup` for sockets and pipes. Distinguishing between a socket fileno and a fakefs fileno may require separating the domain of integers used by the fake os module.

Pull request:
Add support for fake os.dup, os.dup2 and os.lseek
- closes #970

#### Tasks
- [x] Unit tests added that reproduce the issue or prove feature is working
- [x] Fix or feature added
- [x] Entry to release notes added
- [x] Pre-commit CI shows no errors
- [x] Unit tests passing


Patch:
diff --git a/CHANGES.md b/CHANGES.md
index cbfb705..cf33df4 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -4,7 +4,7 @@ The released versions correspond to PyPI releases.
 ## Unreleased
 
 ### Changes
-* the handling of file permissions under Posix is should now mostly match the behavior
+* the handling of file permissions under Posix should now mostly match the behavior
   of the real filesystem, which may change the behavior of some tests
 * removed the argument `module_cleanup_mode`, that was introduced as a temporary workaround
   in the previous version - related problems shall be handled using a cleanup handler
@@ -12,6 +12,7 @@ The released versions correspond to PyPI releases.
 ### Enhancements
 * added support for `O_NOFOLLOW` and `O_DIRECTORY` flags in `os.open`
   (see [#972](../../issues/972) and [#974](../../issues/974))
+* added support for fake `os.dup`, `os.dup2` and `os.lseek` (see [#970](../../issues/970))
 
 ### Fixes
 * fixed a specific problem on reloading a pandas-related module (see [#947](../../issues/947)),
diff --git a/docs/troubleshooting.rst b/docs/troubleshooting.rst
index 6cf0b31..2a87cb9 100644
--- a/docs/troubleshooting.rst
+++ b/docs/troubleshooting.rst
@@ -223,6 +223,9 @@ is the convenience argument :ref:`allow_root_user`:
       def setUp(self):
           self.setUpPyfakefs(allow_root_user=False)
 
+``Pyfakefs`` also handles file permissions under UNIX systems while accessing files.
+If accessing files as another user and/or group, the respective group/other file
+permissions are considered.
 
 .. _usage_with_mock_open:
 
diff --git a/pyfakefs/fake_file.py b/pyfakefs/fake_file.py
index 5d7d3ee..b91127f 100644
--- a/pyfakefs/fake_file.py
+++ b/pyfakefs/fake_file.py
@@ -858,7 +858,7 @@ class FakeFileWrapper:
                     # if we get here, we have an open file descriptor
                     # without write permission, which has to be closed
                     assert self.filedes
-                    self._filesystem._close_open_file(self.filedes)
+                    self._filesystem.close_open_file(self.filedes)
                 raise
 
             if self._filesystem.is_windows_fs and self._changed:
@@ -866,7 +866,7 @@ class FakeFileWrapper:
 
         assert self.filedes is not None
         if self._closefd:
-            self._filesystem._close_open_file(self.filedes)
+            self._filesystem.close_open_file(self.filedes)
         else:
             open_files = self._filesystem.open_files[self.filedes]
             assert open_files is not None
@@ -1288,7 +1288,7 @@ class FakeDirWrapper:
     def close(self) -> None:
         """Close the directory."""
         assert self.filedes is not None
-        self._filesystem._close_open_file(self.filedes)
+        self._filesystem.close_open_file(self.filedes)
 
 
 class FakePipeWrapper:
diff --git a/pyfakefs/fake_filesystem.py b/pyfakefs/fake_filesystem.py
index 4ff0879..108c751 100644
--- a/pyfakefs/fake_filesystem.py
+++ b/pyfakefs/fake_filesystem.py
@@ -814,7 +814,7 @@ class FakeFilesystem:
         if ns is not None and len(ns) != 2:
             raise TypeError("utime: 'ns' must be a tuple of two ints")
 
-    def add_open_file(self, file_obj: AnyFileWrapper) -> int:
+    def add_open_file(self, file_obj: AnyFileWrapper, new_fd: int = -1) -> int:
         """Add file_obj to the list of open files on the filesystem.
         Used internally to manage open files.
 
@@ -822,10 +822,31 @@ class FakeFilesystem:
 
         Args:
             file_obj: File object to be added to open files list.
+            new_fd: The optional new file descriptor.
 
         Returns:
             File descriptor number for the file object.
         """
+        if new_fd >= 0:
+            size = len(self.open_files)
+            if new_fd < size:
+                open_files = self.open_files[new_fd]
+                if open_files:
+                    for f in open_files:
+                        try:
+                            f.close()
+                        except OSError:
+                            pass
+                if new_fd in self._free_fd_heap:
+                    self._free_fd_heap.remove(new_fd)
+                self.open_files[new_fd] = [file_obj]
+            else:
+                for fd in range(size, new_fd):
+                    self.open_files.append([])
+                    heapq.heappush(self._free_fd_heap, fd)
+                self.open_files.append([file_obj])
+            return new_fd
+
         if self._free_fd_heap:
             open_fd = heapq.heappop(self._free_fd_heap)
             self.open_files[open_fd] = [file_obj]
@@ -834,7 +855,7 @@ class FakeFilesystem:
         self.open_files.append([file_obj])
         return len(self.open_files) - 1
 
-    def _close_open_file(self, file_des: int) -> None:
+    def close_open_file(self, file_des: int) -> None:
         """Remove file object with given descriptor from the list
         of open files.
 
diff --git a/pyfakefs/fake_os.py b/pyfakefs/fake_os.py
index 02eeb69..565fb8a 100644
--- a/pyfakefs/fake_os.py
+++ b/pyfakefs/fake_os.py
@@ -102,12 +102,15 @@ class FakeOsModule:
             "chmod",
             "chown",
             "close",
+            "dup",
+            "dup2",
             "fstat",
             "fsync",
             "getcwd",
             "lchmod",
             "link",
             "listdir",
+            "lseek",
             "lstat",
             "makedirs",
             "mkdir",
@@ -321,6 +324,16 @@ class FakeOsModule:
         file_handle = self.filesystem.get_open_file(fd)
         file_handle.close()
 
+    def dup(self, fd: int) -> int:
+        file_handle = self.filesystem.get_open_file(fd)
+        return self.filesystem.add_open_file(file_handle)
+
+    def dup2(self, fd: int, fd2: int, inheritable: bool = True) -> int:
+        if fd == fd2:
+            return fd
+        file_handle = self.filesystem.get_open_file(fd)
+        return self.filesystem.add_open_file(file_handle, fd2)
+
     def read(self, fd: int, n: int) -> bytes:
         """Read number of bytes from a file descriptor, returns bytes read.
 
@@ -370,6 +383,13 @@ class FakeOsModule:
         file_handle.flush()
         return len(contents)
 
+    def lseek(self, fd: int, pos: int, whence: int):
+        file_handle = self.filesystem.get_open_file(fd)
+        if isinstance(file_handle, FakeFileWrapper):
+            file_handle.seek(pos, whence)
+        else:
+            raise OSError(errno.EBADF, "Bad file descriptor for fseek")
+
     def pipe(self) -> Tuple[int, int]:
         read_fd, write_fd = os.pipe()
         read_wrapper = FakePipeWrapper(self.filesystem, read_fd, False)


Test patch:
diff --git a/pyfakefs/tests/fake_os_test.py b/pyfakefs/tests/fake_os_test.py
index b89ab70..c9f69c1 100644
--- a/pyfakefs/tests/fake_os_test.py
+++ b/pyfakefs/tests/fake_os_test.py
@@ -3002,6 +3002,65 @@ class FakeOsModuleTest(FakeOsModuleTestBase):
             os.stat in os.supports_effective_ids,
         )
 
+    def test_dup(self):
+        with self.assertRaises(OSError) as cm:
+            self.os.dup(500)
+        self.assertEqual(errno.EBADF, cm.exception.errno)
+        file_path = self.make_path("test.txt")
+        self.create_file(file_path, contents="heythere")
+        fd1 = self.os.open(file_path, os.O_RDONLY)
+        fd2 = self.os.dup(fd1)
+        self.assertEqual(b"hey", self.os.read(fd1, 3))
+        self.assertEqual(b"there", self.os.read(fd1, 10))
+        self.os.close(fd1)
+        self.os.close(fd2)
+
+    def test_dup_uses_freed_fd(self):
+        file_path1 = self.make_path("foo.txt")
+        file_path2 = self.make_path("bar.txt")
+        self.create_file(file_path1, contents="foo here")
+        self.create_file(file_path2, contents="bar here")
+        fd1 = self.os.open(file_path1, os.O_RDONLY)
+        fd2 = self.os.open(file_path2, os.O_RDONLY)
+        self.os.close(fd1)
+        fd3 = self.os.dup(fd2)
+        self.assertEqual(fd1, fd3)
+        self.os.close(fd2)
+
+    def test_dup2_uses_existing_fd(self):
+        with self.assertRaises(OSError) as cm:
+            self.os.dup2(500, 501)
+        self.assertEqual(errno.EBADF, cm.exception.errno)
+
+        file_path1 = self.make_path("foo.txt")
+        file_path2 = self.make_path("bar.txt")
+        self.create_file(file_path1, contents="foo here")
+        self.create_file(file_path2, contents="bar here")
+        fd1 = self.os.open(file_path1, os.O_RDONLY)
+        fd2 = self.os.open(file_path2, os.O_RDONLY)
+        self.assertEqual(b"bar", self.os.read(fd2, 3))
+        fd2 = self.os.dup2(fd1, fd2)
+        self.assertEqual(b"foo", self.os.read(fd2, 3))
+        self.os.lseek(fd2, 0, 0)
+        self.assertEqual(b"foo", self.os.read(fd1, 3))
+        self.os.close(fd2)
+
+    def test_dup2_with_new_fd(self):
+        file_path1 = self.make_path("foo.txt")
+        file_path2 = self.make_path("bar.txt")
+        self.create_file(file_path1)
+        self.create_file(file_path2)
+        fd1 = self.os.open(file_path1, os.O_RDONLY)
+        fd2 = fd1 + 2
+        self.assertEqual(fd2, self.os.dup2(fd1, fd2))
+        fd3 = self.os.open(file_path2, os.O_RDONLY)
+        fd4 = self.os.dup(fd3)
+        self.os.close(fd4)
+        self.os.close(fd2)
+        # we have a free position before fd2 that is now filled
+        self.assertEqual(fd1 + 1, fd3)
+        self.assertEqual(fd1 + 3, fd4)
+
 
 class RealOsModuleTest(FakeOsModuleTest):
     def use_real_fs(self):