I want to patch a method by running the original method with additional code before and after. In particular, I'm running tests within a pyfakefs
in-memory file system, but I want to sometime use the real file system because some packages will not work on the fake file system (pybedtools
in my case).
There is probably simple way to do this, but I can't figure it out after many, many tries. Is this possible?
Just for an example, below I'm trying to patch to_csv
in pandas.
import os
import tempfile
from unittest.mock import patch
import pandas as pd
from pyfakefs.fake_filesystem_unittest import Patcher
df_intervals = pd.DataFrame([
['1', 10, 20],
['20', 45, 55]],
columns=['chrom', 'start', 'end'])
with Patcher(use_known_patches=True) as patcher:
# As expecte writing to fake filesystem works
fname = tempfile.NamedTemporaryFile()
df_intervals.to_csv(fname.name)
assert not os.path.exists(fname.name)
assert patcher.fs.isfile(fname.name)
# But, how do I patch `to_csv` to write to the real filesystem? My failed attempts:
# Attempt 1
# TypeError: super(type, obj): obj must be an instance or subtype of type
class patched_DataFrame(pd.DataFrame):
def to_csv(self, fname):
print('Pausing fake file system')
patcher.pause()
super().to_csv(fname)
print('Resuming fake file system')
patcher.resume()
with patch.object(pd.core.generic.NDFrame, 'to_csv', new=patched_DataFrame.to_csv):
df_intervals.to_csv(fname.name)
# Attempt 2: TypeError: 'patched_DataFrame' object is not callable
with patch('pandas.core.frame.DataFrame', new_callable=patched_DataFrame):
df_intervals.to_csv(fname.name)
# Attempt 3: infinite recursion
def patched_to_csv(self, fname):
print('Pausing fake file system')
patcher.pause()
self.to_csv(fname)
print('Resuming fake file system')
patcher.resume()
with patch.object(pd.core.generic.NDFrame, 'to_csv', new=patched_to_csv):
df_intervals.to_csv(fname.name)
One (not very elegant) possibility would be to use the third approach and avoid the recursion by using the old saved to_csv
method:
from pyfakefs.fake_filesystem_unittest import Patcher, Pause
with Patcher() as patcher:
...
def patched_to_csv(self, fname):
with Pause(patcher.fs):
original_to_csv(self, fname)
original_to_csv = pd.core.generic.NDFrame.to_csv
with patch.object(pd.core.generic.NDFrame, 'to_csv', new=patched_to_csv):
df_intervals.to_csv(fname.name)
Note that I used the context manager for pause/resume - this would allow to easily propagate a return value of the patches function if needed and is less error-prone.
Also note that use_known_patches
is True
by default.
Disclaimer:
I'm a contributor to pyfakefs
.
Update: I changed the answer, because the previous attempt to avoid the recursion was wrong.