Sansan/MusicPlayer
1
0 Fork 0
Code Issues 1 Pull Requests 1 Actions Packages Projects Releases 7 Wiki Activity
Files
f7619baf34af05eec89a593f87c65c8eb05e9f49
MusicPlayer/tests/unit/test_music_server_client.gd
T
Sansan f7619baf34 Use /api/v1 with token authentication and typed responses 
Closes #38, #39 and #40; adds the client side of #41.

- Every request goes to /api/v1. MusicServerClient gets a token with
  Settings.api_password the first time one is needed, reuses it, and on a
  401 gets a new one and sends the request once more. Switching server gets
  a new token too. Callers never see the token.
- Sync is started with POST, as MusicServer 5.0 requires.
- The client has every endpoint in the server's OpenAPI spec, including
  statistics, soundtracks and songs by UUID, version, downloads and tokens.
- tools/generate_api_types.gd generates a typed class per schema in
  server/openapi.json into server/generated/. Song info, statistics and
  versions are passed to callers as these classes. A list the server sends
  as null is an empty list.
- FakeHttpTransport hands out tokens and answers 401 without one, so every
  test goes through the token flow. New tests cover the token handling,
  every endpoint and each generated class.
2026-10-06 22:45:28 +02:00

545 lines
20 KiB
GDScript
Raw Blame History

extends GdUnitTestSuite
## MusicServerClient: which paths it asks for, how it handles the answers and how it
## gets and renews its API token.
var _server: FakeHttpTransport
func before_test() -> void:
_server = FakeHttpTransport.new().install()
func after_test() -> void:
_server.uninstall()
func test_fetch_song_list_passes_song_infos() -> void:
var songs: Array = [{"SongNo": 0, "Soundtrack": "Chrono Trigger", "Song": "Corridors of Time"}]
_server.respond_json("/music/list", songs)
var received: Array[SongInfo] = []
MusicServerClient.fetch_song_list(
func(data: Array[SongInfo]) -> void: received.append_array(data))
assert_int(received.size()).is_equal(1)
assert_int(received[0].song_no).is_equal(0)
assert_str(received[0].soundtrack).is_equal("Chrono Trigger")
assert_str(received[0].song).is_equal("Corridors of Time")
func test_fetch_song_info_passes_song_info() -> void:
_server.respond_json("/music/info", {"SongNo": 7, "Soundtrack": "Terranigma",
"Song": "Underworld", "SoundtrackUuid": "st-1", "SongUuid": "so-1"})
var received: Array[SongInfo] = []
MusicServerClient.fetch_song_info(func(info: SongInfo) -> void: received.append(info))
assert_int(received[0].song_no).is_equal(7)
assert_str(received[0].song_uuid).is_equal("so-1")
func test_empty_song_list_sent_as_null_gives_empty_array() -> void:
_server.respond_bytes("/music/list", "null".to_utf8_buffer())
var received: Array = []
MusicServerClient.fetch_song_list(func(data: Array[SongInfo]) -> void: received.append(data))
assert_int(received.size()).is_equal(1)
assert_array(received[0]).is_empty()
func test_json_list_that_is_not_an_array_calls_on_failed() -> void:
_server.respond_json("/music/list", {"error": "not a list"})
var outcome: Array[String] = []
MusicServerClient.fetch_song_list(
func(_data: Array[SongInfo]) -> void: outcome.append("done"),
func() -> void: outcome.append("failed"))
assert_array(outcome).contains_exactly(["failed"])
func test_fetch_random_song_uses_normal_endpoint() -> void:
MusicServerClient.fetch_random_song(false, func(_body: PackedByteArray) -> void: pass)
assert_array(_server.requested_paths).contains_exactly(["/music/rand"])
func test_fetch_random_song_uses_low_played_endpoint() -> void:
MusicServerClient.fetch_random_song(true, func(_body: PackedByteArray) -> void: pass)
assert_array(_server.requested_paths).contains_exactly(["/music/rand/low"])
func test_fetch_song_passes_raw_bytes() -> void:
var mp3: PackedByteArray = PackedByteArray([0x49, 0x44, 0x33])
_server.respond_bytes("/music?song=4", mp3)
var received: Array[PackedByteArray] = []
MusicServerClient.fetch_song(4, func(body: PackedByteArray) -> void: received.append(body))
assert_array(received).contains_exactly([mp3])
func test_failed_request_does_not_call_back() -> void:
_server.respond_json("/music/list", [], 500)
var called: Array[bool] = [false]
MusicServerClient.fetch_song_list(func(_data: Variant) -> void: called[0] = true)
assert_bool(called[0]).is_false()
func test_invalid_json_does_not_call_back() -> void:
_server.respond_bytes("/music/info", "not json".to_utf8_buffer())
var called: Array[bool] = [false]
MusicServerClient.fetch_song_info(func(_data: SongInfo) -> void: called[0] = true)
assert_bool(called[0]).is_false()
func test_failed_request_calls_on_failed() -> void:
_server.respond_bytes("/music/rand", PackedByteArray(), 503)
var outcome: Array[String] = []
MusicServerClient.fetch_random_song(false,
func(_body: PackedByteArray) -> void: outcome.append("done"),
func() -> void: outcome.append("failed"))
assert_array(outcome).contains_exactly(["failed"])
func test_invalid_json_calls_on_failed() -> void:
_server.respond_bytes("/music/info", "not json".to_utf8_buffer())
var outcome: Array[String] = []
MusicServerClient.fetch_song_info(
func(_data: SongInfo) -> void: outcome.append("done"),
func() -> void: outcome.append("failed"))
assert_array(outcome).contains_exactly(["failed"])
func test_successful_request_does_not_call_on_failed() -> void:
_server.respond_bytes("/music/addQue", PackedByteArray())
var outcome: Array[String] = []
MusicServerClient.add_to_queue(
func() -> void: outcome.append("done"),
func() -> void: outcome.append("failed"))
assert_array(outcome).contains_exactly(["done"])
func test_send_calls_back_without_arguments() -> void:
_server.respond_bytes("/music/reset", PackedByteArray())
var called: Array[bool] = [false]
MusicServerClient.reset_playlist(func() -> void: called[0] = true)
assert_bool(called[0]).is_true()
func test_mark_song_played_sends_put() -> void:
_server.respond_bytes("/music/played?song=3", PackedByteArray())
MusicServerClient.mark_song_played(3)
assert_array(_server.requested_paths).contains_exactly(["/music/played?song=3"])
assert_array(_server.requested_methods).contains_exactly([HTTPClient.METHOD_PUT])
func test_character_name_is_url_encoded() -> void:
MusicServerClient.fetch_character_image("Paper Mario.png",
func(_body: PackedByteArray) -> void: pass)
assert_array(_server.requested_paths).contains_exactly(["/character?name=Paper%20Mario.png"])
func test_start_sync_sends_post() -> void:
_server.respond_json("/sync", "Sync started")
var called: Array[bool] = [false]
MusicServerClient.start_sync(func() -> void: called[0] = true)
assert_array(_server.requested_paths).contains_exactly(["/sync"])
assert_array(_server.requested_methods).contains_exactly([HTTPClient.METHOD_POST])
assert_bool(called[0]).is_true()
func test_start_full_sync_sends_post() -> void:
MusicServerClient.start_full_sync(func() -> void: pass)
assert_array(_server.requested_paths).contains_exactly(["/sync/full"])
assert_array(_server.requested_methods).contains_exactly([HTTPClient.METHOD_POST])
func test_sync_already_running_calls_on_failed() -> void:
_server.respond_json("/sync", {"error": "Sync is already running"}, 423)
var outcome: Array[String] = []
MusicServerClient.start_sync(
func() -> void: outcome.append("done"),
func() -> void: outcome.append("failed"))
assert_array(outcome).contains_exactly(["failed"])
func test_requests_go_to_selected_server_under_api_path() -> void:
var previous_path: String = Settings.default_path
Settings.default_path = "http://localhost:8080"
var requested_urls: Array[String] = []
MusicServerClient.transport = _RecordingTransport.new(requested_urls)
MusicServerClient.fetch_song_info(func(_data: SongInfo) -> void: pass)
Settings.default_path = previous_path
assert_array(requested_urls).contains_exactly([
"http://localhost:8080/api/v1/token",
"http://localhost:8080/api/v1/music/info",
])
# Tokens
func test_request_sends_token_in_authorization_header() -> void:
_server.respond_json("/music/info", {})
MusicServerClient.fetch_song_info(func(_data: SongInfo) -> void: pass)
assert_int(_server.token_requests).is_equal(1)
assert_array(_server.requested_headers[0]) \
.contains_exactly(["Authorization: Bearer fake-token-1"])
func test_token_request_sends_password_and_client_type() -> void:
var previous_password: String = Settings.api_password
Settings.api_password = "secret"
MusicServerClient.fetch_song_info(func(_data: SongInfo) -> void: pass)
Settings.api_password = previous_password
assert_dict(_server.last_token_request) \
.is_equal({"password": "secret", "client_type": "godot"})
func test_token_is_reused() -> void:
_server.respond_json("/music/info", {})
_server.respond_json("/music/list", [])
MusicServerClient.fetch_song_info(func(_data: SongInfo) -> void: pass)
MusicServerClient.fetch_song_list(func(_data: Array[SongInfo]) -> void: pass)
MusicServerClient.fetch_song_info(func(_data: SongInfo) -> void: pass)
assert_int(_server.token_requests).is_equal(1)
assert_array(_server.requested_paths) \
.contains_exactly(["/music/info", "/music/list", "/music/info"])
func test_requests_while_getting_a_token_share_it() -> void:
_server.respond_json("/music/info", {})
_server.respond_json("/music/list", [])
_server.hold("/token")
var done: Array[String] = []
MusicServerClient.fetch_song_info(func(_data: SongInfo) -> void: done.append("info"))
MusicServerClient.fetch_song_list(func(_data: Array[SongInfo]) -> void: done.append("list"))
assert_array(_server.requested_paths).is_empty()
_server.release("/token")
assert_int(_server.token_requests).is_equal(1)
assert_array(done).contains_exactly(["info", "list"])
func test_expired_token_is_renewed_and_request_sent_again() -> void:
_server.respond_json("/music/info", {"SongNo": 3})
MusicServerClient.fetch_song_info(func(_data: SongInfo) -> void: pass)
_server.expire_tokens()
var received: Array[SongInfo] = []
MusicServerClient.fetch_song_info(func(info: SongInfo) -> void: received.append(info))
assert_int(_server.token_requests).is_equal(2)
assert_array(_server.requested_paths) \
.contains_exactly(["/music/info", "/music/info", "/music/info"])
assert_array(_server.requested_headers[2]) \
.contains_exactly(["Authorization: Bearer fake-token-2"])
assert_int(received.size()).is_equal(1)
assert_int(received[0].song_no).is_equal(3)
func test_request_is_sent_again_only_once() -> void:
_server.respond_json("/music/info", {"error": "Unauthorized"}, 401)
var outcome: Array[String] = []
MusicServerClient.fetch_song_info(
func(_data: SongInfo) -> void: outcome.append("done"),
func() -> void: outcome.append("failed"))
assert_int(_server.token_requests).is_equal(2)
assert_int(_server.count("/music/info")).is_equal(2)
assert_array(outcome).contains_exactly(["failed"])
func test_refused_password_calls_on_failed_without_sending_request() -> void:
_server.respond_json("/token", {"error": "Invalid password"}, 401)
var outcome: Array[String] = []
MusicServerClient.fetch_song_info(
func(_data: SongInfo) -> void: outcome.append("done"),
func() -> void: outcome.append("failed"))
assert_array(_server.requested_paths).is_empty()
assert_array(outcome).contains_exactly(["failed"])
func test_rate_limited_token_request_calls_on_failed() -> void:
_server.respond_json("/token", {"error": "Too many requests"}, 429)
var outcome: Array[String] = []
MusicServerClient.add_to_queue(
func() -> void: outcome.append("done"),
func() -> void: outcome.append("failed"))
assert_array(outcome).contains_exactly(["failed"])
func test_new_token_after_a_failed_token_request() -> void:
_server.respond_json("/token", {"error": "Too many requests"}, 429)
MusicServerClient.add_to_queue(func() -> void: pass)
_server.respond_json("/token", {"token": "second", "expires_at": "", "client_type": "godot"})
_server.respond_bytes("/music/addQue", PackedByteArray())
_server.expire_tokens()
MusicServerClient.add_to_queue(func() -> void: pass)
assert_int(_server.token_requests).is_equal(2)
assert_array(_server.requested_headers.back()) \
.contains_exactly(["Authorization: Bearer second"])
func test_changing_server_gets_a_new_token() -> void:
var previous_path: String = Settings.default_path
_server.respond_json("/music/info", {})
MusicServerClient.fetch_song_info(func(_data: SongInfo) -> void: pass)
Settings.default_path = "http://localhost:8080"
MusicServerClient.fetch_song_info(func(_data: SongInfo) -> void: pass)
Settings.default_path = previous_path
assert_int(_server.token_requests).is_equal(2)
func test_health_needs_no_token() -> void:
_server.respond_json("/health", {"status": "ok"})
var received: Array = []
MusicServerClient.fetch_health(func(data: Dictionary) -> void: received.append(data))
assert_int(_server.token_requests).is_equal(0)
assert_array(_server.requested_headers[0]).is_empty()
assert_array(received).contains_exactly([{"status": "ok"}])
func test_delete_token_sends_token_and_forgets_it() -> void:
_server.respond_json("/music/info", {})
MusicServerClient.fetch_song_info(func(_data: SongInfo) -> void: pass)
_server.respond_json("/token", {"status": "deleted"})
var called: Array[bool] = [false]
MusicServerClient.delete_token(func() -> void: called[0] = true)
assert_bool(called[0]).is_true()
assert_int(_server.requested_methods.back()).is_equal(HTTPClient.METHOD_DELETE)
assert_array(_server.requested_headers.back()) \
.contains_exactly(["Authorization: Bearer fake-token-1"])
func test_delete_token_without_token_sends_nothing() -> void:
var called: Array[bool] = [false]
MusicServerClient.delete_token(func() -> void: called[0] = true)
assert_bool(called[0]).is_true()
assert_array(_server.requested_paths).is_empty()
# Other endpoints
func test_endpoint_paths_and_methods() -> void:
var noop: Callable = func(_data: Variant = null) -> void: pass
var calls: Array[Array] = [
[func() -> void: MusicServerClient.fetch_random_song_classic(noop),
"/music/rand/classic", HTTPClient.METHOD_GET],
[func() -> void: MusicServerClient.fetch_next_song(noop),
"/music/next", HTTPClient.METHOD_GET],
[func() -> void: MusicServerClient.fetch_previous_song(noop),
"/music/previous", HTTPClient.METHOD_GET],
[func() -> void: MusicServerClient.fetch_sound_test_song(noop),
"/music/soundTest", HTTPClient.METHOD_GET],
[func() -> void: MusicServerClient.mark_current_song_played(noop),
"/music/addPlayed", HTTPClient.METHOD_GET],
[func() -> void: MusicServerClient.fetch_all_soundtracks(noop),
"/music/all", HTTPClient.METHOD_GET],
[func() -> void: MusicServerClient.fetch_all_soundtracks_ordered(noop),
"/music/all/order", HTTPClient.METHOD_GET],
[func() -> void: MusicServerClient.fetch_soundtracks(noop),
"/soundtracks", HTTPClient.METHOD_GET],
[func() -> void: MusicServerClient.fetch_soundtrack("a b", noop),
"/soundtracks/a%20b", HTTPClient.METHOD_GET],
[func() -> void: MusicServerClient.fetch_song_by_uuid("so-1", noop),
"/songs/so-1", HTTPClient.METHOD_GET],
[func() -> void: MusicServerClient.fetch_song_file("so-1", noop),
"/songs/so-1/file", HTTPClient.METHOD_GET],
[func() -> void: MusicServerClient.fetch_character_list(noop),
"/character/list", HTTPClient.METHOD_GET],
[func() -> void: MusicServerClient.fetch_sync_progress(noop),
"/sync/progress", HTTPClient.METHOD_GET],
[func() -> void: MusicServerClient.fetch_version(noop),
"/version", HTTPClient.METHOD_GET],
[func() -> void: MusicServerClient.fetch_version_history(noop),
"/version/history", HTTPClient.METHOD_GET],
[func() -> void: MusicServerClient.fetch_latest_client_version(noop),
"/download", HTTPClient.METHOD_GET],
[func() -> void: MusicServerClient.fetch_client_assets(noop),
"/download/list", HTTPClient.METHOD_GET],
[func() -> void: MusicServerClient.download_client_windows(noop),
"/download/windows", HTTPClient.METHOD_GET],
[func() -> void: MusicServerClient.download_client_linux(noop),
"/download/linux", HTTPClient.METHOD_GET],
[func() -> void: MusicServerClient.cleanup_tokens(noop),
"/token/cleanup", HTTPClient.METHOD_POST],
[func() -> void: MusicServerClient.fetch_statistics_summary(noop),
"/statistics/summary", HTTPClient.METHOD_GET],
[func() -> void: MusicServerClient.fetch_most_played_soundtracks(5, noop),
"/statistics/soundtracks/most-played?limit=5", HTTPClient.METHOD_GET],
[func() -> void: MusicServerClient.fetch_least_played_soundtracks(5, noop),
"/statistics/soundtracks/least-played?limit=5", HTTPClient.METHOD_GET],
[func() -> void: MusicServerClient.fetch_last_played_soundtracks(5, noop),
"/statistics/soundtracks/last-played?limit=5", HTTPClient.METHOD_GET],
[func() -> void: MusicServerClient.fetch_oldest_played_soundtracks(5, noop),
"/statistics/soundtracks/oldest-played?limit=5", HTTPClient.METHOD_GET],
[func() -> void: MusicServerClient.fetch_never_played_soundtracks(noop),
"/statistics/soundtracks/never-played", HTTPClient.METHOD_GET],
[func() -> void: MusicServerClient.fetch_most_played_songs(5, noop),
"/statistics/songs/most-played?limit=5", HTTPClient.METHOD_GET],
[func() -> void: MusicServerClient.fetch_least_played_songs(5, noop),
"/statistics/songs/least-played?limit=5", HTTPClient.METHOD_GET],
]
for entry: Array in calls:
_server.requested_paths.clear()
_server.requested_methods.clear()
entry[0].call()
assert_array(_server.requested_paths).contains_exactly([entry[1]])
assert_array(_server.requested_methods).contains_exactly([entry[2]])
func test_statistics_summary_is_typed() -> void:
_server.respond_json("/statistics/summary", {"total_soundtracks": 120,
"played_soundtracks": 100, "avg_soundtrack_plays": 2.5})
var received: Array[StatisticsSummary] = []
MusicServerClient.fetch_statistics_summary(
func(summary: StatisticsSummary) -> void: received.append(summary))
assert_int(received[0].total_soundtracks).is_equal(120)
assert_int(received[0].played_soundtracks).is_equal(100)
assert_float(received[0].avg_soundtrack_plays).is_equal(2.5)
func test_most_played_soundtracks_are_typed_with_their_songs() -> void:
_server.respond_json("/statistics/soundtracks/most-played?limit=2", [
{"soundtrack_uuid": "st-1", "soundtrack_name": "Chrono Trigger",
"soundtrack_played": 9, "songs": [{"song_name": "Frog's Theme"}]},
{"soundtrack_uuid": "st-2", "soundtrack_name": "Lufia II", "soundtrack_played": 4,
"songs": []},
])
var received: Array[SoundtrackWithSongs] = []
MusicServerClient.fetch_most_played_soundtracks(2,
func(list: Array[SoundtrackWithSongs]) -> void: received.append_array(list))
assert_int(received.size()).is_equal(2)
assert_str(received[0].soundtrack_name).is_equal("Chrono Trigger")
assert_int(received[0].soundtrack_played).is_equal(9)
assert_str(received[0].songs[0].song_name).is_equal("Frog's Theme")
assert_array(received[1].songs).is_empty()
func test_most_played_songs_are_typed() -> void:
_server.respond_json("/statistics/songs/most-played?limit=1", [
{"song_uuid": "so-1", "song_name": "Dancing Mad", "times_played": 12},
])
var received: Array[SongInfoForStats] = []
MusicServerClient.fetch_most_played_songs(1,
func(list: Array[SongInfoForStats]) -> void: received.append_array(list))
assert_str(received[0].song_name).is_equal("Dancing Mad")
assert_int(received[0].times_played).is_equal(12)
func test_statistics_calls_on_failed_on_server_error() -> void:
var outcome: Array[String] = []
var on_done: Callable = func(_data: Variant) -> void: outcome.append("done")
var on_failed: Callable = func() -> void: outcome.append("failed")
for path: String in ["/statistics/summary",
"/statistics/soundtracks/most-played?limit=3",
"/statistics/soundtracks/least-played?limit=3",
"/statistics/soundtracks/last-played?limit=3",
"/statistics/soundtracks/oldest-played?limit=3",
"/statistics/soundtracks/never-played",
"/statistics/songs/most-played?limit=3",
"/statistics/songs/least-played?limit=3"]:
_server.respond_json(path, {"error": "Internal Server Error"}, 500)
_call_every_statistics_endpoint(on_done, on_failed)
assert_array(outcome).contains_exactly(["failed", "failed", "failed", "failed", "failed",
"failed", "failed", "failed"])
func test_statistics_calls_on_failed_when_token_is_refused() -> void:
_server.respond_json("/token", {"error": "Invalid password"}, 401)
var outcome: Array[String] = []
_call_every_statistics_endpoint(func(_data: Variant) -> void: outcome.append("done"),
func() -> void: outcome.append("failed"))
assert_array(_server.requested_paths).is_empty()
assert_array(outcome).contains_exactly(["failed", "failed", "failed", "failed", "failed",
"failed", "failed", "failed"])
func _call_every_statistics_endpoint(on_done: Callable, on_failed: Callable) -> void:
MusicServerClient.fetch_statistics_summary(on_done, on_failed)
MusicServerClient.fetch_most_played_soundtracks(3, on_done, on_failed)
MusicServerClient.fetch_least_played_soundtracks(3, on_done, on_failed)
MusicServerClient.fetch_last_played_soundtracks(3, on_done, on_failed)
MusicServerClient.fetch_oldest_played_soundtracks(3, on_done, on_failed)
MusicServerClient.fetch_never_played_soundtracks(on_done, on_failed)
MusicServerClient.fetch_most_played_songs(3, on_done, on_failed)
MusicServerClient.fetch_least_played_songs(3, on_done, on_failed)
class _RecordingTransport:
extends HttpTransport
var _urls: Array[String]
func _init(urls: Array[String]) -> void:
super(null)
_urls = urls
func request(
url: String,
_method: int,
on_complete: Callable,
_headers: PackedStringArray = PackedStringArray(),
_body: String = "",
) -> void:
_urls.append(url)
if url.ends_with("/token"):
on_complete.call(200, JSON.stringify({"token": "t"}).to_utf8_buffer())
Reference in new issue View Git Blame Copy Permalink