486 lines
17 KiB
Python
486 lines
17 KiB
Python
"""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
|
|
|
|
|
|
|
|
def test_description_and_image_are_spaced_apart(tmp_path):
|
|
tracks = [_track(1, _audio(tmp_path, "a.mp3"))]
|
|
plan = exporter.plan_export("Mix", tracks, tmp_path / "out", exporter.WEB)
|
|
page = exporter.build_index_html(plan, "placeholder.png")
|
|
gap = page[page.index(plan.description) + len(plan.description):
|
|
page.index("placeholder.png")]
|
|
assert "<br><br>" in gap
|
|
|
|
|
|
def test_finished_window_hides_stop(qapp):
|
|
from lintunes.gui.web_upload_progress import WebUploadProgress
|
|
win = WebUploadProgress("Mix", f"{HOST}:/srv/mixes/mix")
|
|
win.show()
|
|
win.finish("done")
|
|
assert win._stop.isHidden() and win._hide.text() == "Close"
|
|
win.close()
|