Something went wrong. Try again.
My personal dotfiles and NixOS configurations.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675import faulthandlerimport importlib.utilfrom http.server import BaseHTTPRequestHandler, ThreadingHTTPServerimport osfrom pathlib import Pathimport shutilimport structimport subprocessimport sysimport tempfileimport threadingimport timeimport typesimport unittestimport wave
import gi
gi.require_version("Gst", "1.0")from gi.repository import GLib, Gst
faulthandler.dump_traceback_later(30, repeat=True)Gst.init(None)if not os.environ.get("POLYPHONY_SOURCE"): raise SystemExit("Set POLYPHONY_SOURCE to the patched Polyphony source tree")SOURCE = Path(os.environ["POLYPHONY_SOURCE"])WORK = tempfile.TemporaryDirectory()ROOT = Path(WORK.name)shutil.copytree(SOURCE / "src", ROOT / "src")for path in (ROOT / "src").rglob("*"): path.chmod(0o755 if path.is_dir() else 0o644)(ROOT / "src").chmod(0o755)spec = importlib.util.spec_from_file_location("playback_backend", ROOT / "src/playback_backend.py")module = importlib.util.module_from_spec(spec)spec.loader.exec_module(module)PlaybackBackend = module.PlaybackBackend
class HTTPFixture: def __init__(self, body, header_delay=0, body_delay=0, failures=()): self.body = body self.header_delay = header_delay self.body_delay = body_delay self.failures = list(failures) self.requests = [] fixture = self
class Handler(BaseHTTPRequestHandler): def log_message(self, *args): pass
def do_GET(self): fixture.requests.append(self.headers.get("Range")) failure = fixture.failures.pop(0) if fixture.failures else None time.sleep(fixture.header_delay) if failure == 500: self.send_error(500) return start = int(self.headers.get("Range", "bytes=0-").split("=")[1].split("-")[0]) body = fixture.body[start:] self.send_response(206 if start else 200) self.send_header("Content-Type", "audio/mp4") self.send_header("Content-Length", str(len(body))) self.send_header("Accept-Ranges", "bytes") if start: self.send_header("Content-Range", f"bytes {start}-{len(fixture.body)-1}/{len(fixture.body)}") self.end_headers() try: if failure == "short": self.wfile.write(body[:32]) self.close_connection = True return for offset in range(0, len(body), 4096): time.sleep(fixture.body_delay) self.wfile.write(body[offset:offset + 4096]) self.wfile.flush() except (BrokenPipeError, ConnectionResetError): pass
self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) self.thread.start() self.uri = f"http://127.0.0.1:{self.server.server_port}/decrypt/test"
def close(self): self.server.shutdown() self.server.server_close() self.thread.join()
class GaplessTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.commits = [] self.errors = [] self.eos = [] self.backend = PlaybackBackend(finish_callback=self.commits.append, error_callback=lambda *args: self.errors.append(args), eos_callback=lambda: self.eos.append(True)) self.sink = Gst.ElementFactory.make("fakesink") self.sink.set_property("sync", True) self.sink.set_property("signal-handoffs", True) self.buffers = [] self.sink.connect("handoff", lambda sink, buf, pad: self.buffers.append((time.monotonic(), buf.get_size()))) self.backend.pipeline.set_property("audio-sink", self.sink) self.backend.pipeline.set_property("video-sink", Gst.ElementFactory.make("fakesink"))
def tearDown(self): self.backend.close() self.tmp.cleanup()
def wav(self, name, rate=48000, seconds=0.6): path = Path(self.tmp.name) / name with wave.open(str(path), "wb") as output: output.setnchannels(1) output.setsampwidth(2) output.setframerate(rate) output.writeframes(struct.pack("<h", 1000) * int(rate * seconds)) return path.as_uri()
def wait(self, predicate, timeout=6): deadline = time.monotonic() + timeout context = GLib.MainContext.default() while not predicate() and time.monotonic() < deadline: while context.pending(): context.iteration(False) time.sleep(0.002) self.assertTrue(predicate(), "pipeline timeout") self.assertEqual(self.errors, [])
def test_gapless_identical_uri_and_early_assignment(self): uri = self.wav("same.wav", seconds=1) token = object() assigned = [] main_thread = threading.get_ident() self.backend.pipeline.connect("about-to-finish", lambda *args: assigned.append((time.monotonic(), threading.get_ident()))) self.backend.queue_uri(uri, token) start = time.monotonic() self.backend.load_uri(uri) self.wait(lambda: bool(self.eos)) self.assertEqual(self.commits, [token]) self.assertGreaterEqual(len(assigned), 2) self.assertNotEqual(assigned[0][1], main_thread) self.assertLess(assigned[0][0] - start, 0.8) self.assertEqual(sum(size for _, size in self.buffers), 2 * 48000 * 2) self.assertGreater(self.buffers[-1][0] - start, 1.8) gaps = [b[0] - a[0] for a, b in zip(self.buffers, self.buffers[1:])] self.assertLess(max(gaps), 0.2)
def test_initial_start_does_not_commit_successor(self): uri = self.wav("initial.wav", seconds=1) token = object() self.backend.queue_uri(uri, token) self.backend.load_uri(uri) self.wait(lambda: bool(self.buffers)) self.assertEqual(self.commits, []) self.wait(lambda: bool(self.commits)) self.assertEqual(self.commits, [token])
def test_unarmed_eos_has_no_commit(self): self.backend.load_uri(self.wav("fallback.wav")) self.wait(lambda: bool(self.eos)) self.assertEqual(self.commits, [])
def test_late_arm_does_not_fabricate_transition(self): uri = self.wav("late.wav", seconds=1) emitted = threading.Event() self.backend.pipeline.connect("about-to-finish", lambda *args: emitted.set()) self.backend.load_uri(uri) self.wait(emitted.is_set) self.backend.queue_uri(uri, object()) self.wait(lambda: bool(self.eos)) self.assertEqual(self.commits, [])
def test_stop_after_assignment_invalidates_pending_start(self): uri = self.wav("stop.wav", seconds=1) token = object() self.backend.queue_uri(uri, token) self.backend.load_uri(uri) self.wait(lambda: self.backend._assigned is token) self.assertTrue(self.backend.clear_next_uri()) self.backend.stop() self.wait(lambda: True) self.assertEqual(self.commits, []) self.assertIsNone(self.backend._assigned) self.assertEqual(list(self.backend._streams), [])
def test_repeated_teardown_with_assigned_successor(self): current = self.wav("teardown-current.wav", seconds=3) successor = self.wav("teardown-next.wav", rate=44100, seconds=1) for paused in [False, True] * 10: token = object() self.backend.queue_uri(successor, token) self.backend.load_uri(current) self.wait(lambda: self.backend._assigned is token) if paused: self.backend.pause() self.wait(lambda: self.backend.pipeline.get_state(0)[1] == Gst.State.PAUSED) started = time.monotonic() self.backend.stop() self.assertLess(time.monotonic() - started, 2) self.assertEqual(self.backend.pipeline.get_state(0)[1], Gst.State.NULL) self.assertIsNone(self.backend._assigned) self.assertEqual(list(self.backend._streams), []) self.assertEqual(self.commits, [])
def test_codec_label_from_legacy_playbin(self): labels = [] self.backend._codec_callback = labels.append self.backend.load_uri(self.wav("codec.wav")) self.wait(lambda: bool(labels)) self.assertIn("PCM · 48 kHz", labels)
def test_manual_next_resets_old_epoch(self): old = self.wav("old.wav", seconds=1) new = self.wav("new.wav", seconds=0.3) token = object() self.backend.queue_uri(old, token) self.backend.load_uri(old) self.wait(lambda: self.backend._assigned is token) self.backend.stop() self.backend.load_uri(new) self.wait(lambda: bool(self.eos)) self.assertEqual(self.commits, [])
def test_clear_before_assignment(self): uri = self.wav("clear.wav") self.backend.queue_uri(uri, object()) self.assertFalse(self.backend.clear_next_uri()) self.backend.load_uri(uri) self.wait(lambda: bool(self.eos)) self.assertEqual(self.commits, [])
def test_changed_rate_stream_starts(self): first = self.wav("first.wav") second = self.wav("second.wav", rate=44100) token = object() self.backend.queue_uri(second, token) self.backend.load_uri(first) self.wait(lambda: bool(self.eos)) self.assertEqual(self.commits, [token])
def test_successor_armed_after_early_signal_falls_back(self): uri = self.wav("triple.wav", seconds=1) first, second = object(), object()
def commit(token): self.commits.append(token) if token is first: self.backend.queue_uri(uri, second)
self.backend._finish_callback = commit self.backend.queue_uri(uri, first) self.backend.load_uri(uri) self.wait(lambda: bool(self.eos)) self.assertEqual(self.commits, [first]) self.assertEqual(sum(size for _, size in self.buffers), 2 * 48000 * 2)
def seek_after_assignment(self, paused): current = self.wav("seek-current.wav", seconds=3) successor = self.wav("seek-next.wav", rate=44100, seconds=1) token = object() self.backend.queue_uri(successor, token) self.backend.load_uri(current) self.wait(lambda: self.backend._assigned is token) if paused: self.backend.pause() self.wait(lambda: self.backend.pipeline.get_state(0)[1] == Gst.State.PAUSED) completed = [] self.backend.seek(1400, completed.append) self.assertEqual(completed, []) self.wait(lambda: bool(completed)) self.assertEqual(completed, [True]) self.assertEqual(self.backend.pipeline.get_property("current-uri"), current) ok, position = self.backend.pipeline.query_position(Gst.Format.TIME) self.assertTrue(ok) self.assertAlmostEqual(position / Gst.MSECOND, 1400, delta=150) expected = Gst.State.PAUSED if paused else Gst.State.PLAYING self.wait(lambda: self.backend.pipeline.get_state(0)[1] == expected) self.assertEqual(self.commits, []) if paused: time.sleep(0.1) self.assertEqual(self.backend.pipeline.get_state(0)[1], Gst.State.PAUSED) self.assertFalse(self.eos)
def test_seek_after_assignment_keeps_playing(self): self.seek_after_assignment(False)
def test_seek_after_assignment_stays_paused(self): self.seek_after_assignment(True)
def test_seek_after_commit_uses_new_current_uri(self): first = self.wav("before-commit.wav", seconds=0.6) second = self.wav("after-commit.wav", seconds=3) token = object() self.backend.queue_uri(second, token) self.backend.load_uri(first) self.wait(lambda: self.commits == [token]) self.backend.pause() completed = [] self.backend.seek(1200, completed.append) self.wait(lambda: bool(completed)) self.assertEqual(completed, [True]) self.assertEqual(self.backend.pipeline.get_property("current-uri"), second) self.assertEqual(self.backend.pipeline.get_state(0)[1], Gst.State.PAUSED)
def test_stop_cancels_pending_seek(self): uri = self.wav("cancel-seek.wav", seconds=3) self.backend.load_uri(uri, False) self.wait(lambda: self.backend.pipeline.get_state(0)[1] == Gst.State.PAUSED) completed = [] self.backend.seek(1200, completed.append) self.backend.stop() context = GLib.MainContext.default() while context.pending(): context.iteration(False) self.assertEqual(completed, []) self.assertIsNone(self.backend._pending_seek) self.assertEqual(self.backend.pipeline.get_state(0)[1], Gst.State.NULL)
def test_failed_seek_does_not_report_success(self): completed = [] self.backend.seek(1200, completed.append) self.assertEqual(completed, [False])
def test_latest_seek_wins(self): uri = self.wav("latest-seek.wav", seconds=3) self.backend.load_uri(uri, False) self.wait(lambda: self.backend.pipeline.get_state(0)[1] == Gst.State.PAUSED) old, latest = [], [] self.backend.seek(500, old.append) self.backend.seek(1700, latest.append) self.wait(lambda: bool(latest)) self.assertEqual(old, []) self.assertEqual(latest, [True]) ok, position = self.backend.pipeline.query_position(Gst.Format.TIME) self.assertTrue(ok) self.assertAlmostEqual(position / Gst.MSECOND, 1700, delta=50)
def mp4(self, seconds=2): ffmpeg = shutil.which("ffmpeg") self.assertIsNotNone(ffmpeg, "ffmpeg is required for HTTP fragmented MP4 coverage") path = Path(self.tmp.name) / "fragmented.mp4" subprocess.run([ffmpeg, "-v", "error", "-y", "-f", "lavfi", "-i", "sine=frequency=440:sample_rate=48000", "-t", str(seconds), "-c:a", "aac", "-movflags", "frag_keyframe+empty_moov+default_base_moof", "-frag_duration", "250000", str(path)], check=True) return path.read_bytes()
def test_http_fragmented_delayed_headers_body_and_seek(self): fixture = HTTPFixture(self.mp4(3), header_delay=0.25, body_delay=0.08) self.addCleanup(fixture.close) self.sink.set_property("qos", True) self.sink.set_property("max-lateness", 40 * Gst.MSECOND) self.backend.load_uri(fixture.uri) self.wait(lambda: len(self.buffers) >= 5) self.backend.pause() completed = [] self.backend.seek(1200, completed.append) self.wait(lambda: bool(completed)) self.assertEqual(completed, [True]) ok, pos = self.backend.pipeline.query_position(Gst.Format.TIME) self.assertTrue(ok) self.assertAlmostEqual(pos / Gst.MSECOND, 1200, delta=100) self.assertEqual(self.backend.pipeline.get_state(0)[1], Gst.State.PAUSED) self.backend.play() self.wait(lambda: bool(self.eos)) self.assertGreater(sum(size for _, size in self.buffers), 48000)
def test_http_fragmented_successor_finite_lateness(self): fixture = HTTPFixture(self.mp4(), header_delay=0.2, body_delay=0.04) self.addCleanup(fixture.close) self.sink.set_property("qos", True) self.sink.set_property("max-lateness", 40 * Gst.MSECOND) token = object() self.backend.queue_uri(fixture.uri, token) self.backend.load_uri(self.wav("current.wav", seconds=2)) self.wait(lambda: bool(self.eos), timeout=10) self.assertEqual(self.commits, [token]) self.assertGreater(sum(size for _, size in self.buffers), 3.5 * 48000 * 2)
def test_buffering_arbiter_intent_live_and_successor(self): self.backend.load_uri(self.wav("arbiter.wav", seconds=3)) self.wait(lambda: bool(self.buffers)) self.backend._current_uri = "http://127.0.0.1/current" states = [] self.backend._state_callback = states.append source = Gst.ElementFactory.make("queue2") self.backend._buffer_uri = lambda src: self.backend._current_uri send = lambda percent: self.backend._on_bus_message(None, Gst.Message.new_buffering(source, percent)) send(10) self.wait(lambda: self.backend.pipeline.get_state(0)[1] == Gst.State.PAUSED) self.assertNotIn(Gst.State.PAUSED, states) self.assertTrue(self.backend._want_playing) self.backend.pause() send(100) self.assertEqual(self.backend.pipeline.get_state(0)[1], Gst.State.PAUSED) send(10) self.backend.play() self.assertEqual(self.backend.pipeline.get_state(0)[1], Gst.State.PAUSED) send(100) self.wait(lambda: self.backend.pipeline.get_state(0)[1] == Gst.State.PLAYING) self.backend._buffer_uri = lambda src: "file:///inactive-successor" send(5) self.assertEqual(list(self.backend._buffering.values()), ["file:///inactive-successor"]) self.assertEqual(self.backend.pipeline.get_state(0)[1], Gst.State.PLAYING) self.backend.set_live(True) self.backend._buffer_uri = lambda src: None send(5) self.assertEqual(self.backend._buffering, {})
def test_local_timeout(self): source = Gst.ElementFactory.make("souphttpsrc") source.set_property("location", "http://127.0.0.1:8129/decrypt/test") self.backend._on_source_setup(self.backend.pipeline, source) self.assertEqual(source.get_property("timeout"), 180)
class PreloaderTests(unittest.TestCase): def setUp(self): package = types.ModuleType("gapless_fixture") package.__path__ = [] sys.modules[package.__name__] = package preparer_module = types.ModuleType("gapless_fixture.playback_preparer") preparer_module.DECRYPT_URL = "http://127.0.0.1:8129/decrypt/"
class Preparer: def prepare(self, song, api, quality, callback): self.callback = callback
def cancel(self): pass
preparer_module.PlaybackPreparer = Preparer song_module = types.ModuleType("gapless_fixture.song") song_module.Song = types.SimpleNamespace(new=lambda entry: entry) sys.modules[preparer_module.__name__] = preparer_module sys.modules[song_module.__name__] = song_module spec = importlib.util.spec_from_file_location("gapless_fixture.preloader", ROOT / "src/preloader.py") self.module = importlib.util.module_from_spec(spec) spec.loader.exec_module(self.module) self.armed = [] self.assigned = False backend = types.SimpleNamespace(queue_uri=lambda *args: self.armed.append(args), clear_next_uri=lambda: self.assigned) self.preloader = self.module.NextSongPreloader(backend)
def test_immediate_prepare_and_unique_token(self): entry = {} self.preloader.consider(999999, entry, None, None) self.preloader._preparer.callback("file:///same", None) first = self.armed[-1][1] self.assertTrue(self.preloader.take(first)) self.preloader.consider(999999, entry, None, None) self.preloader._preparer.callback("file:///same", None) self.assertIsNot(first, self.armed[-1][1]) self.assertFalse(self.preloader.take(first))
def test_invalidation_after_assignment_requests_reset(self): entry = {} self.preloader.consider(999999, entry, None, None) self.preloader._preparer.callback("file:///same", None) self.assigned = True self.assertTrue(self.preloader.queue_changed({})) self.assertIsNone(self.preloader._token)
def test_late_preparation_callback_is_discarded(self): self.preloader.consider(999999, {}, None, None) callback = self.preloader._preparer.callback self.preloader.drop() callback("file:///stale", None) self.assertEqual(self.armed, [])
def test_failed_preparation_retries_after_cooldown(self): from unittest.mock import patch entry = {} with patch.object(self.module.time, "monotonic", return_value=100) as clock: self.preloader.consider(0, entry, None, None) stale = self.preloader._preparer.callback failed_token = self.preloader._token stale(None, RuntimeError("temporary failure")) self.assertIsNone(self.preloader._entry) self.assertFalse(self.preloader.take(failed_token)) self.assertEqual(self.preloader._uri, "") for now in (100.5, 101, 109.9): clock.return_value = now self.preloader.consider(0, entry, None, None) self.assertIsNone(self.preloader._token) clock.return_value = 110 self.preloader.consider(0, entry, None, None) retry_token = self.preloader._token self.assertIsNotNone(retry_token) stale("file:///stale", None) self.assertEqual(self.armed, []) self.preloader._preparer.callback("file:///retry", None) self.assertTrue(self.preloader.take(retry_token))
def test_missing_uri_does_not_pin_entry_or_block_another(self): self.preloader.consider(0, {}, None, None) self.preloader._preparer.callback(None, None) self.assertIsNone(self.preloader._entry) self.assertEqual(self.armed, []) self.preloader.consider(0, {}, None, None) self.assertIsNotNone(self.preloader._token)
def warm_fixture(self, fixture): self.module.DECRYPT_URL = fixture.uri.rsplit("/", 1)[0] + "/" self.module.WARM_BACKOFF = 0.02 self.preloader.consider(999999, {}, None, None) self.preloader._preparer.callback(fixture.uri, None)
def wait_warm(self, predicate, timeout=5): deadline = time.monotonic() + timeout context = GLib.MainContext.default() while not predicate() and time.monotonic() < deadline: while context.pending(): context.iteration(False) time.sleep(0.005) self.assertTrue(predicate())
def test_warm_missing_early_signal_uses_real_eos_fallback(self): playback = GaplessTests() playback.setUp() self.addCleanup(playback.tearDown) fixture = HTTPFixture(playback.mp4(1), header_delay=0.4, body_delay=0.01) self.addCleanup(fixture.close) self.preloader._backend = playback.backend early = threading.Event() playback.backend.pipeline.connect("about-to-finish", lambda *args: early.set()) self.warm_fixture(fixture) playback.backend.load_uri(playback.wav("early.wav", seconds=1)) playback.wait(early.is_set) self.assertIsNone(playback.backend._assigned) self.wait_warm(lambda: bool(self.preloader._uri)) playback.wait(lambda: bool(playback.eos)) self.assertEqual(playback.commits, []) playback.backend.stop() playback.eos.clear() playback.buffers.clear() self.preloader.drop() playback.backend.load_uri(fixture.uri) playback.wait(lambda: bool(playback.eos)) self.assertGreater(sum(size for _, size in playback.buffers), 0.9 * 48000 * 2)
def test_http_500_short_body_retries_before_arming(self): fixture = HTTPFixture(b"asset" * 10000, header_delay=0.05, body_delay=0.01, failures=(500, "short")) self.addCleanup(fixture.close) with self.assertLogs(level="WARNING") as logs: self.warm_fixture(fixture) self.assertEqual(self.armed, []) self.wait_warm(lambda: bool(self.armed)) self.assertEqual(len(fixture.requests), 3) self.assertEqual(len(logs.output), 2) self.assertEqual(len(self.armed), 1)
def test_http_exhaustion_never_arms(self): fixture = HTTPFixture(b"asset", failures=(500,) * 4) self.addCleanup(fixture.close) with self.assertLogs(level="WARNING") as logs: self.warm_fixture(fixture) self.wait_warm(lambda: self.preloader._token is None) self.assertEqual(len(fixture.requests), 4) self.assertEqual(len(logs.output), 4) self.assertEqual(self.armed, []) self.assertIsNotNone(self.preloader._failed)
def test_http_cancel_during_headers_discards_ready(self): fixture = HTTPFixture(b"asset" * 100, header_delay=0.2) self.addCleanup(fixture.close) self.warm_fixture(fixture) self.wait_warm(lambda: bool(fixture.requests)) self.preloader.drop() end = time.monotonic() + 0.4 self.wait_warm(lambda: time.monotonic() >= end) self.assertEqual(self.armed, []) self.assertEqual(len(fixture.requests), 1)
def test_decrypt_prefetch_drains_body_off_main_loop(self): from unittest.mock import patch entered = threading.Event() release = threading.Event() reads = [] main = threading.get_ident()
class Response: headers = {"Content-Length": "10"} status = 200 def __enter__(self): return self
def __exit__(self, *args): pass
def read1(self, size): reads.append(threading.get_ident()) entered.set() release.wait(2) return b"asset" if len(reads) < 3 else b""
self.preloader.consider(999999, {}, None, None) with patch.object(self.module.urllib.request, "urlopen", return_value=Response()) as request: self.preloader._preparer.callback("http://127.0.0.1:8129/decrypt/test", None) self.assertTrue(entered.wait(2)) self.assertEqual(len(self.armed), 0) release.set() deadline = time.monotonic() + 2 while not self.armed and time.monotonic() < deadline: GLib.MainContext.default().iteration(False) time.sleep(0.002) self.assertEqual(len(self.armed), 1) self.assertEqual(len(reads), 3) self.assertTrue(all(ident != main for ident in reads)) request.assert_called_once_with("http://127.0.0.1:8129/decrypt/test", timeout=30)
class PlayerPolicyTests(unittest.TestCase): def method(self, name): import ast tree = ast.parse((ROOT / "src/player.py").read_text()) player = next(node for node in tree.body if isinstance(node, ast.ClassDef) and node.name == "Player") method = next(node for node in player.body if isinstance(node, ast.FunctionDef) and node.name == name) module = ast.Module(body=[method], type_ignores=[]) namespace = {"Any": object, "END_NATURAL_END_OF_TRACK": 1} exec(compile(module, "player.py", "exec"), namespace) return namespace[name]
def test_pause_uses_intent_during_automatic_paused(self): calls = [] player = types.SimpleNamespace(radio_active=False, backend=types.SimpleNamespace(_want_playing=True, pause=lambda: calls.append("pause"))) self.method("pause")(player) self.assertEqual(calls, ["pause"])
def test_seek_only_publishes_applied_target(self): calls, callbacks = [], [] player = types.SimpleNamespace(can_seek=True, timestamp_milli=100, preloader=types.SimpleNamespace(drop=lambda: calls.append("drop")), backend=types.SimpleNamespace(seek=lambda target, cb: callbacks.append((target, cb))), _prepare_successor=lambda: calls.append("prepare"), emit=lambda signal: calls.append(signal)) self.method("seek")(player, 2345) self.assertEqual(player.timestamp_milli, 100) self.assertEqual(calls, ["drop"]) self.assertEqual(callbacks[0][0], 2345) callbacks[0][1](False) self.assertEqual(calls, ["drop"]) self.assertEqual(player.timestamp_milli, 100) callbacks[0][1](True) self.assertEqual(player.timestamp_milli, 2345) self.assertEqual(calls, ["drop", "prepare", "seeked"])
def test_queue_invalidation_resets_pipeline(self): calls = [] player = types.SimpleNamespace(preloader=types.SimpleNamespace(queue_changed=lambda entry: True), _queue_entry=lambda position: {}, track_pos=0, _on_pos_changed=lambda: calls.append("reset")) self.method("_on_queue_changed")(player) self.assertEqual(calls, ["reset"])
def test_eos_never_uses_preloaded_shortcut(self): calls = [] player = types.SimpleNamespace(radio_active=False, episode_active=False, track_pos=0, preloader=types.SimpleNamespace(drop=lambda: calls.append("drop")), activity=types.SimpleNamespace(ended=lambda *args: calls.append("ended")), song=None) self.method("_on_eos")(player) self.assertEqual(player.track_pos, 1) self.assertEqual(calls, ["drop", "ended"])
if __name__ == "__main__": unittest.main()