chuanqi/pipeline_bot/tests/test_pipeline_units.py

"""流水线新增单元测试: SC52 等级解析。"""
import json

from pbot.codec.bitbuf import encode6
from pbot.codec.frame import build_frame, parse_frame
from pbot.world.mirror import WorldMirror
from pbot.world.update import WorldUpdater


def _sc52_frame(level: int, hp: int = 100) -> bytes:
    body = json.dumps({"job": 0, "level": level, "hp": hp,
                       "maxhp": hp, "exp": 0, "maxexp": 100}).encode()
    return build_frame(52, 0, 0, 0, 0, encode6(body))


def test_sc52_updates_level() -> None:
    w = WorldMirror()
    u = WorldUpdater()
    u.apply(w, parse_frame(_sc52_frame(35)))
    assert w.me.level == 35
    u.apply(w, parse_frame(_sc52_frame(36, hp=120)))
    assert w.me.level == 36
    assert w.me.hp == 120  # maxhp/hp 同步(若实现)或仅 level(见下)


def test_sc52_level_zero_ignored() -> None:
    w = WorldMirror()
    u = WorldUpdater()
    u.apply(w, parse_frame(_sc52_frame(0)))
    assert w.me.level == 0  # 0 不产生误报


def test_sc52_level_up_callback_fires_once() -> None:
    w = WorldMirror()
    u = WorldUpdater()
    calls: list[tuple[int, int]] = []
    w.on_level_up = lambda old, new: calls.append((old, new))
    u.apply(w, parse_frame(_sc52_frame(34)))
    u.apply(w, parse_frame(_sc52_frame(35)))  # 升级 → 回调
    u.apply(w, parse_frame(_sc52_frame(35)))  # 同级重推 → 不回调
    u.apply(w, parse_frame(_sc52_frame(34)))  # 降级(异常帧) → 不回调
    assert calls == [(0, 34), (34, 35)]


# ---- naming ----
from pathlib import Path  # noqa: E402

from jiuyangfugu176.ctx import ChecklistWriter  # noqa: E402
from jiuyangfugu176.naming import random_cn_name, safe_filename  # noqa: E402


def test_random_cn_name_all_chinese() -> None:
    for _ in range(50):
        n = random_cn_name()
        assert 2 <= len(n) <= 8
        assert all("\u4e00" <= c <= "\u9fff" for c in n)


def test_safe_filename() -> None:
    assert safe_filename("玖扬复古176.38区/新角色") == "玖扬复古176.38区_新角色"
    assert safe_filename('a\\b:*?"<>|') == "a_b"
    assert safe_filename("   ") == "unnamed"


# ---- checklist ----
def test_checklist_writer_marks(tmp_path: Path) -> None:
    w = ChecklistWriter(tmp_path / "进度.md", "玖扬复古176.38区", "凌风")
    w.mark("T1 喝药保护", True, "state=on")
    w.mark("T2 比奇杀怪", False, "超时")
    txt = (tmp_path / "进度.md").read_text(encoding="utf-8")
    assert "☑" in txt and "☒" in txt and "凌风" in txt
    # header 与 note 行在 mark 重写后仍保留
    w.note("登录成功")
    w.mark("T3 新增项", True)
    txt = (tmp_path / "进度.md").read_text(encoding="utf-8")
    assert "# 传奇流水线进度" in txt
    assert "> " in txt and "登录成功" in txt
    # 二次 mark 同名任务 = 更新而非重复追加
    w.mark("T2 比奇杀怪", True, "20/20")
    txt = (tmp_path / "进度.md").read_text(encoding="utf-8")
    assert txt.count("T2 比奇杀怪") == 1
    assert txt.count("凌风") >= 1


def test_progress_checkpoint_roundtrip(tmp_path: Path) -> None:
    """断点打点: set_progress 写 md 注释, read_progress 取最新值(进程被杀续跑)。"""
    from jiuyangfugu176.ctx import read_progress

    f = tmp_path / "进度2.md"
    w = ChecklistWriter(f, "玖扬复古176.62区", "苍刻")
    assert read_progress(f, "kill") is None  # 无打点 → None(首跑)
    w.set_progress("kill", 5)
    w.set_progress("kill", 18)
    assert read_progress(f, "kill") == 18  # 取最新
    w.set_progress("kill", 0)  # 完成清理
    assert read_progress(f, "kill") == 0  # 0=已完成(调用方 or 0 → 从头)
    assert read_progress(f, "gold") is None  # 其他 key 不受影响


def test_checklist_resume_carries_marks(tmp_path: Path) -> None:
    """断线重跑续传: 新 writer 继承旧文件的勾选(get_mark 可查)。"""
    p = tmp_path / "进度.md"
    w1 = ChecklistWriter(p, "区A", "角色X")
    w1.mark("T1 开启喝药保护", True, "state=on")
    w1.mark("T3 比奇大城杀怪20", True, "比奇击杀 20/20")
    w2 = ChecklistWriter(p, "区A", "角色X")  # 同文件重开 = 续跑
    assert w2.get_mark("T1 开启喝药保护") == (True, "state=on")
    assert w2.get_mark("T3 比奇大城杀怪20") == (True, "比奇击杀 20/20")
    assert w2.get_mark("T9 沃玛寺庙挂机") is None  # 未做过的无记录
    txt = p.read_text(encoding="utf-8")
    assert "续跑" in txt and "2 项勾选" in txt
    # 新勾不丢旧勾
    w2.mark("T9 沃玛寺庙挂机", True, "100杀")
    txt = p.read_text(encoding="utf-8")
    assert "T3 比奇大城杀怪20" in txt and "T9 沃玛寺庙挂机" in txt


def test_checklist_resume_ignores_failed_marks(tmp_path: Path) -> None:
    """☒(失败)任务不得被续跑跳过 — 51区 17:13 实证: T10 失败记录被
    误当已勾跳过, 修复后 ☒ 不继承."""
    p = tmp_path / "进度.md"
    w1 = ChecklistWriter(p, "区A", "角色X")
    w1.mark("T8 开启五行之力", True, "面板已消费")
    w1.mark("T10 回盟重土城在线", False, "未落地盟重(当前 沃玛寺庙一层)")
    w2 = ChecklistWriter(p, "区A", "角色X")
    assert w2.get_mark("T8 开启五行之力") == (True, "面板已消费")
    assert w2.get_mark("T10 回盟重土城在线") is None, "失败记录不得继承"


# ---- zone_ports 新网关字段(2026-10-02 38区抓包定案) ----
def test_zone_entry_parses_gateway_fields() -> None:
    from pbot.login.zone_ports import _parse_zone_entry
    entry = _parse_zone_entry({"d": 14606326, "p": 8021, "s": 1,
                               "n": "裁决38区", "i": "gfgame.dhsf.xqhuyu.com",
                               "x": 12000})
    assert entry == {"port": 8021, "state": 1, "name": "裁决38区",
                     "gw_host": "gfgame.dhsf.xqhuyu.com", "gw_port": 12000}


def test_zone_entry_gateway_fields_default_empty() -> None:
    from pbot.login.zone_ports import _parse_zone_entry
    entry = _parse_zone_entry({"d": 14606325, "p": 8081, "s": 1, "n": "裁决37区"})
    assert entry["gw_host"] == ""
    assert entry["gw_port"] == 0


# ---- gateway host 覆盖(38区新网关架构) ----
def test_gateway_connection_host_override() -> None:
    import socket
    from threading import Thread

    from pbot.transport.gateway import GatewayConnection
    from pbot.types import ZoneId

    def _accept() -> None:
        srv.accept()[0].close()

    srv = socket.socket()
    srv.bind(("127.0.0.1", 0))
    srv.listen(1)
    port = srv.getsockname()[1]
    t = Thread(target=_accept, daemon=True)
    t.start()
    conn = GatewayConnection(ZoneId(14606326), port, host="127.0.0.1")
    assert conn.host == "127.0.0.1"
    conn.close()
    srv.close()
    t.join(timeout=2)


# ---- protocol_map_name: 洞穴移动帧 body(2026-10-01 洞内 392 帧静默根因回归) ----
from pbot.game.messages import protocol_map_name  # noqa: E402


def test_protocol_map_name_cave_display_passthrough() -> None:
    """天然洞穴: 服务器 MapId=显示名 → 移动帧 body 必须用中文名(d011 是
    2026-09-26 未验证假设, 真机 21040 帧定案任务图 body=显示名;
    2026-10-01 实跑 d011 全静默实证)。"""
    assert protocol_map_name("天然洞穴一层", "天然洞穴一层") == "天然洞穴一层"
    assert protocol_map_name("天然洞穴", "天然洞穴") == "天然洞穴"


def test_protocol_map_name_known_numeric_ids() -> None:
    """数字文件 ID(比奇=0/盟重=3)仍直传 — 与洞穴中文透传互不干扰。"""
    assert protocol_map_name("0", "比奇大城") == "0"
    assert protocol_map_name("3", "盟重土城") == "3"
    assert protocol_map_name("", "比奇矿区") == "比奇矿区"


# ---- bootstrap 纯函数 ----
from jiuyangfugu176.bootstrap import _checklist_fname, _zone_label  # noqa: E402


def test_zone_label() -> None:
    assert _zone_label("38区") == "玖扬复古176·38区"


def test_checklist_fname() -> None:
    assert _checklist_fname("38区", "凌风") == "进度_玖扬复古176_38区_凌风.md"
    assert _checklist_fname('38/区:*?"x', "凌\\风") == "进度_玖扬复古176_38_区_x_凌_风.md"


# ---- 抢怪避让: SC31 attacker≠我 → contested_t 打标, 我打 → 清标 ----
import json as _json  # noqa: E402

from pbot.codec.bitbuf import encode6 as _encode6  # noqa: E402
from pbot.codec.frame import Frame as _Frame  # noqa: E402
from pbot.world.actor import ActorKey as _ActorKey  # noqa: E402
from pbot.world.actor import Monster as _Monster  # noqa: E402
from pbot.world.mirror import WorldMirror as _WorldMirror  # noqa: E402


def _sc31(victim: str, attacker: str) -> _Frame:
    body = _json.dumps({"Hp": 10, "MaxHp": 20, "Damage": 5,
                        "UserID": victim, "AttackUserID": attacker}).encode()
    return _Frame(msg_id=31, recog=0, p1=0, p2=0, p3=0, pre=0, tail=0,
                  body=_encode6(body))


def test_sc31_other_attacker_marks_contended() -> None:
    w = _new_world_with_monster()
    WorldUpdater().apply(w, _sc31("44444", "9999000179080001"))
    assert w.monsters[44444].contested_t > 0.0


def test_sc31_my_hit_clears_contended() -> None:
    w = _new_world_with_monster()
    w.monsters[44444] = _Monster(key=_ActorKey(44444), x=5, y=5, race=0,
                                 name="鸡", is_dead=False, contested_t=123.0)
    WorldUpdater().apply(w, _sc31("44444", w.me.uid))
    assert w.monsters[44444].contested_t == 0.0


def _new_world_with_monster() -> "_WorldMirror":
    from pbot.world.mirror import WorldMirror
    w = WorldMirror()
    w.me.uid = "14590001790847124000069434"
    w.monsters[44444] = _Monster(key=_ActorKey(44444), x=5, y=5, race=0,
                                 name="鸡", is_dead=False)
    return w