76 lines
2.6 KiB
Python
76 lines
2.6 KiB
Python
"""端到端互操作：Python 客户端 ←→ Node 服务端（跨实现）。
|
||
|
||
价值：这是**唯一能证明协议层真通**的测试 ——
|
||
两边是完全独立的实现，若我的 protodef 有偏差，握手不可能成功。
|
||
|
||
用法：
|
||
1) cd /home/newqqagent/plugin_data/mc/runtime && node testserver.js &
|
||
2) python3 tests/test_e2e_interop.py
|
||
"""
|
||
import sys, os, socket
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
from mcfold.client import Client
|
||
|
||
HOST, PORT = "127.0.0.1", 25599
|
||
|
||
def reachable():
|
||
try:
|
||
socket.create_connection((HOST, PORT), timeout=2).close(); return True
|
||
except OSError:
|
||
return False
|
||
|
||
def _start_server():
|
||
"""自动起 Node 服务端（独立实现），让本测试**总能真跑**而不是跳过。"""
|
||
import subprocess, time, shutil
|
||
runtime = "/home/newqqagent/plugin_data/mc/runtime"
|
||
script = os.path.join(os.path.dirname(os.path.abspath(__file__)), "support_testserver.js")
|
||
if not shutil.which("node"):
|
||
return None
|
||
shutil.copy(script, os.path.join(runtime, "support_testserver.js"))
|
||
p = subprocess.Popen(["node", "support_testserver.js", str(PORT)], cwd=runtime,
|
||
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
|
||
for _ in range(50):
|
||
if reachable(): return p
|
||
if p.poll() is not None: return None
|
||
time.sleep(0.2)
|
||
return None
|
||
|
||
def test_interop():
|
||
proc = None
|
||
if not reachable():
|
||
proc = _start_server()
|
||
if not reachable():
|
||
print(f" - 服务端 {HOST}:{PORT} 不可用，跳过"); return
|
||
try:
|
||
_do_interop()
|
||
finally:
|
||
if proc:
|
||
proc.terminate()
|
||
try: proc.wait(timeout=5)
|
||
except Exception: proc.kill()
|
||
try: os.remove("/home/newqqagent/plugin_data/mc/runtime/support_testserver.js")
|
||
except OSError: pass
|
||
|
||
def _do_interop():
|
||
c = Client(HOST, PORT, "PyProbe", "1.20.1")
|
||
try:
|
||
r = c.run_smoke()
|
||
assert r["protocol"] == 763, r
|
||
assert r["login_ok"], r
|
||
assert "success" in r["packets"], r
|
||
print(f" ✓ 跨实现互操作: 协议={r['protocol']} 包={r['packets']} 登录成功")
|
||
finally:
|
||
c.close()
|
||
|
||
def test_protocol_version_from_folder():
|
||
"""协议版本必须由「实例文件夹」决定，而不是硬编码。"""
|
||
c = Client(HOST, PORT, "x", "1.20.1")
|
||
c.load_protocol()
|
||
assert c.proto_num == 763, c.proto_num
|
||
print(f" ✓ 版本来自 version.json: 1.20.1 -> 协议 {c.proto_num}")
|
||
|
||
if __name__ == "__main__":
|
||
test_protocol_version_from_folder()
|
||
test_interop()
|
||
print("\n端到端互操作：通过")
|