mtgodot-poc/project/playable_combat_test.gd

extends SceneTree

## CBT-01 离线事件与输入断言：真实 Quickbar + PlayerSkill + SkillTable 发出的请求，
## 经真实 PlayableProbe / PlayableFlow 归档。只有假客户端记录请求，不连服务器。
##
##   godot --headless --path project --script playable_combat_test.gd
##
## 覆盖 §7.2：拒绝不发有效施法请求、正常请求不重复、无目标、超距、冷却、资源不足、
## 目标死亡、自身死亡、断线/切图、施法中换目标。

const SkillTable = preload("res://ui/skill_table.gd")
const Quickbar = preload("res://ui/quickbar.gd")
const Probe = preload("res://testing/playable_probe.gd")
const Flow = preload("res://testing/playable_flow.gd")
const Report = preload("res://testing/playable_report.gd")
const Config = preload("res://testing/playable_config.gd")
const Matrix = preload("res://playable_skill_matrix.gd")

const MAIN := 1000
const MOB := 2000
const MOB_B := 2001
const MOB_RACE := 101
## Warrior "Three-Way Cut": attack skill, needs a target, TargetRange 400cm.
const SKILL := 1
const SLOT := 7

class FakeClient extends Node:
	signal skill_cooldown_end(skill: int)
	signal quickslots_changed()
	signal skills_changed()
	signal entity_dead(vid: int)
	signal damage(vid: int, amount: int, flag: int)
	signal disconnected(reason: String)
	signal entered_game()
	var target := 0
	var in_game := true
	var entities := {
		MAIN: {"vid": MAIN, "kind": 0, "ch_type": 0, "pos_cm": Vector3.ZERO, "dead": false},
		MOB: {"vid": MOB, "kind": 2, "ch_type": 2, "race": MOB_RACE, "pos_cm": Vector3(300, 0, 0), "dead": false, "hp": 100},
		MOB_B: {"vid": MOB_B, "kind": 2, "ch_type": 2, "race": MOB_RACE, "pos_cm": Vector3(350, 0, 0), "dead": false, "hp": 100},
	}
	var skills := [{"id": SKILL, "level": 5, "master": 0}]
	## Every outgoing request, in order: [name, args...].
	var requests: Array = []
	func is_in_game() -> bool: return in_game
	func get_main_vid() -> int: return MAIN
	func get_entity(vid) -> Dictionary: return entities.get(int(vid), {})
	func get_entities() -> Array: return entities.values()
	func get_ground_items() -> Array: return []
	func get_target() -> Dictionary: return {"vid": target}
	func get_skills() -> Array: return skills
	func get_inventory() -> Array: return []
	func get_quickslots() -> Array: return []
	func use_skill(id, vid) -> bool:
		requests.append(["use_skill", int(id), int(vid)])
		return in_game
	func cast_skill(motion, rot, x, y) -> bool:
		requests.append(["cast_skill", int(motion)])
		return in_game
	func quickslot_add(pos, type, ref) -> bool:
		requests.append(["quickslot_add", pos, type, ref])
		return true

## Only the NetPlay surface Quickbar/PlayableFlow touch. Range reservation itself
## (walk in, then use_skill_hook) is covered by netplay_test.
class FakeNetPlay extends Node:
	var client: FakeClient
	var context := {}
	var reservations: Array = []
	func select_target(vid: int) -> bool:
		client.target = vid
		return not client.get_entity(vid).is_empty()
	func skill_context() -> Dictionary:
		return context.merged({"target_distance": Callable(self, "skill_target_distance_cm")})
	func skill_target_distance_cm(vid: int) -> float:
		var e: Dictionary = client.get_entity(vid)
		var me: Dictionary = client.get_entity(MAIN)
		if e.is_empty() or me.is_empty():
			return -1.0
		return Vector3(e.pos_cm).distance_to(Vector3(me.pos_cm))
	func reserve_use_skill(vid: int, slot: int, range_cm: float) -> void:
		reservations.append([vid, slot, int(range_cm)])
	func is_use_skill_reserved(slot: int) -> bool:
		return not reservations.is_empty() and int(reservations[-1][1]) == slot
	func set_attack_key(_down: bool) -> void:
		pass

class FakeApp extends Node:
	var context := {}
	func get_playable_context() -> Dictionary: return context
	func get_playable_snapshot() -> Dictionary: return {"scene_ready": true}

class Rig:
	var client: FakeClient
	var net_play: FakeNetPlay
	var app: FakeApp
	var quickbar: Node
	var ui: Control
	var player: Node3D
	var probe: Node
	var report: RefCounted
	var flow: RefCounted
	var events: Array = []
	var now := 1_000_000
	func observed(kind: String) -> Array:
		return events.filter(func(e: Array) -> bool: return e[0] == kind)
	func count(name: String) -> int:
		return client.requests.filter(func(r: Array) -> bool: return r[0] == name).size()

var failures := 0
var table: RefCounted
var _rigs: Array = []

func _init() -> void:
	call_deferred("run")

func check(ok: bool, message: String) -> void:
	if not ok:
		failures += 1
		printerr("FAIL: " + message)

func run() -> void:
	table = SkillTable.new()
	var loaded := false
	for lang in ["en", "common"]:
		if table.load_file(AssetRoot.path().path_join("locale/locale/%s/skilldesc.txt" % lang)):
			loaded = true
			break
	if not loaded:
		# Missing skill data makes every assertion meaningless; never report a silent pass.
		check(false, "skilldesc.txt is required for playable_combat_test")
	else:
		check(table.target_range(SKILL) == 400 and table.is_need_target(SKILL), "fixture skill 1 is a 400cm targeted attack")
		test_single_request_per_activation()
		test_cooldown_holds_without_request()
		test_no_target_rejected_without_request()
		test_out_of_range_reserves_without_request()
		test_not_enough_sp_rejected_without_request()
		test_dead_target_rejected_without_request()
		test_dead_self_holds_without_request()
		test_disconnect_fails_case_and_stops_requests()
		test_target_switch_keeps_cast_target()
		test_skill_matrix()
	for rig in _rigs:
		rig.probe.disconnect_all()
		rig.flow.clock_us = Callable()
		rig.flow = null
		for node in [rig.quickbar, rig.ui, rig.player, rig.probe, rig.net_play, rig.app, rig.client]:
			node.free()
	_rigs.clear()
	print("playable_combat_test: failures=%d" % failures)
	quit(1 if failures else 0)

func make_rig(repeats := 2) -> Rig:
	var rig := Rig.new()
	rig.client = FakeClient.new()
	root.add_child(rig.client)
	rig.net_play = FakeNetPlay.new()
	rig.net_play.client = rig.client
	rig.net_play.context = {"cur_sp": 500, "cur_hp": 500}
	rig.player = Node3D.new()
	root.add_child(rig.player)
	rig.ui = Control.new()
	root.add_child(rig.ui)
	rig.quickbar = Quickbar.new()
	rig.quickbar.net_play = rig.net_play
	root.add_child(rig.quickbar)
	rig.quickbar.setup(rig.client, table, rig.ui, func() -> Node: return rig.player)
	rig.app = FakeApp.new()
	rig.app.context = {"quickbar": rig.quickbar, "net_play": rig.net_play}
	rig.probe = Probe.new()
	root.add_child(rig.probe)
	rig.probe.setup(rig.app, rig.client)
	rig.probe.watch_local(rig.quickbar, "skill_cast_started", "cast_started")
	rig.probe.watch_local(rig.quickbar, "skill_rejected", "skill_rejected")
	rig.probe.observed.connect(func(kind: String, data: Dictionary) -> void: rig.events.append([kind, data]))
	var config := {"allowed_mob_vnums": [MOB_RACE], "timeout_seconds": 900,
		"skill_cases": [{"case_id": "three-way-cut", "skill_id": SKILL, "target": "enemy",
			"repeats": repeats, "required_evidence": ["cast_started", "damage"]}]}
	rig.report = Report.new()
	rig.report.begin("combat-%d" % _rigs.size(), "full", {})
	rig.report.set_required_cases(["CBT-three-way-cut"])
	rig.flow = Flow.new()
	rig.flow.report = rig.report
	rig.flow.probe = rig.probe
	rig.flow.flow = rig.app
	rig.flow.client = rig.client
	rig.flow.config = config
	rig.flow.suite = "full"
	rig.flow.allow_gameplay = true
	rig.flow.clock_us = func() -> int: return rig.now
	rig.probe.observed.connect(rig.flow.on_observed)
	# Enter the skill stage the way the closed loop leaves PICKUP: bound main + epoch.
	rig.flow._start_us = rig.now
	rig.flow._main_vid = MAIN
	rig.flow._bound_epoch = rig.probe.connection_epoch
	rig.flow._begin_skill_case()
	_rigs.append(rig)
	return rig

func tick(rig: Rig, advance_us := 0) -> void:
	rig.now += advance_us
	rig.flow.tick()

# --- input assertions ----------------------------------------------------------

func test_single_request_per_activation() -> void:
	var rig := make_rig()
	check(rig.client.requests.is_empty(), "assigning the test slot sends no CG_QUICKSLOT_ADD")
	tick(rig)
	check(rig.count("use_skill") == 1 and rig.count("cast_skill") == 1,
		"one activation sends exactly one use_skill and one cast_skill: %s" % [rig.client.requests])
	check(rig.client.requests[0] == ["use_skill", SKILL, MOB], "use_skill carries the resolved target")
	var casts := rig.observed("cast_started")
	check(casts.size() == 1 and int(casts[0][1].target_vid) == MOB and casts[0][1].source == "local",
		"cast_started is a local event bound to the cast target")
	tick(rig, 100_000)
	check(rig.count("use_skill") == 1, "a pending cast is not re-requested while waiting for server evidence")

func test_cooldown_holds_without_request() -> void:
	var rig := make_rig()
	tick(rig)
	rig.client.damage.emit(MOB, 30, 1)
	tick(rig)
	check(rig.report.case_status("CBT-three-way-cut") == "" and rig.flow._skill_successes == 1, "first cast verified by server damage")
	tick(rig, 600_000)
	check(rig.count("use_skill") == 1 and rig.observed("skill_rejected").is_empty(),
		"local cooldown holds the retry without a request or a rejection event")
	rig.client.skill_cooldown_end.emit(SKILL)
	tick(rig, 600_000)
	check(rig.count("use_skill") == 2 and rig.count("cast_skill") == 2, "GC_SKILL_COOLTIME_END unlocks exactly one new request")

func test_no_target_rejected_without_request() -> void:
	var rig := make_rig()
	rig.client.target = 0
	rig.quickbar.activate(SLOT)
	var rejected := rig.observed("skill_rejected")
	check(rejected.size() == 1 and rejected[0][1].code == "NEED_TARGET", "no target -> NEED_TARGET")
	check(rig.client.requests.is_empty(), "NEED_TARGET sends no skill request")

func test_out_of_range_reserves_without_request() -> void:
	var rig := make_rig()
	rig.client.entities[MOB].pos_cm = Vector3(900, 0, 0)
	rig.client.entities[MOB_B].pos_cm = Vector3(950, 0, 0)
	tick(rig)
	check(rig.count("use_skill") == 0 and rig.count("cast_skill") == 0, "target beyond TargetRange sends no skill request")
	check(rig.net_play.reservations.size() == 1 and rig.net_play.reservations[0][0] == MOB,
		"out-of-range cast becomes a MODE_USE_SKILL reservation: %s" % [rig.net_play.reservations])
	check(rig.observed("skill_rejected").is_empty() and rig.observed("cast_started").is_empty(),
		"reservation is neither a rejection nor a cast")
	# NetPlay walked into range and fired use_skill_hook -> activate_reserved.
	rig.client.entities[MOB].pos_cm = Vector3(300, 0, 0)
	var global_slot: int = rig.net_play.reservations[0][1] if not rig.net_play.reservations.is_empty() else SLOT
	check(bool(rig.quickbar.activate_reserved(global_slot)), "reserved skill fires once in range")
	check(rig.count("use_skill") == 1 and rig.observed("cast_started").size() == 1, "in-range reserved cast sends exactly one request")
	rig.client.damage.emit(MOB, 30, 1)
	tick(rig)
	check(rig.flow._skill_successes == 1, "the asynchronous reserved cast is credited to the case")

func test_not_enough_sp_rejected_without_request() -> void:
	var rig := make_rig()
	rig.net_play.context = {"cur_sp": 10, "cur_hp": 500}
	tick(rig)
	var rejected := rig.observed("skill_rejected")
	check(rejected.size() == 1 and rejected[0][1].code == "NOT_ENOUGH_SP", "SP below GetNeedSP -> NOT_ENOUGH_SP")
	check(rig.client.requests.is_empty(), "NOT_ENOUGH_SP sends no skill request")
	check(rig.report.case_status("CBT-three-way-cut") == "", "resource shortage is retried, not a fixture BLOCKED")

func test_dead_target_rejected_without_request() -> void:
	var rig := make_rig()
	rig.client.entities[MOB].dead = true
	rig.client.target = MOB
	rig.quickbar.activate(SLOT)
	var rejected := rig.observed("skill_rejected")
	check(rejected.size() == 1 and rejected[0][1].code == "CANNOT_ATTACK", "dead target -> CANNOT_ATTACK")
	check(rig.client.requests.is_empty(), "dead target sends no skill request")
	tick(rig, 600_000)
	check(rig.client.requests.size() == 2 and rig.client.requests[0] == ["use_skill", SKILL, MOB_B],
		"flow re-targets a living monster instead of the corpse: %s" % [rig.client.requests])

func test_dead_self_holds_without_request() -> void:
	var rig := make_rig()
	rig.client.entities[MAIN].dead = true
	tick(rig)
	check(rig.client.requests.is_empty() and rig.observed("skill_rejected").is_empty(), "dead main character sends nothing")
	rig.client.entities[MAIN].dead = false
	rig.net_play.context = {"cur_sp": 500, "can_act": false}
	tick(rig, 600_000)
	check(rig.client.requests.is_empty(), "stunned/knocked-down (CANNOT_ACT) sends nothing")

func test_disconnect_fails_case_and_stops_requests() -> void:
	var rig := make_rig()
	tick(rig)
	check(rig.count("use_skill") == 1, "cast requested before the disconnect")
	rig.client.in_game = false
	rig.client.disconnected.emit("socket closed")
	check(rig.report.case_status("CBT-three-way-cut") == "FAIL" and rig.flow.done, "disconnect during a skill case fails that case")
	# Server damage after reconnect belongs to the new epoch and the finished case.
	rig.client.in_game = true
	rig.client.entered_game.emit()
	rig.client.damage.emit(MOB, 30, 1)
	tick(rig, 600_000)
	check(rig.count("use_skill") == 1 and rig.flow._skill_successes == 0, "no request or credit after the disconnect")

func test_target_switch_keeps_cast_target() -> void:
	var rig := make_rig(1)
	tick(rig)
	check(rig.observed("cast_started").size() == 1 and int(rig.observed("cast_started")[0][1].target_vid) == MOB, "cast bound to target A")
	rig.net_play.select_target(MOB_B)
	rig.client.damage.emit(MOB_B, 30, 1)
	tick(rig)
	check(rig.flow._skill_successes == 0, "damage on the newly selected target B is not evidence for the cast on A")
	rig.client.damage.emit(MOB, 30, 1)
	tick(rig)
	check(rig.flow._skill_successes == 1 and rig.report.case_status("CBT-three-way-cut") == "PASS",
		"damage on the cast-time target A completes the case")
	check(rig.count("use_skill") == 1, "switching the selection never re-sends the cast")

# --- test/playable/skill-cases.json ---------------------------------------------

func test_skill_matrix() -> void:
	var committed := ProjectSettings.globalize_path("res://").path_join("../test/playable/skill-cases.json").simplify_path()
	for job in Config.SKILL_MATRIX_JOBS:
		var r: Dictionary = Matrix.check(committed, job)
		check(r.status == "BLOCKED" and r.errors.is_empty() and r.blocked.size() == 4 and r.skill_cases.is_empty(),
			"committed matrix carries no invented skill ids and blocks %s: %s" % [job, r])
	var evidence := ["cast_started", "damage"]
	var matrix := {"schema_version": 1, "jobs": {"WARRIOR": {
		"single_target": {"status": "confirmed", "skill_id": SKILL, "target": "enemy", "required_evidence": evidence},
		"area": {"status": "not_applicable", "checklist_ref": "REL-01 checklist row"},
		"self_buff": {"status": "not_applicable", "checklist_ref": "REL-01 checklist row"},
		"flying": {"status": "not_applicable", "checklist_ref": "REL-01 checklist row"}}}}
	var ok: Dictionary = Config.skill_matrix_cases(matrix, "WARRIOR", table)
	check(ok.errors.is_empty() and ok.blocked.is_empty() and ok.cases.size() == 1, "confirmed matrix resolves: %s" % [ok])
	if ok.cases.size() == 1:
		check(ok.cases[0].case_id == "warrior-single-target" and int(ok.cases[0].repeats) == Config.SKILL_MATRIX_REPEATS,
			"matrix case defaults to 10 successful casts")
		check("CBT-warrior-single-target" in Config.required_cases({"skill_cases": ok.cases}, "full"),
			"matrix case becomes a required full-suite case")
	var bad := [
		["unconfirmed entry with a skill id", "area", {"status": "unconfirmed", "skill_id": 3}],
		["skill from another job", "single_target", {"status": "confirmed", "skill_id": 32, "required_evidence": evidence}],
		["fewer than 10 casts", "single_target", {"status": "confirmed", "skill_id": SKILL, "repeats": 3, "required_evidence": evidence}],
		["local-only evidence", "single_target", {"status": "confirmed", "skill_id": SKILL, "required_evidence": ["cast_started", "fx_spawned"]}],
		["not applicable without checklist record", "area", {"status": "not_applicable"}],
		["unknown status", "flying", {"status": "skip"}],
	]
	for row in bad:
		var mutated: Dictionary = matrix.duplicate(true)
		mutated.jobs.WARRIOR[row[1]] = row[2]
		check(not Config.skill_matrix_cases(mutated, "WARRIOR", table).errors.is_empty(), "matrix rejects %s" % row[0])
	var none: Dictionary = matrix.duplicate(true)
	none.jobs.WARRIOR.single_target = {"status": "not_applicable", "checklist_ref": "x"}
	check(not Config.skill_matrix_cases(none, "WARRIOR", table).errors.is_empty(), "a job with nothing confirmed is not READY")
	check(not Config.skill_matrix_cases(matrix, "SURA", table).errors.is_empty(), "a job absent from the matrix is an error")