test_mpd_report.py (5537B)
1 import datetime 2 import os 3 import socketserver 4 import subprocess 5 import tempfile 6 import threading 7 import unittest 8 from pathlib import Path 9 10 BIN_OVERRIDE = os.environ.get("MPD_TEST_BIN") 11 12 MPD_RESPONSE = ( 13 "directory: Artist\n" 14 "Last-Modified: 2026-08-07T10:00:00Z\n" 15 "directory: Artist/Album\n" 16 "Last-Modified: 2026-08-08T10:00:00Z\n" 17 "file: Artist/Album/track.flac\n" 18 "Last-Modified: 2026-08-08T11:00:00Z\n" 19 "Time: 60\n" 20 "duration: 60.0\n" 21 ) 22 23 24 class MpdHandler(socketserver.StreamRequestHandler): 25 def handle(self): 26 self.wfile.write(b"OK MPD 0.23.0\n") 27 self.wfile.flush() 28 while raw := self.rfile.readline(): 29 command = raw.decode().rstrip("\r\n") 30 self.server.commands.append(command) 31 if command.startswith("listallinfo "): 32 self.wfile.write(MPD_RESPONSE.encode()) 33 self.wfile.write(b"OK\n") 34 else: 35 name = command.split(maxsplit=1)[0] 36 self.wfile.write( 37 f"ACK [5@0] {{{name}}} unsupported test command\n".encode() 38 ) 39 self.wfile.flush() 40 41 42 class MpdServer(socketserver.ThreadingTCPServer): 43 allow_reuse_address = True 44 daemon_threads = True 45 46 def __init__(self): 47 self.commands = [] 48 super().__init__(("127.0.0.1", 0), MpdHandler) 49 50 51 def epoch(day): 52 value = datetime.datetime.fromisoformat(day).replace(tzinfo=datetime.UTC) 53 return int(value.timestamp()) 54 55 56 class MpdReportTest(unittest.TestCase): 57 @classmethod 58 def setUpClass(cls): 59 if BIN_OVERRIDE is not None: 60 cls.temporary_directory = None 61 cls.binary = Path(BIN_OVERRIDE) / "mpd_report" 62 return 63 cls.temporary_directory = tempfile.TemporaryDirectory() 64 cls.binary = Path(cls.temporary_directory.name) / "mpd_report" 65 source = Path(__file__).resolve().parents[1] / "mpd_report.c" 66 common = Path(__file__).resolve().parents[1] / "mpd_common.c" 67 subprocess.run( 68 [ 69 "cc", 70 "-std=c17", 71 "-D_POSIX_C_SOURCE=200809L", 72 "-Wall", 73 "-Wextra", 74 "-Wconversion", 75 "-Wstrict-prototypes", 76 "-o", 77 str(cls.binary), 78 str(source), 79 str(common), 80 "-lmpdclient", 81 ], 82 check=True, 83 ) 84 85 @classmethod 86 def tearDownClass(cls): 87 if cls.temporary_directory is not None: 88 cls.temporary_directory.cleanup() 89 90 def setUp(self): 91 self.server = MpdServer() 92 self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) 93 self.thread.start() 94 95 def tearDown(self): 96 self.server.shutdown() 97 self.server.server_close() 98 self.thread.join(timeout=2) 99 100 def run_tool(self, *arguments, cwd=None): 101 environment = os.environ.copy() 102 environment.update( 103 MPD_HOST="127.0.0.1", 104 MPD_PORT=str(self.server.server_address[1]), 105 MPD_CONFIG_HELPER="off", 106 ) 107 return subprocess.run( 108 [str(self.binary), *arguments], 109 text=True, 110 capture_output=True, 111 cwd=cwd, 112 env=environment, 113 timeout=5, 114 ) 115 116 def test_all_time_writes_plain_directories_and_tracks(self): 117 with tempfile.TemporaryDirectory() as directory: 118 report = Path(directory) / "report.txt" 119 tracks = Path(directory) / "all_tracks.txt" 120 result = self.run_tool( 121 "--output", str(report), "--tracks-output", str(tracks) 122 ) 123 124 self.assertEqual(result.returncode, 0, result.stderr) 125 self.assertEqual(result.stdout, "") 126 self.assertEqual(report.read_text(), "Artist\nArtist/Album\n") 127 self.assertEqual(tracks.read_text(), "Artist/Album/track.flac\n") 128 self.assertEqual(self.server.commands, ['listallinfo ""']) 129 130 def test_multiple_windows_are_indexed_and_can_include_files(self): 131 first = f"{epoch('2026-08-07')},{epoch('2026-08-08')}" 132 second = f"{epoch('2026-08-08')},{epoch('2026-08-09')}" 133 result = self.run_tool( 134 "--output", "-", "--files", "--window", first, "--window", second 135 ) 136 137 self.assertEqual(result.returncode, 0, result.stderr) 138 self.assertEqual( 139 result.stdout, 140 "0\tArtist\n" "1\tArtist/Album\n" "1\tArtist/Album/track.flac\n", 141 ) 142 143 def test_overlapping_windows_emit_each_matching_index(self): 144 broad = f"{epoch('2026-08-07')},{epoch('2026-08-09')}" 145 narrow = f"{epoch('2026-08-08')},{epoch('2026-08-09')}" 146 result = self.run_tool("--output", "-", "--window", broad, "--window", narrow) 147 148 self.assertEqual(result.returncode, 0, result.stderr) 149 self.assertEqual( 150 result.stdout, 151 "0\tArtist\n" "0\tArtist/Album\n" "1\tArtist/Album\n", 152 ) 153 154 def test_default_output_is_all_txt_and_stdout_stays_empty(self): 155 with tempfile.TemporaryDirectory() as directory: 156 result = self.run_tool(cwd=directory) 157 158 self.assertEqual(result.returncode, 0, result.stderr) 159 self.assertEqual(result.stdout, "") 160 self.assertEqual( 161 (Path(directory) / "all.txt").read_text(), 162 "Artist\nArtist/Album\n", 163 ) 164 165 166 if __name__ == "__main__": 167 unittest.main()