模拟类方法，仍然使用一些封装的代码运行原始方法

Question

我想通过在之前和之后运行带有附加代码的原始方法来修补方法。特别是，我在内存中的pyfakefs文件系统中运行测试，但有时我想使用真实的文件系统，因为有些包在伪文件系统上不起作用(在我的情况下是pybedtools(。

可能有一种简单的方法可以做到这一点，但经过多次尝试，我还是想不通。这可能吗？

举个例子，下面我将尝试在熊猫中修补to_csv。

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)

Answer 1

一种(不太优雅的(可能性是使用第三种方法，并通过使用旧的保存的to_csv方法来避免递归：

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)

请注意，我使用了暂停/恢复的上下文管理器——如果需要，这将允许轻松地传播补丁函数的返回值，并且不太容易出错
还要注意，默认情况下use_known_patches为True。

免责声明：
我是pyfakefs的撰稿人。

更新：我更改了答案，因为之前避免递归的尝试是错误的。

Answer 2

这里有一种方法。

df_intervals = pd.DataFrame([
['1', 10, 20],
['20', 45, 55]],
columns=['chrom', 'start', 'end'])

def fakefs_decorator(func, patcher):
""" Force a method to work on the real filesystem """
def fs_wrapper(*args, **kwargs):
patcher.pause()
out = func(*args, **kwargs)
patcher.resume()
return out
if hasattr(func, '__self__'):
def c_wrapper(_, *args, **kwargs):
return fs_wrapper(*args, **kwargs)
return classmethod(c_wrapper)
return fs_wrapper

with Patcher(allow_root_user=False, use_known_patches=True) as patcher:
fs_from_dataframe = fakefs_decorator(pybedtools.BedTool.from_dataframe, patcher)
fs_to_dataframe = fakefs_decorator(pybedtools.BedTool.to_dataframe, patcher)
fs_intersect = fakefs_decorator(pybedtools.BedTool.intersect, patcher)
@patch('pybedtools.BedTool.from_dataframe', new=fs_from_dataframe)
@patch('pybedtools.bedtool.BedTool.to_dataframe', new=fs_to_dataframe)
@patch('pybedtools.bedtool.BedTool.intersect', new=fs_intersect)
def test(df_intervals):
bed_object = pybedtools.BedTool.from_dataframe(df_intervals)
joined_bed_object = bed_object.intersect(bed_object)
df = joined_bed_object.to_dataframe()
return df
df = test(df_intervals)