travandClaude Opus 5.5
0dad4cc95e
First step of friend library sharing (Cassette): - cassette/syncthing_api.py: a typed wrapper over the *local* Syncthing's REST API. Key and address come from Syncthing's config.xml (config.json may override); every failure is a SyncthingError naming the step, what Syncthing said, and whether it's not running / a bad key / a refusal. - cassette/state.py: friends, invites and share selections, kept in the machine-local ~/.local/share/lintunes/cassette, never the synced data dir. - cassette/host.py: one machine runs Cassette. config.json flags it, preferences.json names it, and the other machine greys the menu group out and says where friend sharing lives. - Connections ▸ Sync Settings…: VLC-style list + page, editing a copy that only OK writes. General shows Syncthing's status with a next step for every failure, the display name and an avatar picker. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
366 lines
14 KiB
Python
366 lines
14 KiB
Python
"""Round 64: Cassette's footing — the Syncthing REST client, the host machine,
|
|
the Cassette state file and the Sync Settings shell.
|
|
|
|
* LinTunes only ever talks to *local* Syncthing. The key and address come from
|
|
Syncthing's config.xml, overridable in config.json, and every failure names
|
|
the step and what Syncthing said, sorted into "not running" / "bad key" /
|
|
"it said no" so the UI can offer the right next step.
|
|
* One machine is the Cassette host: config.json flags it, preferences.json
|
|
(synced) names it so the other machine can say where friend sharing lives.
|
|
* Sync Settings edits a copy; Cancel leaves everything as it was.
|
|
"""
|
|
import json
|
|
|
|
import pytest
|
|
import requests
|
|
from PyQt6.QtGui import QColor, QImage
|
|
|
|
from lintunes.cassette import host, probe as probe_mod, syncthing_api as st
|
|
from lintunes.cassette.state import (
|
|
CassetteState, Friend, Invite, Selection, load_state, save_state,
|
|
)
|
|
from lintunes.preferences import Preferences
|
|
|
|
|
|
# ---- a fake Syncthing ----
|
|
|
|
class _Response:
|
|
def __init__(self, status=200, body=None):
|
|
self.status_code = status
|
|
self._body = body
|
|
if body is None:
|
|
self.content = b""
|
|
self.text = ""
|
|
elif isinstance(body, str):
|
|
self.content = body.encode()
|
|
self.text = body
|
|
else:
|
|
self.text = json.dumps(body)
|
|
self.content = self.text.encode()
|
|
|
|
def json(self):
|
|
if isinstance(self._body, str):
|
|
raise ValueError("not json")
|
|
return self._body
|
|
|
|
|
|
class FakeSession:
|
|
"""Routes (method, path) to canned responses and records every call."""
|
|
|
|
def __init__(self, routes=None, error=None):
|
|
self.headers = {}
|
|
self.routes = routes or {}
|
|
self.error = error
|
|
self.calls = []
|
|
|
|
def request(self, method, url, params=None, json=None, timeout=None,
|
|
verify=None):
|
|
path = url.split("8384", 1)[-1]
|
|
self.calls.append((method, path, params, json))
|
|
if self.error is not None:
|
|
raise self.error
|
|
route = self.routes.get((method, path))
|
|
if callable(route):
|
|
return route(params, json)
|
|
return route if route is not None else _Response(404, "404 page not found")
|
|
|
|
|
|
def _client(routes=None, error=None):
|
|
session = FakeSession(routes, error)
|
|
return st.SyncthingClient(st.Endpoint("127.0.0.1:8384", "KEY"), session), session
|
|
|
|
|
|
# ---- endpoint discovery ----
|
|
|
|
CONFIG_XML = """<configuration version="37">
|
|
<gui enabled="true" tls="{tls}">
|
|
<address>{address}</address>
|
|
<apikey>{key}</apikey>
|
|
</gui>
|
|
</configuration>"""
|
|
|
|
|
|
def _write_config(tmp_path, address="127.0.0.1:8384", key="abc", tls="false"):
|
|
(tmp_path / "config.xml").write_text(
|
|
CONFIG_XML.format(address=address, key=key, tls=tls))
|
|
return tmp_path
|
|
|
|
|
|
class TestEndpoint:
|
|
def test_reads_config_xml(self, tmp_path):
|
|
endpoint = st.read_gui_config(_write_config(tmp_path))
|
|
assert endpoint == st.Endpoint("127.0.0.1:8384", "abc", False)
|
|
assert endpoint.base_url == "http://127.0.0.1:8384"
|
|
|
|
def test_any_address_means_loopback(self, tmp_path):
|
|
endpoint = st.read_gui_config(
|
|
_write_config(tmp_path, address="0.0.0.0:9999", tls="true"))
|
|
assert endpoint.address == "127.0.0.1:9999"
|
|
assert endpoint.base_url.startswith("https://")
|
|
|
|
def test_no_key_is_none(self, tmp_path):
|
|
assert st.read_gui_config(_write_config(tmp_path, key="")) is None
|
|
assert st.read_gui_config(tmp_path / "nowhere") is None
|
|
|
|
def test_config_json_overrides(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr(st, "config_dir", lambda: _write_config(tmp_path))
|
|
assert st.resolve_endpoint({}).api_key == "abc"
|
|
endpoint = st.resolve_endpoint({st.API_KEY_KEY: "mine",
|
|
st.ADDRESS_KEY: "127.0.0.1:1"})
|
|
assert (endpoint.api_key, endpoint.address) == ("mine", "127.0.0.1:1")
|
|
|
|
def test_no_key_anywhere(self, monkeypatch):
|
|
monkeypatch.setattr(st, "config_dir", lambda: None)
|
|
assert st.resolve_endpoint({}) is None
|
|
|
|
|
|
# ---- the client ----
|
|
|
|
class TestClient:
|
|
def test_sends_the_key(self):
|
|
client, session = _client({("GET", "/rest/system/version"):
|
|
_Response(body={"version": "v1.30.0"})})
|
|
assert client.version() == "v1.30.0"
|
|
assert session.headers["X-API-Key"] == "KEY"
|
|
|
|
def test_not_running(self):
|
|
client, _ = _client(error=requests.ConnectionError("refused"))
|
|
with pytest.raises(st.SyncthingError) as info:
|
|
client.version()
|
|
assert info.value.kind == st.NOT_RUNNING
|
|
assert info.value.step == "Checking Syncthing's version"
|
|
|
|
def test_bad_key(self):
|
|
client, _ = _client({("GET", "/rest/system/version"): _Response(403, "CSRF")})
|
|
with pytest.raises(st.SyncthingError) as info:
|
|
client.version()
|
|
assert info.value.kind == st.BAD_KEY
|
|
|
|
def test_syncthing_said_no_is_quoted(self):
|
|
client, _ = _client({("PUT", "/rest/config/devices/DEV"):
|
|
_Response(400, "invalid device ID")})
|
|
with pytest.raises(st.SyncthingError) as info:
|
|
client.add_device("DEV", "Sam")
|
|
assert info.value.kind == st.OTHER
|
|
assert "Sam" in info.value.step
|
|
assert info.value.detail == "invalid device ID"
|
|
|
|
def test_missing_folder_is_none(self):
|
|
client, _ = _client()
|
|
assert client.folder("nope") is None
|
|
|
|
def test_share_folder_is_idempotent(self):
|
|
folder = {"id": "f", "devices": [{"deviceID": "ME"}]}
|
|
patched = []
|
|
routes = {
|
|
("GET", "/rest/config/folders/f"): lambda p, j: _Response(body=folder),
|
|
("PATCH", "/rest/config/folders/f"):
|
|
lambda p, j: (patched.append(j), folder.update(j), _Response())[-1],
|
|
}
|
|
client, _ = _client(routes)
|
|
client.share_folder("f", "SAM")
|
|
client.share_folder("f", "SAM")
|
|
assert len(patched) == 1
|
|
assert [d["deviceID"] for d in folder["devices"]] == ["ME", "SAM"]
|
|
|
|
def test_add_folder_body(self):
|
|
client, session = _client({("PUT", "/rest/config/folders/x"): _Response()})
|
|
client.add_folder("x", "Cassette", "/tmp/x", "sendonly", ["SAM"])
|
|
_, _, _, body = session.calls[-1]
|
|
assert body["type"] == "sendonly"
|
|
assert body["devices"] == [{"deviceID": "SAM"}]
|
|
assert body["path"] == "/tmp/x"
|
|
|
|
def test_min_disk_free_falls_back_to_defaults(self):
|
|
client, _ = _client({("GET", "/rest/config/defaults/folder"):
|
|
_Response(body={"minDiskFree": {"value": 1, "unit": "%"}})})
|
|
assert client.min_disk_free("missing") == (1.0, "%")
|
|
|
|
def test_disk_floor_bytes(self):
|
|
assert st.disk_floor_bytes(1, "%", 1000) == 10
|
|
assert st.disk_floor_bytes(2, "GB", 0) == 2 * 10**9
|
|
assert st.disk_floor_bytes(1, "GiB", 0) == 2**30
|
|
|
|
|
|
# ---- probe ----
|
|
|
|
class TestProbe:
|
|
def test_ok(self, monkeypatch):
|
|
monkeypatch.setattr(st, "config_dir", lambda: None)
|
|
session = FakeSession({
|
|
("GET", "/rest/system/version"): _Response(body={"version": "v1.30.0"}),
|
|
("GET", "/rest/system/status"): _Response(body={"myID": "AAA"}),
|
|
})
|
|
result = probe_mod.probe({st.API_KEY_KEY: "k"}, session=session)
|
|
assert result.ok and result.version == "v1.30.0" and result.my_id == "AAA"
|
|
|
|
def test_no_key(self, monkeypatch):
|
|
monkeypatch.setattr(st, "config_dir", lambda: None)
|
|
assert probe_mod.probe({}).state == probe_mod.NO_API
|
|
|
|
def test_not_running(self, monkeypatch):
|
|
monkeypatch.setattr(st, "config_dir", lambda: None)
|
|
session = FakeSession(error=requests.ConnectionError("refused"))
|
|
result = probe_mod.probe({st.API_KEY_KEY: "k"}, session=session)
|
|
assert result.state == probe_mod.NOT_RUNNING
|
|
|
|
|
|
# ---- state ----
|
|
|
|
class TestState:
|
|
def test_round_trip(self, tmp_path):
|
|
state = CassetteState(display_name="trav", share_same_for_all=True)
|
|
state.general = Selection(all_library=True, playlist_ids={"P1"})
|
|
state.invites["t1"] = Invite(token="t1", created="2026-09-28")
|
|
state.friends["t2"] = Friend(token="t2", device_id="DEV", name="Sam",
|
|
selection=Selection(track_ids={3, 1}),
|
|
cassetted=[5, 6])
|
|
save_state(state, tmp_path)
|
|
loaded = load_state(tmp_path)
|
|
assert loaded.to_dict() == state.to_dict()
|
|
assert loaded.friends["t2"].selection.track_ids == {1, 3}
|
|
assert loaded.selection_for(loaded.friends["t2"]) is loaded.general
|
|
|
|
def test_missing_or_broken_is_fresh(self, tmp_path):
|
|
assert load_state(tmp_path).friends == {}
|
|
(tmp_path / "state.json").write_text("{nope")
|
|
assert load_state(tmp_path).invites == {}
|
|
|
|
def test_copy_is_deep(self):
|
|
state = CassetteState()
|
|
state.general.track_ids.add(1)
|
|
clone = state.copy()
|
|
clone.general.track_ids.add(2)
|
|
assert state.general.track_ids == {1}
|
|
|
|
|
|
# ---- host ----
|
|
|
|
@pytest.fixture
|
|
def isolated_config(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "cfg"))
|
|
return tmp_path
|
|
|
|
|
|
class TestHost:
|
|
def test_unclaimed_then_claimed(self, qapp, isolated_config):
|
|
prefs = Preferences(isolated_config / "data")
|
|
assert host.host_status(prefs)[0] == host.UNCLAIMED
|
|
host.claim(prefs)
|
|
assert host.host_status(prefs) == (host.HERE, host.hostname())
|
|
assert prefs.get("cassette_host")["hostname"] == host.hostname()
|
|
|
|
def test_elsewhere_names_the_host(self, qapp, isolated_config):
|
|
prefs = Preferences(isolated_config / "data")
|
|
prefs.set("cassette_host", {"machine_id": "other", "hostname": "tummult"})
|
|
assert host.host_status(prefs) == (host.ELSEWHERE, "tummult")
|
|
assert not host.is_host(prefs)
|
|
|
|
|
|
# ---- the dialog ----
|
|
|
|
class _NoProbe:
|
|
"""Stands in for SyncthingProbe: never touches a socket."""
|
|
|
|
def __init__(self):
|
|
from PyQt6.QtCore import QObject, pyqtSignal
|
|
|
|
class _Sig(QObject):
|
|
finished = pyqtSignal(object)
|
|
self._sig = _Sig()
|
|
self.finished = self._sig.finished
|
|
self.starts = []
|
|
|
|
def start(self, config=None):
|
|
self.starts.append(config)
|
|
|
|
|
|
class TestSyncSettings:
|
|
def _dialog(self, tmp_path, state=None, config=None):
|
|
from lintunes.gui.sync_settings_dialog import SyncSettingsDialog
|
|
return SyncSettingsDialog(state or CassetteState(display_name="trav"),
|
|
tmp_path, config or {}, probe=_NoProbe())
|
|
|
|
def test_general_first_then_friends(self, qapp, tmp_path):
|
|
state = CassetteState()
|
|
state.friends["b"] = Friend(token="b", device_id="B", name="zed")
|
|
state.friends["a"] = Friend(token="a", device_id="A", name="Amy")
|
|
dialog = self._dialog(tmp_path, state)
|
|
labels = [dialog.pages_list.item(i).text()
|
|
for i in range(dialog.pages_list.count())]
|
|
assert labels == ["General", "Amy", "zed"]
|
|
|
|
def test_cancel_changes_nothing(self, qapp, tmp_path):
|
|
original = CassetteState(display_name="trav")
|
|
dialog = self._dialog(tmp_path, original)
|
|
dialog.general.display_name.setText("someone else")
|
|
dialog.reject()
|
|
assert original.display_name == "trav"
|
|
|
|
def test_ok_takes_the_name(self, qapp, tmp_path):
|
|
dialog = self._dialog(tmp_path)
|
|
dialog.general.display_name.setText(" T ")
|
|
dialog._accept()
|
|
assert dialog.state.display_name == "T"
|
|
|
|
def test_status_text_always_has_a_next_step(self):
|
|
from lintunes.gui.sync_settings_dialog import status_text
|
|
for state in (probe_mod.NOT_RUNNING, probe_mod.NO_API):
|
|
headline, hint = status_text(probe_mod.ProbeResult(state))
|
|
assert headline and hint
|
|
assert "Retry" in status_text(probe_mod.ProbeResult(probe_mod.NOT_RUNNING))[1]
|
|
assert status_text(probe_mod.ProbeResult(probe_mod.OK, version="v1"))[0] \
|
|
== "Syncthing v1 is running."
|
|
|
|
def test_key_row_only_when_the_api_is_the_problem(self, qapp, tmp_path):
|
|
dialog = self._dialog(tmp_path)
|
|
dialog.general.show_probe(probe_mod.ProbeResult(probe_mod.NO_API))
|
|
assert not dialog.general._key_row.isHidden()
|
|
dialog.general.show_probe(probe_mod.ProbeResult(probe_mod.OK, version="v"))
|
|
assert dialog.general._key_row.isHidden()
|
|
assert dialog.general._retry.isHidden()
|
|
|
|
def test_api_key_change_goes_to_config(self, qapp, tmp_path):
|
|
dialog = self._dialog(tmp_path, config={st.API_KEY_KEY: "old"})
|
|
assert dialog.config_changes() == {}
|
|
dialog.general.api_key.setText("new")
|
|
assert dialog.config_changes() == {st.API_KEY_KEY: "new"}
|
|
|
|
def test_avatar_is_cropped_square_and_written_on_apply(self, qapp, tmp_path):
|
|
from lintunes.gui.sync_settings_dialog import AVATAR_PX, load_avatar_image
|
|
wide = QImage(400, 200, QImage.Format.Format_RGB32)
|
|
wide.fill(QColor("red"))
|
|
source = tmp_path / "me.png"
|
|
wide.save(str(source))
|
|
image = load_avatar_image(source)
|
|
assert (image.width(), image.height()) == (AVATAR_PX, AVATAR_PX)
|
|
assert load_avatar_image(tmp_path / "missing.png") is None
|
|
|
|
root = tmp_path / "root"
|
|
dialog = self._dialog(root)
|
|
dialog.general.new_avatar = image
|
|
dialog._accept()
|
|
dialog.apply()
|
|
assert dialog.state.avatar == "avatar.png"
|
|
assert not QImage(str(root / "avatar.png")).isNull()
|
|
|
|
|
|
class TestMenu:
|
|
def test_other_machine_greys_the_group_and_says_where(self, qapp, isolated_config):
|
|
from PyQt6.QtWidgets import QMainWindow, QMenu
|
|
from lintunes.gui.cassette_ui import CassetteUi
|
|
prefs = Preferences(isolated_config / "data")
|
|
window = QMainWindow()
|
|
ui = CassetteUi(window, prefs)
|
|
menu = QMenu()
|
|
ui.build_menu(menu)
|
|
ui.refresh_menu()
|
|
assert all(a.isEnabled() for a in ui._actions)
|
|
assert not ui._elsewhere_action.isVisible()
|
|
|
|
prefs.set("cassette_host", {"machine_id": "other", "hostname": "tummult"})
|
|
ui.refresh_menu()
|
|
assert not any(a.isEnabled() for a in ui._actions)
|
|
assert ui._elsewhere_action.isVisible()
|
|
assert "tummult" in ui._elsewhere_action.text()
|