chuanqi/pipeline_bot/tests/test_wander.py

"""WanderBehavior/UnlockBehavior 走链行为验证(不发真实网络 — 桩发送)。"""

import json
import time
from typing import final

from pbot.bot.unlock import UnlockBehavior
from pbot.bot.wander import DIRS, RUN_INTERVAL_S, STEP_INTERVAL_S, WAIT_MAP, WanderBehavior
from pbot.codec.frame import Frame, build_frame, parse_frame
from pbot.world.actor import FindTarget
from pbot.world.mirror import WorldMirror


@final
class SpyConn:
    """记录 send_frame 参数; recv 不产生帧。"""

    def __init__(self) -> None:
        self.sent: list[tuple[int, int, int, int, int, bytes]]
        self.pres: list[int]
        self.sent = []
        self.pres = []
        # 看门狗字段(AutoFight.tick 检查): 恒"最近收帧"= 不触发死线
        self.last_recv_t: float = __import__("time").time()

    def send_frame(self, msg_id: int, recog: int, p1: int, p2: int, p3: int,
                   body: bytes = b"", tail: int | None = None, pre: int = 0) -> int:
        del tail
        self.sent.append((msg_id, recog, p1, p2, p3, body))
        self.pres.append(pre)
        return 0

    def send_raw(self, wire: bytes) -> None:
        import struct

        from pbot.codec.bitbuf import decode6
        pos = 0
        while pos < len(wire):
            h = struct.unpack("<9i", decode6(wire[pos:pos + 48]))
            blen = h[7]
            self.sent.append((h[1], h[2], h[3], h[4], h[5],
                              wire[pos + 48:pos + 48 + blen]))
            pos += 48 + blen

    def recv_frames(self, timeout: float) -> list[Frame]:
        del timeout
        return []


def ready_world(x: int, y: int) -> WorldMirror:
    """SC50/SC51 已就绪的世界。"""
    world = WorldMirror()
    world.me.map_name = WAIT_MAP
    world.me.actor_key = 7
    world.me.x, world.me.y = x, y
    return world


def fast_wander(sid: int, spy: SpyConn, world: WorldMirror) -> WanderBehavior:
    """已 on_enter 且时间门已到点的 WanderBehavior。"""
    b = WanderBehavior(sid=sid)
    b.on_enter(spy, world)  # pyright: ignore[reportArgumentType]
    b._last_step_t = time.time() - STEP_INTERVAL_S - 0.05  # pyright: ignore[reportPrivateUsage]
    return b


def fast_walker_no_run(spy: SpyConn, world: WorldMirror) -> WanderBehavior:
    """use_run=False 且时间门已到点的 WanderBehavior。"""
    b = WanderBehavior(sid=1, use_run=False)
    b.on_enter(spy, world)  # pyright: ignore[reportArgumentType]
    b._last_step_t = time.time() - STEP_INTERVAL_S  # pyright: ignore[reportPrivateUsage]
    return b


# ---- WanderBehavior: 单帧制(cap_unlock 定案) ----


def test_wander_sends_single_3013_frame() -> None:
    """Given: 世界就绪 When: 一拍到点 Then: 发三帧组 3013+CS1+CS28x2(cap_gold t=4985)。"""
    world = ready_world(23, 27)
    spy = SpyConn()
    b = fast_wander(sid=159, spy=spy, world=world)
    _ = b.tick(spy, world)  # pyright: ignore[reportArgumentType]
    kinds = [s[0] for s in spy.sent]
    assert kinds == [3011]  # 单帧制(cap_live 14 步铁证: 零 CS1/CS28 上行)
    msg_id, recog, p1, p2, p3, body = spy.sent[0]
    del msg_id
    assert recog == 0
    assert p1 == 5  # 初始朝向
    assert p2 == (27 << 16) | 23  # 起点
    assert p3 == 159  # p3 = sid(cap_unlock/cap_live 双定案)
    assert body == WAIT_MAP.encode()


def test_wander_walk_mode_uses_3011() -> None:
    """Given: use_run=False When: tick Then: 发 CS3011 走 1 格。"""
    world = ready_world(23, 27)
    spy = SpyConn()
    b = fast_walker_no_run(spy, world)
    _ = b.tick(spy, world)  # pyright: ignore[reportArgumentType]
    assert spy.sent[0][0] == 3011


def test_wander_3011_carries_landing_prediction_in_pre() -> None:
    """Given: 走一步 When: 发 3011 Then: pre=预测落点(Y<<16|X)非零。

    回归: 2026-09-26 破案定案 — cap_round2 30 会话统计, pre=0 的 3011
    215/215 被 SC28 拒, pre≠0 的 60/60 放行。真客户端 pre=落点预测。
    """
    world = ready_world(23, 27)
    spy = SpyConn()
    b = fast_wander(sid=1, spy=spy, world=world)
    _ = b.tick(spy, world)  # pyright: ignore[reportArgumentType]
    x, y = 23, 27
    dx, dy = DIRS[5]  # 初始朝向 5
    expected_pre = ((y + dy) << 16) | (x + dx)
    assert spy.pres[0] == expected_pre
    assert spy.pres[0] != 0


def test_wander_run_mode_sends_3013_two_cells() -> None:
    """Given: use_run=True When: tick Then: 发 CS3013 且落点=起点+2格。

    协议定案(cap_round2 sid=88 放行实证): 3013 与 3011 同构,
    跑一步 = 沿 dir 跳 2 格, 步距 0.55s, 服务器仍回 SC1。
    """
    world = ready_world(23, 27)
    spy = SpyConn()
    b = WanderBehavior(sid=1, use_run=True)
    b.on_enter(spy, world)  # pyright: ignore[reportArgumentType]
    b._last_step_t = time.time() - RUN_INTERVAL_S * 2  # pyright: ignore[reportPrivateUsage]
    _ = b.tick(spy, world)  # pyright: ignore[reportArgumentType]
    assert spy.sent[0][0] == 3013
    _, recog, p1, p2, p3, body = spy.sent[0]
    del recog, p3, body
    dx, dy = DIRS[5]  # 初始朝向 5
    assert p1 == 5 and p2 == (27 << 16) | 23
    assert spy.pres[0] == ((27 + dy * 2) << 16) | (23 + dx * 2)


def test_wander_holds_until_sc1_confirm() -> None:
    """Given: 上一步未获 SC1 确认 When: 到点再 tick Then: 不发下一步(链式咬合)。"""
    world = ready_world(23, 27)
    spy = SpyConn()
    b = fast_wander(sid=1, spy=spy, world=world)
    _ = b.tick(spy, world)  # pyright: ignore[reportArgumentType]
    assert len(spy.sent) == 1
    b._last_step_t = time.time()  # pyright: ignore[reportPrivateUsage] - 步距刚走完
    world.record("self_move_confirmed")  # SC1 到达(释放 pending)
    _ = b.tick(spy, world)  # pyright: ignore[reportArgumentType]
    assert len(spy.sent) == 1  # 步距未到, 不发
    b._last_step_t = time.time() - STEP_INTERVAL_S - 0.05  # pyright: ignore[reportPrivateUsage]
    _ = b.tick(spy, world)  # pyright: ignore[reportArgumentType]
    assert len(spy.sent) == 2  # 确认+到点 → 放行下一步


def test_wander_pull_back_releases_pending() -> None:
    """Given: SC28 拉回 When: tick Then: pending 释放且拒绝计数。"""
    world = ready_world(23, 27)
    spy = SpyConn()
    b = fast_wander(sid=1, spy=spy, world=world)
    _ = b.tick(spy, world)  # pyright: ignore[reportArgumentType]
    world.record("pull_back")
    b._last_step_t = time.time() - STEP_INTERVAL_S - 0.05  # pyright: ignore[reportPrivateUsage] - 余量防时钟抖动
    _ = b.tick(spy, world)  # pyright: ignore[reportArgumentType]
    assert len(spy.sent) == 2  # 拉回后以服务器位置重发(pending 已释放)
    del b


def test_wander_anchor_echo_releases_pending() -> None:
    """Given: SC28 锚点回显(anchor_echo 事件) When: 到点 tick Then: 放行重发。

    回归: 2026-09-24 冻结 bug — _absorb_events 不认 anchor_echo 导致
    pending 永挂, dir 恒 5 坐标不变(一发即停)。
    新行为: 同向锁 3 步 — 撞回后保持同方向, 链式 3011 沿同方向发,
    SC1 一旦放行就跑通(避免 (200,30)↔(198,28) A↔B 振荡)。
    """
    world = ready_world(23, 27)
    spy = SpyConn()
    b = fast_wander(sid=1, spy=spy, world=world)
    _ = b.tick(spy, world)  # pyright: ignore[reportArgumentType]
    assert len(spy.sent) == 1
    world.record("anchor_echo")  # update.py 对 SC28 p3=32368232 记的事件名
    b._last_step_t = time.time() - STEP_INTERVAL_S - 0.05  # pyright: ignore[reportPrivateUsage]
    _ = b.tick(spy, world)  # pyright: ignore[reportArgumentType]
    assert len(spy.sent) == 2
    assert spy.sent[1][2] == spy.sent[0][2]  # p1=heading 同向锁住
    assert spy.sent[1][0] == 3011  # 仍发新帧(非冻结)


def test_wander_pending_timeout_frees_silence() -> None:
    """Given: 帧被静默丢(无任何回包) When: 超过 PENDING_TIMEOUT_S Then: 重发。"""
    world = ready_world(23, 27)
    spy = SpyConn()
    b = fast_wander(sid=1, spy=spy, world=world)
    _ = b.tick(spy, world)  # pyright: ignore[reportArgumentType]
    assert len(spy.sent) == 1
    b._last_step_t = time.time() - STEP_INTERVAL_S - 0.05  # pyright: ignore[reportPrivateUsage]
    b._pending_t = time.time() - 10.0  # pyright: ignore[reportPrivateUsage] - 模拟超时
    _ = b.tick(spy, world)  # pyright: ignore[reportArgumentType]
    assert len(spy.sent) == 2


def test_wander_new_heading_avoids_occupied_cells() -> None:
    """Given: 视野内 NPC/怪占满部分邻格 When: 换向 Then: 选中未占据方向。"""
    from pbot.types import ActorId
    from pbot.world.actor import ActorKey, Npc

    world = ready_world(23, 27)
    spy = SpyConn()
    b = fast_wander(sid=1, spy=spy, world=world)
    for h, (dx, dy) in enumerate(DIRS):
        if h in (5, 2):  # 5=当前朝向(排除), 2=留作唯一空闲候选
            continue
        world.npcs[100 + h] = Npc(key=ActorKey(100 + h), aid=ActorId(f"a{h}"),
                                  name=f"n{h}", x=23 + dx, y=27 + dy, index=h)
    b._rejected = 1  # pyright: ignore[reportPrivateUsage] - 触发换向分支
    heading = b._pick_new_heading(world)  # pyright: ignore[reportPrivateUsage]
    assert heading == 2


def test_wander_waits_until_world_known() -> None:
    """Given: SC50/SC51 未到 When: tick Then: 不发帧。"""
    world = WorldMirror()
    spy = SpyConn()
    b = fast_wander(sid=1, spy=spy, world=world)
    _ = b.tick(spy, world)  # pyright: ignore[reportArgumentType]
    assert spy.sent == []


def test_dirs_table_matches_lua_source() -> None:
    """Given: Lua mmoConstants 方向表 When: 检查 Then: 与 backup 源码一致。"""
    assert DIRS[0] == (0, -1)  # 上
    assert DIRS[2] == (1, 0)  # 右
    assert DIRS[4] == (0, 1)  # 下
    assert DIRS[6] == (-1, 0)  # 左


# ---- UnlockBehavior: 黄金时序状态机 ----


def test_unlock_renwu_golden_timing() -> None:
    """Given: 已传比奇 When: renwu链 Then: 3031→[1.2s]→1011 分步发, 不连发。"""
    world = ready_world(289, 620)
    world.me.map_name = "比奇大城"
    world.me.map_id = "0"
    spy = SpyConn()
    u = UnlockBehavior(sid=159)
    u.on_enter(spy, world)  # pyright: ignore[reportArgumentType]
    u._phase_t = 0.0  # pyright: ignore[reportPrivateUsage] - 跳过 settle 等待
    _ = u.tick(spy, world)  # pyright: ignore[reportArgumentType]
    kinds = [s[0] for s in spy.sent]
    assert kinds == [3031]  # 第一拍只发 3031
    u._phase = "renwu_1011"  # pyright: ignore[reportPrivateUsage] - 模拟对话框已点
    u._phase_t = 0.0  # pyright: ignore[reportPrivateUsage]
    _ = u.tick(spy, world)  # pyright: ignore[reportArgumentType]
    kinds = [s[0] for s in spy.sent]
    assert kinds == [3031, 1011]
    body = spy.sent[1][5].decode()
    assert json.loads(body)["Act"] == "@renwu_mov,1"


def test_unlock_retention_band_starts_npc_chain() -> None:
    """Given: 等待图滞留带(y<5) When: on_enter Then: 先走 NPC 视野等待, 不直发 renwu。"""
    world = ready_world(30, 0)
    spy = SpyConn()
    u = UnlockBehavior(sid=159)
    u.on_enter(spy, world)  # pyright: ignore[reportArgumentType]
    _ = u.tick(spy, world)  # pyright: ignore[reportArgumentType]
    assert spy.sent == []  # 等NPC视野, 不盲发


def test_unlock_uses_full_cs3031_body() -> None:
    """Given: 无 SC55 缓存 When: 发 3031 Then: body 为 562B 完整体(≥500B)。"""
    world = ready_world(289, 620)
    world.me.map_name = "比奇大城"
    spy = SpyConn()
    u = UnlockBehavior(sid=1)
    u.on_enter(spy, world)  # pyright: ignore[reportArgumentType]
    u._phase_t = 0.0  # pyright: ignore[reportPrivateUsage]
    _ = u.tick(spy, world)  # pyright: ignore[reportArgumentType]
    body = spy.sent[0][5]
    assert len(body) >= 500  # 完整任务体, 非 200B 截断


def test_unlock_accept_signal_enters_run() -> None:
    """Given: 1011 已发 When: SC20102 到达 Then: 进入冲刺并发 3013。"""
    world = ready_world(24, 27)
    spy = SpyConn()
    u = UnlockBehavior(sid=159)
    u.on_enter(spy, world)  # pyright: ignore[reportArgumentType]
    u._phase = "accept"  # pyright: ignore[reportPrivateUsage]
    u._phase_t = time.time()  # pyright: ignore[reportPrivateUsage]
    world.find_target = FindTarget(map_id=WAIT_MAP, x=16, y=34)
    world.record("auto_find")
    _ = u.tick(spy, world)  # pyright: ignore[reportArgumentType]
    assert u._phase == "run"  # pyright: ignore[reportPrivateUsage]
    kinds = [s[0] for s in spy.sent]
    assert kinds == [3013]  # 受理同拍即起跑(事件把 last_step_t 归零)
    p2 = spy.sent[0][3]
    assert p2 == (27 << 16) | 24


def test_unlock_arrival_sends_exit() -> None:
    """Given: 冲刺中距目标≤2 When: tick Then: 发 CS1011 @exit 收尾。"""
    world = ready_world(16, 32)
    world.find_target = FindTarget(map_id=WAIT_MAP, x=16, y=34)
    spy = SpyConn()
    u = UnlockBehavior(sid=1)
    u.on_enter(spy, world)  # pyright: ignore[reportArgumentType]
    u._phase = "run"  # pyright: ignore[reportPrivateUsage]
    u._phase_t = time.time()  # pyright: ignore[reportPrivateUsage]
    u._last_step_t = 0.0  # pyright: ignore[reportPrivateUsage]
    _ = u.tick(spy, world)  # pyright: ignore[reportArgumentType]
    kinds = [s[0] for s in spy.sent]
    assert kinds == [1011]
    body = spy.sent[0][5].decode()
    assert json.loads(body)["Act"] == "@exit"


def test_unlock_done_when_confirmed_in_walkable() -> None:
    """Given: SC1 已确认且 y≥5 When: tick Then: DONE。"""
    world = ready_world(23, 27)
    spy = SpyConn()
    u = UnlockBehavior(sid=1)
    u.on_enter(spy, world)  # pyright: ignore[reportArgumentType]
    u._move_confirmed = 1  # pyright: ignore[reportPrivateUsage]
    world.record("self_move_confirmed")
    result = u.tick(spy, world)  # pyright: ignore[reportArgumentType]
    assert result.value == "done"


# ---- update.py 新增 SC 帧解析 ----


def test_sc1_confirms_position() -> None:
    """Given: SC1 确认帧 When: updater.apply Then: 位置更新+事件。"""
    from pbot.world.update import WorldUpdater

    world = WorldMirror()
    world.me.uid = "u1"
    updater = WorldUpdater()
    wire = build_frame(1, 21, 28, 5, 32368232, b"", tail=817)
    f = parse_frame(wire)
    updater.apply(world, f)
    assert world.me.x == 21 and world.me.y == 28
    assert world.events[-1].kind == "self_move_confirmed"


def test_sc28_anchor_echo_not_reject() -> None:
    """Given: SC28 p3=32368232(锚点回显) When: apply Then: 不计拒绝。"""
    from pbot.world.update import WorldUpdater

    world = WorldMirror()
    world.me.map_name = WAIT_MAP
    updater = WorldUpdater()
    wire = build_frame(28, 0, 23, 27, 32368232, b"")
    f = parse_frame(wire)
    updater.apply(world, f)
    kinds = [e.kind for e in world.events]
    assert "pull_back" not in kinds
    assert "anchor_echo" in kinds


def test_sc31_my_strike_feeds_hp() -> None:
    """Given: SC31 受击者是我 When: apply Then: me.hp 实时更新。

    回归: SC53 面板帧稀疏(仅登录/死亡推), 不喂 SC31 血量则
    低血逃跑永远只在 hp=0(已死)时触发(d401 死亡循环)。
    """
    from pbot.codec.bitbuf import encode6
    from pbot.world.update import WorldUpdater

    world = WorldMirror()
    world.me.uid = "14558825001790500354000067138"
    world.me.actor_key = 21889350
    updater = WorldUpdater()
    body = encode6(json.dumps(
        {"Hp": 80, "MaxHp": 305, "Damage": 60,
         "UserID": "21889350",
         "AttackUserID": "61526"}).encode())
    wire = build_frame(31, 0, 0, 0, 0, body)
    f = parse_frame(wire)
    updater.apply(world, f)
    assert world.me.hp == 80 and world.me.max_hp == 305


def test_sc27_revive_restores_full_hp() -> None:
    """Given: 死亡 hp=0 收 SC27 When: apply Then: 复活+回满血。

    回归: 回城复活后 SC53 不重推, hp 残留 0 会立即误触发低血逃跑。
    """
    from pbot.world.update import WorldUpdater

    world = WorldMirror()
    world.me.uid = "u1"
    world.me.hp, world.me.max_hp = 0, 305
    world.me.is_dead = True
    updater = WorldUpdater()
    wire = build_frame(27, 0, 0, 0, 0, b"")
    f = parse_frame(wire)
    updater.apply(world, f)
    assert world.me.is_dead is False
    assert world.me.hp == 305


def test_sc31_my_lethal_strike_records_me_die() -> None:
    """Given: SC31 受击者是我且 Hp=0 When: apply Then: is_dead + me_die 事件。

    回归: 我被怪打死时服务器不发 SC32, 只有 SC31(hp=0)+SC634 回城 —
    缺 me_die 则逃跑阶段继续发移动帧("跑尸")。
    """
    from pbot.codec.bitbuf import encode6
    from pbot.world.update import WorldUpdater

    world = WorldMirror()
    world.me.uid = "14558825001790500354000067138"
    world.me.actor_key = 21889350
    updater = WorldUpdater()
    body = encode6(json.dumps(
        {"Hp": 0, "MaxHp": 407, "Damage": 72,
         "UserID": "21889350",
         "AttackUserID": "62471"}).encode())
    wire = build_frame(31, 0, 0, 0, 0, body)
    f = parse_frame(wire)
    updater.apply(world, f)
    assert world.me.is_dead is True
    assert world.me.hp == 0
    assert "me_die" in [e.kind for e in world.events]


def test_sc20102_sets_find_target() -> None:
    """Given: SC20102(body=encode6 JSON) When: apply Then: find_target 就位。"""
    from pbot.codec.bitbuf import encode6
    from pbot.world.update import WorldUpdater

    world = WorldMirror()
    updater = WorldUpdater()
    body = encode6(json.dumps(
        {"mapID": "等待地图", "x": 16, "y": 34,
         "targetType": 100, "targetIndex": [34]}).encode())
    wire = build_frame(20102, 0, 0, 0, 0, body)
    f = parse_frame(wire)
    updater.apply(world, f)
    assert world.find_target is not None
    assert world.find_target.x == 16 and world.find_target.y == 34


def test_cs3013_wire_layout() -> None:
    """Given: 跑帧 When: build_frame Then: 布局同 3011(cap_unlock 同构)。"""
    wire = build_frame(3013, 0, 5, (27 << 16) | 23, 159, WAIT_MAP.encode())
    f = parse_frame(wire)
    assert f.msg_id == 3013 and f.p2 == (27 << 16) | 23 and f.p3 == 159