"""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()