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

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)