chuanqi/protocol_bot/tests/test_automove.py

"""AutoMoveBehavior(5013 自动寻路)行为验证(不发真实网络 — 桩发送)。"""

import time
from typing import TYPE_CHECKING, cast

from pbot.bot.automove import (
    AUTO_MOVE_ABORT,
    AUTO_MOVE_FINISH,
    AUTO_MOVE_START,
    AutoMoveBehavior,
)
from pbot.world.mirror import WorldMirror
from tests.test_wander import SpyConn

if TYPE_CHECKING:
    from pbot.bot.behavior import TickResult
    from pbot.transport.gateway import GatewayConnection


def tick(b: AutoMoveBehavior, spy: SpyConn,
         world: WorldMirror) -> "TickResult":
    conn = cast("object", spy)
    return b.tick(cast("GatewayConnection", conn), world)


def ready_world(x: int, y: int) -> WorldMirror:
    world = WorldMirror()
    # 用非映射图名: 本文件测帧布局机制(START/FINISH/ABORT/3013), 不测真实
    # 地形 — "等待地图"已映射 0139.map(46x50) 会加载真实墙, 坐标会越界
    world.me.map_name = "测试机制图"
    world.me.actor_key = 7
    world.me.x, world.me.y = x, y
    return world


def make(spy: SpyConn, world: WorldMirror,
         dest: tuple[int, int]) -> AutoMoveBehavior:
    b = AutoMoveBehavior(sid=159, dest_x=dest[0], dest_y=dest[1])
    conn = cast("object", spy)
    b.on_enter(cast("GatewayConnection", conn), world)
    return b


def test_start_notify_layout() -> None:
    """CS5013 START 帧布局: recog=X p1=Y p2=1 p3=sid(5013,1,x,y Lua 真源)。"""
    world = ready_world(23, 27)
    spy = SpyConn()
    b = make(spy, world, (40, 60))
    _ = tick(b, spy, world)  # init → notify
    assert spy.sent == []
    _ = tick(b, spy, world)  # notify → START + moving
    assert len(spy.sent) == 1
    msg_id, recog, p1, p2, p3, body = spy.sent[0]
    assert msg_id == 5013
    assert recog == 40
    assert p1 == 60
    assert p2 == AUTO_MOVE_START
    assert p3 == 159
    assert body == b""


def test_arrive_sends_finish() -> None:
    """到达目标(切比雪夫≤2) → CS5013 act=3 FINISH → DONE。"""
    world = ready_world(38, 59)
    spy = SpyConn()
    b = make(spy, world, (40, 60))
    _ = tick(b, spy, world)  # init → notify
    _ = tick(b, spy, world)  # notify: START + 规划
    _ = tick(b, spy, world)  # moving: 到达判定
    kinds = [s[0] for s in spy.sent]
    assert kinds[-1] == 5013
    finish = spy.sent[-1]
    assert finish[3] == AUTO_MOVE_FINISH  # p2=3
    del b


def test_abort_when_no_path() -> None:
    """目标被 2 格厚墙带完全隔开(A* 无路) → 不发 START, 直接 ABORT(2)。

    注: A* 步长 2 格可越过 1 格薄墙(mid 检查只看中间格), 围墙须 ≥2 格厚。
    """
    world = ready_world(23, 27)
    spy = SpyConn()
    b = make(spy, world, (23, 45))
    for y in range(33, 38):
        for x in range(1024):
            b._grid.mark_blocked((x, y))  # pyright: ignore[reportPrivateUsage]
    _ = tick(b, spy, world)  # init → notify
    _ = tick(b, spy, world)  # notify: 规划失败
    kinds = [(s[0], s[3]) for s in spy.sent]
    assert kinds == [(5013, AUTO_MOVE_ABORT)]
    del b


def test_moving_sends_3013_with_prediction() -> None:
    """移动中逐腿 CS3013, pre=预测落点(2 格), 位置前进。"""
    world = ready_world(23, 27)
    spy = SpyConn()
    b = make(spy, world, (23, 45))
    _ = tick(b, spy, world)  # init
    _ = tick(b, spy, world)  # notify(START)
    assert spy.sent[0][0] == 5013
    _ = tick(b, spy, world)  # moving: 第一腿
    legs = [s for s in spy.sent if s[0] == 3013]
    assert len(legs) == 1
    _, recog, p1, p2, p3, body = legs[0]
    del recog, body
    assert p1 == 4  # 向下 dir=4
    assert p2 == (27 << 16) | 23  # 起点
    assert p3 == 159
    assert spy.pres[-1] == ((27 + 2) << 16) | 23  # pre=预测落点


def test_pull_back_marks_blocked_and_replans() -> None:
    """SC28 弹回 → 禁行格记录 + 连续 2 次触发重规划(路径更新)。"""
    world = ready_world(23, 27)
    spy = SpyConn()
    b = make(spy, world, (23, 45))
    _ = tick(b, spy, world)  # init
    _ = tick(b, spy, world)  # notify
    _ = tick(b, spy, world)  # 第一腿(dir=4)
    world.record("pull_back")
    b._last_step_t = time.time() - 2.0  # pyright: ignore[reportPrivateUsage]
    b._pending_t = time.time() - 2.0  # pyright: ignore[reportPrivateUsage]
    _ = tick(b, spy, world)  # 第二腿(重规划后)
    world.record("pull_back")
    b._last_step_t = time.time() - 2.0  # pyright: ignore[reportPrivateUsage]
    b._pending_t = time.time() - 2.0  # pyright: ignore[reportPrivateUsage]
    _ = tick(b, spy, world)  # 重规划触发
    blocked = (23, 27 + 2)  # dir=4 弹回落点: me 未推进仍 (23,27) + 2 格
    assert blocked in b._grid.blocked  # pyright: ignore[reportPrivateUsage]
    legs = [s for s in spy.sent if s[0] == 3013]
    assert len(legs) >= 2
    del b


def test_server_auto_move_event_consumed() -> None:
    """SC5013 下行事件(server_auto_move)可被消费且不崩溃。"""
    world = ready_world(23, 27)
    spy = SpyConn()
    b = make(spy, world, (23, 45))
    world.record("server_auto_move", "(30,30)")
    _ = tick(b, spy, world)
    assert len(world.events) == 0  # 事件已清空
    del b


def test_waits_until_world_known() -> None:
    """SC50/SC51 未就绪 → 不发帧。"""
    world = WorldMirror()
    spy = SpyConn()
    b = make(spy, world, (40, 60))
    _ = tick(b, spy, world)
    assert spy.sent == []
    del b