mtgodot-poc/project/live_smoke_test.gd

# live_smoke_test —— 真实服务器端到端冒烟测试。
#
# 由 script/live_smoke_test.sh 以 MT_TEST_MODE=smoke 启动。测试流程：
#   自动登录 → 角色列表 → 进入游戏 → 世界/属性读取 → 移动/停止 → 聊天
#   → 可选重连 → 输出 JSON 报告并以 PASS/FAIL 退出。
#
# 只使用可逆的常规操作，不执行删角、丢弃、强化、交易等破坏性动作。
extends Node

const FUNC_WAIT := 0
const FUNC_MOVE := 1
const CHAT_TALKING := 0
const SETTLE_SECONDS := 2.5
const STEP_TIMEOUT_SECONDS := 25.0
const TOTAL_TIMEOUT_SECONDS := 120.0

var _flow: Node
var _client: Node
var _state := "boot"
var _state_since_ms := 0
var _started_ms := 0
var _entered_count := 0
var _phase_history: Array[String] = []
var _checks: Array[Dictionary] = []
var _failures: Array[String] = []
var _events := {
	"char_list": 0,
	"entered_game": 0,
	"entity_moved": 0,
	"chat": 0,
	"disconnected": 0,
}
var _origin := Vector3.ZERO
var _origin_valid := false
var _mutations := true
var _reconnect := true
var _finished := false
var _finish_queued := false
var _report_path := ""

func setup(flow: Node, m2client: Node) -> void:
	_flow = flow
	_client = m2client
	_started_ms = Time.get_ticks_msec()
	_state_since_ms = _started_ms
	_mutations = OS.get_environment("MT_SMOKE_MUTATIONS") != "0"
	_reconnect = OS.get_environment("MT_SMOKE_RECONNECT") != "0"
	_report_path = OS.get_environment("MT_TEST_REPORT").strip_edges()
	if _report_path.is_empty():
		_report_path = "user://live_smoke_report.json"

	if _client == null:
		_fail("M2Client 未创建")
		_finish_later()
		return
	if _client.has_signal("stage_changed"):
		_client.stage_changed.connect(_on_stage_changed)
	if _client.has_signal("phase_changed"):
		_client.phase_changed.connect(_on_phase_changed)
	if _client.has_signal("char_list"):
		_client.char_list.connect(_on_char_list)
	if _client.has_signal("entered_game"):
		_client.entered_game.connect(_on_entered_game)
	if _client.has_signal("login_failed"):
		_client.login_failed.connect(_on_login_failed)
	if _client.has_signal("disconnected"):
		_client.disconnected.connect(_on_disconnected)
	if _client.has_signal("entity_moved"):
		_client.entity_moved.connect(_on_entity_moved)
	if _client.has_signal("chat"):
		_client.chat.connect(_on_chat)

	_state = "waiting_login"
	_log("START mutations=%s reconnect=%s report=%s" % [_mutations, _reconnect, _report_path])
	_check(OS.get_environment("MT_PROTOCOL") == "classic", "classic 协议已启用")
	_check(OS.get_environment("MT_ACCOUNT") != "", "账号环境变量已提供")
	_check(OS.get_environment("MT_PASSWORD") != "", "密码环境变量已提供")

func _process(_delta: float) -> void:
	if _finished:
		return
	var now := Time.get_ticks_msec()
	var total_s := float(now - _started_ms) / 1000.0
	if total_s > TOTAL_TIMEOUT_SECONDS:
		_fail("总超时，当前状态=%s" % _state)
		_finish_later()
		return

	match _state:
		"settle_first":
			if _seconds_in_state(now) >= SETTLE_SECONDS:
				_run_world_checks()
		"wait_move":
			if _seconds_in_state(now) >= 1.5:
				_send_stop()
		"wait_chat":
			if _seconds_in_state(now) >= 0.5:
				_send_chat()
		"wait_reconnect":
			if _seconds_in_state(now) >= STEP_TIMEOUT_SECONDS:
				_fail("重连后未重新进入游戏")
				_finish_later()
		"settle_second":
			if _seconds_in_state(now) >= SETTLE_SECONDS:
				_check(_client != null and _client.is_in_game(), "重连后客户端仍处于游戏状态")
				var vid := int(_client.get_main_vid())
				_check(vid != 0 and not _client.get_entity(vid).is_empty(), "重连后主角重新建立")
				_check(not _client.get_points().is_empty() and _has_method_and_array("get_skills")
					and _has_method_and_array("get_quickslots"), "重连后属性/技能/快捷栏可读取")
				var scene: Node = _flow.get("_game")
				_check(is_instance_valid(scene) and bool(scene.get("_model_built")), "重连后真实角色模型建立")
				_finish_later()
		"waiting_login", "waiting_first_game":
			if _seconds_in_state(now) >= STEP_TIMEOUT_SECONDS:
				_fail("阶段超时：%s" % _state)
				_finish_later()

func _seconds_in_state(now: int) -> float:
	return float(now - _state_since_ms) / 1000.0

func _set_state(next_state: String) -> void:
	_state = next_state
	_state_since_ms = Time.get_ticks_msec()
	_log("STATE %s" % _state)

func _on_stage_changed(stage: String) -> void:
	_log("stage=%s" % stage)

func _on_phase_changed(phase: String) -> void:
	_phase_history.append(phase)
	_log("phase=%s" % phase)

func _on_char_list(characters: Array) -> void:
	_events["char_list"] += 1
	_check(not characters.is_empty(), "角色列表非空")
	if characters.is_empty():
		_fail("服务器返回空角色列表")
		_finish_later()
	else:
		_log("char_list count=%d" % characters.size())

func _on_entered_game() -> void:
	_entered_count += 1
	_events["entered_game"] = _entered_count
	_check(true, "第 %d 次进入游戏" % _entered_count)
	if _entered_count == 1:
		_set_state("settle_first")
	elif _state == "wait_reconnect":
		_set_state("settle_second")

func _on_login_failed(reason: String) -> void:
	_fail("登录失败：%s" % reason)
	_finish_later()

func _on_disconnected(reason: String) -> void:
	_events["disconnected"] += 1
	_log("disconnected=%s" % reason)
	if _state != "wait_reconnect" and _state != "settle_second":
		_fail("测试期间意外断线：%s" % reason)
		_finish_later()

func _on_entity_moved(_vid: int) -> void:
	_events["entity_moved"] += 1

func _on_chat(_type: int, _vid: int, _text: String) -> void:
	_events["chat"] += 1

func _run_world_checks() -> void:
	var in_game: bool = _client != null and _client.has_method("is_in_game") and _client.is_in_game()
	_check(in_game, "客户端处于 in_game")
	var main_vid := int(_client.get_main_vid()) if _client.has_method("get_main_vid") else 0
	_check(main_vid != 0, "主角色 VID 有效")
	if main_vid == 0:
		_fail("主角色不存在，无法继续常规测试")
		_finish_later()
		return

	var main_entity: Dictionary = _client.get_entity(main_vid)
	_check(not main_entity.is_empty(), "主角色实体可读取")
	var entities: Array = _client.get_entities()
	_check(not entities.is_empty(), "世界实体列表可读取")
	var points: Dictionary = _client.get_points()
	_check(not points.is_empty(), "属性数据可读取")
	_check(_has_method_and_array("get_inventory"), "背包接口可读取")
	_check(_has_method_and_array("get_skills"), "技能接口可读取")
	_check(_has_method_and_array("get_quickslots"), "快捷栏接口可读取")
	_check(_has_method_and_array("get_affects"), "状态接口可读取")

	var p: Variant = main_entity.get("pos_cm", null)
	if p is Vector3:
		_origin = p
		_origin_valid = true
		_log("main_vid=%d pos_cm=%s entities=%d" % [main_vid, p, entities.size()])
	else:
		_fail("主角色缺少 pos_cm")

	if not _mutations:
		_start_reconnect()
		return
	_send_move()

func _has_method_and_array(method_name: String) -> bool:
	if _client == null or not _client.has_method(method_name):
		return false
	var value: Variant = _client.call(method_name)
	return value is Array

func _send_move() -> void:
	if not _origin_valid:
		_finish_later()
		return
	var ok := bool(_client.move(FUNC_MOVE, 0, 0.0, int(_origin.x) + 100, int(_origin.y)))
	_check(ok, "发送 1 米移动包")
	_set_state("wait_move")

func _send_stop() -> void:
	var ok := bool(_client.move(FUNC_WAIT, 0, 0.0, int(_origin.x), int(_origin.y)))
	_check(ok, "发送停止移动包")
	_set_state("wait_chat")

func _send_chat() -> void:
	var ok := bool(_client.say(CHAT_TALKING, "[smoke] client live smoke test"))
	_check(ok, "发送普通聊天包")
	_start_reconnect()

func _start_reconnect() -> void:
	if not _reconnect:
		_finish_later()
		return
	var reconnect_ok := bool(_client.reconnect())
	_check(reconnect_ok, "调用客户端重连")
	if reconnect_ok:
		_set_state("wait_reconnect")
	else:
		_finish_later()

func _check(ok: bool, name: String) -> void:
	_checks.append({"name": name, "ok": ok})
	if ok:
		print("LIVE_SMOKE PASS: " + name)
	else:
		_fail(name)

func _fail(message: String) -> void:
	if not _failures.has(message):
		_failures.append(message)
	printerr("LIVE_SMOKE FAIL: " + message)

func _log(message: String) -> void:
	print("LIVE_SMOKE: " + message)

func _finish_later() -> void:
	if _finish_queued or _finished:
		return
	_finish_queued = true
	call_deferred("_finish")

func _finish() -> void:
	if _finished:
		return
	_finished = true
	var passed := _failures.is_empty()
	var report := {
		"status": "PASS" if passed else "FAIL",
		"started_at_ms": _started_ms,
		"duration_seconds": float(Time.get_ticks_msec() - _started_ms) / 1000.0,
		"entered_game_count": _entered_count,
		"phase_history": _phase_history,
		"events": _events,
		"checks": _checks,
		"failures": _failures,
		"mutations": _mutations,
		"reconnect": _reconnect,
	}
	var file := FileAccess.open(_report_path, FileAccess.WRITE)
	if file:
		file.store_string(JSON.stringify(report, "  "))
		file.close()
	else:
		printerr("LIVE_SMOKE: 无法写入报告：" + _report_path)
	if passed:
		print("LIVE_SMOKE RESULT: PASS")
	else:
		printerr("LIVE_SMOKE RESULT: FAIL (%d failures)" % _failures.size())
	get_tree().quit(0 if passed else 1)