test_mpd_tools.py (24292B)
1 import json 2 import os 3 import shlex 4 import socketserver 5 import subprocess 6 import sys 7 import tempfile 8 import threading 9 import time 10 import unittest 11 from pathlib import Path 12 13 ROOT = Path(__file__).resolve().parents[2] 14 BIN_OVERRIDE = os.environ.get("MPD_TEST_BIN") 15 BIN = Path(BIN_OVERRIDE) if BIN_OVERRIDE else ROOT / "bin" 16 TOOLS = ( 17 "mpd_add_to_playlist", 18 "mpd_add_to_queue", 19 "mpd_edit_queue", 20 "mpd_update_library", 21 "mpd_update_queue", 22 "mpd_insert_next", 23 "mpd_edit_playlist", 24 "mpd_now_playing", 25 "mpd_trim_queue", 26 "mpd_report", 27 ) 28 29 30 class MPDState: 31 def __init__(self): 32 self.lock = threading.Lock() 33 self.queue = [] 34 self.playlists = {} 35 self.next_id = 1 36 self.version = 1 37 self.state = "stop" 38 self.current = None 39 self.commands = [] 40 self.updating = False 41 self.advance_on_current_song = False 42 43 def add_song(self, uri, position=None): 44 song = {"uri": uri, "id": self.next_id} 45 self.next_id += 1 46 if position is None or position >= len(self.queue): 47 self.queue.append(song) 48 else: 49 self.queue.insert(position, song) 50 if self.current is not None and position <= self.current: 51 self.current += 1 52 self.version += 1 53 return song 54 55 def response(self, raw): 56 args = shlex.split(raw) 57 if not args: 58 return [] 59 command = args[0] 60 self.commands.append((command, args[1:])) 61 62 if command == "status": 63 lines = [ 64 "volume: -1", 65 "repeat: 0", 66 "random: 0", 67 "single: 0", 68 "consume: 0", 69 f"playlist: {self.version}", 70 f"playlistlength: {len(self.queue)}", 71 f"state: {self.state}", 72 ] 73 if self.current is not None and self.current < len(self.queue): 74 song = self.queue[self.current] 75 lines += [ 76 f"song: {self.current}", 77 f"songid: {song['id']}", 78 "elapsed: 12.500", 79 "duration: 180.000", 80 ] 81 if self.updating: 82 lines.append("updating_db: 1") 83 return lines 84 85 if command == "currentsong": 86 if self.advance_on_current_song: 87 self.advance_on_current_song = False 88 if self.current is not None and self.current + 1 < len(self.queue): 89 self.current += 1 90 if self.current is None or self.current >= len(self.queue): 91 return [] 92 song = self.queue[self.current] 93 return [ 94 f"file: {song['uri']}", 95 'Artist: Artist "quoted"', 96 "AlbumArtist: Album Artist", 97 "Album: Test Album", 98 "Title: Test Title", 99 "Time: 180", 100 "duration: 180.000", 101 f"Pos: {self.current}", 102 f"Id: {song['id']}", 103 ] 104 105 if command == "playlistinfo": 106 lines = [] 107 for position, song in enumerate(self.queue): 108 lines += [ 109 f"file: {song['uri']}", 110 "Time: 180", 111 f"Pos: {position}", 112 f"Id: {song['id']}", 113 ] 114 return lines 115 116 if command == "listplaylist": 117 return [f"file: {uri}" for uri in self.playlists.get(args[1], [])] 118 119 if command in ("add", "addid"): 120 uri = args[1] 121 if uri == "FAIL": 122 raise ValueError("No such song") 123 position = int(args[2]) if command == "addid" and len(args) > 2 else None 124 song = self.add_song(uri, position) 125 return [f"Id: {song['id']}"] if command == "addid" else [] 126 127 if command == "delete": 128 spec = args[1] 129 if ":" in spec: 130 start_text, end_text = spec.split(":", 1) 131 start = int(start_text) 132 end = len(self.queue) if not end_text else int(end_text) 133 else: 134 start = int(spec) 135 end = start + 1 136 removed = end - start 137 del self.queue[start:end] 138 if self.current is not None: 139 if start <= self.current < end: 140 self.current = None 141 self.state = "stop" 142 elif self.current >= end: 143 self.current -= removed 144 self.version += 1 145 return [] 146 147 if command == "move": 148 source, destination = int(args[1]), int(args[2]) 149 song = self.queue.pop(source) 150 self.queue.insert(destination, song) 151 if self.current == source: 152 self.current = destination 153 elif self.current is not None and source < self.current <= destination: 154 self.current -= 1 155 elif self.current is not None and destination <= self.current < source: 156 self.current += 1 157 self.version += 1 158 return [] 159 160 if command == "clear": 161 self.queue.clear() 162 self.current = None 163 self.state = "stop" 164 self.version += 1 165 return [] 166 167 if command == "play": 168 if self.queue: 169 self.current = int(args[1]) if len(args) > 1 else 0 170 self.state = "play" 171 return [] 172 173 if command == "pause": 174 self.state = "pause" if args[1] == "1" else "play" 175 return [] 176 177 if command == "playlistclear": 178 self.playlists[args[1]] = [] 179 return [] 180 181 if command == "playlistadd": 182 if args[2] == "FAIL": 183 raise ValueError("No such song") 184 self.playlists.setdefault(args[1], []).append(args[2]) 185 return [] 186 187 if command == "rename": 188 source, destination = args[1], args[2] 189 if source not in self.playlists: 190 raise ValueError("No such playlist") 191 if destination in self.playlists: 192 raise ValueError("Playlist already exists") 193 self.playlists[destination] = self.playlists.pop(source) 194 return [] 195 196 if command == "rm": 197 if args[1] not in self.playlists: 198 raise ValueError("No such playlist") 199 del self.playlists[args[1]] 200 return [] 201 202 if command in ("update", "rescan"): 203 self.updating = True 204 return ["updating_db: 1"] 205 206 if command == "idle": 207 self.updating = False 208 return ["changed: update"] 209 210 raise ValueError(f"unsupported command: {raw}") 211 212 213 class MPDHandler(socketserver.StreamRequestHandler): 214 def handle(self): 215 self.wfile.write(b"OK MPD 0.23.0\n") 216 self.wfile.flush() 217 command_list = None 218 responses = [] 219 while True: 220 raw = self.rfile.readline() 221 if not raw: 222 return 223 line = raw.decode().rstrip("\r\n") 224 if line in ("command_list_begin", "command_list_ok_begin"): 225 command_list = line 226 responses = [] 227 with self.server.state.lock: 228 self.server.state.commands.append((line, [])) 229 continue 230 if line == "command_list_end": 231 with self.server.state.lock: 232 self.server.state.commands.append((line, [])) 233 for response in responses: 234 for item in response: 235 self.wfile.write(f"{item}\n".encode()) 236 if command_list == "command_list_ok_begin": 237 self.wfile.write(b"list_OK\n") 238 self.wfile.write(b"OK\n") 239 self.wfile.flush() 240 command_list = None 241 responses = [] 242 continue 243 try: 244 with self.server.state.lock: 245 response = self.server.state.response(line) 246 except (ValueError, IndexError) as error: 247 self.wfile.write(f"ACK [50@0] {{{line.split()[0]}}} {error}\n".encode()) 248 self.wfile.flush() 249 command_list = None 250 responses = [] 251 continue 252 if command_list is not None: 253 responses.append(response) 254 else: 255 for item in response: 256 self.wfile.write(f"{item}\n".encode()) 257 self.wfile.write(b"OK\n") 258 self.wfile.flush() 259 260 261 class MPDServer(socketserver.ThreadingTCPServer): 262 allow_reuse_address = True 263 daemon_threads = True 264 265 def __init__(self): 266 self.state = MPDState() 267 super().__init__(("127.0.0.1", 0), MPDHandler) 268 269 def handle_error(self, request, client_address): 270 if isinstance(sys.exception(), BrokenPipeError): 271 return 272 super().handle_error(request, client_address) 273 274 275 class MPDToolsTest(unittest.TestCase): 276 @classmethod 277 def setUpClass(cls): 278 if BIN_OVERRIDE is None: 279 subprocess.run( 280 ["knit", *(f"bin/{tool}" for tool in TOOLS)], 281 cwd=ROOT, 282 check=True, 283 ) 284 285 def setUp(self): 286 self.server = MPDServer() 287 self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) 288 self.thread.start() 289 290 def tearDown(self): 291 self.server.shutdown() 292 self.server.server_close() 293 self.thread.join(timeout=2) 294 295 def tool_environment(self, extra_env=None): 296 env = os.environ.copy() 297 env.update( 298 MPD_HOST="127.0.0.1", 299 MPD_PORT=str(self.server.server_address[1]), 300 MPD_TIMEOUT="2", 301 MPD_CONFIG_HELPER="off", 302 ) 303 if extra_env: 304 for key, value in extra_env.items(): 305 if value is None: 306 env.pop(key, None) 307 else: 308 env[key] = value 309 return env 310 311 def run_tool(self, tool, *args, input_text=None, extra_env=None): 312 return subprocess.run( 313 [str(BIN / tool), *args], 314 input=input_text, 315 text=True, 316 capture_output=True, 317 env=self.tool_environment(extra_env), 318 timeout=10, 319 ) 320 321 def seed_queue(self, uris, current=None, state="stop"): 322 for uri in uris: 323 self.server.state.add_song(uri) 324 self.server.state.current = current 325 self.server.state.state = state 326 327 def test_add_batches_and_preserves_line_content(self): 328 tracks = [f"https://example.test/{index}" for index in range(1100)] 329 tracks[5] = " leading and trailing " 330 text = "# ignored\r\n" + "\r\n".join(tracks) + "\r\n" 331 result = self.run_tool("mpd_add_to_queue", input_text=text) 332 self.assertEqual(result.returncode, 0, result.stderr) 333 self.assertIn("Added 1100 tracks", result.stdout) 334 self.assertEqual([song["uri"] for song in self.server.state.queue], tracks) 335 begins = [c for c, _ in self.server.state.commands if c == "command_list_begin"] 336 self.assertEqual(len(begins), 3) 337 338 def test_add_failure_is_nonzero(self): 339 result = self.run_tool("mpd_add_to_queue", input_text="ok\nFAIL\nlater\n") 340 self.assertEqual(result.returncode, 1) 341 self.assertIn("failed input batch", result.stderr) 342 343 def test_replace_uses_ranges_and_preserves_paused_current(self): 344 old = [f"old/{index}" for index in range(100)] 345 self.seed_queue(old, current=50, state="pause") 346 wanted = ["new/a", "new/b", old[50], "new/c", "new/d"] 347 result = self.run_tool("mpd_update_queue", input_text="\n".join(wanted)) 348 self.assertEqual(result.returncode, 0, result.stderr) 349 self.assertEqual([song["uri"] for song in self.server.state.queue], wanted) 350 self.assertEqual(self.server.state.current, 2) 351 self.assertEqual(self.server.state.state, "pause") 352 deletes = [ 353 args for command, args in self.server.state.commands if command == "delete" 354 ] 355 self.assertEqual(len(deletes), 2) 356 self.assertTrue(all(":" in args[0] for args in deletes)) 357 358 def test_failed_queue_replacement_rolls_back(self): 359 self.seed_queue(["old/a", "old/b"], current=1, state="pause") 360 result = self.run_tool("mpd_update_queue", input_text="new/a\nFAIL\n") 361 self.assertEqual(result.returncode, 1) 362 self.assertIn("restoring the previous queue", result.stderr) 363 self.assertNotIn("rollback failed", result.stderr) 364 self.assertEqual( 365 [song["uri"] for song in self.server.state.queue], 366 ["old/a", "old/b"], 367 ) 368 self.assertEqual(self.server.state.current, 1) 369 self.assertEqual(self.server.state.state, "pause") 370 371 def test_player_snapshot_retries_if_song_advances(self): 372 self.seed_queue(["old/a", "old/b", "old/c"], current=0, state="play") 373 self.server.state.advance_on_current_song = True 374 result = self.run_tool("mpd_update_queue", input_text="new/a\nold/b\nnew/b\n") 375 self.assertEqual(result.returncode, 0, result.stderr) 376 self.assertEqual( 377 [song["uri"] for song in self.server.state.queue], 378 ["new/a", "old/b", "new/b"], 379 ) 380 self.assertEqual(self.server.state.current, 1) 381 382 def test_insert_next_and_trim(self): 383 self.seed_queue(["a", "b", "c", "d", "e"], current=1, state="play") 384 result = self.run_tool("mpd_insert_next", input_text="x\ny\n") 385 self.assertEqual(result.returncode, 0, result.stderr) 386 self.assertEqual( 387 [song["uri"] for song in self.server.state.queue], 388 ["a", "b", "x", "y", "c", "d", "e"], 389 ) 390 self.server.state.current = 5 391 result = self.run_tool("mpd_trim_queue", "--keep", "1") 392 self.assertEqual(result.returncode, 0, result.stderr) 393 self.assertEqual( 394 [song["uri"] for song in self.server.state.queue], ["c", "d", "e"] 395 ) 396 self.assertEqual(self.server.state.current, 1) 397 398 def test_playlist_add_and_edit(self): 399 result = self.run_tool( 400 "mpd_add_to_playlist", "favorites", "-", input_text="one\ntwo\n" 401 ) 402 self.assertEqual(result.returncode, 0, result.stderr) 403 self.assertEqual(self.server.state.playlists["favorites"], ["one", "two"]) 404 405 with tempfile.TemporaryDirectory() as directory: 406 editor = Path(directory) / "editor.sh" 407 editor.write_text("#!/bin/sh\nprintf 'three\\nfour\\n' > \"$1\"\n") 408 editor.chmod(0o755) 409 result = self.run_tool( 410 "mpd_edit_playlist", 411 "favorites", 412 extra_env={"VISUAL": str(editor)}, 413 ) 414 self.assertEqual(result.returncode, 0, result.stderr) 415 self.assertEqual(self.server.state.playlists["favorites"], ["three", "four"]) 416 417 def test_failed_playlist_staging_preserves_original(self): 418 self.server.state.playlists["favorites"] = ["one", "two"] 419 with tempfile.TemporaryDirectory() as directory: 420 editor = Path(directory) / "editor.sh" 421 editor.write_text("#!/bin/sh\nprintf 'new\\nFAIL\\n' > \"$1\"\n") 422 editor.chmod(0o755) 423 result = self.run_tool( 424 "mpd_edit_playlist", 425 "favorites", 426 extra_env={"VISUAL": str(editor)}, 427 ) 428 self.assertEqual(result.returncode, 1) 429 self.assertEqual(self.server.state.playlists["favorites"], ["one", "two"]) 430 self.assertEqual( 431 [ 432 name 433 for name in self.server.state.playlists 434 if name.startswith("mpd-tools-") 435 ], 436 [], 437 ) 438 439 def test_queue_editor_preserves_current_and_aborts_on_editor_failure(self): 440 self.seed_queue(["one", "two"], current=0, state="play") 441 before = list(self.server.state.queue) 442 result = self.run_tool("mpd_edit_queue", extra_env={"VISUAL": "/bin/false"}) 443 self.assertEqual(result.returncode, 1) 444 self.assertEqual(self.server.state.queue, before) 445 446 with tempfile.TemporaryDirectory() as directory: 447 editor = Path(directory) / "editor.sh" 448 editor.write_text("#!/bin/sh\nprintf 'two\\none\\n' > \"$1\"\n") 449 editor.chmod(0o755) 450 result = self.run_tool("mpd_edit_queue", extra_env={"VISUAL": str(editor)}) 451 self.assertEqual(result.returncode, 0, result.stderr) 452 self.assertEqual( 453 [song["uri"] for song in self.server.state.queue], ["two", "one"] 454 ) 455 self.assertEqual(self.server.state.current, 1) 456 self.assertEqual(self.server.state.state, "play") 457 458 def test_now_playing_json(self): 459 self.seed_queue(["track.ogg"], current=0, state="play") 460 result = self.run_tool("mpd_now_playing", "--json") 461 self.assertEqual(result.returncode, 0, result.stderr) 462 value = json.loads(result.stdout) 463 self.assertEqual(value["state"], "playing") 464 self.assertEqual(value["artist"], 'Artist "quoted"') 465 self.assertEqual(value["uri"], "track.ogg") 466 self.assertEqual(value["elapsed_seconds"], 12.5) 467 self.assertEqual(value["song_id"], 1) 468 469 def test_now_playing_is_silent_when_paused_or_stopped(self): 470 for state in ("pause", "stop"): 471 with self.subTest(state=state): 472 self.seed_queue(["track.ogg"], current=0, state=state) 473 for arguments in ((), ("--json",)): 474 with self.subTest(arguments=arguments): 475 result = self.run_tool("mpd_now_playing", *arguments) 476 self.assertEqual(result.returncode, 0, result.stderr) 477 self.assertEqual(result.stdout, "") 478 479 def test_explicit_config_helper_overrides_connection_environment(self): 480 self.seed_queue(["helper.ogg"], current=0, state="play") 481 with tempfile.TemporaryDirectory() as directory: 482 helper = Path(directory) / "helper" 483 helper.write_text( 484 "#!/bin/sh\n" 485 "printf 'MPD_HOST=127.0.0.1\\n'\n" 486 f"printf 'MPD_PORT={self.server.server_address[1]}\\n'\n" 487 ) 488 helper.chmod(0o755) 489 result = self.run_tool( 490 "mpd_now_playing", 491 "--json", 492 extra_env={ 493 "MPD_CONFIG_HELPER": str(helper), 494 "MPD_HOST": "invalid.example", 495 "MPD_PORT": "1", 496 }, 497 ) 498 self.assertEqual(result.returncode, 0, result.stderr) 499 self.assertEqual(json.loads(result.stdout)["uri"], "helper.ogg") 500 501 def test_amen_is_auto_discovered_and_runs_once_across_reconnects(self): 502 self.seed_queue(["one"], current=0, state="pause") 503 with tempfile.TemporaryDirectory() as directory: 504 home = Path(directory) 505 (home / "bin").mkdir() 506 count = home / "calls" 507 helper = home / "bin" / "amen" 508 helper.write_text( 509 "#!/bin/sh\n" 510 f"printf x >> {shlex.quote(str(count))}\n" 511 "printf 'MPD_HOST=127.0.0.1\\n'\n" 512 f"printf 'MPD_PORT={self.server.server_address[1]}\\n'\n" 513 ) 514 helper.chmod(0o755) 515 result = self.run_tool( 516 "mpd_edit_queue", 517 extra_env={ 518 "HOME": str(home), 519 "MPD_CONFIG_HELPER": None, 520 "MPD_HOST": "invalid.example", 521 "MPD_PORT": "1", 522 "VISUAL": "/bin/true", 523 }, 524 ) 525 calls = count.read_text() 526 self.assertEqual(result.returncode, 0, result.stderr) 527 self.assertEqual(calls, "x") 528 529 def test_invalid_config_helper_output_fails_without_fallback(self): 530 with tempfile.TemporaryDirectory() as directory: 531 helper = Path(directory) / "invalid-helper" 532 helper.write_text("#!/bin/sh\nprintf 'MPD_HOST=127.0.0.1\\n'\n") 533 helper.chmod(0o755) 534 result = self.run_tool( 535 "mpd_now_playing", 536 extra_env={"MPD_CONFIG_HELPER": str(helper)}, 537 ) 538 self.assertEqual(result.returncode, 1) 539 self.assertIn("must print exactly valid MPD_HOST", result.stderr) 540 541 def test_config_helper_timeout_is_bounded(self): 542 with tempfile.TemporaryDirectory() as directory: 543 helper = Path(directory) / "slow-helper" 544 helper.write_text("#!/bin/sh\nexec sleep 30\n") 545 helper.chmod(0o755) 546 started = time.monotonic() 547 result = self.run_tool( 548 "mpd_now_playing", 549 extra_env={ 550 "MPD_CONFIG_HELPER": str(helper), 551 "MPD_CONFIG_HELPER_TIMEOUT": "1", 552 }, 553 ) 554 elapsed = time.monotonic() - started 555 self.assertEqual(result.returncode, 1) 556 self.assertIn("timed out after 1 seconds", result.stderr) 557 self.assertLess(elapsed, 3) 558 559 def test_editor_uses_tmpdir(self): 560 self.server.state.playlists["favorites"] = ["one"] 561 with tempfile.TemporaryDirectory() as directory: 562 editor = Path(directory) / "editor.sh" 563 editor.write_text( 564 "#!/bin/sh\n" 565 f'case "$1" in {shlex.quote(directory)}/*) exit 0;; *) exit 1;; esac\n' 566 ) 567 editor.chmod(0o755) 568 result = self.run_tool( 569 "mpd_edit_playlist", 570 "favorites", 571 extra_env={"VISUAL": str(editor), "TMPDIR": directory}, 572 ) 573 self.assertEqual(result.returncode, 0, result.stderr) 574 575 def test_watch_returns_signal_specific_status(self): 576 process = subprocess.Popen( 577 [str(BIN / "mpd_now_playing"), "--watch"], 578 stdout=subprocess.DEVNULL, 579 stderr=subprocess.PIPE, 580 text=True, 581 env=self.tool_environment(), 582 ) 583 try: 584 time.sleep(0.1) 585 process.terminate() 586 _, stderr = process.communicate(timeout=3) 587 finally: 588 if process.poll() is None: 589 process.kill() 590 process.wait() 591 self.assertEqual(process.returncode, 128 + 15, stderr) 592 593 def test_empty_replacement_clears_queue(self): 594 self.seed_queue(["one", "two"], current=0, state="pause") 595 result = self.run_tool("mpd_update_queue", input_text="") 596 self.assertEqual(result.returncode, 0, result.stderr) 597 self.assertEqual(self.server.state.queue, []) 598 self.assertEqual(self.server.state.state, "stop") 599 600 def test_library_update_deduplicates_without_fixed_delay(self): 601 result = self.run_tool( 602 "mpd_update_library", 603 "--no-wait", 604 input_text="rock/a.ogg\nrock/b.ogg\njazz/c.ogg\n", 605 ) 606 self.assertEqual(result.returncode, 0, result.stderr) 607 updates = [ 608 args for command, args in self.server.state.commands if command == "update" 609 ] 610 self.assertEqual(updates, [["jazz"], ["rock"]]) 611 612 def test_library_update_waits_on_idle_event(self): 613 result = self.run_tool("mpd_update_library", input_text="rock/a.ogg\n") 614 self.assertEqual(result.returncode, 0, result.stderr) 615 commands = [command for command, _ in self.server.state.commands] 616 self.assertIn("idle", commands) 617 618 def test_help_and_invalid_usage(self): 619 for tool in TOOLS: 620 result = self.run_tool(tool, "--help") 621 self.assertEqual(result.returncode, 0, f"{tool}: {result.stderr}") 622 self.assertIn("Usage:", result.stdout) 623 result = self.run_tool("mpd_trim_queue", "--keep", "nope") 624 self.assertEqual(result.returncode, 2) 625 result = self.run_tool("mpd_add_to_queue", "--unknown") 626 self.assertEqual(result.returncode, 2) 627 628 629 if __name__ == "__main__": 630 unittest.main()