模拟 class 方法,仍然 运行 带有一些包装代码的原始方法
Mocking class methods, still running the original method with some wrapped code
我想通过 运行 原始方法在前后添加代码来修补一个方法。特别是,我在 pyfakefs
内存文件系统中进行了 运行 测试,但有时我想使用真实的文件系统,因为某些软件包无法在假文件系统上运行 (pybedtools
在我的例子中)。
可能有一个简单的方法可以做到这一点,但我尝试了很多很多次后还是无法弄清楚。这可能吗?
举个例子,下面我试图在 pandas 中修补 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)
一种(不是很优雅)的可能性是使用第三种方法并通过使用旧保存的 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)
请注意,我为 pause/resume 使用了上下文管理器 - 这将允许在需要时轻松传播补丁函数的 return 值并且不易出错。
另请注意 use_known_patches
默认为 True
。
免责声明:
我是 pyfakefs
.
的贡献者
更新:我修改了答案,因为之前试图避免递归的尝试是错误的。
这是一种方法。
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)
我想通过 运行 原始方法在前后添加代码来修补一个方法。特别是,我在 pyfakefs
内存文件系统中进行了 运行 测试,但有时我想使用真实的文件系统,因为某些软件包无法在假文件系统上运行 (pybedtools
在我的例子中)。
可能有一个简单的方法可以做到这一点,但我尝试了很多很多次后还是无法弄清楚。这可能吗?
举个例子,下面我试图在 pandas 中修补 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)
一种(不是很优雅)的可能性是使用第三种方法并通过使用旧保存的 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)
请注意,我为 pause/resume 使用了上下文管理器 - 这将允许在需要时轻松传播补丁函数的 return 值并且不易出错。
另请注意 use_known_patches
默认为 True
。
免责声明:
我是 pyfakefs
.
更新:我修改了答案,因为之前试图避免递归的尝试是错误的。
这是一种方法。
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)