289 lines
10 KiB
Python
289 lines
10 KiB
Python
"""chongming_request.Client 语义回归：信封组装 / 换副本重试 / 冷静期 / 错误文案。
|
||
|
||
不依赖真实 Zenoh 会话与 worker：假 Session 捕获出站载荷、模拟各类回复。
|
||
运行: python tests/test_client.py
|
||
"""
|
||
import asyncio
|
||
import sys
|
||
import time
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
|
||
|
||
import orjson # noqa: E402
|
||
import chongming_request.client as client_module # noqa: E402
|
||
from chongming_request import Client # noqa: E402
|
||
|
||
|
||
class FakeOk:
|
||
def __init__(self, payload):
|
||
self.payload = payload
|
||
|
||
|
||
class FakeErr:
|
||
def __init__(self, payload):
|
||
self.payload = payload
|
||
|
||
|
||
class FakeReply:
|
||
def __init__(self, payload=b"", err=None):
|
||
"""err 给出时构造 err 回复（ok=None，与 zenoh 超时答复同形）"""
|
||
self.ok = None if err is not None else FakeOk(payload)
|
||
self.err = err
|
||
|
||
|
||
class BlockingIter:
|
||
"""模拟半死副本：迭代阻塞超过等待超时"""
|
||
|
||
def __init__(self, delay):
|
||
self.delay = delay
|
||
|
||
def __iter__(self):
|
||
time.sleep(self.delay)
|
||
return iter(())
|
||
|
||
|
||
class StubSession:
|
||
"""按副本给回复；记录每次 get 的 key 与入参"""
|
||
|
||
def __init__(self, workers, replies=None, blocking=None):
|
||
self.workers = workers
|
||
self.replies = replies or {} # worker_key -> FakeReply 或其列表（按调用次序弹出）
|
||
self.blocking = blocking or {} # worker_key -> 阻塞秒数
|
||
self.calls = [] # (full_key, payload_bytes, kwargs)
|
||
|
||
def liveliness(self):
|
||
class LV:
|
||
def get(_self, key):
|
||
return [
|
||
type("R", (), {"ok": type("O", (), {"key_expr": w})()})()
|
||
for w in self.workers
|
||
]
|
||
return LV()
|
||
|
||
def get(self, full_key, **kwargs):
|
||
self.calls.append((full_key, kwargs.get("payload"), kwargs))
|
||
worker_key = full_key.rsplit("/", 1)[0]
|
||
if worker_key in self.blocking:
|
||
return BlockingIter(self.blocking[worker_key])
|
||
item = self.replies.get(worker_key, [FakeReply(b'{"ok":true}')])
|
||
if isinstance(item, list):
|
||
item = item.pop(0)
|
||
return iter([item])
|
||
|
||
def sent_keys(self):
|
||
return [k for k, _, _ in self.calls]
|
||
|
||
|
||
class FirstPick:
|
||
"""把 random.choice 固定为取第一个候选，使副本选择在测试里确定"""
|
||
|
||
def choice(self, seq):
|
||
return seq[0]
|
||
|
||
|
||
def _pin_first_pick():
|
||
original = client_module.random
|
||
client_module.random = FirstPick()
|
||
return original
|
||
|
||
|
||
# ── 信封组装 ────────────────────────────────────────────────────────────────
|
||
|
||
def test_wrap_business_payload():
|
||
sess = StubSession(["worker/demo/edge-01"])
|
||
payload = {"a": 1, "b": {"c": 2}}
|
||
out = asyncio.run(Client(session=sess).call("demo", "calc.add", payload, method="post", timeout=1.0))
|
||
assert out == {"ok": True}
|
||
key, sent, kwargs = sess.calls[0]
|
||
assert key == "worker/demo/edge-01/calc.add"
|
||
assert orjson.loads(sent) == {"meta": {"method": "POST"}, "data": {"a": 1, "b": {"c": 2}}}
|
||
assert kwargs["timeout"] == 1.0 # zenoh 查询自身超时已对齐
|
||
assert payload == {"a": 1, "b": {"c": 2}} # 调用方 dict 未被修改
|
||
|
||
|
||
def test_envelope_reuse_verbatim():
|
||
"""调用方信封：保留全部字段、falsy 值不改写，仅补 meta.method，不改调用方 dict"""
|
||
sess = StubSession(["worker/demo/edge-01"])
|
||
payload = {"meta": {"trace": "t1"}, "data": 0, "trace_id": "z"}
|
||
asyncio.run(Client(session=sess).call("demo", "calc.add", payload, method="POST"))
|
||
assert orjson.loads(sess.calls[0][1]) == {
|
||
"meta": {"trace": "t1", "method": "POST"}, "data": 0, "trace_id": "z",
|
||
}
|
||
assert payload["meta"] == {"trace": "t1"} # 嵌套 meta 也未被写回
|
||
|
||
|
||
def test_envelope_method_not_overridden():
|
||
sess = StubSession(["worker/demo/edge-01"])
|
||
asyncio.run(Client(session=sess).call("demo", "calc.add", {"meta": {"method": "put"}, "data": {}}, method="POST"))
|
||
assert orjson.loads(sess.calls[0][1])["meta"]["method"] == "put"
|
||
|
||
|
||
def test_data_only_dict_stays_envelope():
|
||
"""只有 data 键的 dict 按信封处理（协议兼容规则），补上 meta.method"""
|
||
sess = StubSession(["worker/demo/edge-01"])
|
||
asyncio.run(Client(session=sess).call("demo", "calc.add", {"data": {"x": 1}}, method="POST"))
|
||
assert orjson.loads(sess.calls[0][1]) == {"data": {"x": 1}, "meta": {"method": "POST"}}
|
||
|
||
|
||
def test_non_dict_payload_wrapped():
|
||
sess = StubSession(["worker/demo/edge-01"])
|
||
asyncio.run(Client(session=sess).call("demo", "calc.add", [1, 2]))
|
||
assert orjson.loads(sess.calls[0][1]) == {"meta": {"method": "GET"}, "data": [1, 2]}
|
||
|
||
|
||
def test_meta_not_dict_raises_before_send():
|
||
sess = StubSession(["worker/demo/edge-01"])
|
||
try:
|
||
asyncio.run(Client(session=sess).call("demo", "calc.add", {"meta": "hello", "n": 1}))
|
||
except ValueError as e:
|
||
assert "meta" in str(e)
|
||
else:
|
||
raise AssertionError("应抛 ValueError")
|
||
assert sess.calls == [] # 发送前就失败
|
||
|
||
|
||
# ── 回复语义 ────────────────────────────────────────────────────────────────
|
||
|
||
def test_empty_reply_returns_empty_dict():
|
||
sess = StubSession(["worker/demo/edge-01"], replies={"worker/demo/edge-01": [FakeReply(b"")]})
|
||
assert asyncio.run(Client(session=sess).call("demo", "calc.add", {})) == {}
|
||
|
||
|
||
def test_no_workers_raises_runtime_error():
|
||
sess = StubSession([])
|
||
try:
|
||
asyncio.run(Client(session=sess).call("demo", "calc.add", {}))
|
||
except RuntimeError as e:
|
||
assert "没有可用 Worker" in str(e) and "demo" in str(e)
|
||
else:
|
||
raise AssertionError("应抛 RuntimeError")
|
||
|
||
|
||
def test_call_sync():
|
||
sess = StubSession(["worker/demo/edge-01"])
|
||
assert Client(session=sess).call_sync("demo", "calc.add", {}) == {"ok": True}
|
||
|
||
|
||
# ── 换副本重试 / 冷静期 ─────────────────────────────────────────────────────
|
||
|
||
def test_timeout_switches_worker_and_marks_suspect():
|
||
sess = StubSession(
|
||
["worker/demo/edge-01", "worker/demo/edge-02"],
|
||
blocking={"worker/demo/edge-01": 0.5},
|
||
replies={"worker/demo/edge-02": [FakeReply(b'{"who":"edge-02"}')]},
|
||
)
|
||
c = Client(session=sess)
|
||
original = _pin_first_pick()
|
||
try:
|
||
out = asyncio.run(c.call("demo", "calc.add", {}, timeout=0.05, retries=1))
|
||
finally:
|
||
client_module.random = original
|
||
assert out == {"who": "edge-02"}
|
||
assert sess.sent_keys() == [
|
||
"worker/demo/edge-01/calc.add",
|
||
"worker/demo/edge-02/calc.add",
|
||
]
|
||
assert "worker/demo/edge-01" in c._suspect_until # 超时副本进入冷静期
|
||
|
||
|
||
def test_suspect_skipped_on_next_call():
|
||
sess = StubSession(["worker/demo/edge-01", "worker/demo/edge-02"])
|
||
c = Client(session=sess)
|
||
c._suspect_until["worker/demo/edge-01"] = time.monotonic() + 30
|
||
original = _pin_first_pick()
|
||
try:
|
||
asyncio.run(c.call("demo", "calc.add", {}))
|
||
finally:
|
||
client_module.random = original
|
||
assert sess.sent_keys() == ["worker/demo/edge-02/calc.add"]
|
||
|
||
|
||
def test_err_reply_switches_then_success():
|
||
sess = StubSession(
|
||
["worker/demo/edge-01", "worker/demo/edge-02"],
|
||
replies={
|
||
"worker/demo/edge-01": [FakeReply(err=FakeErr(b'{"error":"boom"}'))],
|
||
"worker/demo/edge-02": [FakeReply(b'{"ok":2}')],
|
||
},
|
||
)
|
||
c = Client(session=sess)
|
||
original = _pin_first_pick()
|
||
try:
|
||
out = asyncio.run(c.call("demo", "calc.add", {}))
|
||
finally:
|
||
client_module.random = original
|
||
assert out == {"ok": 2}
|
||
assert sess.sent_keys() == [
|
||
"worker/demo/edge-01/calc.add",
|
||
"worker/demo/edge-02/calc.add",
|
||
]
|
||
# 业务错误不列入冷静期（与 worker_manager 一致：只有超时/无回复进冷静期）
|
||
assert "worker/demo/edge-01" not in c._suspect_until
|
||
|
||
|
||
def test_single_replica_fails_fast_with_detail():
|
||
sess = StubSession(["worker/demo/edge-01"], blocking={"worker/demo/edge-01": 0.3})
|
||
try:
|
||
asyncio.run(Client(session=sess).call("demo", "calc.add", {}, timeout=0.05, retries=2))
|
||
except RuntimeError as e:
|
||
msg = str(e)
|
||
assert "已尝试 1 个 Worker" in msg
|
||
assert "上限 3 次" in msg
|
||
assert "无其它未试副本可换" in msg and "超时" in msg # last_error 保留
|
||
else:
|
||
raise AssertionError("应抛 RuntimeError")
|
||
assert len(sess.calls) == 1 # 无未试副本立即结束，不白等剩余重试
|
||
|
||
|
||
def test_zenoh_timeout_reply_treated_as_error():
|
||
"""zenoh 查询自身超时给出的答复（ok=None, err.payload=b"Timeout"）走错误通道"""
|
||
sess = StubSession(
|
||
["worker/demo/edge-01"],
|
||
replies={"worker/demo/edge-01": [FakeReply(err=FakeErr(b"Timeout"))]},
|
||
)
|
||
try:
|
||
asyncio.run(Client(session=sess).call("demo", "calc.add", {}))
|
||
except RuntimeError as e:
|
||
assert "返回错误: Timeout" in str(e)
|
||
else:
|
||
raise AssertionError("应抛 RuntimeError")
|
||
|
||
|
||
def test_empty_discovery_not_cached():
|
||
"""发现结果为空不写缓存：副本随后上线，下一次调用立即可见（不等 15s TTL）"""
|
||
sess = StubSession([])
|
||
c = Client(session=sess)
|
||
try:
|
||
asyncio.run(c.call("demo", "calc.add", {}))
|
||
except RuntimeError:
|
||
pass
|
||
sess.workers.append("worker/demo/edge-01")
|
||
assert asyncio.run(c.call("demo", "calc.add", {})) == {"ok": True}
|
||
|
||
|
||
def test_exhausted_error_message_contains_worker_error():
|
||
sess = StubSession(
|
||
["worker/demo/edge-01"],
|
||
replies={"worker/demo/edge-01": [FakeReply(err=FakeErr(b"boom"))]},
|
||
)
|
||
try:
|
||
asyncio.run(Client(session=sess).call("demo", "calc.add", {}, retries=1))
|
||
except RuntimeError as e:
|
||
assert "返回错误: boom" in str(e)
|
||
assert "已尝试 1 个 Worker" in str(e)
|
||
else:
|
||
raise AssertionError("应抛 RuntimeError")
|
||
|
||
|
||
def _main():
|
||
tests = [v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)]
|
||
for t in tests:
|
||
t()
|
||
print(f"PASS {t.__name__}")
|
||
print(f"\n{len(tests)} tests passed")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
_main()
|