chuanqi/tests/test_env_stub.py

# tests/test_env_stub.py
"""Task 5: EnvData + install_env (L_GameEnvManager/LuaBridgeCtl/FileUtils/gameEnviroment) + install_all 装配。"""
import socket
import threading
import time

import lupa.luajit21 as lj
import pytest


def test_env_manager_keys(tmp_path):
    from apilot.stubs.env import EnvData, install_env

    rt = lj.LuaRuntime()
    env = EnvData({
        "uid": "1186115037",
        "token": "jwt.x",
        "isBoxLogin": 1,
        "signkey": "k" * 32,
    })
    install_env(rt, env, str(tmp_path))
    assert rt.eval("L_GameEnvManager:GetEnvDataByKey('uid')") == "1186115037"
    assert rt.eval("L_GameEnvManager:GetEnvDataByKey('isBoxLogin')") == 1
    assert rt.eval("L_GameEnvManager:GetEnvDataByKey('missing')") is None
    rt.execute("global = global or {}; global.L_GameEnvManager = L_GameEnvManager")
    assert rt.eval("global.L_GameEnvManager:GetEnvDataByKey('token')") == "jwt.x"


def test_bridge_ctl_and_fileutils(tmp_path):
    from apilot.stubs.env import EnvData, install_env

    rt = lj.LuaRuntime()
    install_env(rt, EnvData({}), str(tmp_path))
    # 心跳开关默认开
    assert rt.eval("LuaBridgeCtl:Inst():GetModulesSwitch(2)") == 1
    rt.execute("LuaBridgeCtl:Inst():SetModulesSwitch(2, 1)")
    assert rt.eval("LuaBridgeCtl:Inst():GetModulesSwitch(2)") == 1
    # FileUtils 写入相对路径
    rt.execute("cc.FileUtils:getInstance():writeStringToFile('hi', 'a.txt')")
    assert (tmp_path / "a.txt").read_text() == "hi"
    # GET_STRING 未命中返回空串
    assert rt.eval("GET_STRING(99999)") == ""


def test_fileutils_path_safety(tmp_path):
    """writeStringToFile/isFileExist 拒绝逃逸 state_dir 的相对路径。"""
    from apilot.stubs.env import EnvData, install_env

    rt = lj.LuaRuntime()
    install_env(rt, EnvData({}), str(tmp_path))
    # 子目录写入 OK
    rt.execute("cc.FileUtils:getInstance():writeStringToFile('ok', 'sub/b.txt')")
    assert (tmp_path / "sub" / "b.txt").read_text() == "ok"
    # writeStringToFile 路径逃逸被拒
    with pytest.raises(Exception):
        rt.execute("cc.FileUtils:getInstance():writeStringToFile('x', '../escape.txt')")
    # isFileExist 逃逸路径返回 false
    assert rt.eval("cc.FileUtils:getInstance():isFileExist('../escape.txt')") is False


def test_install_all_networkctl_roundtrip(tmp_path):
    """install_all 装配后 Lua 侧可经 global.networkCtl 走 netClient 抵达环回网关。"""
    from apilot.stubs import install_all
    from apilot.stubs.env import EnvData
    from apilot.stubs.netclient import NetClientStub
    from apilot.stubs.networkctl import NetworkCtlStub
    from apilot.stubs.scheduler import Scheduler

    received = bytearray()
    finished = threading.Event()

    def gateway(server_sock):
        conn, _ = server_sock.accept()
        try:
            conn.settimeout(2.0)
            while not finished.is_set():
                try:
                    chunk = conn.recv(4096)
                except (socket.timeout, OSError):
                    break
                if not chunk:
                    break
                received.extend(chunk)
        finally:
            conn.close()
            server_sock.close()

    srv = socket.socket()
    srv.bind(("127.0.0.1", 0))
    srv.listen(1)
    port = srv.getsockname()[1]
    threading.Thread(target=gateway, args=(srv,), daemon=True).start()

    nc = NetClientStub()
    nc.connect("127.0.0.1", port)
    ctl = NetworkCtlStub(nc)
    sched = Scheduler()
    env = EnvData({"uid": "u1"})

    rt = lj.LuaRuntime()
    install_all(rt, nc, ctl, sched, env, str(tmp_path))
    rt.execute("global.networkCtl:SendMsg(529, 12345, 1, 2, 3, 'hi', 2)")

    time.sleep(0.2)
    finished.set()
    nc.disconnect()
    time.sleep(0.1)

    # 48 字节 encoded header + 2 字节 body
    assert len(received) >= 50, f"expected 50 bytes, got {len(received)}: {bytes(received)!r}"
    assert bytes(received[48:50]) == b"hi"