Repository: jmcgeheeiv/pyfakefs
Pull request #451 (2018-11-08 18:29:21)

Added support for pause/resume patching of the file system modules
- can be called from the unit test, the patcher or the fs
- see #448

mrbean-bremen: @jmcgeheeiv - please have a look!

Patch:
diff --git a/CHANGES.md b/CHANGES.md
index 6bd39b7..bc3f688 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -4,6 +4,8 @@ The release versions are PyPi releases.
 ## Version 3.6 (as yet unreleased)
 
 #### New Features
+  * added support for pause/resume of patching the file system modules 
+    ([#448](../../issues/448))
   * allow to set current group ID, set current user ID and group ID as 
     `st_uid` and `st_gid` in new files ([#449](../../issues/449))
 
@@ -45,7 +47,7 @@ The release versions are PyPi releases.
     ([#423](../../issues/423)) 
   * handle `contents=None` in `create_file()` as empty contents if size not 
     set ([#424](../../issues/424))
-  * added pathlib2 support ([#408](../../issues/408)) ([#422](../../issues/422))
+  * added `pathlib2` support ([#408](../../issues/408)) ([#422](../../issues/422))
   * added support for null device ([#418](../../issues/418))
   * improved error message for "Bad file descriptor in fake filesystem"
     ([#419](../../issues/419))
@@ -55,7 +57,7 @@ The release versions are PyPi releases.
     ([#441](../../issues/441))
   * file timestamps are now updated more according to the real behavior
     ([#435](../../issues/435))
-  * fixed a problem related with patching shutil functions using zipfile 
+  * fixed a problem related to patching `shutil` functions using `zipfile` 
     ([#427](../../issues/427))  
 
 ## [Version 3.4.3](https://pypi.python.org/pypi/pyfakefs/3.4.3)
diff --git a/docs/modules.rst b/docs/modules.rst
index e37b489..620b431 100644
--- a/docs/modules.rst
+++ b/docs/modules.rst
@@ -16,7 +16,7 @@ Fake filesystem classes
         get_disk_usage, set_disk_usage,
         add_real_directory, add_real_file, add_real_paths,
         create_dir, create_file, create_symlink,
-        get_object
+        get_object, pause, resume
 
 .. autoclass:: pyfakefs.fake_filesystem.FakeFile
     :members: byte_contents, contents, set_contents,
@@ -29,10 +29,10 @@ Unittest module classes
 -----------------------
 
 .. autoclass:: pyfakefs.fake_filesystem_unittest.TestCase
-    :members: fs, setUpPyfakefs
+    :members: fs, setUpPyfakefs, pause, resume
 
 .. autoclass:: pyfakefs.fake_filesystem_unittest.Patcher
-    :members: setUp, tearDown
+    :members: setUp, tearDown, pause, resume
 
 Faked module classes
 --------------------
diff --git a/pyfakefs/fake_filesystem.py b/pyfakefs/fake_filesystem.py
index 0b93911..b3afb2b 100644
--- a/pyfakefs/fake_filesystem.py
+++ b/pyfakefs/fake_filesystem.py
@@ -838,9 +838,12 @@ class FakeFilesystem(object):
         root: The root :py:class:`FakeDirectory` entry of the file system.
         cwd: The current working directory path.
         umask: The umask used for newly created files, see `os.umask`.
+        patcher: Holds the Patcher object if created from it. Allows access
+        to the patcher object if using the pytest fs fixture.
     """
 
-    def __init__(self, path_separator=os.path.sep, total_size=None):
+    def __init__(self, path_separator=os.path.sep, total_size=None,
+                 patcher=None):
         """
         Args:
             path_separator:  optional substitute for os.path.sep
@@ -855,6 +858,7 @@ class FakeFilesystem(object):
         """
         self.path_separator = path_separator
         self.alternative_path_separator = os.path.altsep
+        self.patcher = patcher
         if path_separator != os.sep:
             self.alternative_path_separator = None
 
@@ -908,6 +912,35 @@ class FakeFilesystem(object):
         self.add_mount_point(self.root.name, total_size)
         self._add_standard_streams()
 
+    def pause(self):
+        """Pause the patching of the file system modules until `resume` is
+        called. After that call, all file system calls are executed in the
+        real file system.
+        Calling pause() twice is silently ignored.
+        Only allowed if the file system object was created by a
+        Patcher object. This is also the case for the pytest `fs` fixture.
+
+        Raises:
+            RuntimeError: if the file system was not created by a Patcher.
+        """
+        if self.patcher is None:
+            raise RuntimeError('pause() can only be called from a fake file '
+                               'system object created by a Patcher object')
+        self.patcher.pause()
+
+    def resume(self):
+        """Resume the patching of the file system modules if `pause` has
+        been called before. After that call, all file system calls are
+        executed in the fake file system.
+        Does nothing if patching is not paused.
+        Raises:
+            RuntimeError: if the file system has not been created by `Patcher`.
+        """
+        if self.patcher is None:
+            raise RuntimeError('resume() can only be called from a fake file '
+                               'system object created by a Patcher object')
+        self.patcher.resume()
+
     def line_separator(self):
         return '\r\n' if self.is_windows_fs else '\n'
 


Test patch:
diff --git a/pyfakefs/fake_filesystem_unittest.py b/pyfakefs/fake_filesystem_unittest.py
index 02a48f6..4f1d356 100644
--- a/pyfakefs/fake_filesystem_unittest.py
+++ b/pyfakefs/fake_filesystem_unittest.py
@@ -189,6 +189,23 @@ class TestCaseMixin(object):
         self._stubber.setUp()
         self.addCleanup(self._stubber.tearDown)
 
+    def pause(self):
+        """Pause the patching of the file system modules until `resume` is
+        called. After that call, all file system calls are executed in the
+        real file system.
+        Calling pause() twice is silently ignored.
+
+        """
+        self._stubber.pause()
+
+    def resume(self):
+        """Resume the patching of the file system modules if `pause` has
+        been called before. After that call, all file system calls are
+        executed in the fake file system.
+        Does nothing if patching is not paused.
+        """
+        self._stubber.resume()
+
 
 class TestCase(unittest.TestCase, TestCaseMixin):
     """Test case class that automatically replaces file-system related
@@ -371,6 +388,7 @@ class Patcher(object):
 
         # _isStale is set by tearDown(), reset by _refresh()
         self._isStale = True
+        self._patching = False
 
     def __enter__(self):
         """Context manager for usage outside of
@@ -428,7 +446,7 @@ class Patcher(object):
             self._stubs.smart_unset_all()
         self._stubs = mox3_stubout.StubOutForTesting()
 
-        self.fs = fake_filesystem.FakeFilesystem()
+        self.fs = fake_filesystem.FakeFilesystem(patcher=self)
         for name in self._fake_module_classes:
             self.fake_modules[name] = self._fake_module_classes[name](self.fs)
         self.fake_modules[PATH_MODULE] = self.fake_modules['os'].path
@@ -447,36 +465,37 @@ class Patcher(object):
         if doctester is not None:
             doctester.globs = self.replace_globs(doctester.globs)
 
-        if sys.version_info < (3, ):
-            # file() was eliminated in Python3
-            self._stubs.smart_set(builtins, 'file', self.fake_open)
-        self._stubs.smart_set(builtins, 'open', self.fake_open)
-        for name, modules in self._modules.items():
-            for module, attr in modules:
-                self._stubs.smart_set(module, name, self.fake_modules[attr])
-
-        for name, modules in self._fct_modules.items():
-            _, method, mod_name = self._fake_module_functions[name]
-            fake_module = self.fake_modules[mod_name]
-            attr = method.__get__(fake_module, fake_module.__class__)
-            for module in modules:
-                self._stubs.smart_set(module, name, attr)
-
-        self._dyn_patcher = DynamicPatcher(self)
-        sys.meta_path.insert(0, self._dyn_patcher)
-
-        for module in self.modules_to_reload:
-            if module.__name__ in sys.modules:
-                reload(module)
-
-        if not self._use_dynamic_patch:
-            self._dyn_patcher.cleanup()
-            sys.meta_path.pop(0)
+        self.start_patching()
 
         # the temp directory is assumed to exist at least in `tempfile1`,
         # so we create it here for convenience
         self.fs.create_dir(temp_dir)
 
+    def start_patching(self):
+        if not self._patching:
+            self._patching = True
+            if sys.version_info < (3,):
+                # file() was eliminated in Python3
+                self._stubs.smart_set(builtins, 'file', self.fake_open)
+            self._stubs.smart_set(builtins, 'open', self.fake_open)
+            for name, modules in self._modules.items():
+                for module, attr in modules:
+                    self._stubs.smart_set(module, name, self.fake_modules[attr])
+            for name, modules in self._fct_modules.items():
+                _, method, mod_name = self._fake_module_functions[name]
+                fake_module = self.fake_modules[mod_name]
+                attr = method.__get__(fake_module, fake_module.__class__)
+                for module in modules:
+                    self._stubs.smart_set(module, name, attr)
+            self._dyn_patcher = DynamicPatcher(self)
+            sys.meta_path.insert(0, self._dyn_patcher)
+            for module in self.modules_to_reload:
+                if module.__name__ in sys.modules:
+                    reload(module)
+            if not self._use_dynamic_patch:
+                self._dyn_patcher.cleanup()
+                sys.meta_path.pop(0)
+
     def replace_globs(self, globs_):
         globs = globs_.copy()
         if self._isStale:
@@ -488,11 +507,33 @@ class Patcher(object):
 
     def tearDown(self, doctester=None):
         """Clear the fake filesystem bindings created by `setUp()`."""
-        self._isStale = True
-        self._stubs.smart_unset_all()
-        if self._use_dynamic_patch:
-            self._dyn_patcher.cleanup()
-            sys.meta_path.pop(0)
+        self.stop_patching()
+
+    def stop_patching(self):
+        if self._patching:
+            self._isStale = True
+            self._patching = False
+            self._stubs.smart_unset_all()
+            if self._use_dynamic_patch:
+                self._dyn_patcher.cleanup()
+                sys.meta_path.pop(0)
+
+    def pause(self):
+        """Pause the patching of the file system modules until `resume` is
+        called. After that call, all file system calls are executed in the
+        real file system.
+        Calling pause() twice is silently ignored.
+
+        """
+        self.stop_patching()
+
+    def resume(self):
+        """Resume the patching of the file system modules if `pause` has
+        been called before. After that call, all file system calls are
+        executed in the fake file system.
+        Does nothing if patching is not paused.
+        """
+        self.start_patching()
 
 
 class DynamicPatcher(object):
diff --git a/pyfakefs/tests/fake_filesystem_unittest_test.py b/pyfakefs/tests/fake_filesystem_unittest_test.py
index 432e45f..1c0e45f 100644
--- a/pyfakefs/tests/fake_filesystem_unittest_test.py
+++ b/pyfakefs/tests/fake_filesystem_unittest_test.py
@@ -25,6 +25,7 @@ import os
 import multiprocessing
 import shutil
 import sys
+import tempfile
 import unittest
 from unittest import TestCase
 
@@ -280,6 +281,64 @@ class PatchModuleTest(fake_filesystem_unittest.TestCase):
             4, pyfakefs.tests.import_as_example.system_stat(file_path).st_size)
 
 
+class PauseResumeTest(TestPyfakefsUnittestBase):
+    def test_pause_resume(self):
+        fake_temp_file = tempfile.NamedTemporaryFile()
+        self.assertTrue(self.fs.exists(fake_temp_file.name))
+        self.assertTrue(os.path.exists(fake_temp_file.name))
+        self.pause()
+        self.assertTrue(self.fs.exists(fake_temp_file.name))
+        self.assertFalse(os.path.exists(fake_temp_file.name))
+        real_temp_file = tempfile.NamedTemporaryFile()
+        self.assertFalse(self.fs.exists(real_temp_file.name))
+        self.assertTrue(os.path.exists(real_temp_file.name))
+        self.resume()
+        self.assertFalse(os.path.exists(real_temp_file.name))
+        self.assertTrue(os.path.exists(fake_temp_file.name))
+
+    def test_pause_resume_fs(self):
+        fake_temp_file = tempfile.NamedTemporaryFile()
+        self.assertTrue(self.fs.exists(fake_temp_file.name))
+        self.assertTrue(os.path.exists(fake_temp_file.name))
+        # resume does nothing if not paused
+        self.fs.resume()
+        self.assertTrue(os.path.exists(fake_temp_file.name))
+        self.fs.pause()
+        self.assertTrue(self.fs.exists(fake_temp_file.name))
+        self.assertFalse(os.path.exists(fake_temp_file.name))
+        real_temp_file = tempfile.NamedTemporaryFile()
+        self.assertFalse(self.fs.exists(real_temp_file.name))
+        self.assertTrue(os.path.exists(real_temp_file.name))
+        # pause does nothing if already paused
+        self.fs.pause()
+        self.assertFalse(self.fs.exists(real_temp_file.name))
+        self.assertTrue(os.path.exists(real_temp_file.name))
+        self.fs.resume()
+        self.assertFalse(os.path.exists(real_temp_file.name))
+        self.assertTrue(os.path.exists(fake_temp_file.name))
+
+    def test_pause_resume_without_patcher(self):
+        fs = fake_filesystem.FakeFilesystem()
+        self.assertRaises(RuntimeError, fs.resume)
+
+
+class PauseResumePatcherTest(fake_filesystem_unittest.TestCase):
+    def test_pause_resume(self):
+        with Patcher() as p:
+            fake_temp_file = tempfile.NamedTemporaryFile()
+            self.assertTrue(p.fs.exists(fake_temp_file.name))
+            self.assertTrue(os.path.exists(fake_temp_file.name))
+            p.pause()
+            self.assertTrue(p.fs.exists(fake_temp_file.name))
+            self.assertFalse(os.path.exists(fake_temp_file.name))
+            real_temp_file = tempfile.NamedTemporaryFile()
+            self.assertFalse(p.fs.exists(real_temp_file.name))
+            self.assertTrue(os.path.exists(real_temp_file.name))
+            p.resume()
+            self.assertFalse(os.path.exists(real_temp_file.name))
+            self.assertTrue(os.path.exists(fake_temp_file.name))
+
+
 class TestCopyOrAddRealFile(TestPyfakefsUnittestBase):
     """Tests the `fake_filesystem_unittest.TestCase.copyRealFile()` method.
     Note that `copyRealFile()` is deprecated in favor of
diff --git a/pyfakefs/tests/import_as_example.py b/pyfakefs/tests/import_as_example.py
index 18d6a88..d58b13e 100644
--- a/pyfakefs/tests/import_as_example.py
+++ b/pyfakefs/tests/import_as_example.py
@@ -46,9 +46,9 @@ if Path:
         return Path(filepath).exists()
 
 
-def check_if_exists4(filepath, exists=my_os.path.exists):
+def check_if_exists4(filepath, file_exists=my_os.path.exists):
     # this is a similar case as in the tempfile implementation under Posix
-    return exists(filepath)
+    return file_exists(filepath)
 
 
 def check_if_exists5(filepath):
diff --git a/pyfakefs/tests/pytest/pytest_plugin_test.py b/pyfakefs/tests/pytest/pytest_plugin_test.py
index d722ce7..ce2174c 100644
--- a/pyfakefs/tests/pytest/pytest_plugin_test.py
+++ b/pyfakefs/tests/pytest/pytest_plugin_test.py
@@ -1,7 +1,24 @@
 """Tests that the pytest plugin properly provides the "fs" fixture"""
 import os
+import tempfile
 
 
 def test_fs_fixture(fs):
     fs.create_file('/var/data/xx1.txt')
     assert os.path.exists('/var/data/xx1.txt')
+
+
+def test_pause_resume(fs):
+    fake_temp_file = tempfile.NamedTemporaryFile()
+    assert fs.exists(fake_temp_file.name)
+    assert os.path.exists(fake_temp_file.name)
+    fs.pause()
+    assert fs.exists(fake_temp_file.name)
+    assert not os.path.exists(fake_temp_file.name)
+    real_temp_file = tempfile.NamedTemporaryFile()
+    assert not fs.exists(real_temp_file.name)
+    assert os.path.exists(real_temp_file.name)
+    fs.resume()
+    assert not os.path.exists(real_temp_file.name)
+    assert os.path.exists(fake_temp_file.name)
+