spaceship/tests/test_sfx.gd

extends SceneTree

# Headless test: run with
#   godot --headless -s res://tests/test_sfx.gd
# Forgejo #146: shots and destruction should be heard. Enemy destruction and
# incoming narration are covered too, so visual events cannot silently miss
# the shared sound pool.
#
# The interesting failures here are not "no sound came out" - they are the
# ones a play-through would not reveal for weeks. A voice pool that runs dry
# under a firing fleet and silently drops every shot after the twelfth. A
# sound started by a node that is being freed on the same frame, which is
# every death in the game. So those are what is checked.
#
# Covers:
#   A. every sound the catalog names actually exists and loads
#   B. firing plays a shot, and it survives a fleet firing together - no
#      voice is lost, none is left stuck
#   C. a dying ship is heard, from a node that is freed on the same frame
#   D. enemy and structure destruction are heard
#   E. levels are in proportion: a shot fires constantly and a death happens
#      once, so a shot must be the quieter of the two
#   F. sound keeps working while a dialog has the game stopped (#128)

# 20ms is well clear of what the generator actually produces (1.5ms at
# worst, on the black hole) while still catching a real lazy attack.
const MAX_LEAD_SILENCE_MS: float = 20.0

var game_state: Node
var scene: Node
var fleet: Node2D
var stage: int = 0
var elapsed: float = 0.0


func _initialize() -> void:
	# GameState is an autoload: a bare GameState.<instance member> does not
	# compile in a SceneTree script, so the node is fetched by name instead.
	game_state = root.get_node("GameState")
	game_state.reset_new_game()
	game_state.intro_active = false
	game_state.reset_quests()
	game_state.rocks = 500


func _fail(message: String) -> bool:
	print("TEST_SFX: FAIL (%s)" % message)
	quit(1)
	return true


func _pool() -> Node:
	return root.get_node_or_null(Sfx.NODE_NAME)


func _sounding() -> int:
	var pool: Node = _pool()
	if pool == null:
		return 0
	var n: int = 0
	for voice in pool.get_children():
		if (voice as AudioStreamPlayer).playing:
			n += 1
	return n


# Every sample committed in audio/, not just the ones the catalog names.
func _wav_file_names() -> Array:
	var names: Array = []
	var dir: DirAccess = DirAccess.open("res://audio")
	if dir == null:
		return names
	for file_name in dir.get_files():
		# The editor leaves .import siblings; take the real files only.
		if file_name.ends_with(".wav"):
			names.append(file_name)
	names.sort()
	return names


# Leading samples that are exactly zero, in milliseconds. Negative if the
# file could not be parsed, which the caller treats as a failure rather than
# quietly passing.
func _leading_silence_ms(path: String) -> float:
	var f: FileAccess = FileAccess.open(path, FileAccess.READ)
	if f == null:
		return -1.0
	if f.get_buffer(4).get_string_from_ascii() != "RIFF":
		return -1.0
	f.get_32()
	if f.get_buffer(4).get_string_from_ascii() != "WAVE":
		return -1.0
	var channels: int = 0
	var rate: int = 0
	var bits: int = 0
	while f.get_position() + 8 <= f.get_length():
		var chunk_id: String = f.get_buffer(4).get_string_from_ascii()
		var chunk_size: int = f.get_32()
		var next: int = f.get_position() + chunk_size + (chunk_size & 1)
		if chunk_id == "fmt ":
			f.get_16()
			channels = f.get_16()
			rate = f.get_32()
			f.get_32()
			f.get_16()
			bits = f.get_16()
		elif chunk_id == "data":
			if bits != 16 or channels <= 0 or rate <= 0:
				return -1.0
			var silent_frames: int = 0
			var data_end: int = f.get_position() + chunk_size
			while f.get_position() + 2 * channels <= data_end:
				var quiet: bool = true
				for c in range(channels):
					if f.get_16() != 0:
						quiet = false
				if not quiet:
					break
				silent_frames += 1
			return float(silent_frames) / float(rate) * 1000.0
		f.seek(next)
	return -1.0


func _process(delta: float) -> bool:
	elapsed += delta
	match stage:
		0:
			# --- A. the catalog is real -----------------------------
			for id in Sfx.SOUNDS:
				# Preloaded, so this asserts the catalog is already resident
				# rather than merely loadable - play() must never touch the
				# disk on the frame a sound is asked for.
				var stream: AudioStream = Sfx.SOUNDS[id] as AudioStream
				if stream == null:
					return _fail("the \"%s\" sound is not a preloaded stream" % id)
				if stream.get_length() <= 0.02:
					return _fail("the \"%s\" sound is %.3fs long - that is silence" \
						% [id, stream.get_length()])
				if not Sfx.LEVELS.has(id):
					return _fail("the \"%s\" sound has no level set" % id)
			print("%d sounds, all loading, all with a level" % Sfx.SOUNDS.size())

			# --- F. and none of them opens on silence ---------------
			# Reported from the game as sounds "sometimes starting a bit
			# late". Measured, the samples were clean and the cause was
			# elsewhere - play() used to load() on the frame the sound was
			# asked for. But make_sfx.py regenerates these, and an envelope
			# that eased in from zero would put the delay back somewhere
			# nobody would think to look.
			#
			# Reads the source .wav rather than the imported stream: these
			# import as QOA (compress/mode=2), so AudioStreamWAV.data is
			# not PCM and cannot be inspected sample by sample. True digital
			# silence only - the black hole deliberately SWELLS (see its
			# docstring in make_sfx.py) and a soft attack is a choice.
			var checked: int = 0
			var worst_id: String = ""
			var worst_ms: float = 0.0
			for file_name in _wav_file_names():
				var lead: float = _leading_silence_ms("res://audio/" + file_name)
				if lead < 0.0:
					return _fail("could not read %s to check its attack" % file_name)
				checked += 1
				if lead > worst_ms:
					worst_ms = lead
					worst_id = file_name
				if lead > MAX_LEAD_SILENCE_MS:
					return _fail("%s opens with %.1fms of dead air" % [file_name, lead])
			if checked < Sfx.SOUNDS.size() - 1:
				return _fail("only %d wav files checked for dead air - that proves nothing" % checked)
			print("%d samples checked, worst attack %s at %.1fms" % [checked, worst_id, worst_ms])

			# --- E. and in proportion -------------------------------
			if float(Sfx.LEVELS[Sfx.SHOT]) >= float(Sfx.LEVELS[Sfx.SHIP_DEATH]):
				return _fail("a shot is as loud as a death - one happens constantly, the other once")
			print("shot at %.0fdB, ship death at %.0fdB, structure at %.0fdB" \
				% [Sfx.LEVELS[Sfx.SHOT], Sfx.LEVELS[Sfx.SHIP_DEATH],
					Sfx.LEVELS[Sfx.BUILDING_DEATH]])

			scene = load("res://world/Main.tscn").instantiate()
			root.add_child(scene)
			current_scene = scene
			fleet = scene.get_node("Fleet")
			elapsed = 0.0
			stage = 1
			return false
		1:
			if elapsed < 0.4:
				return false
			scene.get_node("AsteroidField").set_process(false)
			scene.get_node("EnemySpawner").set_process(false)
			if _pool() != null and _sounding() > 0:
				return _fail("something was already sounding before anything happened")

			# --- B. more at once than there are voices ---------------
			Sfx.play(self, Sfx.SHOT)
			if _sounding() != 1:
				return _fail("firing played %d sounds, expected 1" % _sounding())
			if _pool().get_parent() != root:
				return _fail("the sound pool hangs off the scene - a death would take it with it")
			# Driven by a DEATH, not a shot. #229 coalesces shots - a volley
			# is one event to the ear and now gets one voice - so shots can
			# no longer saturate anything. Deaths carry no window, because
			# several things dying together really is several events, and
			# they are what the pool has to absorb now.
			for i in range(Sfx.VOICES * 3):
				Sfx.play(self, Sfx.SHIP_DEATH)
			var busy: int = _sounding()
			if busy != Sfx.VOICES:
				return _fail("%d voices sounding after %d deaths, the pool holds %d" \
					% [busy, Sfx.VOICES * 3, Sfx.VOICES])
			if _pool().get_child_count() != Sfx.VOICES:
				return _fail("the pool grew to %d voices" % _pool().get_child_count())
			print("%d deaths into a %d-voice pool: %d sounding, none leaked" \
				% [Sfx.VOICES * 3, Sfx.VOICES, busy])

			# --- G. and a volley does not stack (#229) ----------------
			for voice in _pool().get_children():
				(voice as AudioStreamPlayer).stop()
			Sfx._last_started_msec.clear()
			for i in range(Sfx.VOICES * 3):
				Sfx.play(self, Sfx.SHOT)
			if _sounding() != 1:
				return _fail("%d simultaneous shots took %d voices - that is the volley getting louder per ship" \
					% [Sfx.VOICES * 3, _sounding()])
			print("%d simultaneous shots still take 1 voice" % (Sfx.VOICES * 3))

			# --- F. and a dialog does not silence it -----------------
			for voice in _pool().get_children():
				(voice as AudioStreamPlayer).stop()
			var narration: Control = scene.get_node("UI/NarrationBanner")
			narration.say("Everything stops but this.", "TEST")
			if _sounding() == 0:
				return _fail("a narration line appeared in silence")
			if not paused:
				return _fail("the dialog did not stop the game, so this proves nothing")
			for voice in _pool().get_children():
				if not (voice as AudioStreamPlayer).can_process():
					return _fail("sound stops whenever a dialog is up")
			print("a dialog stops the game and sound carries on")
			while narration.dismiss():
				pass

			for voice in _pool().get_children():
				(voice as AudioStreamPlayer).stop()
			elapsed = 0.0
			stage = 2
			return false
		2:
			if elapsed < 0.3:
				return false
			# --- C. a death, from a node freed on the same frame -----
			fleet.spawn_ships(2, Vector2.ZERO,
				PackedStringArray([GameState.VESSEL_INTERCEPTOR,
					GameState.VESSEL_INTERCEPTOR]))
			for voice in _pool().get_children():
				(voice as AudioStreamPlayer).stop()
			var doomed: Node2D = fleet.ships[0]
			doomed.take_damage(1)
			if not is_instance_valid(doomed) or not doomed.is_queued_for_deletion():
				return _fail("the ship was not freed, so this is not the case that matters")
			if _sounding() == 0:
				return _fail("a vessel died in silence")
			print("a vessel freed on the frame it dies is still heard")

			# An enemy uses the same persistent pool. Its own node is freed on
			# this frame too, so the call must happen before queue_free().
			for voice in _pool().get_children():
				(voice as AudioStreamPlayer).stop()
			var enemy: Node2D = load("res://enemies/Enemy.tscn").instantiate()
			scene.add_child(enemy)
			enemy.take_damage(enemy.hp)
			if not enemy.is_queued_for_deletion():
				return _fail("the enemy was not destroyed")
			if _sounding() == 0:
				return _fail("an enemy vessel exploded in silence")
			print("an enemy vessel explosion is heard")

			# --- D. a structure ------------------------------------
			for voice in _pool().get_children():
				(voice as AudioStreamPlayer).stop()
			var field: Node2D = scene.get_node("BuildingField")
			var id: int = game_state.add_building(GameState.BUILDING_SHIP_CRAFTER,
				Vector2(400.0, 0.0))
			field.rebuild_all()
			game_state.destroy_building(id)
			if _sounding() == 0:
				return _fail("a structure came down in silence")
			print("a structure coming down is heard")

			# --- G. the things #210/#213/#214 asked to be able to hear -----
			for voice in _pool().get_children():
				(voice as AudioStreamPlayer).stop()
			var rock: Node2D = load("res://world/Asteroid.tscn").instantiate()
			scene.add_child(rock)
			rock.shatter(false)
			if _sounding() == 0:
				return _fail("an asteroid broke apart in silence")
			print("a destroyed asteroid is heard")

			for voice in _pool().get_children():
				(voice as AudioStreamPlayer).stop()
			var bomb: Node2D = load("res://vfx/GasBombEffect.gd").launch(
				scene, Vector2.ZERO, Vector2(300.0, 0.0), 6)
			if _sounding() != 0:
				return _fail("the gas bomb made its noise on the throw, not the blast")
			bomb._explode()
			if _sounding() == 0:
				return _fail("a gas bomb bloomed in silence")
			print("a gas bomb is heard when it blooms, and not before")

			# The hold has to be able to take the deposit, or there is no
			# arrival to hear.
			game_state.fleet_inventory.total_capacity = -1
			for voice in _pool().get_children():
				(voice as AudioStreamPlayer).stop()
			if game_state.deposit_fleet_resource(GameState.RESOURCE_ORE, 1) != 1:
				return _fail("the fleet hold refused a single rock")
			var after_first: int = _sounding()
			if after_first == 0:
				return _fail("a rock reached the fleet hold in silence")
			# Mining lands one unit at a time, so the window is the whole of
			# what stops this being a machine-gun.
			game_state.deposit_fleet_resource(GameState.RESOURCE_ORE, 1)
			game_state.deposit_fleet_resource(GameState.RESOURCE_ORE, 1)
			if _sounding() != after_first:
				return _fail("a burst of pickups opened %d voices, not the %d one blip needs" \
					% [_sounding(), after_first])
			print("a rock arriving in the hold is heard, and a burst of them only once")

			# The bank is a number moving in a building the player is not in.
			for voice in _pool().get_children():
				(voice as AudioStreamPlayer).stop()
			game_state._pickup_sound_at = 0
			game_state.deposit_bank_resource(GameState.RESOURCE_ORE, 5)
			if _sounding() != 0:
				return _fail("a deposit into the bank made a pickup noise")
			print("banked resources stay silent")

			print("TEST_SFX: PASS")
			quit(0)
			return true
	return false