Added ability to set user ID (#432)

- can be used to test root user
- see #431

CHANGES.md

@@ -7,6 +7,7 @@ This version of pyfakefs does not support Python 3.3. Python 3.3 users shall
 keep using pyfakefs 3.4.3, or upgrade to a newer Python version.
 
 #### New Features
+  * added possibility to set root user ([#431](../../issues/431))
   * automatically patch `Path` if imported like `from pathlib import Path` 
     ([#440](../../issues/440))
   * added side_effect option to fake files ([#433](../../pull/433))

pyfakefs/fake_filesystem.py

@@ -161,6 +161,16 @@
 )
 
 NR_STD_STREAMS = 3
+USER_ID = None
+
+
+def set_uid(uid):
+    global USER_ID
+    USER_ID = uid
+
+
+def is_root():
+    return USER_ID == 0
 
 
 class FakeLargeFileIoException(Exception):
@@ -607,7 +617,8 @@ def add_entry(self, path_object):
             OSError: if the directory has no write permission (Posix only)
             OSError: if the file or directory to be added already exists
         """
-        if not self.st_mode & PERM_WRITE and not self.filesystem.is_windows_fs:
+        if (not is_root() and not self.st_mode & PERM_WRITE and
+                not self.filesystem.is_windows_fs):
             exception = IOError if IS_PY2 else OSError
             raise exception(errno.EACCES, 'Permission Denied', self.path)
 
@@ -668,7 +679,8 @@ def remove_entry(self, pathname_name, recursive=True):
             if self.filesystem.has_open_file(entry):
                 self.filesystem.raise_os_error(errno.EACCES, pathname_name)
         else:
-            if self.st_mode & (PERM_WRITE | PERM_EXE) != PERM_WRITE | PERM_EXE:
+            if (not is_root() and (self.st_mode & (PERM_WRITE | PERM_EXE) !=
+                                   PERM_WRITE | PERM_EXE)):
                 self.filesystem.raise_os_error(errno.EACCES, pathname_name)
 
         if recursive and isinstance(entry, FakeDirectory):
@@ -3681,7 +3693,7 @@ def chdir(self, target_directory):
         directory = self.filesystem.resolve(target_directory)
         # A full implementation would check permissions all the way
         # up the tree.
-        if not directory.st_mode | PERM_EXE:
+        if not is_root() and not directory.st_mode | PERM_EXE:
             self.filesystem.raise_os_error(errno.EACCES, directory)
         self.filesystem.cwd = target_directory
 
@@ -4973,9 +4985,11 @@ def call(self, file_, mode='r', buffering=-1, encoding=None,
                  not self.filesystem.is_windows_fs)):
             error_fct(errno.EEXIST, file_path)
         if file_object:
-            if ((open_modes.can_read and not file_object.st_mode & PERM_READ)
-                    or (open_modes.can_write and
-                        not file_object.st_mode & PERM_WRITE)):
+            if (not is_root() and
+                    ((open_modes.can_read and
+                      not file_object.st_mode & PERM_READ)
+                     or (open_modes.can_write and
+                         not file_object.st_mode & PERM_WRITE))):
                 error_fct(errno.EACCES, file_path)
             if open_modes.can_write:
                 if open_modes.truncate:

pyfakefs/tests/fake_filesystem_shutil_test.py

@@ -25,6 +25,7 @@
 import unittest
 
 from pyfakefs import fake_filesystem_unittest
+from pyfakefs.fake_filesystem import set_uid
 from pyfakefs.tests.test_utils import RealFsTestMixin
 
 is_windows = sys.platform == 'win32'
@@ -355,6 +356,10 @@ def use_real_fs(self):
 
 
 class FakeCopyFileTest(RealFsTestCase):
+    def tearDown(self):
+        set_uid(None)
+        super(FakeCopyFileTest, self).tearDown()
+
     def test_common_case(self):
         src_file = self.make_path('xyzzy')
         dst_file = self.make_path('xyzzy_copy')
@@ -410,6 +415,12 @@ def test_raises_if_dest_exists_and_is_not_writable(self):
         self.assertTrue(os.path.exists(src_file))
         self.assertTrue(os.path.exists(dst_file))
         self.assertRaises(IOError, shutil.copyfile, src_file, dst_file)
+        if not self.use_real_fs():
+            set_uid(0)
+            shutil.copyfile(src_file, dst_file)
+            self.assertTrue(self.os.path.exists(dst_file))
+            with self.open(dst_file) as f:
+                self.assertEqual('contents of source file', f.read())
         os.chmod(dst_file, 0o666)
 
     @unittest.skipIf(is_windows, 'Posix specific behavior')

pyfakefs/tests/fake_filesystem_test.py

@@ -24,6 +24,7 @@
 import unittest
 
 from pyfakefs import fake_filesystem
+from pyfakefs.fake_filesystem import set_uid
 from pyfakefs.tests.test_utils import DummyTime, TestCase
 
 
@@ -259,6 +260,7 @@ def setUp(self):
             'foobaz', filesystem=self.filesystem)
         self.fake_grandchild = fake_filesystem.FakeDirectory(
             'quux', filesystem=self.filesystem)
+        set_uid(None)
 
     def test_new_filesystem(self):
         self.assertEqual('/', self.filesystem.path_separator)
@@ -491,6 +493,9 @@ def test_create_file_in_read_only_directory_raises_in_posix(self):
         else:
             self.assert_raises_os_error(errno.EACCES,
                                         self.filesystem.create_file, file_path)
+        set_uid(0)
+        self.filesystem.create_file(file_path)
+        self.assertTrue(self.filesystem.exists(file_path))
 
     def test_create_file_in_read_only_directory_possible_in_windows(self):
         self.filesystem.is_windows_fs = True
@@ -1802,6 +1807,7 @@ def setUp(self):
         self.pyfakefs_path = os.path.split(
             os.path.dirname(os.path.abspath(__file__)))[0]
         self.root_path = os.path.split(self.pyfakefs_path)[0]
+        set_uid(None)
 
     def test_add_non_existing_real_file_raises(self):
         nonexisting_path = os.path.join('nonexisting', 'test.txt')

pyfakefs/tests/fake_open_test.py

@@ -27,6 +27,7 @@
 import unittest
 
 from pyfakefs import fake_filesystem
+from pyfakefs.fake_filesystem import set_uid
 from pyfakefs.tests.test_utils import DummyTime, TestCase, RealFsTestCase
 
 
@@ -39,6 +40,7 @@ class FakeFileOpenTest(FakeFileOpenTestBase):
     def setUp(self):
         super(FakeFileOpenTest, self).setUp()
         self.orig_time = time.time
+        set_uid(None)
 
     def tearDown(self):
         super(FakeFileOpenTest, self).tearDown()
@@ -452,6 +454,10 @@ def test_open_flags400(self):
         self.open(file_path, 'r').close()
         self.assert_raises_io_error(errno.EACCES, self.open, file_path, 'w')
         self.assert_raises_io_error(errno.EACCES, self.open, file_path, 'w+')
+        if not self.use_real_fs():
+            set_uid(0)
+            self.open(file_path, 'w').close()
+            self.open(file_path, 'w+').close()
 
     def test_open_flags200(self):
         # set up
@@ -462,16 +468,25 @@ def test_open_flags200(self):
         self.assertRaises(IOError, self.open, file_path, 'r')
         self.open(file_path, 'w').close()
         self.assertRaises(IOError, self.open, file_path, 'w+')
+        if not self.use_real_fs():
+            set_uid(0)
+            self.open(file_path, 'r').close()
+            self.open(file_path, 'w+').close()
 
     def test_open_flags100(self):
         # set up
         self.check_posix_only()
         file_path = self.make_path('target_file')
         self.create_with_permission(file_path, 0o100)
-        # actual tests 4
+        # actual tests
         self.assertRaises(IOError, self.open, file_path, 'r')
         self.assertRaises(IOError, self.open, file_path, 'w')
         self.assertRaises(IOError, self.open, file_path, 'w+')
+        if not self.use_real_fs():
+            set_uid(0)
+            self.open(file_path, 'r').close()
+            self.open(file_path, 'w').close()
+            self.open(file_path, 'w+').close()
 
     def test_follow_link_read(self):
         self.skip_if_symlink_not_supported()

pyfakefs/tests/fake_os_test.py

@@ -25,7 +25,7 @@
 import unittest
 
 from pyfakefs import fake_filesystem
-from pyfakefs.fake_filesystem import FakeFileOpen
+from pyfakefs.fake_filesystem import FakeFileOpen, set_uid
 from pyfakefs.extra_packages import use_scandir, use_scandir_package
 
 from pyfakefs.tests.test_utils import DummyTime, TestCase, RealFsTestCase
@@ -54,6 +54,7 @@ def setUp(self):
         super(FakeOsModuleTest, self).setUp()
         self.rwx = self.os.R_OK | self.os.W_OK | self.os.X_OK
         self.rw = self.os.R_OK | self.os.W_OK
+        set_uid(None)
 
     def test_chdir(self):
         """chdir should work on a directory."""
@@ -225,6 +226,9 @@ def test_fdopen_mode(self):
         self.os.fdopen(fileno1, 'r')
         exception = OSError if self.is_python2 else IOError
         self.assertRaises(exception, self.os.fdopen, fileno1, 'w')
+        set_uid(0)
+        self.os.fdopen(fileno1, 'w')
+        self.os.close(fileno1)
 
     def test_fstat(self):
         directory = self.make_path('xyzzy')
@@ -1603,6 +1607,10 @@ def test_mkdir_raises_if_parent_is_read_only(self):
 
         directory = self.make_path('a', 'b')
         self.assert_raises_os_error(errno.EACCES, self.os.mkdir, directory)
+        if not self.use_real_fs():
+            set_uid(0)
+            self.os.mkdir(directory)
+            self.assertTrue(self.os.path.exists(directory))
 
     def test_mkdir_with_with_symlink_parent(self):
         self.check_posix_only()
@@ -1672,7 +1680,12 @@ def test_makedirs_raises_if_access_denied(self):
         self.os.chmod(directory, 0o400)
 
         directory = self.make_path('a', 'b')
+        set_uid(None)
         self.assertRaises(Exception, self.os.makedirs, directory)
+        if not self.use_real_fs():
+            set_uid(0)
+            self.os.makedirs(directory)
+            self.assertTrue(self.os.path.exists(directory))
 
     @unittest.skipIf(sys.version_info < (3, 2),
                      'os.makedirs(exist_ok) argument new in version 3.2')
@@ -2670,7 +2683,10 @@ def test_fdopen_mode(self):
         self.os.fdopen(fileno1)
         self.os.fdopen(fileno1, 'r')
         exception = OSError if self.is_python2 else IOError
+        set_uid(None)
         self.assertRaises(exception, self.os.fdopen, fileno1, 'w')
+        set_uid(0)
+        self.os.fdopen(fileno1, 'w')
 
     def test_stat(self):
         directory = self.make_path('xyzzy')