NixMsg/sdk/python/tests/test_checklist.py

"""DEVELOPMENT 第 9 节接入清单（对真实 nixmsg；跳过仅 JS 跨域）。"""

from __future__ import annotations

import threading
import time
import unittest
from typing import Optional

from nixmsg import (
    Body,
    Client,
    ConnectionState,
    NixMsgError,
    Receipt,
    RegisterOptions,
    SendOptions,
    Target,
)
from nixmsg.uuid7 import new_uuid7

from .harness import NixMsgServer

IMMEDIATE = SendOptions(delay_ms=0)


def wait_until(pred, timeout: float = 15.0, interval: float = 0.05) -> bool:
    deadline = time.time() + timeout
    while time.time() < deadline:
        if pred():
            return True
        time.sleep(interval)
    return False


class MessageBox:
    def __init__(self) -> None:
        self.items: list = []
        self.lock = threading.Lock()
        self.event = threading.Event()

    def on_message(self, msg) -> None:
        with self.lock:
            self.items.append(msg)
        self.event.set()

    def wait_n(self, n: int, timeout: float = 15.0):
        deadline = time.time() + timeout
        while time.time() < deadline:
            with self.lock:
                if len(self.items) >= n:
                    return list(self.items)
            self.event.wait(0.1)
            self.event.clear()
        with self.lock:
            return list(self.items)

    def clear(self) -> None:
        with self.lock:
            self.items.clear()
        self.event.clear()


class ReceiptBox:
    def __init__(self) -> None:
        self.items: list[Receipt] = []
        self.lock = threading.Lock()
        self.event = threading.Event()

    def on_receipt(self, r: Receipt) -> None:
        with self.lock:
            self.items.append(r)
        self.event.set()

    def wait_state(self, state: str, timeout: float = 15.0) -> Optional[Receipt]:
        deadline = time.time() + timeout
        while time.time() < deadline:
            with self.lock:
                for r in self.items:
                    if r.state == state:
                        return r
            self.event.wait(0.1)
            self.event.clear()
        return None


class ChecklistIT(unittest.TestCase):
    srv: NixMsgServer
    seq = 0

    @classmethod
    def setUpClass(cls) -> None:
        cls.srv = NixMsgServer()
        cls.srv.start()

    @classmethod
    def tearDownClass(cls) -> None:
        cls.srv.stop()

    def _uid(self, prefix: str) -> str:
        ChecklistIT.seq += 1
        return f"{prefix}{ChecklistIT.seq:04d}"

    def _register(self, eid: str, password: str = "password12", name: str = "", code: Optional[str] = None):
        return Client.register(
            self.srv.ws_url,
            code if code is not None else self.srv.reg_code,
            RegisterOptions(id=eid, login_password=password, name=name or eid),
        )

    def _connect(self, eid: str, password: str = "password12", **kwargs) -> Client:
        c = Client(**kwargs)
        c.connect(self.srv.ws_url, eid, password=password, wait=True)
        self.assertEqual(c.state, ConnectionState.ONLINE)
        return c

    def test_01_handshake(self) -> None:
        eid = self._uid("hs")
        self._register(eid)
        tokens: list[str] = []
        c = Client()
        c.on_session(lambda t: tokens.append(t))
        c.connect(self.srv.ws_url, eid, password="password12")
        self.assertEqual(c.state, ConnectionState.ONLINE)
        self.assertTrue(c.limits.server_time_ms > 0)
        self.assertGreaterEqual(c.limits.max_body_bytes, 256 * 1024)
        self.assertTrue(tokens and tokens[0].startswith("nst_"))
        c.close()

    def test_02_dm_callback_once(self) -> None:
        a, b = self._uid("a2"), self._uid("b2")
        self._register(a)
        self._register(b)
        ca, cb = self._connect(a), self._connect(b)
        box = MessageBox()
        cb.on_message(box.on_message)
        mid = new_uuid7()
        ca.send(Target("endpoint", b), Body(data="hello-once"), SendOptions(delay_ms=0, message_id=mid))
        got = box.wait_n(1, 10)
        self.assertEqual(len(got), 1)
        self.assertEqual(got[0].id, mid)
        self.assertEqual(got[0].body.data, "hello-once")
        time.sleep(0.5)
        self.assertEqual(len(box.wait_n(1, 0.2)), 1)
        ca.close()
        cb.close()

    def test_03_send_while_disconnected_no_dup(self) -> None:
        a, b = self._uid("a3"), self._uid("b3")
        self._register(a)
        self._register(b)
        ca, cb = self._connect(a), self._connect(b)
        box = MessageBox()
        cb.on_message(box.on_message)
        mid = new_uuid7()
        # 断开发送方传输，触发重连；期间入队发送
        ca._transport.disconnect()
        self.assertTrue(wait_until(lambda: ca.state == ConnectionState.RECONNECTING, 5))

        err: list[BaseException] = []
        result: list = []

        def do_send() -> None:
            try:
                result.append(
                    ca.send(
                        Target("endpoint", b),
                        Body(data="queued"),
                        SendOptions(delay_ms=0, message_id=mid),
                    )
                )
            except BaseException as e:
                err.append(e)

        th = threading.Thread(target=do_send, daemon=True)
        th.start()
        th.join(timeout=60)
        self.assertFalse(err, err)
        self.assertTrue(result)
        self.assertEqual(result[0].id, mid)
        self.assertTrue(wait_until(lambda: ca.state == ConnectionState.ONLINE, 30))
        got = box.wait_n(1, 15)
        self.assertEqual(len(got), 1)
        self.assertEqual(got[0].id, mid)
        time.sleep(0.8)
        self.assertEqual(len(box.items), 1)
        ca.close()
        cb.close()

    def test_04_same_message_id_and_dedup_unit_covered(self) -> None:
        """同消息号重交：断线入队后送达一次。ack 丢失重推依赖 FakeTransport 单测（真实 broker 无法选择性丢 ack）。"""
        a, b = self._uid("a4"), self._uid("b4")
        self._register(a)
        self._register(b)
        ca, cb = self._connect(a), self._connect(b)
        box = MessageBox()
        cb.on_message(box.on_message)
        mid = new_uuid7()
        r1 = ca.send(Target("endpoint", b), Body(data="idem"), SendOptions(delay_ms=0, message_id=mid))
        self.assertEqual(r1.id, mid)
        got = box.wait_n(1, 10)
        self.assertEqual(len(got), 1)
        # 同号同内容再发：服务器防重，回调仍只有一次
        r2 = ca.send(Target("endpoint", b), Body(data="idem"), SendOptions(delay_ms=0, message_id=mid))
        self.assertEqual(r2.id, mid)
        time.sleep(0.8)
        self.assertEqual(len(box.items), 1)
        ca.close()
        cb.close()

    def test_05_recall_within_delay(self) -> None:
        a, b = self._uid("a5"), self._uid("b5")
        self._register(a)
        self._register(b)
        ca, cb = self._connect(a), self._connect(b)
        box = MessageBox()
        revoked: list = []
        cb.on_message(box.on_message)
        cb.on_revoked(lambda e: revoked.append(e))
        mid = new_uuid7()
        r = ca.send(
            Target("endpoint", b),
            Body(data="will-recall"),
            SendOptions(delay_ms=10_000, message_id=mid),
        )
        self.assertEqual(r.state, "scheduled")
        ca.recall(mid)
        time.sleep(1.2)
        self.assertEqual(box.items, [])
        self.assertEqual(revoked, [])
        ca.close()
        cb.close()

    def test_06_scheduled_about_2s(self) -> None:
        a, b = self._uid("a6"), self._uid("b6")
        self._register(a)
        self._register(b)
        ca, cb = self._connect(a), self._connect(b)
        box = MessageBox()
        cb.on_message(box.on_message)
        mid = new_uuid7()
        t0 = time.monotonic()
        ca.send(Target("endpoint", b), Body(data="later"), SendOptions(delay_ms=2000, message_id=mid))
        got = box.wait_n(1, 12)
        elapsed = time.monotonic() - t0
        self.assertEqual(len(got), 1)
        self.assertEqual(got[0].id, mid)
        self.assertGreaterEqual(elapsed, 1.5)
        self.assertLess(elapsed, 8.0)
        ca.close()
        cb.close()

    def test_07_offline_keep(self) -> None:
        a, b_ok, b_miss = self._uid("a7"), self._uid("bok"), self._uid("bms")
        self._register(a)
        self._register(b_ok)
        self._register(b_miss)
        ca = self._connect(a)
        # 对方晚约 1 秒上线能收到
        mid1 = new_uuid7()
        ca.send(
            Target("endpoint", b_ok),
            Body(data="keep-ok"),
            SendOptions(delay_ms=0, keep=True, ttl_seconds=86400, message_id=mid1),
        )
        time.sleep(1.0)
        cb1 = self._connect(b_ok)
        box1 = MessageBox()
        cb1.on_message(box1.on_message)
        got1 = box1.wait_n(1, 10)
        self.assertEqual(len(got1), 1)
        self.assertEqual(got1[0].id, mid1)
        cb1.close()

        # 保留 1 秒且 3 秒后才上线则收不到，发送方收到过期回执
        receipts = ReceiptBox()
        ca.on_receipt(receipts.on_receipt)
        mid2 = new_uuid7()
        ca.send(
            Target("endpoint", b_miss),
            Body(data="keep-expire"),
            SendOptions(delay_ms=0, keep=True, ttl_seconds=1, message_id=mid2),
        )
        time.sleep(3.5)
        cb2 = self._connect(b_miss)
        box2 = MessageBox()
        cb2.on_message(box2.on_message)
        time.sleep(1.5)
        self.assertEqual(box2.items, [])
        exp = receipts.wait_state("expired", 15)
        if exp is None:
            # 回执可能略慢：用 status 核对投递已过期
            st = ca.status(mid2)
            items = (st.get("data") or st).get("items") if isinstance(st.get("data") or st, dict) else None
            # status 顶层即 data
            data = st if "items" in st else (st.get("data") or {})
            items = data.get("items") or data.get("deliveries") or []
            states = [str(i.get("state", "")) for i in items] if isinstance(items, list) else []
            self.assertTrue(
                "expired" in states or any(r.state == "expired" for r in receipts.items),
                f"want expired receipt/status, receipts={[r.state for r in receipts.items]} status={st}",
            )
        else:
            self.assertEqual(exp.id, mid2)
        ca.close()
        cb2.close()

    def test_08_group_sender_no_echo(self) -> None:
        a, b, c = self._uid("a8"), self._uid("b8"), self._uid("c8")
        self._register(a)
        self._register(b)
        self._register(c)
        ca, cb, cc = self._connect(a), self._connect(b), self._connect(c)
        gid = f"g_{a}"
        ca.group_create("G", [{"id": b}, {"id": c}], group_id=gid)
        time.sleep(0.4)
        box_a, box_b, box_c = MessageBox(), MessageBox(), MessageBox()
        ca.on_message(box_a.on_message)
        cb.on_message(box_b.on_message)
        cc.on_message(box_c.on_message)
        mid = new_uuid7()
        ca.send(Target("group", gid), Body(data="hi-g"), SendOptions(delay_ms=0, message_id=mid))
        gb = box_b.wait_n(1, 10)
        gc = box_c.wait_n(1, 10)
        self.assertEqual(len(gb), 1)
        self.assertEqual(len(gc), 1)
        self.assertEqual(gb[0].id, mid)
        self.assertEqual(gc[0].id, mid)
        time.sleep(0.8)
        self.assertEqual(box_a.items, [])
        ca.close()
        cb.close()
        cc.close()

    def test_09_talk_password(self) -> None:
        a, b = self._uid("a9"), self._uid("b9")
        self._register(a)
        self._register(b)
        ca, cb = self._connect(a), self._connect(b)
        cb.set_talk_password("talk99")
        # 拒绝
        with self.assertRaises(NixMsgError) as cm:
            ca.send(Target("endpoint", b), Body(data="no"), IMMEDIATE)
        self.assertIn(cm.exception.code, ("talk_password_required", "talk_password_invalid"))
        # 解锁
        ca.unlock(b, "talk99")
        mid = new_uuid7()
        ca.send(Target("endpoint", b), Body(data="ok"), SendOptions(delay_ms=0, message_id=mid))
        box = MessageBox()
        cb.on_message(box.on_message)
        self.assertEqual(len(box.wait_n(1, 10)), 1)
        # 改密后失效
        cb.set_talk_password("talk00")
        with self.assertRaises(NixMsgError) as cm2:
            ca.send(Target("endpoint", b), Body(data="fail"), IMMEDIATE)
        self.assertIn(cm2.exception.code, ("talk_password_required", "talk_password_invalid"))
        # 对方先发则可以回复
        ca.set_talk_password("alicepw")
        box2 = MessageBox()
        ca.on_message(box2.on_message)
        cb.send(
            Target("endpoint", a),
            Body(data="first"),
            SendOptions(delay_ms=0, talk_password="alicepw", message_id=new_uuid7()),
        )
        self.assertEqual(len(box2.wait_n(1, 10)), 1)
        # a 可回 b（b 曾主动发过）
        mid3 = new_uuid7()
        box3 = MessageBox()
        cb.on_message(box3.on_message)
        ca.send(Target("endpoint", b), Body(data="reply"), SendOptions(delay_ms=0, message_id=mid3))
        self.assertEqual(len(box3.wait_n(1, 10)), 1)
        ca.close()
        cb.close()

    def test_10_kick_no_reconnect(self) -> None:
        eid = self._uid("k10")
        self._register(eid)
        c1 = self._connect(eid)
        states: list[ConnectionState] = []
        c1.on_connection(lambda e: states.append(e.state))
        c2 = self._connect(eid)
        self.assertTrue(wait_until(lambda: c1.state == ConnectionState.KICKED, 15))
        time.sleep(2.5)
        self.assertEqual(c1.state, ConnectionState.KICKED)
        self.assertNotEqual(c1.state, ConnectionState.ONLINE)
        self.assertEqual(c2.state, ConnectionState.ONLINE)
        c1.close()
        c2.close()

    def test_11_body_too_large_local(self) -> None:
        eid = self._uid("big")
        self._register(eid)
        c = self._connect(eid)
        big = "x" * (256 * 1024 + 1)
        with self.assertRaises(NixMsgError) as cm:
            c.send(Target("endpoint", eid), Body(data=big), IMMEDIATE)
        self.assertEqual(cm.exception.code, "body_too_large")
        c.close()

    def test_12_registration_toggle(self) -> None:
        code = self.srv.reg_code
        # 关闭时失败
        self.srv.set_registration(enabled=False)
        with self.assertRaises(NixMsgError) as cm:
            Client.register(self.srv.ws_url, code, RegisterOptions(id=self._uid("r12a"), login_password="password12"))
        self.assertEqual(cm.exception.code, "registration_closed")
        # 错码
        self.srv.set_registration(enabled=True, code=code)
        with self.assertRaises(NixMsgError) as cm2:
            Client.register(
                self.srv.ws_url,
                "wrong-code-xx",
                RegisterOptions(id=self._uid("r12b"), login_password="password12"),
            )
        self.assertEqual(cm2.exception.code, "registration_code_invalid")
        # 成功后能登录
        eid = self._uid("r12c")
        Client.register(self.srv.ws_url, code, RegisterOptions(id=eid, login_password="password12"))
        c = self._connect(eid)
        c.close()
        # 换码后旧码失败、已注册照常登录
        new_code = "s2py-new-code1"
        self.srv.set_registration(enabled=True, code=new_code)
        with self.assertRaises(NixMsgError):
            Client.register(
                self.srv.ws_url,
                code,
                RegisterOptions(id=self._uid("r12d"), login_password="password12"),
            )
        c2 = self._connect(eid)
        c2.close()
        # 恢复默认码供后续用例
        self.srv.set_registration(enabled=True, code=code)
        self.srv.reg_code = code

    def test_13_change_login_password(self) -> None:
        eid = self._uid("pw13")
        self._register(eid, password="password12")
        c = self._connect(eid, password="password12")
        c.change_login_password("password12", "password99")
        c.close()
        # 新密码成功
        c2 = self._connect(eid, password="password99")
        c2.close()
        # 旧密码失败且不再重连
        c3 = Client()
        states: list[ConnectionState] = []
        c3.on_connection(lambda e: states.append(e.state))
        with self.assertRaises(NixMsgError) as cm:
            c3.connect(self.srv.ws_url, eid, password="password12", wait=True)
        self.assertIn(cm.exception.code, ("bad_credentials", "auth_failed"))
        time.sleep(2.0)
        self.assertEqual(c3.state, ConnectionState.AUTH_FAILED)
        c3.close()

    def test_15_session_token(self) -> None:
        eid = self._uid("tok")
        self._register(eid)
        tokens: list[str] = []
        c = Client()
        c.on_session(lambda t: tokens.append(t))
        c.connect(self.srv.ws_url, eid, password="password12")
        self.assertTrue(tokens)
        token = tokens[0]
        c.close()
        # 令牌重连
        c2 = Client()
        c2.connect(self.srv.ws_url, eid, session_token=token)
        self.assertEqual(c2.state, ConnectionState.ONLINE)
        c2.close()
        # 另一处密码登录使旧令牌失效
        c3 = self._connect(eid, password="password12")
        new_token = c3.session_token
        self.assertTrue(new_token and new_token != token)
        c3.close()
        c4 = Client()
        with self.assertRaises(NixMsgError) as cm:
            c4.connect(self.srv.ws_url, eid, session_token=token, wait=True)
        self.assertEqual(cm.exception.code, "session_invalid")
        time.sleep(1.5)
        self.assertEqual(c4.state, ConnectionState.AUTH_FAILED)
        c4.close()
        # logout 后令牌失效
        c5 = self._connect(eid, password="password12")
        tok5 = c5.session_token
        assert tok5
        c5.logout()
        time.sleep(0.3)
        c6 = Client()
        with self.assertRaises(NixMsgError) as cm2:
            c6.connect(self.srv.ws_url, eid, session_token=tok5, wait=True)
        self.assertEqual(cm2.exception.code, "session_invalid")
        c6.close()


if __name__ == "__main__":
    unittest.main()