无法 monkeypatch rpc 服务器 class 方法
Unable to monkeypatch an rpc server class method
我需要用 jsonrpcserver
库的 @method
注释修饰一个 class 方法。 class 正在实现一个作为 asyncio 服务器启动的 rpc 服务器,并使用像这样的 pytest fixture 启动
# conftest.py
@pytest.fixture(autouse=True, scope="module")
@pytest.mark.asyncio
async def rpc_server(...):
rpc_server = RpcServer(
addr="127.0.0.1",
port=9500,
...
)
task = asyncio.create_task(rpc_server.start())
yield rpc_server
task.cancel()
测试应该monkeypatch的方法之一 RpcServer
class
# test_rpc.py
@pytest.mark.asyncio
async def test_rpc_server_exception(
rpc_server: RpcServer,
...
monkeypatch: MonkeyPatch,
):
async def raise_runtime_error():
raise RuntimeError()
monkeypatch.setattr(
RpcServer, "method_to_be_patched", raise_runtime_error, raising=True
)
... # making an rpc request to launch method_to_be_patched
assert ...
method_to_be_patched
在收到新请求后由 jsonrpcserver
库的 async_dispatch
调用,如下所示
# rpc_server.py
@method
async def method_to_be_patched(self, ...) -> str:
...
return ...
问题是 monkeypatch 没有修补任何东西并且测试通过而没有引发任何异常(就像我需要的那样)。我已经尝试过 monkeypatch RpcServer
并且实例从 pytest 夹具中产生但没有任何成功,但通过调试似乎 class 方法正确指向虚拟函数但仍然调用了原始函数。
编辑:问题的出现是因为 python 导入工作。据我所知,在像 from ... import ...
导入时,我正在创建一个新参考,所以基本上我是在修补从 test_rpc.py
创建的参考,而不是 rpc_server.py
中的参考(如果我我错了)。
所以我尝试了
# test_rpc.py
@pytest.mark.asyncio
async def test_rpc_server_exception(
rpc_server: RpcServer,
...
monkeypatch: MonkeyPatch,
):
async def raise_runtime_error():
raise RuntimeError()
import network # the package containing rpc_server.py
monkeypatch.setattr(
network.rpc_server.RpcServer, "method_to_be_patched", raise_runtime_error, raising=True
)
... # making an rpc request to launch method_to_be_patched
assert ...
但仍然没有得到预期的行为。
项目树是这样的
/src
|rpc_server.py
/test
|conftest.py
/e2e
|test_rpc.py
解决方案是 monkeypatch where 方法被调用所以因为我在这里使用 jsonrpcserver
我不得不 monkeypatch 里面定义的 call
方法async_dispatch
模块,现在它按我预期的方式工作
我需要用 jsonrpcserver
库的 @method
注释修饰一个 class 方法。 class 正在实现一个作为 asyncio 服务器启动的 rpc 服务器,并使用像这样的 pytest fixture 启动
# conftest.py
@pytest.fixture(autouse=True, scope="module")
@pytest.mark.asyncio
async def rpc_server(...):
rpc_server = RpcServer(
addr="127.0.0.1",
port=9500,
...
)
task = asyncio.create_task(rpc_server.start())
yield rpc_server
task.cancel()
测试应该monkeypatch的方法之一 RpcServer
class
# test_rpc.py
@pytest.mark.asyncio
async def test_rpc_server_exception(
rpc_server: RpcServer,
...
monkeypatch: MonkeyPatch,
):
async def raise_runtime_error():
raise RuntimeError()
monkeypatch.setattr(
RpcServer, "method_to_be_patched", raise_runtime_error, raising=True
)
... # making an rpc request to launch method_to_be_patched
assert ...
method_to_be_patched
在收到新请求后由 jsonrpcserver
库的 async_dispatch
调用,如下所示
# rpc_server.py
@method
async def method_to_be_patched(self, ...) -> str:
...
return ...
问题是 monkeypatch 没有修补任何东西并且测试通过而没有引发任何异常(就像我需要的那样)。我已经尝试过 monkeypatch RpcServer
并且实例从 pytest 夹具中产生但没有任何成功,但通过调试似乎 class 方法正确指向虚拟函数但仍然调用了原始函数。
编辑:问题的出现是因为 python 导入工作。据我所知,在像 from ... import ...
导入时,我正在创建一个新参考,所以基本上我是在修补从 test_rpc.py
创建的参考,而不是 rpc_server.py
中的参考(如果我我错了)。
所以我尝试了
# test_rpc.py
@pytest.mark.asyncio
async def test_rpc_server_exception(
rpc_server: RpcServer,
...
monkeypatch: MonkeyPatch,
):
async def raise_runtime_error():
raise RuntimeError()
import network # the package containing rpc_server.py
monkeypatch.setattr(
network.rpc_server.RpcServer, "method_to_be_patched", raise_runtime_error, raising=True
)
... # making an rpc request to launch method_to_be_patched
assert ...
但仍然没有得到预期的行为。
项目树是这样的
/src
|rpc_server.py
/test
|conftest.py
/e2e
|test_rpc.py
解决方案是 monkeypatch where 方法被调用所以因为我在这里使用 jsonrpcserver
我不得不 monkeypatch 里面定义的 call
方法async_dispatch
模块,现在它按我预期的方式工作