v0.48.0: web mixes upload to a web server (rsync over ssh), and the player shuffles
Export Web Mix gets an 'Upload to web server' box: the mix is built in a cache staging folder and mirrored with rsync --delete into <path>/<slug>/ on the host set in Preferences → Integrations → Web Server (config.json, this machine only). An ssh check asks before replacing an existing mix; a status window follows check, export and upload. The web player gains a shuffle button (and the S key).
This commit is contained in:
@@ -0,0 +1,467 @@
|
||||
"""Round 81: Export Web Mix can upload to a web server, and the player shuffles.
|
||||
|
||||
- The server (host + folder) lives in config.json, per machine.
|
||||
- Upload = rsync mirror over ssh into <folder>/<slug>/, never the folder
|
||||
itself, from a staging tree that is rebuilt every time.
|
||||
- No test ever runs ssh or rsync: Popen and which are stubbed.
|
||||
"""
|
||||
import io
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from PyQt6.QtCore import QObject, pyqtSignal
|
||||
|
||||
from lintunes import config
|
||||
from lintunes.export import exporter, upload
|
||||
from lintunes.models import Track
|
||||
|
||||
HOST = "example-host"
|
||||
|
||||
|
||||
def _track(tid, path):
|
||||
return Track(track_id=tid, name=f"Song {tid}", artist="Artist",
|
||||
location=str(path), kind="MPEG audio file",
|
||||
total_time=180_000)
|
||||
|
||||
|
||||
def _audio(tmp_path, name):
|
||||
path = tmp_path / "local" / name
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_bytes(b"x" * 100)
|
||||
return path
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _xdg(tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "cfg"))
|
||||
monkeypatch.setenv("XDG_CACHE_HOME", str(tmp_path / "cache"))
|
||||
|
||||
|
||||
# ---- pure helpers ----
|
||||
|
||||
@pytest.mark.parametrize("title, slug", [
|
||||
("Summer Mix 2026!", "summer-mix-2026"),
|
||||
("Étoile Filante", "etoile-filante"),
|
||||
(" rock / roll ", "rock-roll"),
|
||||
("🎵 Night 🎵", "night"),
|
||||
("!!!", ""),
|
||||
])
|
||||
def test_slugify(title, slug):
|
||||
assert upload.slugify(title) == slug
|
||||
|
||||
|
||||
@pytest.mark.parametrize("slug, ok", [
|
||||
("mix", True), ("", False), (".", False), ("..", False), ("a/b", False)])
|
||||
def test_valid_slug(slug, ok):
|
||||
assert upload.valid_slug(slug) is ok
|
||||
|
||||
|
||||
def test_server_settings_needs_both():
|
||||
assert upload.server_settings({}) is None
|
||||
assert upload.server_settings({upload.HOST_KEY: HOST}) is None
|
||||
assert upload.server_settings(
|
||||
{upload.HOST_KEY: f" {HOST} ", upload.PATH_KEY: "/srv/mixes "}) \
|
||||
== (HOST, "/srv/mixes")
|
||||
|
||||
|
||||
def test_remote_dir_is_always_inside_the_folder():
|
||||
assert upload.remote_dir("/srv/mixes/", "mix") == "/srv/mixes/mix"
|
||||
assert upload.remote_dir("~/www/mixes", "mix") == "www/mixes/mix"
|
||||
assert upload.remote_dir("~", "mix") == "mix"
|
||||
for bad in ("", "..", "a/b"):
|
||||
with pytest.raises(upload.UploadError):
|
||||
upload.remote_dir("/srv/mixes", bad)
|
||||
|
||||
|
||||
def test_exists_command_quotes_the_path():
|
||||
argv = upload.exists_command(HOST, "/srv/my mixes/a")
|
||||
assert argv[0] == "ssh" and "BatchMode=yes" in argv
|
||||
assert argv[-3:] == ["test", "-d", "'/srv/my mixes/a'"]
|
||||
assert HOST in argv
|
||||
|
||||
|
||||
def test_rsync_command_mirrors_into_the_slug_folder(tmp_path):
|
||||
argv = upload.rsync_command(HOST, tmp_path / "mix", "/srv/mixes/mix")
|
||||
assert argv[0] == "rsync"
|
||||
assert "--delete" in argv and "--checksum" in argv
|
||||
assert "BatchMode=yes" in argv[argv.index("-e") + 1]
|
||||
assert argv[-2] == str(tmp_path / "mix") + "/" # contents, not folder
|
||||
assert argv[-1] == f"{HOST}:/srv/mixes/mix/"
|
||||
|
||||
|
||||
def test_reset_staging_stays_inside_the_staging_root(tmp_path):
|
||||
root = upload.staging_root()
|
||||
built = root / "mix" / "audios"
|
||||
built.mkdir(parents=True)
|
||||
(built / "old.mp3").write_bytes(b"x")
|
||||
target = upload.reset_staging("mix")
|
||||
assert target == root / "mix" and not target.exists()
|
||||
assert root.exists()
|
||||
for bad in ("", "..", "../x"):
|
||||
with pytest.raises(upload.UploadError):
|
||||
upload.reset_staging(bad)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("line, percent", [
|
||||
(" 32,768 0% 0.00kB/s 0:00:00 (xfr#0, to-chk=10/12)", 0),
|
||||
(" 1,234,567 42% 1.23MB/s 0:00:03 (xfr#2, to-chk=5/9)", 42),
|
||||
(" 12.34M 100% 10.00MB/s 0:00:01", 100),
|
||||
("rsync: [Receiver] mkdir \"/srv/x\" failed: No such file (2)", None),
|
||||
("sending incremental file list", None),
|
||||
])
|
||||
def test_parse_progress2(line, percent):
|
||||
assert upload.parse_progress2(line) == percent
|
||||
|
||||
|
||||
def test_plan_export_folder_override(tmp_path):
|
||||
tracks = [_track(1, _audio(tmp_path, "a.mp3"))]
|
||||
plan = exporter.plan_export("My Mix", tracks, tmp_path / "out",
|
||||
exporter.WEB, folder="my-mix")
|
||||
assert plan.dest_dir == tmp_path / "out" / "my-mix"
|
||||
assert plan.audio_dir == tmp_path / "out" / "my-mix" / "audios"
|
||||
assert exporter.plan_export("My Mix", tracks, tmp_path / "out",
|
||||
exporter.WEB).dest_dir.name == "My Mix"
|
||||
|
||||
|
||||
# ---- workers, with a fake subprocess ----
|
||||
|
||||
class _FakeProc:
|
||||
def __init__(self, output: bytes, returncode: int):
|
||||
self.stdout = io.BufferedReader(io.BytesIO(output))
|
||||
self.returncode = None
|
||||
self._code = returncode
|
||||
self.killed = False
|
||||
|
||||
def poll(self):
|
||||
return self.returncode
|
||||
|
||||
def wait(self):
|
||||
self.returncode = self._code
|
||||
return self._code
|
||||
|
||||
def communicate(self):
|
||||
self.returncode = self._code
|
||||
return self.stdout.read(), None
|
||||
|
||||
def kill(self):
|
||||
self.killed = True
|
||||
|
||||
|
||||
def _fake_popen(monkeypatch, output=b"", code=0):
|
||||
calls = []
|
||||
|
||||
def popen(argv, **kwargs):
|
||||
proc = _FakeProc(output, code)
|
||||
calls.append((argv, proc))
|
||||
return proc
|
||||
|
||||
monkeypatch.setattr(upload.subprocess, "Popen", popen)
|
||||
monkeypatch.setattr(upload.shutil, "which", lambda name: f"/usr/bin/{name}")
|
||||
return calls
|
||||
|
||||
|
||||
def _collect(worker, *names):
|
||||
seen = {n: [] for n in names}
|
||||
for n in names:
|
||||
getattr(worker, n).connect(lambda *a, n=n: seen[n].append(a))
|
||||
return seen
|
||||
|
||||
|
||||
@pytest.mark.parametrize("code, outcome", [(0, True), (1, False)])
|
||||
def test_server_check_reports_existence(qapp, monkeypatch, code, outcome):
|
||||
calls = _fake_popen(monkeypatch, code=code)
|
||||
worker = upload.ServerCheckWorker(HOST, "/srv/mixes/mix")
|
||||
seen = _collect(worker, "result", "failed")
|
||||
worker._run()
|
||||
assert seen["result"] == [(outcome,)] and not seen["failed"]
|
||||
assert calls[0][0][0] == "ssh"
|
||||
|
||||
|
||||
def test_server_check_connection_error(qapp, monkeypatch):
|
||||
_fake_popen(monkeypatch, b"ssh: Could not resolve hostname\n", 255)
|
||||
worker = upload.ServerCheckWorker(HOST, "/srv/mixes/mix")
|
||||
seen = _collect(worker, "result", "failed")
|
||||
worker._run()
|
||||
assert not seen["result"]
|
||||
assert "Could not resolve hostname" in seen["failed"][0][0]
|
||||
|
||||
|
||||
def test_missing_rsync_is_an_error_not_a_crash(qapp, monkeypatch, tmp_path):
|
||||
monkeypatch.setattr(upload.shutil, "which", lambda name: None)
|
||||
worker = upload.UploadWorker(HOST, tmp_path, "/srv/mixes/mix")
|
||||
seen = _collect(worker, "failed", "finished")
|
||||
worker._run()
|
||||
assert "rsync" in seen["failed"][0][0] and not seen["finished"]
|
||||
|
||||
|
||||
def test_upload_reports_progress_and_files(qapp, monkeypatch, tmp_path):
|
||||
out = (b"sending incremental file list\n"
|
||||
b" 32,768 10% 1.00MB/s 0:00:01 (xfr#1, to-chk=3/5)\r"
|
||||
b" 327,680 100% 1.00MB/s 0:00:01 (xfr#3, to-chk=0/5)\n")
|
||||
_fake_popen(monkeypatch, out, 0)
|
||||
worker = upload.UploadWorker(HOST, tmp_path, "/srv/mixes/mix")
|
||||
seen = _collect(worker, "progress", "finished", "failed")
|
||||
worker._run()
|
||||
assert [p[0] for p in seen["progress"]] == [10, 100]
|
||||
assert seen["finished"][0][0]["files_sent"] == 3
|
||||
assert not seen["failed"]
|
||||
|
||||
|
||||
def test_upload_failure_carries_rsyncs_message(qapp, monkeypatch, tmp_path):
|
||||
_fake_popen(monkeypatch,
|
||||
b'rsync: [Receiver] mkdir "/srv/mixes/mix" failed: '
|
||||
b"No such file or directory (2)\n", 11)
|
||||
worker = upload.UploadWorker(HOST, tmp_path, "/srv/mixes/mix")
|
||||
seen = _collect(worker, "failed", "finished")
|
||||
worker._run()
|
||||
assert "No such file or directory" in seen["failed"][0][0]
|
||||
assert "exit 11" in seen["failed"][0][0]
|
||||
|
||||
|
||||
def test_cancel_kills_the_process(qapp, monkeypatch, tmp_path):
|
||||
calls = _fake_popen(monkeypatch, b"", 0)
|
||||
worker = upload.UploadWorker(HOST, tmp_path, "/srv/mixes/mix")
|
||||
worker._spawn(["rsync"])
|
||||
worker.cancel()
|
||||
assert calls[0][1].killed
|
||||
|
||||
|
||||
# ---- the dialogs ----
|
||||
|
||||
class _FakeLastFm(QObject):
|
||||
login_finished = pyqtSignal(bool, str)
|
||||
status_message = pyqtSignal(str)
|
||||
|
||||
def is_logged_in(self):
|
||||
return False
|
||||
|
||||
|
||||
def test_upload_box_is_off_by_default_and_slug_follows_title(qapp):
|
||||
from lintunes.gui.export_dialog import WebMixDialog
|
||||
dialog = WebMixDialog("Summer Mix", None, {})
|
||||
assert not dialog._upload.isChecked()
|
||||
assert dialog.values()["slug"] == "summer-mix"
|
||||
dialog._title.textEdited.emit("Fall Mix")
|
||||
assert dialog.values()["slug"] == "fall-mix"
|
||||
dialog._slug.textEdited.emit("custom")
|
||||
dialog._slug.setText("custom")
|
||||
dialog._title.textEdited.emit("Winter Mix")
|
||||
assert dialog.values()["slug"] == "custom"
|
||||
|
||||
|
||||
def test_ticked_without_a_server_stays_open(qapp, monkeypatch):
|
||||
from lintunes.gui import export_dialog
|
||||
told = []
|
||||
monkeypatch.setattr(export_dialog.QMessageBox, "warning",
|
||||
lambda *a, **k: told.append(a[2]))
|
||||
dialog = export_dialog.WebMixDialog("Mix", None, {})
|
||||
dialog._upload.setChecked(True)
|
||||
dialog.accept()
|
||||
assert told and "Preferences" in told[0] and "uncheck" in told[0]
|
||||
assert dialog.result() != dialog.DialogCode.Accepted
|
||||
|
||||
|
||||
def test_ticked_with_a_server_accepts(qapp, monkeypatch):
|
||||
from lintunes.gui import export_dialog
|
||||
monkeypatch.setattr(export_dialog.QMessageBox, "warning",
|
||||
lambda *a, **k: pytest.fail("should not warn"))
|
||||
dialog = export_dialog.WebMixDialog(
|
||||
"Mix", None, {upload.HOST_KEY: HOST, upload.PATH_KEY: "/srv/mixes"})
|
||||
dialog._upload.setChecked(True)
|
||||
assert f"{HOST}:/srv/mixes/mix" in dialog._upload_note.text()
|
||||
dialog.accept()
|
||||
assert dialog.result() == dialog.DialogCode.Accepted
|
||||
assert dialog.values()["upload"] is True
|
||||
|
||||
|
||||
def test_preferences_save_the_server_to_config_json(qapp):
|
||||
from lintunes.gui.preferences_dialog import PreferencesDialog
|
||||
|
||||
from lintunes.preferences import Preferences
|
||||
prefs = Preferences(Path(config.config_path()).parent / "data")
|
||||
dialog = PreferencesDialog(prefs, _FakeLastFm())
|
||||
dialog._web_host.setText(HOST)
|
||||
dialog._web_path.setText("/srv/mixes")
|
||||
dialog._save_web_server()
|
||||
saved = config.load_config()
|
||||
assert saved[upload.HOST_KEY] == HOST
|
||||
assert saved[upload.PATH_KEY] == "/srv/mixes"
|
||||
# Nothing about the server reaches the synced preferences.
|
||||
pref_file = Path(config.config_path()).parent / "data" / "preferences.json"
|
||||
assert not pref_file.exists() or HOST not in pref_file.read_text()
|
||||
|
||||
|
||||
# ---- the whole flow through MainWindow, with fake workers ----
|
||||
|
||||
class _FakeCheck:
|
||||
exists = False
|
||||
|
||||
def __init__(self, host, rdir, parent=None):
|
||||
from PyQt6.QtCore import QObject, pyqtSignal
|
||||
|
||||
class S(QObject):
|
||||
result = pyqtSignal(bool)
|
||||
failed = pyqtSignal(str)
|
||||
cancelled = pyqtSignal()
|
||||
self._s = S()
|
||||
self.result, self.failed = self._s.result, self._s.failed
|
||||
self.cancelled = self._s.cancelled
|
||||
|
||||
def busy(self):
|
||||
return False
|
||||
|
||||
def cancel(self):
|
||||
pass
|
||||
|
||||
def start(self):
|
||||
self.result.emit(_FakeCheck.exists)
|
||||
|
||||
|
||||
class _FakeUpload:
|
||||
runs = []
|
||||
|
||||
def __init__(self, host, local, rdir, parent=None):
|
||||
from PyQt6.QtCore import QObject, pyqtSignal
|
||||
|
||||
class S(QObject):
|
||||
progress = pyqtSignal(int, str)
|
||||
finished = pyqtSignal(dict)
|
||||
failed = pyqtSignal(str)
|
||||
cancelled = pyqtSignal()
|
||||
self._s = S()
|
||||
for n in ("progress", "finished", "failed", "cancelled"):
|
||||
setattr(self, n, getattr(self._s, n))
|
||||
self.args = (host, Path(local), rdir)
|
||||
|
||||
def busy(self):
|
||||
return False
|
||||
|
||||
def cancel(self):
|
||||
pass
|
||||
|
||||
def start(self):
|
||||
host, local, rdir = self.args
|
||||
_FakeUpload.runs.append(
|
||||
(host, rdir, sorted(p.relative_to(local).as_posix()
|
||||
for p in local.rglob("*") if p.is_file())))
|
||||
self.finished.emit({"host": host, "dest": rdir, "files_sent": 2})
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def window(qapp, tmp_path, monkeypatch):
|
||||
from lintunes.gui import main_window as mw
|
||||
from lintunes.library_manager import LibraryManager
|
||||
from lintunes.models import Library, Playlist
|
||||
from lintunes.models.playlist import PlaylistType
|
||||
from lintunes.preferences import Preferences
|
||||
|
||||
library = Library(
|
||||
tracks={1: _track(1, _audio(tmp_path, "a.mp3")),
|
||||
2: _track(2, _audio(tmp_path, "b.mp3"))},
|
||||
playlists={"abcd1234": Playlist(
|
||||
name="My Mix", persistent_id="abcd1234",
|
||||
playlist_type=PlaylistType.REGULAR, track_ids=[1, 2])})
|
||||
manager = LibraryManager(library, tmp_path / "data")
|
||||
win = mw.MainWindow(manager, Preferences(tmp_path / "data"))
|
||||
|
||||
class Kind:
|
||||
DialogCode = mw.ExportKindDialog.DialogCode
|
||||
|
||||
def __init__(self, *a):
|
||||
pass
|
||||
|
||||
def exec(self):
|
||||
return self.DialogCode.Accepted
|
||||
|
||||
def kind(self):
|
||||
return exporter.WEB
|
||||
|
||||
class Web(Kind):
|
||||
def values(self):
|
||||
return {"title": "My Mix", "description": "", "image": None,
|
||||
"accent": exporter.DEFAULT_ACCENT, "upload": True,
|
||||
"slug": "my-mix"}
|
||||
|
||||
monkeypatch.setattr(mw, "ExportKindDialog", Kind)
|
||||
monkeypatch.setattr(mw, "WebMixDialog", Web)
|
||||
monkeypatch.setattr(mw.upload, "ServerCheckWorker", _FakeCheck)
|
||||
monkeypatch.setattr(mw.upload, "UploadWorker", _FakeUpload)
|
||||
monkeypatch.setattr(mw.QFileDialog, "getExistingDirectory",
|
||||
lambda *a, **k: pytest.fail("no folder picker"))
|
||||
monkeypatch.setattr(mw.sounds, "play_done", lambda *a: None)
|
||||
_FakeUpload.runs = []
|
||||
_FakeCheck.exists = False
|
||||
yield win
|
||||
if win._web_progress is not None:
|
||||
win._web_progress.close()
|
||||
win.close()
|
||||
|
||||
|
||||
def _wait_for(qapp, cond, timeout=5):
|
||||
end = time.monotonic() + timeout
|
||||
while not cond() and time.monotonic() < end:
|
||||
qapp.processEvents()
|
||||
time.sleep(0.01)
|
||||
assert cond()
|
||||
|
||||
|
||||
def test_upload_flow_exports_to_staging_then_uploads(qapp, window):
|
||||
config.save_config({upload.HOST_KEY: HOST, upload.PATH_KEY: "/srv/mixes"})
|
||||
window._export_playlist("abcd1234")
|
||||
_wait_for(qapp, lambda: _FakeUpload.runs)
|
||||
host, rdir, files = _FakeUpload.runs[0]
|
||||
assert (host, rdir) == (HOST, "/srv/mixes/my-mix")
|
||||
assert "index.html" in files
|
||||
assert "audios/Artist - Song 1.mp3" in files
|
||||
assert window._web_upload is None
|
||||
assert window._web_progress.ended
|
||||
|
||||
|
||||
def test_a_removed_song_doesnt_survive_in_staging(qapp, window):
|
||||
config.save_config({upload.HOST_KEY: HOST, upload.PATH_KEY: "/srv/mixes"})
|
||||
window._export_playlist("abcd1234")
|
||||
_wait_for(qapp, lambda: _FakeUpload.runs)
|
||||
window._manager.library.playlists["abcd1234"].track_ids = [1]
|
||||
window._export_playlist("abcd1234")
|
||||
_wait_for(qapp, lambda: len(_FakeUpload.runs) == 2)
|
||||
files = _FakeUpload.runs[1][2]
|
||||
assert "audios/Artist - Song 2.mp3" not in files
|
||||
|
||||
|
||||
def test_existing_folder_asks_and_no_stops(qapp, window, monkeypatch):
|
||||
from lintunes.gui import main_window as mw
|
||||
config.save_config({upload.HOST_KEY: HOST, upload.PATH_KEY: "/srv/mixes"})
|
||||
_FakeCheck.exists = True
|
||||
asked = []
|
||||
|
||||
def question(*a, **k):
|
||||
asked.append(a[2])
|
||||
return mw.QMessageBox.StandardButton.No
|
||||
monkeypatch.setattr(mw.QMessageBox, "question", question)
|
||||
window._export_playlist("abcd1234")
|
||||
assert asked and "already exists" in asked[0]
|
||||
assert window._web_upload is None and not _FakeUpload.runs
|
||||
assert window._web_progress.status(2) == "Waiting"
|
||||
|
||||
|
||||
def test_unticked_export_is_the_old_flow(qapp, window, monkeypatch, tmp_path):
|
||||
from lintunes.gui import main_window as mw
|
||||
picked = []
|
||||
monkeypatch.setattr(mw.QFileDialog, "getExistingDirectory",
|
||||
lambda *a, **k: picked.append(1) or "")
|
||||
original = mw.WebMixDialog.values
|
||||
monkeypatch.setattr(mw.WebMixDialog, "values",
|
||||
lambda self: {**original(self), "upload": False})
|
||||
window._export_playlist("abcd1234")
|
||||
assert picked and window._web_upload is None
|
||||
|
||||
|
||||
# ---- the web player's shuffle ----
|
||||
|
||||
def test_player_ships_a_shuffle_button():
|
||||
js = (exporter.templates() / "player.js").read_text()
|
||||
css = (exporter.templates() / "player.css").read_text()
|
||||
assert 'class="shuffle"' in js and "setShuffle" in js
|
||||
assert ".audiojs .shuffle.on" in css
|
||||
|
||||
Reference in New Issue
Block a user