Files
TheTree_mock/tests/ui/test_ui_audio_feedback.gd
T
Takeshi b84bade045 Save The Tree implementation and acceptance work in progress
Preserve production code, regression fixtures, audio assets and provenance, test infrastructure, and the original-completion plan and progress ledger. Final game and Windows acceptance remain pending; excluded local artifacts and native experiments are not part of this snapshot.
2026-10-11 15:58:00 +09:00

339 lines
21 KiB
GDScript

extends SceneTree
const Feedback := preload("res://scripts/ui/ui_audio_feedback.gd")
const AudioTeardown := preload("res://tests/helpers/ui_audio_teardown.gd")
var _checks := 0
var _failures := 0
var _runtime: Variant
var _clock: Variant
var _feedback: Variant
var _cues: Array[StringName] = []
var _case_events: Array[Dictionary] = []
var _case_proofs: Array[Dictionary] = []
var _proof_dir := ""
var _settings_path := "user://tests/audio_feedback_%d/settings.cfg" % OS.get_process_id()
func _initialize() -> void:
call_deferred(&"_run")
func _run() -> void:
_runtime = root.get_node("SimulationRuntime")
_clock = root.get_node("GameClock")
var state: Variant = root.get_node("GameState")
_clock.reset(1, 6, 0)
state.reset()
_runtime.initialize(_clock, state, 822251)
_clock.set_paused(true)
_expect(_runtime.register_command_executor(&"audio_test_ok", _complete), "register real Runtime diagnostic executor")
_expect(_runtime.register_command_executor(&"audio_test_noop", _noop_complete), "register explicit noop result executor")
_feedback = Feedback.new()
_feedback.configure(_runtime, _settings_path, false)
root.add_child(_feedback)
_feedback.cue_requested.connect(_record_cue)
_runtime.command_completed.connect(_record_completion)
var ui_bus := AudioServer.get_bus_index(Feedback.UI_BUS)
_expect(ui_bus >= 0 and AudioServer.get_bus_effect_count(ui_bus) == 1 and AudioServer.get_bus_effect(ui_bus, 0) is AudioEffectHardLimiter, "UI cue bus bounds overlapping voices with one limiter")
_test_ogg_decode()
var before: Dictionary = _begin_case(&"paused_success_exact_once")
var result: CommandResult = _feedback.submit_command(&"audio_test_ok", {}, &"player", &"", 0, -1, &"audio-pair")
_expect(result.is_success() and _cues == [&"success"], "paused emit-before-return has exactly one committed success")
var with_audio: Dictionary = _runtime.create_snapshot()
_end_case(&"paused_success_exact_once", before, result)
_expect(_runtime.restore_snapshot(before), "restore exact before snapshot for paired control")
_feedback.begin_loaded_epoch()
_runtime.submit_command(&"audio_test_ok", {}, &"player", &"", 0, -1, &"audio-pair")
_expect(_record_direct_control(with_audio), "audio shim leaves full snapshot bytes/RNG/IDs identical to direct command")
var duplicate_before := _begin_case(&"actual_duplicate_refusal")
var duplicate: CommandResult = _feedback.submit_command(&"audio_test_ok", {}, &"player", &"", 0, -1, &"audio-pair")
_expect(duplicate.code == CommandResult.Code.DUPLICATE_IDEMPOTENCY_KEY and not duplicate.is_replay, "real repeated submit is a failed duplicate refusal, not a replay DTO")
_expect(_cues == [&"refusal"], "actual duplicate enqueue refusal sounds exactly once")
_expect(var_to_bytes(duplicate_before) == var_to_bytes(_runtime.create_snapshot()), "duplicate refusal keeps every Runtime snapshot byte")
_end_case(&"actual_duplicate_refusal", duplicate_before, duplicate)
# Existing public Authority executor owns correlation; no fabricated replay flag/ID.
var replay_payload := {&"incident_id": &"audio.incident.replay", &"incident_type": &"illegal_facility", &"detector_id": &"audio.detector", &"options": {&"severity": 0, &"heat_delta": 0}}
var incident_setup: CommandResult = _runtime.submit_command(&"report_authority_incident", replay_payload, &"player", &"", 0, -1, &"audio-incident-setup")
_expect(incident_setup.is_success() and not incident_setup.is_replay, "real public Authority handler establishes diagnostic incident")
var replay_before := _begin_case(&"actual_public_authority_replay_silent")
var prior: CommandResult = _runtime.command_queue.get_result_by_idempotency_key(&"audio-pair")
_expect(prior.is_success() and prior.is_replay, "actual processed queue lookup returns a successful replay copy")
var replayed: CommandResult = _feedback.submit_command(&"report_authority_incident", replay_payload, &"player", &"", 0, -1, &"audio-replay-diagnostic")
_expect(replayed.is_success() and replayed.is_replay, "existing public Authority handler returns actual replay flag")
var stored_replay: CommandResult = _runtime.command_queue.get_processed_result(replayed.command_id)
_expect(stored_replay.to_dictionary() == replayed.to_dictionary(), "public replay DTO and stored command result have exact correlation")
_expect(_cues.is_empty(), "actual is_replay completion stays silent during paused emit-before-return")
_expect(_feedback.get(&"_armed").is_empty() and _feedback.get(&"_active_submissions").is_empty(), "replay retains no pending audio intent after returning")
var replay_with_audio: Dictionary = _runtime.create_snapshot()
_end_case(&"actual_public_authority_replay_silent", replay_before, replayed, {&"public_command_type": &"report_authority_incident", &"payload": replay_payload, &"setup_result": incident_setup.to_dictionary(), &"observed_queue_replay_copy": prior.to_dictionary(), &"stored_result": stored_replay.to_dictionary(), &"fabricated_result_fields": 0})
_expect(_runtime.restore_snapshot(replay_before), "restore whole public replay before snapshot for direct control")
_feedback.begin_loaded_epoch()
var direct_replay: CommandResult = _runtime.submit_command(&"report_authority_incident", replay_payload, &"player", &"", 0, -1, &"audio-replay-diagnostic")
_case_proofs[-1][&"extra"][&"direct_control_result"] = direct_replay.to_dictionary()
_expect(_record_direct_control(replay_with_audio), "public replay audio shim preserves every gameplay and queue snapshot byte against direct command")
var noop_before := _begin_case(&"actual_noop_silent")
var noop: CommandResult = _feedback.submit_command(&"audio_test_noop")
_expect(noop.is_success() and not noop.is_replay and noop.details.get(&"unchanged", false), "explicit noop is actual successful unchanged result")
_expect(_cues.is_empty(), "explicit successful unchanged result has no completion audio")
_end_case(&"actual_noop_silent", noop_before, noop)
var invalid_before := _begin_case(&"two_actual_enqueue_validation_refusals")
var invalid_tick: int = _clock.absolute_minute - 1
var invalid_first: CommandResult = _feedback.submit_command(&"audio_test_ok", {}, &"player", &"", 0, invalid_tick)
var invalid_second: CommandResult = _feedback.submit_command(&"audio_test_ok", {}, &"player", &"", 0, invalid_tick)
_expect(invalid_first.code == CommandResult.Code.INVALID_COMMAND and invalid_second.code == CommandResult.Code.INVALID_COMMAND and not invalid_first.is_replay and not invalid_second.is_replay and invalid_first.command_id == invalid_second.command_id, "both invalid execute-at times fail enqueue before consuming the same command ID")
_expect(_cues == [&"refusal", &"refusal"], "separate real enqueue validation failures each refuse exactly once")
_expect(var_to_bytes(invalid_before) == var_to_bytes(_runtime.create_snapshot()), "enqueue validation refusals keep every Runtime snapshot byte")
_end_case(&"two_actual_enqueue_validation_refusals", invalid_before, invalid_second, {&"first_result": invalid_first.to_dictionary(), &"execute_at": invalid_tick, &"issued_tick": _clock.absolute_minute, &"command_type": &"audio_test_ok"})
var empty_before := _begin_case(&"two_actual_empty_executor_refusals")
var empty_first: CommandResult = _feedback.submit_command(&"")
var empty_second: CommandResult = _feedback.submit_command(&"")
_expect(empty_first.code == CommandResult.Code.INVALID_COMMAND and empty_second.code == CommandResult.Code.INVALID_COMMAND and not empty_first.is_replay and not empty_second.is_replay and empty_second.command_id == empty_first.command_id + 1, "empty command types are real accepted executor failures with separate consumed IDs")
_expect(_cues == [&"refusal", &"refusal"], "two accepted empty executor failures each refuse exactly once")
var empty_with_audio: Dictionary = _runtime.create_snapshot()
_end_case(&"two_actual_empty_executor_refusals", empty_before, empty_second, {&"first_result": empty_first.to_dictionary(), &"command_type": &""})
_expect(_runtime.restore_snapshot(empty_before), "restore whole empty executor before snapshot for exact direct control")
_feedback.begin_loaded_epoch()
var direct_empty_first: CommandResult = _runtime.submit_command(&"")
var direct_empty_second: CommandResult = _runtime.submit_command(&"")
_case_proofs[-1][&"extra"][&"direct_control_first_result"] = direct_empty_first.to_dictionary()
_case_proofs[-1][&"extra"][&"direct_control_second_result"] = direct_empty_second.to_dictionary()
_expect(_record_direct_control(empty_with_audio), "accepted empty executor failures preserve every full Runtime byte against the same direct submissions")
_cues.clear()
var failed: CommandResult = _feedback.submit_command(&"unknown_audio_executor")
_expect(not failed.is_success() and _cues == [&"refusal"], "paused real executor failure refuses once")
_runtime.command_completed.emit(failed)
_expect(_cues == [&"refusal"], "duplicate completed callback does not repeat consumed failure")
_cues.clear()
_clock.set_paused(false)
var queued: CommandResult = _feedback.submit_command(&"audio_test_ok", {}, &"player", &"", 0, -1, &"audio-queued")
_expect(queued.is_success() and _cues.is_empty(), "queued acceptance has no early success")
_runtime.submit_command(&"audio_test_ok", {}, &"system", &"", 0, -1, &"audio-internal")
_clock.advance_minutes(1)
_expect(_cues == [&"success"], "mixed UI/internal Runtime tick sounds only the armed UI completion")
_runtime.command_completed.emit(_runtime.command_queue.get_processed_result(queued.command_id))
_expect(_cues == [&"success"], "repeat tick callback remains silent")
_cues.clear()
_feedback.submit_command(&"audio_test_ok", {}, &"player", &"", 0, -1, &"audio-earlier")
_runtime.submit_command(&"audio_test_ok", {}, &"system", &"", 0, -1, &"audio-other-earlier")
_clock.set_paused(true)
_feedback.submit_command(&"audio_test_ok", {}, &"player", &"", 0, -1, &"audio-sync-drain")
_expect(_cues == [&"success", &"success"], "paused submit drains prior armed UI/internal/current commands with exact correlation")
_cues.clear()
_clock.set_paused(false)
_feedback.submit_command(&"unknown_audio_queued_executor")
_expect(_cues.is_empty(), "queued executor failure has no premature refusal")
_clock.advance_minutes(1)
_expect(_cues == [&"refusal"], "actual queued failure refuses once on domain tick")
_cues.clear()
_feedback.submit_command(&"audio_test_ok", {}, &"player", &"", 0, -1, &"audio-restored")
var pending: Dictionary = _runtime.create_snapshot()
_expect(_runtime.restore_snapshot(pending), "restore real queued Runtime snapshot")
_feedback.begin_loaded_epoch()
_clock.advance_minutes(1)
_expect(_cues.is_empty(), "loaded pending completion is not a new UI intent")
_clock.set_paused(true)
_feedback.update_settings({&"master_volume": 1.0, &"ui_volume": 0.5, &"muted": true})
_feedback.submit_command(&"audio_test_ok")
_feedback.update_settings(Feedback.DEFAULT_SETTINGS)
_expect(_cues.is_empty(), "muted completion consumes intent and unmute never replays history")
_feedback.bind_runtime(_runtime)
_feedback.bind_runtime(_runtime)
_feedback.submit_command(&"audio_test_ok")
_expect(_cues == [&"success"], "repeated configure connects real Runtime only once")
_cues.clear()
var event := EventInstance.new().configure(&"audio_event_A", &"definition", &"definition", [], 0, 360, -1, true, &"audio-test-event")
event.state = EventInstance.State.WAITING_CHOICE
_feedback.observe_foreground(event)
_feedback.observe_foreground(event)
_feedback.observe_foreground(null)
_feedback.observe_foreground(event)
_expect(_cues == [&"notification"], "same foreground refresh and A-none-A notify once")
_feedback.begin_loaded_epoch(event)
_feedback.observe_foreground(event)
_expect(_cues == [&"notification"], "loaded current foreground seeds silently")
event.event_instance_id = &"audio_event_B"
_feedback.observe_foreground(event)
_expect(_cues == [&"notification", &"notification"], "new actual instance ID notifies once")
_test_settings()
_write_case_report()
_runtime.command_completed.disconnect(_record_completion)
var teardown: Dictionary = await AudioTeardown.free_owner_and_wait(self, _feedback)
_expect(bool(teardown.get(&"passed", false)) and not bool(teardown.get(&"deadline_reached", true)), "owned audio references release before bounded teardown deadline with whole Runtime unchanged")
print("UI audio feedback: %d checks, %d failures" % [_checks, _failures])
quit(_failures)
func _test_ogg_decode() -> void:
var rows: Array[Dictionary] = []
var mix_rate := int(AudioServer.get_mix_rate())
for role: StringName in Feedback.CUES:
var player: AudioStreamPlayer = _feedback.get(&"_players").get(role)
_expect(player != null and player.stream is AudioStreamOggVorbis, "Godot loads original Ogg resource " + String(role))
if player == null or not player.stream is AudioStreamOggVorbis:
continue
var stream := player.stream as AudioStreamOggVorbis
var decoder: AudioStreamPlayback = stream.instantiate_playback()
decoder.start()
var frames := 0
var finite := true
var peak := 0.0
while decoder.is_playing() and frames < mix_rate * 2:
var chunk: PackedVector2Array = decoder.mix_audio(1.0, 512)
if chunk.is_empty():
break
frames += chunk.size()
for sample: Vector2 in chunk:
finite = finite and is_finite(sample.x) and is_finite(sample.y)
peak = maxf(peak, maxf(absf(sample.x), absf(sample.y)))
var ended := not decoder.is_playing()
decoder.stop()
_expect(frames > 0 and finite and peak > 0.0 and peak < 1.0 and ended, "Godot actual finite one-shot decode ends " + String(role))
_feedback.playback_enabled = true
_feedback.call("_request_cue", role, {&"diagnostic": true})
var requested_playing := player.playing
_expect(requested_playing and not stream.loop, "Godot actual cue requests one-shot player state " + String(role))
_feedback.begin_loaded_epoch()
_expect(not player.playing, "loaded epoch stops current cue voice " + String(role))
rows.append({&"role": role, &"encoded_source_sha256": FileAccess.get_sha256(Feedback.CUES[role]), &"class": stream.get_class(), &"stream_seconds": stream.get_length(), &"mix_rate": mix_rate, &"mixed_frames": frames, &"finite": finite, &"sample_peak": peak, &"one_shot_ended": ended, &"player_playing_after_request": requested_playing, &"player_stopped_on_loaded_epoch": not player.playing})
_feedback.playback_enabled = false
_cues.clear()
var report_path := OS.get_environment("THE_TREE_AUDIO_PROOF_REPORT")
if report_path.is_empty():
report_path = "user://tests/audio_feedback_%d/ogg-proof.json" % OS.get_process_id()
DirAccess.make_dir_recursive_absolute(report_path.get_base_dir())
var file := FileAccess.open(report_path, FileAccess.WRITE)
_expect(file != null, "retain actual Godot Ogg decode and request/state proof")
if file != null:
file.store_string(JSON.stringify({&"rows": rows, &"Godot_pid": OS.get_process_id(), &"audio_driver": AudioServer.get_driver_name(), &"human_hearing_verified": false, &"native_Windows": &"pending"}, "\t") + "\n")
file.close()
func _test_settings() -> void:
_feedback.playback_enabled = true
_feedback.request_click()
var click_player: AudioStreamPlayer = _feedback.get(&"_players")[&"click"]
_expect(click_player.playing, "Godot actual imported Ogg player starts a one-shot")
var value := {&"master_volume": 0.75, &"ui_volume": 0.25, &"muted": true}
_expect(_feedback.update_settings(value), "finite routine volume settings accepted")
_expect(not click_player.playing, "mute stops an existing voice so unmute cannot resume its tail")
_feedback.playback_enabled = false
_expect(_feedback.save_settings().get(&"ok", false), "isolated presentation settings save")
var restored := Feedback.new()
restored.configure(_runtime, _settings_path, false)
_expect(restored.settings == value, "presentation settings persist without gameplay save")
_expect(not restored.update_settings({&"master_volume": NAN, &"ui_volume": 0.5, &"muted": false}), "nonfinite volume rejected")
_expect(not restored.update_settings({&"master_volume": -0.1, &"ui_volume": 0.5, &"muted": false}), "out-of-range volume rejected")
var future := ConfigFile.new()
future.set_value("schema", "version", 999)
future.set_value("future", "preserve", "original bytes")
_expect(future.save(_settings_path) == OK, "write isolated future settings specimen")
var future_sha := FileAccess.get_sha256(_settings_path)
_expect(not restored.save_settings().get(&"ok", false), "external future settings change blocks stale writer")
_expect(FileAccess.get_sha256(_settings_path) == future_sha, "blocked stale writer preserves future bytes")
restored.read_settings()
_expect(not restored.settings_writable and not restored.save_settings().get(&"ok", false), "fresh reader keeps unknown settings version read-only")
_expect(FileAccess.get_sha256(_settings_path) == future_sha, "fresh future reader preserves exact original bytes")
_expect(AudioServer.is_bus_mute(AudioServer.get_bus_index(&"Master")), "routine mute connects to actual AudioServer Master bus")
_feedback.update_settings(Feedback.DEFAULT_SETTINGS)
restored.free()
func _complete(command: SimulationCommand) -> CommandResult:
return CommandResult.succeeded(command.command_id)
func _noop_complete(command: SimulationCommand) -> CommandResult:
return CommandResult.succeeded(command.command_id, {&"unchanged": true})
func _record_cue(role: StringName, correlation: Dictionary) -> void:
_cues.append(role)
_case_events.append({&"kind": &"cue_requested", &"role": role, &"correlation": correlation.duplicate(true), &"audio_state": _audio_state()})
func _record_completion(result: CommandResult) -> void:
# Connected after Feedback, so the actual paused intent buffer is observable.
_case_events.append({&"kind": &"command_completed", &"result": result.to_dictionary(), &"audio_state": _audio_state()})
func _dto(value: Variant) -> Variant:
if value is CommandResult:
return value.to_dictionary()
if value is Dictionary:
var mapped := {}
for key: Variant in value:
mapped[key] = _dto(value[key])
return mapped
if value is Array:
var mapped_array: Array = []
for item: Variant in value:
mapped_array.append(_dto(item))
return mapped_array
return value
func _audio_state() -> Dictionary:
return {&"epoch": _feedback.get(&"_epoch"), &"nonce": _feedback.get(&"_nonce"), &"armed": _dto(_feedback.get(&"_armed")), &"active_submissions": _dto(_feedback.get(&"_active_submissions"))}
func _begin_case(label: StringName) -> Dictionary:
_cues.clear()
_case_events.clear()
var before: Dictionary = _runtime.create_snapshot()
_case_events.append({&"kind": &"before_submit", &"case": label, &"audio_state": _audio_state()})
return before
func _end_case(label: StringName, before: Dictionary, result: CommandResult, extra: Dictionary = {}) -> void:
if _proof_dir.is_empty():
var report := OS.get_environment("THE_TREE_AUDIO_COMMAND_PROOF_REPORT")
_proof_dir = report.get_basename() + "-typed" if not report.is_empty() else "user://tests/audio_feedback_%d/commands-typed" % OS.get_process_id()
DirAccess.make_dir_recursive_absolute(_proof_dir)
var after: Dictionary = _runtime.create_snapshot()
var row := {&"case": label, &"returned_result": result.to_dictionary(), &"cues": _cues.duplicate(), &"events": _case_events.duplicate(true), &"after_audio_state": _audio_state(), &"extra": extra.duplicate(true), &"snapshots": {}}
for part: StringName in {&"before": before, &"after": after}:
var path := _proof_dir.path_join("%s-%s.varbytes" % [label, part])
var bytes := var_to_bytes(before if part == &"before" else after)
var raw := FileAccess.open(path, FileAccess.WRITE)
_expect(raw != null, "retain complete typed Runtime snapshot " + String(label) + " " + String(part))
if raw != null:
raw.store_buffer(bytes)
raw.close()
row[&"snapshots"][part] = {&"path": path, &"bytes": bytes.size(), &"sha256": FileAccess.get_sha256(path), &"full_gameplay_dictionary": true, &"field_deletions": 0}
_case_proofs.append(row)
func _record_direct_control(with_audio: Dictionary) -> bool:
var control: Dictionary = _runtime.create_snapshot()
var matched := var_to_bytes(control) == var_to_bytes(with_audio)
var row: Dictionary = _case_proofs[-1]
var path := _proof_dir.path_join("%s-direct_control.varbytes" % row[&"case"])
var bytes := var_to_bytes(control)
var raw := FileAccess.open(path, FileAccess.WRITE)
_expect(raw != null, "retain complete typed same-command direct control " + String(row[&"case"]))
if raw != null:
raw.store_buffer(bytes)
raw.close()
row[&"snapshots"][&"direct_control"] = {&"path": path, &"bytes": bytes.size(), &"sha256": FileAccess.get_sha256(path), &"full_gameplay_dictionary": true, &"field_deletions": 0}
row[&"extra"][&"direct_control_pair_verified"] = matched
return matched
func _write_case_report() -> void:
var path := OS.get_environment("THE_TREE_AUDIO_COMMAND_PROOF_REPORT")
if path.is_empty():
path = "user://tests/audio_feedback_%d/commands.json" % OS.get_process_id()
DirAccess.make_dir_recursive_absolute(path.get_base_dir())
var file := FileAccess.open(path, FileAccess.WRITE)
_expect(file != null, "retain actual result DTO, paused intent buffers and whole typed snapshot evidence")
if file != null:
file.store_string(JSON.stringify({&"cases": _case_proofs, &"Godot_pid": OS.get_process_id(), &"buffer_CommandResult_objects_recorded_as_full_DTO": true, &"gameplay_snapshot_field_deletions": 0, &"source_fallback": false, &"formal_P22_P25_acceptance": false}, "\t") + "\n")
file.close()
func _expect(condition: bool, label: String) -> void:
_checks += 1
if not condition:
_failures += 1
push_error(label)