settled-reach/client/tests/test_input_gate_live.gd

## T-1146: implant-occlusion gate — LIVE tier. Connects the real SimBridge/
## InputMapper/HudGroups autoloads (not stubs, not TestHarness) to a spawned
## server binary and proves the actual client-side pipeline end to end: WASD
## held while a FULLSCREEN implant screen is open must not move the character
## on the server, and closing the screen must require a re-press before
## movement resumes.
##
## Reuses test_input_roundtrip.gd's/test_sprint2_proof.gd's hard-won live-
## server conventions verbatim (port-collision retry, wall-clock deadlines,
## COOLDOWN_TICKS pacing, handshake dance) — see their own header comments for
## the "why" behind each one. The one structural difference from those files:
## this test drives input through the REAL client pipeline (InputMapper poll +
## SimBridge.send_input, exactly mirroring main.gd's dispatch loop) instead of
## LocalBridge.send_message() directly — the gate under test lives in that
## pipeline, so a raw-protocol test would not exercise it at all.
class_name TestInputGateLive
extends GdUnitTestSuite

const CONNECT_TIMEOUT: float = 3.0
const RESPONSE_TIMEOUT: float = 10.0
# Server move cooldown: the proof-room player is Walk stance (ticks_per_move =
# 2, server stance.rs) — see test_input_roundtrip.gd's own header comment for
# the full "why" (a dropped move stays dropped no matter how long the wait).
const COOLDOWN_TICKS: int = 3
const MAX_PORT_ATTEMPTS: int = 5
const TEST_APP := "implant/t1146_live_gate_test"

var _server_pid: int = -1
var _test_port: int = 0

# Autoload state to restore in after_test() — SimBridge/InputMapper/HudGroups
# are persistent singletons shared with every other suite in the run.
var _saved_test_mode: bool = true
var _saved_state: int = 0
var _saved_bridge = null
var _saved_server_path: String = ""
var _saved_server_port: int = 9876


func _server_binary_path() -> String:
	var project_dir := ProjectSettings.globalize_path("res://")
	return project_dir.path_join("../server/target/debug/settled-reach-server")


static func _random_test_port() -> int:
	return 49152 + (randi() % (65535 - 49152 + 1))


## Spawn the server binary directly (not through SimBridge.connect_to_sim's
## own spawn path) so this test controls port retry exactly like the sibling
## live-server tests — SimBridge is then pointed at the already-bound port
## with server_path left empty so it skips its own subprocess spawn and goes
## straight to the TCP-connect retry loop in its _process().
func _spawn_server(server_path: String) -> bool:
	for _attempt in range(MAX_PORT_ATTEMPTS):
		_test_port = _random_test_port()
		var addr := "127.0.0.1:%d" % _test_port
		_server_pid = OS.create_process(server_path, [addr])
		if _server_pid <= 0:
			continue
		await get_tree().create_timer(0.15).timeout
		if OS.is_process_running(_server_pid):
			return true
		_server_pid = -1
	return false


func before_test() -> void:
	_saved_test_mode = SimBridge.test_mode
	_saved_state = SimBridge.state
	_saved_bridge = SimBridge._bridge
	_saved_server_path = SimBridge.server_path
	_saved_server_port = SimBridge.server_port
	InputMapper.input_queue.clear()
	InputMapper._gameplay_occluded = false
	InputMapper._suppress_move_until_release = false
	_release_all_movement_keys()


func after_test() -> void:
	if SimBridge._bridge != null and SimBridge._bridge != _saved_bridge:
		SimBridge._bridge.disconnect_from_server()
	if _server_pid > 0 and OS.is_process_running(_server_pid):
		OS.kill(_server_pid)
	_server_pid = -1

	SimBridge.test_mode = _saved_test_mode
	SimBridge.state = _saved_state
	SimBridge._bridge = _saved_bridge
	SimBridge.server_path = _saved_server_path
	SimBridge.server_port = _saved_server_port
	SimBridge.reset_test_state()

	InputMapper.input_queue.clear()
	InputMapper._gameplay_occluded = false
	InputMapper._suppress_move_until_release = false
	_release_all_movement_keys()

	HudGroups._active_app = ""
	HudGroups._active_mode = HudGroups.Mode.GAMEPLAY
	HudGroups._groups.erase(TEST_APP)


func _release_all_movement_keys() -> void:
	Input.action_release("move_north")
	Input.action_release("move_south")
	Input.action_release("move_east")
	Input.action_release("move_west")


## Drives SimBridge from DISCONNECTED all the way to CONNECTED against a
## freshly spawned server, awaiting real frames so SimBridge's own _process()
## (autoload, ticks automatically) does its TCP-connect + handshake dance.
func _connect_live(server_path: String) -> bool:
	var spawned := await _spawn_server(server_path)
	assert_bool(spawned).override_failure_message(
		"server spawn failed after %d port attempts" % MAX_PORT_ATTEMPTS
	).is_true()
	if not spawned:
		return false

	SimBridge.test_mode = false
	SimBridge.server_path = ""  # already spawned ourselves — SimBridge must not spawn again
	SimBridge.server_port = _test_port
	SimBridge._bridge = null
	SimBridge.state = SimBridge.ConnectionState.DISCONNECTED
	SimBridge.connect_to_sim()

	var deadline_ms: int = Time.get_ticks_msec() + int(CONNECT_TIMEOUT * 1000.0) * 2
	while Time.get_ticks_msec() < deadline_ms:
		if SimBridge.state == SimBridge.ConnectionState.CONNECTED:
			return true
		if SimBridge.state == SimBridge.ConnectionState.ERROR:
			break
		await get_tree().process_frame
	return SimBridge.state == SimBridge.ConnectionState.CONNECTED


## Find the player entity in a decoded snapshot Dictionary.
static func _find_player(snapshot: Dictionary) -> Dictionary:
	for entity in snapshot.get("entities", []):
		if entity.kind.variant == "Player":
			return entity
	return {}


## Poll SimBridge.poll_snapshot() until one arrives (or the timeout elapses).
## Returns the last-seen snapshot Dictionary, or {} on timeout.
func _await_next_snapshot(timeout: float = RESPONSE_TIMEOUT) -> Dictionary:
	var deadline_ms: int = Time.get_ticks_msec() + int(timeout * 1000.0)
	while Time.get_ticks_msec() < deadline_ms:
		var snap: Variant = SimBridge.poll_snapshot()
		if snap != null:
			return snap
		await get_tree().process_frame
	return {}


## Mirrors main.gd's dispatch loop exactly for the actions this test exercises
## (movement + SET_FACING): flush InputMapper's queue and forward each entry
## to SimBridge.send_input(). main.gd's client-only continues (BUG_REPORT,
## OPEN_JOURNAL, OPEN_MENU, INTERACT's target-resolution) are irrelevant here
## since this test never presses those keys — nothing to special-case.
func _flush_input_to_server() -> void:
	var inputs: Array[Dictionary] = InputMapper.flush_queue()
	for input in inputs:
		SimBridge.send_input(input)


## Drive InputMapper._process() + the flush-to-server forwarding for one
## simulated frame, then await a real engine frame so SimBridge's own
## _process() gets a chance to actually transmit the outbound buffer.
func _tick(delta: float = 0.05) -> void:
	InputMapper._process(delta)
	_flush_input_to_server()
	await get_tree().process_frame


# -- Live: WASD held while a FULLSCREEN implant is open must not move the player --

func test_wasd_held_during_implant_occlusion_does_not_move_player(
	_do_skip := not FileAccess.file_exists(_server_binary_path()),
	_skip_reason := "server binary not built — run `cargo build` in server/"
) -> void:
	var server_path := _server_binary_path()
	var connected := await _connect_live(server_path)
	assert_bool(connected).override_failure_message(
		"live SimBridge connection to spawned server failed within timeout"
	).is_true()
	if not connected:
		return

	GameState.player_stance = "Walk"
	InputMapper._last_move_msec = Time.get_ticks_msec() - 1000

	# Baseline: proof-room player starts at (16,16) -> render coords (16.5, 16.5).
	var baseline: Dictionary = await _await_next_snapshot()
	var start_player := _find_player(baseline)
	assert_that(start_player.size()).override_failure_message(
		"must receive at least one snapshot with a Player entity before testing"
	).is_greater(0)
	var start_x: float = start_player.x
	var start_y: float = start_player.y
	assert_float(start_x).is_equal_approx(16.5, 0.001)
	assert_float(start_y).is_equal_approx(16.5, 0.001)

	# Open a FULLSCREEN implant screen — gameplay is now occluded.
	HudGroups.open_app(TEST_APP, HudGroups.Mode.FULLSCREEN)
	assert_bool(InputMapper._gameplay_occluded).override_failure_message(
		"gate precondition: HudGroups.open_app(FULLSCREEN) must occlude gameplay"
	).is_true()

	# Hold W (move_north) and drive several ticks — comfortably past the Walk
	# stance's 400ms move interval, so if the gate were absent this would have
	# produced multiple MoveNorth sends by now.
	Input.action_press("move_north")
	for _i in range(20):
		await _tick(0.05)
	Input.action_release("move_north")

	# Drain whatever the server actually streamed during the hold and confirm
	# the player never left the starting tile.
	var settle: Dictionary = await _await_next_snapshot(2.0)
	if settle.is_empty():
		settle = baseline
	var after_hold := _find_player(settle)
	assert_that(after_hold.size()).is_greater(0)
	assert_float(after_hold.x).override_failure_message(
		(
			"player must NOT move while WASD is held during implant occlusion — x drifted from %.2f to %.2f"
			% [start_x, after_hold.x]
		)
	).is_equal_approx(start_x, 0.001)
	assert_float(after_hold.y).override_failure_message(
		(
			"player must NOT move while WASD is held during implant occlusion — y drifted from %.2f to %.2f"
			% [start_y, after_hold.y]
		)
	).is_equal_approx(start_y, 0.001)


# -- Live: close-while-held must not resume movement without a re-press --------

func test_close_while_held_requires_repress_before_movement_resumes_live(
	_do_skip := not FileAccess.file_exists(_server_binary_path()),
	_skip_reason := "server binary not built — run `cargo build` in server/"
) -> void:
	var server_path := _server_binary_path()
	var connected := await _connect_live(server_path)
	assert_bool(connected).override_failure_message(
		"live SimBridge connection to spawned server failed within timeout"
	).is_true()
	if not connected:
		return

	GameState.player_stance = "Walk"
	InputMapper._last_move_msec = Time.get_ticks_msec() - 1000

	var baseline: Dictionary = await _await_next_snapshot()
	var start_player := _find_player(baseline)
	assert_that(start_player.size()).is_greater(0)
	var start_x: float = start_player.x
	var start_y: float = start_player.y

	# Open the implant, hold W THROUGH the close transition (W is still down
	# when close_app() fires — this is the "forgot their finger on W" case).
	HudGroups.open_app(TEST_APP, HudGroups.Mode.FULLSCREEN)
	Input.action_press("move_north")
	for _i in range(6):
		await _tick(0.05)

	HudGroups.close_app()
	assert_bool(InputMapper._suppress_move_until_release).override_failure_message(
		"the release-latch must arm the instant the implant closes"
	).is_true()

	# Still held for several more ticks post-close — must NOT resume.
	for _i in range(10):
		await _tick(0.05)

	var settle_still_held: Dictionary = await _await_next_snapshot(2.0)
	if settle_still_held.is_empty():
		settle_still_held = baseline
	var still_held_player := _find_player(settle_still_held)
	assert_that(still_held_player.size()).is_greater(0)
	assert_float(still_held_player.x).override_failure_message(
		"player must NOT resume moving while W stays held through the close transition (no re-press yet)"
	).is_equal_approx(start_x, 0.001)
	assert_float(still_held_player.y).override_failure_message(
		"player must NOT resume moving while W stays held through the close transition (no re-press yet)"
	).is_equal_approx(start_y, 0.001)

	# Release, then re-press — movement must resume normally now.
	Input.action_release("move_north")
	await _tick(0.05)  # observes the release, clears the latch
	assert_bool(InputMapper._suppress_move_until_release).override_failure_message(
		"the latch must clear once every movement key has been observed released"
	).is_false()

	InputMapper._last_move_msec = Time.get_ticks_msec() - 1000
	Input.action_press("move_north")
	var resumed := false
	for _i in range(30):
		await _tick(0.05)
		var snap: Dictionary = await _await_next_snapshot(0.5)
		if not snap.is_empty():
			var p := _find_player(snap)
			if (
				p.size() > 0
				and not (is_equal_approx(p.x, start_x) and is_equal_approx(p.y, start_y))
			):
				resumed = true
				break
	Input.action_release("move_north")
	assert_bool(resumed).override_failure_message(
		"movement must resume after a fresh re-press following the close-while-held transition"
	).is_true()