feat: deltachat_rpc_client.Rpc accepts rpc_server_path for using a particular deltachat-rpc-server (#7493)

also simplifies and make more readable popen-call to rpc-server 

addresses #7428
This commit is contained in:
holger krekel
2025-11-25 11:02:42 +01:00
committed by GitHub
parent fdff90eba4
commit b9ae74fab2
2 changed files with 30 additions and 17 deletions

View File

@@ -54,7 +54,7 @@ class RpcMethod:
class Rpc:
"""RPC client."""
def __init__(self, accounts_dir: Optional[str] = None, **kwargs):
def __init__(self, accounts_dir: Optional[str] = None, rpc_server_path="deltachat-rpc-server", **kwargs):
"""Initialize RPC client.
The given arguments will be passed to subprocess.Popen().
@@ -66,6 +66,7 @@ class Rpc:
}
self._kwargs = kwargs
self.rpc_server_path = rpc_server_path
self.process: subprocess.Popen
self.id_iterator: Iterator[int]
self.event_queues: dict[int, Queue]
@@ -79,24 +80,16 @@ class Rpc:
def start(self) -> None:
"""Start RPC server subprocess."""
popen_kwargs = {"stdin": subprocess.PIPE, "stdout": subprocess.PIPE}
if sys.version_info >= (3, 11):
self.process = subprocess.Popen(
"deltachat-rpc-server",
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
# Prevent subprocess from capturing SIGINT.
process_group=0,
**self._kwargs,
)
# Prevent subprocess from capturing SIGINT.
popen_kwargs["process_group"] = 0
else:
self.process = subprocess.Popen(
"deltachat-rpc-server",
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
# `process_group` is not supported before Python 3.11.
preexec_fn=os.setpgrp, # noqa: PLW1509
**self._kwargs,
)
# `process_group` is not supported before Python 3.11.
popen_kwargs["preexec_fn"] = os.setpgrp # noqa: PLW1509
popen_kwargs.update(self._kwargs)
self.process = subprocess.Popen(self.rpc_server_path, **popen_kwargs)
self.id_iterator = itertools.count(start=1)
self.event_queues = {}
self.request_results = {}

View File

@@ -0,0 +1,20 @@
import subprocess
import sys
from platform import system # noqa
import pytest
from deltachat_rpc_client import DeltaChat, Rpc
@pytest.mark.skipif("system() == 'Windows'")
def test_install_venv_and_use_other_core(tmp_path):
venv = tmp_path.joinpath("venv1")
subprocess.check_call([sys.executable, "-m", "venv", venv])
python = venv / "bin" / "python"
subprocess.check_call([python, "-m", "pip", "install", "deltachat-rpc-server==2.20.0"])
rpc = Rpc(accounts_dir=tmp_path.joinpath("accounts"), rpc_server_path=venv.joinpath("bin", "deltachat-rpc-server"))
with rpc:
dc = DeltaChat(rpc)
assert dc.rpc.get_system_info()["deltachat_core_version"] == "v2.20.0"