MusicPlayer/tests/unit/test_updater.gd

extends GdUnitTestSuite
## Updating the game in place: which file to get, checking it, swapping it in, and
## cleaning up after.

const WINDOWS_GAME: PackedByteArray = [0x4D, 0x5A, 0x90, 0x00, 0x03]
const LINUX_GAME: PackedByteArray = [0x7F, 0x45, 0x4C, 0x46, 0x02]

var _dir: String
var _updater: Updater


func before_test() -> void:
	_dir = create_temp_dir("updater_%d" % randi())
	_updater = auto_free(Updater.new())
	_updater.in_editor = false
	_updater.os_name = "Windows"
	_updater.executable_path = _dir.path_join("MusicPlayer.exe")


func _write(path: String, bytes: PackedByteArray) -> String:
	var file: FileAccess = FileAccess.open(path, FileAccess.WRITE)
	file.store_buffer(bytes)
	file.close()
	return path


func _read(path: String) -> PackedByteArray:
	return FileAccess.get_file_as_bytes(path)


@warning_ignore("unused_parameter")
func test_asset_for(os_name: String, in_place: bool, asset: String, test_parameters: Array = [
	["Windows", true, "windows"],
	["Windows", false, "windows"],
	["Linux", true, "linux"],
	["Linux", false, "linux-package"],
	["macOS", true, ""],
	["macOS", false, ""],
]) -> void:
	assert_str(Updater.asset_for(os_name, in_place)).is_equal(asset)


@warning_ignore("unused_parameter")
func test_check_file(bytes: PackedByteArray, os_name: String, ok: bool,
		test_parameters: Array = [
	[WINDOWS_GAME, "Windows", true],
	[LINUX_GAME, "Linux", true],
	[LINUX_GAME, "Windows", false],
	[WINDOWS_GAME, "Linux", false],
	["<html>Not Found</html>".to_utf8_buffer(), "Windows", false],
	[PackedByteArray(), "Linux", false],
]) -> void:
	var path: String = _write(_dir.path_join("game"), bytes)

	assert_bool(Updater.check_file(path, os_name)).is_equal(ok)


func test_missing_file_is_not_the_game() -> void:
	assert_bool(Updater.check_file(_dir.path_join("nothing"), "Windows")).is_false()


func test_windows_moves_the_running_game_aside() -> void:
	var exe: String = _write(_dir.path_join("MusicPlayer.exe"), "old".to_utf8_buffer())
	var update: String = _write(exe + Updater.UPDATE_SUFFIX, "new".to_utf8_buffer())

	assert_int(Updater.install(update, exe, "Windows")).is_equal(OK)

	assert_str(_read(exe).get_string_from_utf8()).is_equal("new")
	assert_str(_read(exe + Updater.OLD_SUFFIX).get_string_from_utf8()).is_equal("old")
	assert_bool(FileAccess.file_exists(update)).is_false()


func test_windows_replaces_an_old_version_left_from_before() -> void:
	var exe: String = _write(_dir.path_join("MusicPlayer.exe"), "old".to_utf8_buffer())
	_write(exe + Updater.OLD_SUFFIX, "older".to_utf8_buffer())
	var update: String = _write(exe + Updater.UPDATE_SUFFIX, "new".to_utf8_buffer())

	assert_int(Updater.install(update, exe, "Windows")).is_equal(OK)

	assert_str(_read(exe + Updater.OLD_SUFFIX).get_string_from_utf8()).is_equal("old")


func test_windows_puts_the_game_back_when_the_new_one_cannot_be_moved_in() -> void:
	var exe: String = _write(_dir.path_join("MusicPlayer.exe"), "old".to_utf8_buffer())

	var error: Error = Updater.install(exe + Updater.UPDATE_SUFFIX, exe, "Windows")

	assert_int(error).is_not_equal(OK)
	assert_str(_read(exe).get_string_from_utf8()).is_equal("old")
	assert_bool(FileAccess.file_exists(exe + Updater.OLD_SUFFIX)).is_false()


func test_linux_replaces_the_game_and_makes_it_runnable() -> void:
	var exe: String = _write(_dir.path_join("MusicPlayer.x86_64"), "old".to_utf8_buffer())
	var update: String = _write(exe + Updater.UPDATE_SUFFIX, "new".to_utf8_buffer())

	assert_int(Updater.install(update, exe, "Linux")).is_equal(OK)

	assert_str(_read(exe).get_string_from_utf8()).is_equal("new")
	assert_bool(FileAccess.file_exists(update)).is_false()
	assert_int(FileAccess.get_unix_permissions(exe) & FileAccess.UNIX_EXECUTE_OWNER) \
			.is_not_equal(0)


func test_leftovers_are_deleted_at_start() -> void:
	var exe: String = _write(_dir.path_join("MusicPlayer.exe"), "new".to_utf8_buffer())
	_write(exe + Updater.OLD_SUFFIX, "old".to_utf8_buffer())
	_write(exe + Updater.UPDATE_SUFFIX, "half".to_utf8_buffer())

	Updater.remove_old_files(exe)

	assert_bool(FileAccess.file_exists(exe)).is_true()
	assert_bool(FileAccess.file_exists(exe + Updater.OLD_SUFFIX)).is_false()
	assert_bool(FileAccess.file_exists(exe + Updater.UPDATE_SUFFIX)).is_false()


func test_can_update_in_a_writable_folder() -> void:
	assert_str(_updater.can_update_in_place()).is_empty()
	assert_array(DirAccess.get_files_at(_dir)).is_empty()


func test_not_from_the_editor() -> void:
	_updater.in_editor = true

	assert_str(_updater.can_update_in_place()).contains("editor")


func test_not_on_other_platforms() -> void:
	_updater.os_name = "macOS"

	assert_str(_updater.can_update_in_place()).contains("Windows and Linux")


func test_not_when_the_folder_cannot_be_written_to() -> void:
	_updater.executable_path = _dir.path_join("missing/MusicPlayer.exe")

	assert_str(_updater.can_update_in_place()).contains("can't be written to")


func test_restart_starts_the_game_and_quits() -> void:
	var started: Array[String] = []
	var quits: Array[bool] = []
	_updater.start_process = func(path: String, _arguments: PackedStringArray) -> int:
		started.append(path)
		return 1
	_updater.quit = func() -> void: quits.append(true)

	_updater.restart()

	assert_array(started).contains_exactly([_updater.executable_path])
	assert_array(quits).has_size(1)


# Answers one HTTP request on a local port with [param status] and [param body], and
# downloads it with the updater. Returns [done paths, failure reasons].
func _download_from_local_server(status: String, body: PackedByteArray) -> Array:
	var server: TCPServer = TCPServer.new()
	assert_int(server.listen(0, "127.0.0.1")).is_equal(OK)
	add_child(_updater)
	var done: Array[String] = []
	var failed: Array[String] = []
	var progress: Array[int] = []
	_updater.download("http://127.0.0.1:%d/MusicPlayer.exe" % server.get_local_port(),
			func(bytes: int, _total: int) -> void: progress.append(bytes),
			func(path: String) -> void: done.append(path),
			func(reason: String) -> void: failed.append(reason))
	var connection: StreamPeerTCP = null
	for i: int in 100:
		await await_millis(20)
		if connection == null && server.is_connection_available():
			connection = server.take_connection()
		if connection != null && connection.get_available_bytes() > 0:
			connection.get_data(connection.get_available_bytes())
			var head: String = "HTTP/1.1 %s\r\nContent-Length: %d\r\nConnection: close\r\n\r\n" \
					% [status, body.size()]
			connection.put_data(head.to_ascii_buffer() + body)
			break
	for i: int in 100:
		await await_millis(20)
		if !done.is_empty() || !failed.is_empty():
			break
	if connection != null:
		connection.disconnect_from_host()
	server.stop()
	remove_child(_updater)
	return [done, failed, progress]


func test_download_keeps_a_checked_game() -> void:
	var game: PackedByteArray = WINDOWS_GAME.duplicate()
	game.resize(4096)

	var result: Array = await _download_from_local_server("200 OK", game)

	assert_array(result[1]).is_empty()
	assert_array(result[0]).contains_exactly([_updater.update_path()])
	assert_array(_read(_updater.update_path())).is_equal(game)


func test_download_of_an_error_page_is_deleted() -> void:
	var result: Array = await _download_from_local_server("404 Not Found",
			"Not Found".to_utf8_buffer())

	assert_array(result[0]).is_empty()
	assert_str(result[1][0]).contains("404")
	assert_bool(FileAccess.file_exists(_updater.update_path())).is_false()


func test_download_that_is_not_the_game_is_deleted() -> void:
	var result: Array = await _download_from_local_server("200 OK",
			"<html>Sign in</html>".to_utf8_buffer())

	assert_array(result[0]).is_empty()
	assert_str(result[1][0]).contains("not the game")
	assert_bool(FileAccess.file_exists(_updater.update_path())).is_false()