146 lines
4.9 KiB
Python
146 lines
4.9 KiB
Python
"""RpcClient / RpcServer（JSON RPC 约定层）单元测试。
|
||
|
||
纯标准库，直接运行：
|
||
|
||
cd pymsg
|
||
python -m unittest discover -s tests -v
|
||
"""
|
||
|
||
import json
|
||
import socket
|
||
import threading
|
||
import time
|
||
import unittest
|
||
from datetime import datetime, timezone
|
||
|
||
from pymsg import RpcClient, RpcError, RpcServer, send
|
||
|
||
|
||
def _free_port() -> int:
|
||
with socket.socket() as s:
|
||
s.bind(("127.0.0.1", 0))
|
||
return s.getsockname()[1]
|
||
|
||
|
||
class _RpcFixture:
|
||
"""后台线程运行 RpcServer；routes: {name: fn}。"""
|
||
|
||
def __init__(self, routes, workers=0, json_default=None):
|
||
self.port = _free_port()
|
||
self.server = RpcServer("127.0.0.1", self.port, workers=workers,
|
||
json_default=json_default)
|
||
for name, fn in routes.items():
|
||
self.server.add_route(name, fn)
|
||
self.thread = threading.Thread(target=self.server.start, daemon=True)
|
||
|
||
def __enter__(self):
|
||
self.thread.start()
|
||
deadline = time.time() + 5
|
||
while time.time() < deadline:
|
||
try:
|
||
with socket.create_connection(("127.0.0.1", self.port), timeout=0.2):
|
||
return self
|
||
except OSError:
|
||
time.sleep(0.02)
|
||
raise RuntimeError("server did not start in time")
|
||
|
||
def __exit__(self, *exc):
|
||
self.server.stop()
|
||
self.thread.join(timeout=5)
|
||
return False
|
||
|
||
|
||
class TestRpc(unittest.TestCase):
|
||
|
||
def test_call_returns_data(self):
|
||
with _RpcFixture({"echo": lambda req: {"got": req["msg"]}}) as fx:
|
||
rpc = RpcClient("127.0.0.1", fx.port, "t")
|
||
self.assertEqual(rpc.call("echo", {"msg": "hi"}), {"got": "hi"})
|
||
|
||
def test_route_field_is_injected(self):
|
||
seen = {}
|
||
|
||
def handler(req):
|
||
seen.update(req)
|
||
return None
|
||
|
||
with _RpcFixture({"r": handler}) as fx:
|
||
RpcClient("127.0.0.1", fx.port).call("r", {"a": 1})
|
||
# _route 已被服务端 pop，handler 只看到业务字段
|
||
self.assertEqual(seen, {"a": 1})
|
||
|
||
def test_business_error_raises_rpcerror(self):
|
||
def handler(req):
|
||
raise ValueError("bad input")
|
||
|
||
with _RpcFixture({"x": handler}) as fx:
|
||
with self.assertRaises(RpcError) as cm:
|
||
RpcClient("127.0.0.1", fx.port).call("x")
|
||
self.assertIn("bad input", str(cm.exception))
|
||
|
||
def test_unknown_route_raises_rpcerror(self):
|
||
with _RpcFixture({}) as fx:
|
||
with self.assertRaises(RpcError) as cm:
|
||
RpcClient("127.0.0.1", fx.port).call("nope")
|
||
self.assertIn("未知路由", str(cm.exception))
|
||
|
||
def test_malformed_json_gets_error_response(self):
|
||
"""服务端收到非法 JSON：返回 success=false 且不崩溃，后续请求正常。"""
|
||
with _RpcFixture({"ok": lambda req: "fine"}) as fx:
|
||
with socket.create_connection(("127.0.0.1", fx.port), timeout=5) as s:
|
||
send(s, "not-json{{{")
|
||
from pymsg import recv
|
||
resp = json.loads(recv(s, timeout=5))
|
||
self.assertFalse(resp["success"])
|
||
self.assertIn("JSON 解析失败", resp["error"])
|
||
# 服务仍然存活
|
||
self.assertEqual(RpcClient("127.0.0.1", fx.port).call("ok"), "fine")
|
||
|
||
def test_data_none_when_absent(self):
|
||
with _RpcFixture({"n": lambda req: None}) as fx:
|
||
self.assertIsNone(RpcClient("127.0.0.1", fx.port).call("n"))
|
||
|
||
def test_json_default_serialization(self):
|
||
def default(o):
|
||
if isinstance(o, datetime):
|
||
return o.isoformat()
|
||
return str(o)
|
||
|
||
now = datetime(2026, 7, 30, 12, 0, tzinfo=timezone.utc)
|
||
with _RpcFixture({"t": lambda req: {"ts": now}}, json_default=default) as fx:
|
||
data = RpcClient("127.0.0.1", fx.port).call("t")
|
||
self.assertEqual(data["ts"], now.isoformat())
|
||
|
||
def test_concurrent_rpc_with_workers(self):
|
||
def handler(req):
|
||
time.sleep(0.3)
|
||
return req["i"] * 2
|
||
|
||
with _RpcFixture({"dbl": handler}, workers=4) as fx:
|
||
results = [None] * 6
|
||
|
||
def w(i):
|
||
results[i] = RpcClient("127.0.0.1", fx.port).call("dbl", {"i": i})
|
||
|
||
t0 = time.monotonic()
|
||
ts = [threading.Thread(target=w, args=(i,)) for i in range(6)]
|
||
for t in ts:
|
||
t.start()
|
||
for t in ts:
|
||
t.join(10)
|
||
elapsed = time.monotonic() - t0
|
||
|
||
self.assertEqual(results, [i * 2 for i in range(6)])
|
||
self.assertLess(elapsed, 1.2, f"workers=4 应并发，实际 {elapsed:.2f}s")
|
||
|
||
def test_per_call_timeout_override(self):
|
||
with _RpcFixture({"slow": lambda req: time.sleep(2)}) as fx:
|
||
rpc = RpcClient("127.0.0.1", fx.port, timeout=10.0)
|
||
from pymsg import TimeoutError
|
||
with self.assertRaises(TimeoutError):
|
||
rpc.call("slow", timeout=0.3)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|