chuanqi/tests/test_host_unit.py

# tests/test_host_unit.py
"""Task 7: host.py — 登录链拼包纯函数(零真实网络)。"""
import json
import socket
import time
from types import SimpleNamespace

from apilot.host import Host


def test_login_2001_body_shape(tmp_path):
    """2001 body = b"uid/" + encode6(jwt) + b"/" + encode6(channel) + b"/1/0"。
    与 cq2_pure_enter.login_gate() 第 124-125 行 逐字节一致。"""
    h = Host(14481607, "longwu988", "Longwu1988", state_dir=str(tmp_path))
    uid, jwt = "1186115037", "eyJfake.jwt.token"
    body = h.build_2001_body(uid, jwt)
    import sys
    sys.path.insert(0, "protocol_bot")
    from encode6bitbuf_final import encode6  # type: ignore
    assert body == f"{uid}/".encode() + encode6(jwt.encode()) + b"/" + encode6(b"110585") + b"/1/0"


def test_create_role_101_body(tmp_path):
    """101 body = {"server_name":zone,"chr":name,"secsdk":"0","job":0,"sex":0,
    "account":uid,"sdkid":"110585"};中文/账号/区号全保留。"""
    h = Host(14481607, "u", "p", state_dir=str(tmp_path))
    body = json.loads(h.build_101_body("萌新1234", uid="u1", job=0, sex=0))
    assert body["chr"] == "萌新1234" and body["sdkid"] == "110585"
    assert body["server_name"] == 14481607
    assert body["account"] == "u1" and body["job"] == 0 and body["sex"] == 0
    assert body["secsdk"] == "0"


def test_recv_with_pre_silent_server(tmp_path):
    """B-1 回归: 静默服务端 1s → _recv_with_pre 不抛 NameError,
    超时后正常返回 (None, None)。"""
    h = Host(14481607, "u", "p", state_dir=str(tmp_path))
    a, b = socket.socketpair()
    fake = SimpleNamespace(sock=a, _buf=b"")
    h.nc.wire = fake
    a.settimeout(1.0)
    try:
        start = time.time()
        frame, pre = h._recv_with_pre(530, 1.0)
        assert frame is None and pre is None
        assert time.time() - start >= 0.9
    finally:
        b.close()
        a.close()


def test_recv_with_pre_fragments_across_buf(tmp_path):
    """M-1 回归: 530 帧头 48 字节被 TCP 分片、前半留在 wire._buf,
    下一轮 _buf+新数据重组后仍能命中; sid 取 m[0] 而非 recog。"""
    import struct
    import sys
    sys.path.insert(0, "protocol_bot")
    from encode6bitbuf_final import encode6  # type: ignore
    h = Host(14481607, "u", "p", state_dir=str(tmp_path))
    a, b = socket.socketpair()
    fake = SimpleNamespace(sock=a, _buf=b"")
    h.nc.wire = fake
    a.settimeout(1.0)
    m = (777, 530, 888, 0, 0, 0, 0, 0, 0)
    pkt = encode6(struct.pack("<9i", *m))
    assert len(pkt) == 48
    b.sendall(pkt[:20])
    time.sleep(0.15)
    frame, pre = h._recv_with_pre(530, 1.0)
    assert frame is None and pre is None
    assert fake._buf == pkt[:20]
    b.sendall(pkt[20:])
    try:
        frame, pre = h._recv_with_pre(530, 1.0)
        assert frame is not None and frame.msg_id == 530
        assert frame.recog == 888
        assert pre == 777
        assert fake._buf == b""
    finally:
        b.close()
        a.close()