chuanqi/protocol_bot/tests/test_monster_discovery.py

"""怪物发现机制修复验证(2026-09-30 run8 破案)。

抓包定案(cap_mine_walk 370s 矿区实走 / cap_live2 盟重挂机):
- 本服 SC10 只推玩家(0帧怪物出现); 怪物被动源 = SC11 移动广播(5位数短ID)
  + SC14 攻击广播 — 旧 _on_actor_move 只更新已知怪, 未知 5 位数 ID 直接
  丢弃 = "路过怪不杀"直接根因
- SC10 真实布局(客户端 handle_InOfView + converActorData 同构):
  key=body.info[0], 坐标=body.X/body.Y, race 分类 0/1/150=玩家 50=NPC
  其余=怪; 帧头 p1=(y<<16)|x 只是坐标回显, 绝不是实体 ID
- SC10 6位数短id(288999 等, recog 5xxxx 人形外形)=人形怪/英雄/宠物,
  不入怪表(玩家量级)
"""

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))

import json

from pbot.codec.bitbuf import encode6
from pbot.codec.frame import Frame
from pbot.world.mirror import WorldMirror
from pbot.world.update import WorldUpdater


def mk_frame(mid: int, recog: int = 0, p1: int = 0, p2: int = 0, p3: int = 0,
             body_json: dict | None = None) -> Frame:
    body = encode6(json.dumps(body_json or {}).encode()) \
        if body_json is not None else b""
    return Frame(msg_id=mid, recog=recog, p1=p1, p2=p2, p3=p3, tail=0,
                 pre=0, body=body)


def sc11(uid: str, x: int, y: int, d: int = 4) -> Frame:
    """SC11 移动广播: 头 recog=X p1=Y, body=encode6(JSON{UserID})."""
    return mk_frame(11, recog=x, p1=y, p2=d, body_json={"UserID": uid})


# ---------- SC11 雷达入表 ----------

def test_sc11_unknown_5digit_monster_enters_table() -> None:
    """未知 5 位数短ID移动广播 → 入怪表(矿区 43217/56431 实证=怪)."""
    w = WorldMirror()
    up = WorldUpdater()
    up.apply(w, sc11("43217", 103, 174))
    m = w.monsters.get(43217)
    assert m is not None, "5位数未知ID必须雷达入表"
    assert (m.x, m.y) == (103, 174)
    assert not m.is_dead


def test_sc11_known_monster_updates_position() -> None:
    """已入表怪: SC11 只更新坐标(不覆盖名字/race)."""
    w = WorldMirror()
    up = WorldUpdater()
    up.apply(w, sc11("43217", 103, 174))
    up.apply(w, sc11("43217", 100, 170))
    m = w.monsters.get(43217)
    assert m is not None and (m.x, m.y) == (100, 170)


def test_sc11_player_uid_ignored() -> None:
    """29位长uid(玩家)移动广播不入怪表(盟重 14504562… 实证)."""
    w = WorldMirror()
    up = WorldUpdater()
    long_uid = "14504562001790120981000064127"
    up.apply(w, sc11(long_uid, 330, 327))
    assert not w.monsters
    w.me.uid = long_uid
    up.apply(w, sc11(long_uid, 331, 328))
    assert not w.monsters


def test_sc11_6digit_monster_enters() -> None:
    """真实客户端抓包定案(cap_client_mine2): 矿区怪短ID为6位数
    (111340/111345/100273 被真客户端攻击, 111377 经SC11广播) —
    旧 >=100000 过滤把整代怪拒之门外."""
    w = WorldMirror()
    up = WorldUpdater()
    up.apply(w, sc11("111340", 65, 40))
    m = w.monsters.get(111340)
    assert m is not None and (m.x, m.y) == (65, 40)


def test_sc11_7digit_still_rejected() -> None:
    """7位数(2293955 宠物/玩家量级, r45 误攻实证)仍不入表."""
    w = WorldMirror()
    up = WorldUpdater()
    up.apply(w, sc11("2293955", 65, 40))
    assert not w.monsters


def test_sc31_attacker_enters_table() -> None:
    """SC31 受击: AttackUserID=攻击怪真实ID(cap_client_mine2 实证
    111340 连打玩家 100+s) — 打我的怪立即入表(哪怕暂无坐标,
    SC14/SC11 到达后坐标补全). 受击揭示=反击的前提."""
    w = WorldMirror()
    w.me.uid = "14570166001790661813000091989"
    up = WorldUpdater()
    body = {"Hp": 457, "MaxHp": 575, "Damage": 6,
            "UserID": w.me.uid, "AttackUserID": "111340"}
    up.apply(w, mk_frame(31, body_json=body))
    m = w.monsters.get(111340)
    assert m is not None, "攻击我的怪必须入表(受击揭示)"
    assert m.name == "attacker"


def test_sc31_attacker_coord_fills_when_known() -> None:
    """受击揭示的怪: 后续 SC14/SC11 到达时坐标补全."""
    w = WorldMirror()
    w.me.uid = "14570166001790661813000091989"
    up = WorldUpdater()
    up.apply(w, mk_frame(31, body_json={
        "Hp": 457, "MaxHp": 575, "Damage": 6,
        "UserID": w.me.uid, "AttackUserID": "111340"}))
    up.apply(w, mk_frame(14, body_json={
        "UserID": "111340", "Dir": 6, "X": 45, "Y": 59,
        "MagicID": 0, "ScriptEffect": 0}))
    m = w.monsters.get(111340)
    assert m is not None and (m.x, m.y) == (45, 59)


def test_sc10_humanoid_registry_blocks_sc11() -> None:
    """SC10 人形(玩家/英雄 race 0/1/150)登记入 humanoids —
    其 SC11 移动广播永不入怪表(盟重 288999 巫师之神实证=英雄)."""
    w = WorldMirror()
    w.me.map_name = "比奇矿区"
    up = WorldUpdater()
    up.apply(w, mon_sc10("288999", 150, 330, 327, name="巫师之神"))
    assert 288999 not in w.monsters
    up.apply(w, sc11("288999", 331, 328))
    assert 288999 not in w.monsters
    assert 288999 in w.humanoids


def test_sc11_7digit_pet_ignored() -> None:
    """7位数(2293955 宠物/玩家量级, r45 误攻实证)不入怪表."""
    w = WorldMirror()
    up = WorldUpdater()
    up.apply(w, sc11("2293955", 330, 327))
    assert not w.monsters


def test_sc30_removes_radar_monster() -> None:
    """SC30 出视野: 雷达入表的怪同样可移除(不残留幽灵目标)."""
    w = WorldMirror()
    up = WorldUpdater()
    up.apply(w, sc11("43217", 103, 174))
    up.apply(w, mk_frame(30, body_json={"UserID": "43217"}))
    assert 43217 not in w.monsters


# ---------- SC10 重写(info[0]/X/Y/race) ----------

def mon_sc10(info0: str, race: int, x: int, y: int, name: str = "矿区僵尸",
             recog: int = 1049) -> Frame:
    return mk_frame(10, recog=recog, p1=(y << 16) | x, p2=4 << 16,
                    body_json={"info": [info0, name, 1, 0, 1],
                               "X": x, "Y": y, "race": race, "rs": race})


def test_sc10_monster_enters_via_info0_and_xy() -> None:
    """真怪物 SC10: key=info[0], 坐标=body.X/Y(非帧头打包值的拆解)."""
    w = WorldMirror()
    w.me.map_name = "比奇矿区"
    up = WorldUpdater()
    up.apply(w, mon_sc10("44871", 1049, 77, 122))
    m = w.monsters.get(44871)
    assert m is not None, "info[0] 才是实体ID"
    assert (m.x, m.y) == (77, 122)
    assert m.name == "矿区僵尸"


def test_sc10_player_race_ignored() -> None:
    """race=0/1/150(玩家)不入怪表 — 即使 info[0] 是 5 位数."""
    w = WorldMirror()
    w.me.map_name = "比奇矿区"
    up = WorldUpdater()
    up.apply(w, mon_sc10("55555", 150, 331, 332, name="某玩家", recog=55543))
    assert not w.monsters


def test_sc10_npc_goes_to_npc_table() -> None:
    """race=50 → NPC 表(不打)."""
    w = WorldMirror()
    w.me.map_name = "比奇矿区"
    up = WorldUpdater()
    up.apply(w, mon_sc10("10002", 50, 342, 338, name="传送员",
                         recog=50))
    assert not w.monsters
    assert any("传送员" in n.name for n in w.npcs.values())


def test_sc10_coordinate_not_used_as_id() -> None:
    """帧头 p1 坐标打包值绝不当实体ID(run7 #7536683=坐标值误当怪)."""
    w = WorldMirror()
    w.me.map_name = "比奇矿区"
    up = WorldUpdater()
    up.apply(w, mon_sc10("44871", 1049, 43, 115))
    assert 44871 in w.monsters
    bad_key = (115 << 16) | 43
    assert bad_key not in w.monsters
    assert all(k < 100000 for k in w.monsters)


# ---------- §2.6 翻案修复: 双层编码 SC10 + p1 坐标回退 (2026-09-30 午后) ----------

def test_batch_frame_double_encoded_sc10() -> None:
    """双层编码 -1 批量帧(解压后整体 6bit 文本态)内的 SC10 必须入镜.

    cap_client_mine2 t=169.89 实证: 矿区怪物 SC10 全藏于此形态,
    旧 apply_batch 解析 0 子帧静默丢弃 = "零推怪"假结论根因.
    """
    import zlib

    from pbot.codec.frame import build_frame
    from pbot.game.session import apply_batch

    sc10_json = json.dumps({
        "info": ["113202", "僵尸"], "rs": 96,
        "IsBoss": False,
    }).encode()
    sc10 = build_frame(10, 1079, (27 << 16) | 39, 7, 134,
                       encode6(sc10_json), pre=(27 << 16) | 39)
    batch = Frame(msg_id=-1, recog=207, p1=299995, p2=0, p3=-10000,
                  pre=0, tail=207, body=zlib.compress(sc10))
    w = WorldMirror()
    w.me.map_name = "比奇矿区"
    apply_batch(w, WorldUpdater(), [batch])
    m = w.monsters.get(113202)
    assert m is not None, "双层编码批量帧内 SC10 必须入怪表"
    assert (m.x, m.y) == (39, 27), "坐标从帧头 p1=(y<<16)|x 回退解析"
    assert m.name == "僵尸"
    assert not m.is_boss


def test_sc10_header_pre_large_value_accepted() -> None:
    """子帧头 pre=(y<<16)|x 压缩坐标(实测 2752563)不得当伪帧头丢弃.

    FrameSplitter 旧版 pre<2**13 约束 = SC10 整代丢失根因之一.
    """
    from pbot.codec.frame import FrameSplitter, build_frame

    sc10_json = json.dumps(
        {"info": ["100989", "僵尸"], "rs": 96}).encode()
    sc10 = build_frame(10, 1079, (70 << 16) | 38, 7, 134,
                       encode6(sc10_json), pre=(70 << 16) | 38)
    frames = FrameSplitter().feed(sc10)
    assert len(frames) == 1, "pre 为大值的合法子帧必须被切出"
    assert frames[0].msg_id == 10


def test_sc10_isboss_flag_entered() -> None:
    """SC10 档案 IsBoss=true → 镜像怪 is_boss=True(BOSS 识别免费通道)."""
    w = WorldMirror()
    w.me.map_name = "比奇矿区"
    up = WorldUpdater()
    up.apply(w, mon_sc10("99999", 96, 40, 70, name="尸王"))
    # mon_sc10 无 IsBoss 字段 — 直接补一帧带 IsBoss 的
    up.apply(w, mk_frame(10, recog=1079, p1=(70 << 16) | 40, p2=4 << 16,
                         body_json={"info": ["88888", "触龙神"], "rs": 96,
                                    "X": 40, "Y": 70, "IsBoss": True}))
    assert w.monsters[88888].is_boss
    assert not w.monsters[99999].is_boss