模拟和side_effect替换 - 保留对原始类及其属性的访问



我想在类的特定实例上模拟方法_subprocess。特别是当任务作为命令触发pip freeze时(在这种情况下,其taskname冻结(。

class Command(object):
    def __init__(self, mgr, taskname, config):
        self.mgr = mgr
        self.taskname = taskname
        self.config = config
        self.append = self.config.get("append", False)
        self.stderr = ""
    def _subprocess(self, cmd, fnp_o, self_=None):
        try:
            mode = "a" if self.append else "w"
            fnp_stderr = self.mgr._get_fnp("log")
            with open(fnp_stderr, "a") as ferr:
                ferr.write("cmd: %snstderr begin:n" % (cmd))
                with open(fnp_o, mode) as fo:
                    proc = subprocess.check_call(
                        cmd.split(),
                        stdout=fo,
                        stderr=ferr,
                        cwd=self.mgr.workdir,
                        encoding="utf-8",
                    )
                ferr.write("stderr endnn")
        except (Exception,) as e:
            if cpdb(): pdb.set_trace()
            raise        

这是测试方法:

def fake_subprocess(self, cmd, fnp_o, self_):
    try:
        raise NotImplementedError("fake_subprocess(%s)" % (locals()))
    except (Exception,) as e:
        pdb.set_trace()
        raise
def test_001_scan(self):
    try:
        with patch.object(Command, '_subprocess', side_effect = self.fake_subprocess) as mock_method:
            options = self.get_options()
            self.mgr = Main(options)
            self.mgr.process()
    except (Exception,) as e:
        pdb.set_trace()
        raise

我的问题是双重的。

首先,fake_subprocess中的self是指UnitTest对象,而不是Command对象。 我对 self_ 参数的使用可以解决这个问题。

其次,在大多数情况下,除了pip freeze我想运行原始子流程,而不是假子流程。

现在,我可以通过保留对Command._subprocess的额外引用并使用self_来解决这个问题。

但是有没有更优雅的方式呢? 说到unittest.Mock很天真.

这就是最终对我有用的:

测试端

def fake_subprocess(self, cmd, fnp_o, self_):
    try:
        if self_.taskname != "freeze":
            return self_._subprocess_actual(cmd, fnp_o, self_)
        with open(fnp_o, self_.mode) as fo:
            fo.write(self.fake_subprocess_payload["freeze"])
    except (Exception,) as e:
        raise
def test_001_scan(self):
    try:
        with patch.object(
            Command, "_subprocess", side_effect=self.fake_subprocess
        ) as mock_method:
            options = self.get_options()
            self.mgr = Main(options)
            self.mgr.process()
    except (Exception,) as e:
        raise

实际代码端

class Command(object):
    def _subprocess(self, cmd, fnp_o, self_=None):
        try:
            fnp_stderr = self.mgr._get_fnp("log")
            with open(fnp_stderr, "a") as ferr:
                ferr.write("cmd: %snstderr begin:n" % (cmd))
                with open(fnp_o, self.mode) as fo:
                    proc = subprocess.check_call(
                        cmd.split(), stdout=fo, stderr=ferr, cwd=self.mgr.workdir
                    )
                ferr.write("stderr endnn")
        except (Exception,) as e:
            if cpdb():
                pdb.set_trace()
            raise
    _subprocess_actual = _subprocess
    def run(self):
        try:
            t_cmd = self.config["cmdline"]  # .replace(r"\","\")
            t_fnp = os.path.join(self.mgr.workdir, self.config["filename"])
            fnp_log = "subprocess.log"
            cmd = sub_template(t_cmd, self, self.mgr.vars)
            fnp_o = sub_template(t_fnp, self, self.mgr.vars)
            self._subprocess(cmd=cmd, fnp_o=fnp_o, self_=self)
        except (Exception,) as e:
            if cpdb():
                pdb.set_trace()
            raise

最新更新