From 0dad4cc95e4e1f6e2da1dfeca3f7252f945a0a4c Mon Sep 17 00:00:00 2001 From: trav Date: Mon, 28 Sep 2026 21:01:17 -0700 Subject: [PATCH] =?UTF-8?q?v0.26.0:=20Cassette=20footing=20=E2=80=94=20Syn?= =?UTF-8?q?cthing=20client,=20host=20machine,=20Sync=20Settings?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- lintunes/__init__.py | 2 +- lintunes/cassette/__init__.py | 0 lintunes/cassette/host.py | 59 +++++ lintunes/cassette/probe.py | 64 +++++ lintunes/cassette/state.py | 232 +++++++++++++++++ lintunes/cassette/syncthing_api.py | 292 +++++++++++++++++++++ lintunes/gui/cassette_ui.py | 94 +++++++ lintunes/gui/main_window.py | 7 +- lintunes/gui/sync_settings_dialog.py | 288 +++++++++++++++++++++ lintunes/preferences.py | 3 + tests/test_round64.py | 365 +++++++++++++++++++++++++++ 11 files changed, 1404 insertions(+), 2 deletions(-) create mode 100644 lintunes/cassette/__init__.py create mode 100644 lintunes/cassette/host.py create mode 100644 lintunes/cassette/probe.py create mode 100644 lintunes/cassette/state.py create mode 100644 lintunes/cassette/syncthing_api.py create mode 100644 lintunes/gui/cassette_ui.py create mode 100644 lintunes/gui/sync_settings_dialog.py create mode 100644 tests/test_round64.py diff --git a/lintunes/__init__.py b/lintunes/__init__.py index 8ec16ed..6136a1c 100644 --- a/lintunes/__init__.py +++ b/lintunes/__init__.py @@ -1,3 +1,3 @@ """LinTunes — iTunes-style music library manager and player for Linux.""" -__version__ = "0.25.0" +__version__ = "0.26.0" diff --git a/lintunes/cassette/__init__.py b/lintunes/cassette/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/lintunes/cassette/host.py b/lintunes/cassette/host.py new file mode 100644 index 0000000..ea4b4ae --- /dev/null +++ b/lintunes/cassette/host.py @@ -0,0 +1,59 @@ +"""Which of my machines runs Cassette. + +trav runs LinTunes on two machines against one synced library. Friend folders, +outboxes and the delivery loop must run in exactly one place, or both machines +would import the same delivery. So one machine is the **host**: + +* ``config.json`` ``cassette_host: true`` on that machine (per machine, never + synced), and +* ``preferences.json`` ``cassette_host: {machine_id, hostname}`` (synced), so + the *other* machine can say "Cassette runs on " instead of just + greying the menu out. + +The first Cassette action on a library with no host asks to claim it. +""" +from __future__ import annotations + +import socket + +from lintunes.config import load_config, save_config +from lintunes.storage.play_journal import machine_id + +CONFIG_KEY = "cassette_host" +PREFS_KEY = "cassette_host" + +HERE = "here" +ELSEWHERE = "elsewhere" +UNCLAIMED = "unclaimed" + + +def hostname() -> str: + try: + return socket.gethostname() or "this computer" + except OSError: + return "this computer" + + +def host_status(prefs) -> tuple[str, str]: + """(HERE | ELSEWHERE | UNCLAIMED, the host's name).""" + record = prefs.get(PREFS_KEY) or {} + flagged = bool(load_config().get(CONFIG_KEY)) + if record.get("machine_id"): + if record["machine_id"] == machine_id(): + return HERE, record.get("hostname") or hostname() + return ELSEWHERE, record.get("hostname") or "another computer" + # No synced record: our own flag still counts (the record is rewritten + # the next time claim() runs), anything else is up for grabs. + return (HERE, hostname()) if flagged else (UNCLAIMED, "") + + +def is_host(prefs) -> bool: + return host_status(prefs)[0] == HERE + + +def claim(prefs): + """Make this machine the Cassette host.""" + config = load_config() + config[CONFIG_KEY] = True + save_config(config) + prefs.set(PREFS_KEY, {"machine_id": machine_id(), "hostname": hostname()}) diff --git a/lintunes/cassette/probe.py b/lintunes/cassette/probe.py new file mode 100644 index 0000000..371a29f --- /dev/null +++ b/lintunes/cassette/probe.py @@ -0,0 +1,64 @@ +"""Is Syncthing there, and can we talk to it? Asked off the GUI thread. + +A stopped Syncthing refuses the connection instantly, but a wedged one can sit +on the socket until the timeout — so the answer arrives as a signal, and every +screen that shows it starts out saying "Checking…". +""" +from __future__ import annotations + +import threading +from dataclasses import dataclass + +from PyQt6.QtCore import QObject, pyqtSignal + +from lintunes.config import load_config +from lintunes.cassette import syncthing_api as st + +# ProbeResult.state +OK = "ok" +NOT_RUNNING = "not_running" +NO_API = "no_api" # no key found, or the key was refused + + +@dataclass +class ProbeResult: + state: str + version: str = "" + my_id: str = "" + detail: str = "" + + @property + def ok(self) -> bool: + return self.state == OK + + +def probe(config: dict | None = None, session=None) -> ProbeResult: + """Synchronous check; ``SyncthingProbe`` runs it on a thread.""" + endpoint = st.resolve_endpoint(config if config is not None else load_config()) + if endpoint is None: + return ProbeResult(NO_API, detail="No API key was found in Syncthing's " + "config.") + client = st.SyncthingClient(endpoint, session=session) + try: + version = client.version() + my_id = client.my_id() + except st.SyncthingError as exc: + if exc.kind == st.NOT_RUNNING: + return ProbeResult(NOT_RUNNING, detail=exc.detail) + if exc.kind == st.BAD_KEY: + return ProbeResult(NO_API, detail=exc.detail) + return ProbeResult(NOT_RUNNING, detail=str(exc)) + return ProbeResult(OK, version=version, my_id=my_id) + + +class SyncthingProbe(QObject): + finished = pyqtSignal(object) # ProbeResult + + def start(self, config: dict | None = None): + def run(): + try: + result = probe(config) + except Exception as exc: # never leave a "Checking…" forever + result = ProbeResult(NOT_RUNNING, detail=str(exc)) + self.finished.emit(result) + threading.Thread(target=run, daemon=True).start() diff --git a/lintunes/cassette/state.py b/lintunes/cassette/state.py new file mode 100644 index 0000000..e0f9b79 --- /dev/null +++ b/lintunes/cassette/state.py @@ -0,0 +1,232 @@ +"""Cassette's own state: who my friends are, what I share with each, what I've +asked each of them for. + +It lives in the machine-local cassette root (``~/.local/share/lintunes/ +cassette/state.json``), **not** in the synced data dir: only the host machine +runs Cassette, the friend folders it describes exist only here, and a second +copy riding Syncthing would just be a conflict waiting to happen. + +Plain dataclasses with ``to_dict``/``from_dict``, like ``lintunes/models``; +``save`` is atomic (``json_storage.write_json``). +""" +from __future__ import annotations + +import copy +import getpass +import os +from dataclasses import dataclass, field +from datetime import datetime, timezone +from pathlib import Path + +from lintunes.config import load_config +from lintunes.storage.json_storage import read_json, write_json + +ROOT_KEY = "cassette_root" # config.json override +STATE_FILE = "state.json" +FORMAT_VERSION = 1 + +# Invite.status +OPEN = "open" +USED = "used" +CANCELLED = "cancelled" + +# Friend.layout — how their followed playlists appear in my sidebar +LAYOUT_MIXED = "mixed" +LAYOUT_FOLDER = "folder" + +# Friend.followed_mode — what happens to a followed playlist's songs +FOLLOW_SEPARATE = "separate" +FOLLOW_LIBRARY = "library" + +# Friend.role — which side of the invite this machine was on +INVITER = "inviter" +INVITEE = "invitee" + + +def now_iso() -> str: + return datetime.now(timezone.utc).replace(tzinfo=None).isoformat( + timespec="seconds") + + +def cassette_root() -> Path: + override = (load_config().get(ROOT_KEY) or "").strip() + if override: + return Path(override).expanduser() + base = os.environ.get("XDG_DATA_HOME", str(Path.home() / ".local/share")) + return Path(base) / "lintunes" / "cassette" + + +def default_display_name() -> str: + try: + return getpass.getuser() + except Exception: + return "LinTunes" + + +@dataclass +class Selection: + """What one friend (or everyone, in share-same mode) may cassette. + + Artists and albums are stored as the track ids they covered when checked + (a snapshot: a new album shows up unchecked). ``all_library`` and shared + playlists are live — they're evaluated each time the copy is built.""" + all_library: bool = False + track_ids: set = field(default_factory=set) + all_playlists: bool = False + playlist_ids: set = field(default_factory=set) + + def to_dict(self) -> dict: + return {"all_library": self.all_library, + "track_ids": sorted(self.track_ids), + "all_playlists": self.all_playlists, + "playlist_ids": sorted(self.playlist_ids)} + + @classmethod + def from_dict(cls, d: dict | None) -> "Selection": + d = d or {} + return cls(all_library=bool(d.get("all_library", False)), + track_ids={int(t) for t in d.get("track_ids", [])}, + all_playlists=bool(d.get("all_playlists", False)), + playlist_ids=set(d.get("playlist_ids", []))) + + def is_empty(self) -> bool: + return not (self.all_library or self.track_ids + or self.all_playlists or self.playlist_ids) + + +@dataclass +class Invite: + token: str + created: str = "" + status: str = OPEN + folder_id: str = "" # my outgoing folder, created with the invite + + def to_dict(self) -> dict: + return {"token": self.token, "created": self.created, + "status": self.status, "folder_id": self.folder_id} + + @classmethod + def from_dict(cls, d: dict) -> "Invite": + return cls(token=d["token"], created=d.get("created", ""), + status=d.get("status", OPEN), + folder_id=d.get("folder_id", "")) + + +@dataclass +class Friend: + token: str + device_id: str + name: str + role: str = INVITEE + out_folder: str = "" # Syncthing folder id I write (send-only) + in_folder: str = "" # Syncthing folder id they write (receive-only here) + added: str = "" + connected_at: str = "" # first time the handshake completed + selection: Selection = field(default_factory=Selection) + layout: str = LAYOUT_FOLDER + followed_mode: str = FOLLOW_SEPARATE + followed: list = field(default_factory=list) # their playlist pids + cassetted: list = field(default_factory=list) # their track ids I want + + def to_dict(self) -> dict: + return {"token": self.token, "device_id": self.device_id, + "name": self.name, "role": self.role, + "out_folder": self.out_folder, "in_folder": self.in_folder, + "added": self.added, "connected_at": self.connected_at, + "selection": self.selection.to_dict(), "layout": self.layout, + "followed_mode": self.followed_mode, + "followed": list(self.followed), + "cassetted": list(self.cassetted)} + + @classmethod + def from_dict(cls, d: dict) -> "Friend": + return cls(token=d["token"], device_id=d.get("device_id", ""), + name=d.get("name", ""), role=d.get("role", INVITEE), + out_folder=d.get("out_folder", ""), + in_folder=d.get("in_folder", ""), + added=d.get("added", ""), + connected_at=d.get("connected_at", ""), + selection=Selection.from_dict(d.get("selection")), + layout=d.get("layout", LAYOUT_FOLDER), + followed_mode=d.get("followed_mode", FOLLOW_SEPARATE), + followed=list(d.get("followed", [])), + cassetted=[int(t) for t in d.get("cassetted", [])]) + + +@dataclass +class CassetteState: + display_name: str = field(default_factory=default_display_name) + avatar: str = "" # filename inside the cassette root, "" = none + share_same_for_all: bool = False + general: Selection = field(default_factory=Selection) + invites: dict = field(default_factory=dict) # token -> Invite + friends: dict = field(default_factory=dict) # token -> Friend + + def to_dict(self) -> dict: + return {"version": FORMAT_VERSION, + "display_name": self.display_name, "avatar": self.avatar, + "share_same_for_all": self.share_same_for_all, + "general": self.general.to_dict(), + "invites": [i.to_dict() for i in self.invites.values()], + "friends": [f.to_dict() for f in self.friends.values()]} + + @classmethod + def from_dict(cls, d: dict | None) -> "CassetteState": + d = d or {} + state = cls(display_name=d.get("display_name") or default_display_name(), + avatar=d.get("avatar", ""), + share_same_for_all=bool(d.get("share_same_for_all", False)), + general=Selection.from_dict(d.get("general"))) + for raw in d.get("invites", []): + invite = Invite.from_dict(raw) + state.invites[invite.token] = invite + for raw in d.get("friends", []): + friend = Friend.from_dict(raw) + state.friends[friend.token] = friend + return state + + def copy(self) -> "CassetteState": + return copy.deepcopy(self) + + def open_invites(self) -> list[Invite]: + return [i for i in self.invites.values() if i.status == OPEN] + + def selection_for(self, friend: Friend) -> Selection: + """The selection that actually applies to ``friend``.""" + return self.general if self.share_same_for_all else friend.selection + + def friend_by_device(self, device_id: str) -> Friend | None: + return next((f for f in self.friends.values() + if f.device_id == device_id), None) + + +def load_state(root: Path | None = None) -> CassetteState: + path = (root or cassette_root()) / STATE_FILE + if not path.exists(): + return CassetteState() + try: + return CassetteState.from_dict(read_json(path)) + except (ValueError, OSError, KeyError): + return CassetteState() + + +def save_state(state: CassetteState, root: Path | None = None): + root = root or cassette_root() + root.mkdir(parents=True, exist_ok=True) + write_json(root / STATE_FILE, state.to_dict()) + + +def friend_dir(root: Path, token: str) -> Path: + return root / "friends" / token + + +def out_dir(root: Path, token: str) -> Path: + return friend_dir(root, token) / "out" + + +def in_dir(root: Path, token: str) -> Path: + return friend_dir(root, token) / "in" + + +def cache_dir(root: Path, token: str) -> Path: + return friend_dir(root, token) / "cache" diff --git a/lintunes/cassette/syncthing_api.py b/lintunes/cassette/syncthing_api.py new file mode 100644 index 0000000..0c56022 --- /dev/null +++ b/lintunes/cassette/syncthing_api.py @@ -0,0 +1,292 @@ +"""The local Syncthing daemon's REST API — the only "network" Cassette uses. + +LinTunes never talks to a friend's machine. It asks *this* machine's Syncthing +to add a device, share a folder, accept one, and to say what it can see; +Syncthing moves every byte. So this module is a thin, typed wrapper over +``http://127.0.0.1:8384/rest/...`` with one rule: **every failure says which +step failed and what Syncthing said** (``SyncthingError``), because the UI +promises never to fail silently. + +Where the API lives and its key come from Syncthing's own ``config.xml`` (the +same file ``sync_identity`` reads for device names). ``config.json`` may +override either, for a Syncthing that isn't where we look. +""" +from __future__ import annotations + +import logging +import xml.etree.ElementTree as ET +from dataclasses import dataclass +from pathlib import Path + +import requests + +from lintunes import sync_identity + +log = logging.getLogger(__name__) + +# config.json keys (per machine — which Syncthing this is is nobody else's +# business, same as ytdlp_cookies_browser). +API_KEY_KEY = "syncthing_api_key" +ADDRESS_KEY = "syncthing_address" + +DEFAULT_ADDRESS = "127.0.0.1:8384" +TIMEOUT_S = 10 + +# SyncthingError.kind +NOT_RUNNING = "not_running" # nothing listening — Syncthing isn't started +BAD_KEY = "bad_key" # it answered 401/403: missing or wrong API key +NO_KEY = "no_key" # we couldn't find a key to send at all +OTHER = "other" # it answered, and said no + + +class SyncthingError(Exception): + """A step that failed, and what Syncthing (or the socket) said.""" + + def __init__(self, step: str, detail: str, kind: str = OTHER, + status: int | None = None): + super().__init__(f"{step}: {detail}") + self.step = step + self.detail = detail + self.kind = kind + self.status = status # HTTP status, when Syncthing answered + + +@dataclass(frozen=True) +class Endpoint: + address: str # host:port + api_key: str + tls: bool = False + + @property + def base_url(self) -> str: + return f"{'https' if self.tls else 'http'}://{self.address}" + + +def config_dir() -> Path | None: + """Syncthing's home directory (holds config.xml), or None.""" + return sync_identity._config_dir() + + +def read_gui_config(directory: Path | None = None) -> Endpoint | None: + """The API endpoint as Syncthing's config.xml describes it, or None when + there is no config.xml or it names no key.""" + directory = directory or config_dir() + if directory is None: + return None + try: + root = ET.parse(directory / "config.xml").getroot() + except (OSError, ET.ParseError): + return None + gui = root.find("gui") + if gui is None: + return None + key = (gui.findtext("apikey") or "").strip() + if not key: + return None + address = (gui.findtext("address") or DEFAULT_ADDRESS).strip() + # A GUI bound to 0.0.0.0 still answers on loopback, which is where we are. + if address.startswith("0.0.0.0:"): + address = "127.0.0.1:" + address.split(":", 1)[1] + tls = (gui.get("tls") or "").lower() == "true" + return Endpoint(address=address, api_key=key, tls=tls) + + +def resolve_endpoint(config: dict) -> Endpoint | None: + """config.json overrides layered over config.xml. None means we have no + key to send — the "Can't reach Syncthing's API" state, not "not running".""" + found = read_gui_config() + key = (config.get(API_KEY_KEY) or "").strip() or (found.api_key if found else "") + if not key: + return None + address = ((config.get(ADDRESS_KEY) or "").strip() + or (found.address if found else DEFAULT_ADDRESS)) + tls = found.tls if found else False + return Endpoint(address=address, api_key=key, tls=tls) + + +class SyncthingClient: + """Typed calls over one Endpoint. Thread-safe enough for our use: each + call is one request on a shared Session (requests' Session is safe for + concurrent simple requests); the event long-poll gets its own client.""" + + def __init__(self, endpoint: Endpoint, session=None): + self.endpoint = endpoint + self._session = session or requests.Session() + self._session.headers["X-API-Key"] = endpoint.api_key + + # ---- plumbing ---- + + def _request(self, method: str, path: str, step: str, *, params=None, + json=None, timeout: float = TIMEOUT_S): + url = self.endpoint.base_url + path + try: + response = self._session.request( + method, url, params=params, json=json, timeout=timeout, + verify=False) # Syncthing's GUI cert is self-signed + except requests.ConnectionError as exc: + raise SyncthingError(step, f"nothing answered at {self.endpoint.address}" + f" ({exc.__class__.__name__})", NOT_RUNNING) from exc + except requests.Timeout as exc: + raise SyncthingError(step, "Syncthing didn't answer in time", + NOT_RUNNING) from exc + except requests.RequestException as exc: + raise SyncthingError(step, str(exc)) from exc + if response.status_code in (401, 403): + raise SyncthingError(step, "Syncthing refused the API key", BAD_KEY) + if response.status_code >= 400: + text = (response.text or "").strip() or f"HTTP {response.status_code}" + raise SyncthingError(step, text, status=response.status_code) + if not response.content: + return None + try: + return response.json() + except ValueError: + return response.text + + # ---- system ---- + + def version(self) -> str: + return self._request("GET", "/rest/system/version", + "Checking Syncthing's version")["version"] + + def my_id(self) -> str: + return self._request("GET", "/rest/system/status", + "Reading this computer's device ID")["myID"] + + def connections(self) -> dict: + """{device id: {"connected": bool, ...}}""" + data = self._request("GET", "/rest/system/connections", + "Checking connections") + return (data or {}).get("connections", {}) + + def device_stats(self) -> dict: + """{device id: {"lastSeen": iso, ...}}""" + return self._request("GET", "/rest/stats/device", + "Reading when devices were last seen") or {} + + def folder_stats(self) -> dict: + """{folder id: {"lastFile": {"at": iso, ...}, "lastScan": iso}}""" + return self._request("GET", "/rest/stats/folder", + "Reading folder activity") or {} + + # ---- devices ---- + + def device(self, device_id: str) -> dict | None: + try: + return self._request("GET", f"/rest/config/devices/{device_id}", + "Looking up a device") + except SyncthingError as exc: + if exc.status == 404: + return None + raise + + def add_device(self, device_id: str, name: str): + """Add (or rename) a device. Idempotent.""" + self._request("PUT", f"/rest/config/devices/{device_id}", + f"Adding {name}'s computer to Syncthing", + json={"deviceID": device_id, "name": name, + "addresses": ["dynamic"]}) + + def remove_device(self, device_id: str, name: str = "the device"): + self._request("DELETE", f"/rest/config/devices/{device_id}", + f"Removing {name}'s computer from Syncthing") + + # ---- folders ---- + + def folder(self, folder_id: str) -> dict | None: + try: + return self._request("GET", f"/rest/config/folders/{folder_id}", + "Looking up a folder") + except SyncthingError as exc: + if exc.status == 404: + return None + raise + + def add_folder(self, folder_id: str, label: str, path: Path, kind: str, + device_ids=(), step: str = "Creating a shared folder"): + """Create or replace a folder. ``kind`` is sendonly/receiveonly/…""" + self._request("PUT", f"/rest/config/folders/{folder_id}", step, json={ + "id": folder_id, "label": label, "path": str(path), "type": kind, + "devices": [{"deviceID": d} for d in device_ids], + "fsWatcherEnabled": True, + }) + + def share_folder(self, folder_id: str, device_id: str, + step: str = "Sharing a folder"): + """Add ``device_id`` to a folder's device list. Idempotent.""" + folder = self.folder(folder_id) + if folder is None: + raise SyncthingError(step, f"folder {folder_id} doesn't exist") + devices = folder.get("devices") or [] + if any(d.get("deviceID") == device_id for d in devices): + return + devices.append({"deviceID": device_id}) + self._request("PATCH", f"/rest/config/folders/{folder_id}", step, + json={"devices": devices}) + + def remove_folder(self, folder_id: str, step: str = "Removing a folder"): + self._request("DELETE", f"/rest/config/folders/{folder_id}", step) + + def min_disk_free(self, folder_id: str | None = None) -> tuple[float, str]: + """(value, unit) of a folder's free-space floor — below it Syncthing + stops syncing. Falls back to the defaults for new folders.""" + data = None + if folder_id: + data = self.folder(folder_id) + if data is None: + data = self._request("GET", "/rest/config/defaults/folder", + "Reading Syncthing's folder defaults") + floor = (data or {}).get("minDiskFree") or {} + return float(floor.get("value", 1)), floor.get("unit", "%") + + # ---- pending (someone knocking) ---- + + def pending_devices(self) -> dict: + """{device id: {"name": str, "time": iso, "address": str}}""" + return self._request("GET", "/rest/cluster/pending/devices", + "Checking for computers asking to connect") or {} + + def pending_folders(self, device_id: str | None = None) -> dict: + """{folder id: {"offeredBy": {device id: {...}}}}""" + params = {"device": device_id} if device_id else None + return self._request("GET", "/rest/cluster/pending/folders", + "Checking for folders offered to us", + params=params) or {} + + # ---- database ---- + + def completion(self, folder_id: str, device_id: str | None = None) -> dict: + params = {"folder": folder_id} + if device_id: + params["device"] = device_id + return self._request("GET", "/rest/db/completion", + "Checking sync progress", params=params) or {} + + def file_info(self, folder_id: str, name: str) -> dict | None: + try: + return self._request("GET", "/rest/db/file", "Checking a file", + params={"folder": folder_id, "file": name}) + except SyncthingError as exc: + if exc.status == 404: + return None + raise + + # ---- events ---- + + def events(self, since: int, types=(), timeout_s: int = 60) -> list: + """Long-poll /rest/events. Returns [] on the server's own timeout.""" + params = {"since": since, "timeout": timeout_s} + if types: + params["events"] = ",".join(types) + return self._request("GET", "/rest/events", "Waiting for Syncthing events", + params=params, timeout=timeout_s + 10) or [] + + +def disk_floor_bytes(value: float, unit: str, total_bytes: int) -> int: + """A minDiskFree setting in bytes, for a disk of ``total_bytes``.""" + if unit == "%": + return int(total_bytes * value / 100) + scale = {"": 1, "B": 1, "kB": 10**3, "MB": 10**6, "GB": 10**9, + "TB": 10**12, "KiB": 2**10, "MiB": 2**20, "GiB": 2**30, + "TiB": 2**40} + return int(value * scale.get(unit, 1)) diff --git a/lintunes/gui/cassette_ui.py b/lintunes/gui/cassette_ui.py new file mode 100644 index 0000000..934d9ff --- /dev/null +++ b/lintunes/gui/cassette_ui.py @@ -0,0 +1,94 @@ +"""Cassette's place in the main window: its Connections-menu group, the +"which machine is the host" gate, and the dialogs those items open. + +Kept out of main_window.py, which is big enough; MainWindow owns one of these +and calls ``build_menu`` / ``refresh_menu``. +""" +from __future__ import annotations + +from PyQt6.QtCore import QObject +from PyQt6.QtGui import QAction +from PyQt6.QtWidgets import QDialog, QMessageBox + +from lintunes.cassette import host +from lintunes.cassette.state import cassette_root, load_state, save_state +from lintunes.config import load_config, save_config + + +class CassetteUi(QObject): + def __init__(self, window, prefs, parent=None): + super().__init__(parent or window) + self._window = window + self._prefs = prefs + self._elsewhere_action: QAction | None = None + self._actions: list[QAction] = [] + + # ---- menu ---- + + def build_menu(self, menu): + """Add the Cassette group, between two separators.""" + menu.addSeparator() + self._elsewhere_action = QAction("", self._window) + self._elsewhere_action.setEnabled(False) + self._elsewhere_action.setVisible(False) + menu.addAction(self._elsewhere_action) + settings = QAction("Sync Settings…", self._window) + settings.triggered.connect(self.show_sync_settings) + menu.addAction(settings) + self._actions.append(settings) + menu.addSeparator() + + def refresh_menu(self): + status, name = host.host_status(self._prefs) + elsewhere = status == host.ELSEWHERE + for action in self._actions: + action.setEnabled(not elsewhere) + if self._elsewhere_action is not None: + self._elsewhere_action.setText(f"Friend sharing runs on {name}") + self._elsewhere_action.setVisible(elsewhere) + + # ---- host gate ---- + + def ensure_host(self) -> bool: + """True when this machine may run Cassette — asking to claim it when + no machine has yet.""" + status, name = host.host_status(self._prefs) + if status == host.HERE: + return True + if status == host.ELSEWHERE: + QMessageBox.information( + self._window, "Friend Sharing", + f"Friend sharing runs on {name}. Open LinTunes there to " + "share with friends or see their libraries.") + return False + answer = QMessageBox.question( + self._window, "Friend Sharing", + f"Use this computer ({host.hostname()}) for sharing with friends?\n\n" + "Only one of your computers handles friends: it keeps their " + "folders and brings in the songs you ask for. Songs it adds to " + "your library still reach your other computers as usual.", + QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.Cancel, + QMessageBox.StandardButton.Yes) + if answer != QMessageBox.StandardButton.Yes: + return False + host.claim(self._prefs) + return True + + # ---- dialogs ---- + + def show_sync_settings(self): + if not self.ensure_host(): + return + from lintunes.gui.sync_settings_dialog import SyncSettingsDialog + root = cassette_root() + config = load_config() + dialog = SyncSettingsDialog(load_state(root), root, config, self._window) + if dialog.exec() != QDialog.DialogCode.Accepted: + return + dialog.apply() + changes = dialog.config_changes() + if changes: + config = load_config() + config.update(changes) + save_config(config) + save_state(dialog.state, root) diff --git a/lintunes/gui/main_window.py b/lintunes/gui/main_window.py index 913f595..4e38064 100644 --- a/lintunes/gui/main_window.py +++ b/lintunes/gui/main_window.py @@ -25,6 +25,7 @@ from lintunes.importers import file_importer from lintunes.gui.album_art_dialog import AlbumArtDialog from lintunes.gui.art_ops import embed_artwork from lintunes.gui.cast_dialog import ChromecastDialog +from lintunes.gui.cassette_ui import CassetteUi from lintunes.gui.device_sync_dialog import DeviceSyncSettingsDialog from lintunes.gui.export_dialog import ExportKindDialog, WebMixDialog from lintunes.gui.identify_dialog import IdentifyDialog @@ -94,6 +95,8 @@ class MainWindow(QMainWindow): # Context playback started from: "library" or "playlist:". Scopes # the now-playing speaker icon to where the song is actually playing. self._now_playing_context = "" + # Friend sharing's menu group and dialogs. + self._cassette = CassetteUi(self, prefs) self.setWindowTitle("LinTunes") self.resize(1280, 800) @@ -375,7 +378,8 @@ class MainWindow(QMainWindow): # "Sync Playlist to Rabbit (Auxio)" was retired in Round 51, once # andTunes could play what it syncs. device_sync's helpers stay: the # export and andTunes code import them. - connections_menu.addSeparator() + # Friend sharing (Cassette) sits between two separators of its own. + self._cassette.build_menu(connections_menu) self._cast_action = self._add_action( connections_menu, "Connect to Chromecast…", "", self._toggle_cast) # Re-checked every time the menu opens: cheap (one gvfs listdir), and @@ -411,6 +415,7 @@ class MainWindow(QMainWindow): # Whole-library sync doesn't care which view is open — only that a # device is plugged in and at least one playlist is ticked. self._andtunes_action.setEnabled(not busy and device is not None) + self._cassette.refresh_menu() # Enabled whenever there's an APK to offer; the click finds the # Rabbit (adb or MTP) and says so if it can't — cheaper than asking # adb on every menu open. diff --git a/lintunes/gui/sync_settings_dialog.py b/lintunes/gui/sync_settings_dialog.py new file mode 100644 index 0000000..cfded5e --- /dev/null +++ b/lintunes/gui/sync_settings_dialog.py @@ -0,0 +1,288 @@ +"""Connections ▸ Sync Settings… — Cassette's settings, laid out like VLC's. + +A list on the left (General first, then one entry per friend) and the chosen +page on the right. Unlike Preferences, which applies as you go, this window +edits a **copy** of the Cassette state and writes nothing until OK: "Select +None" can wipe a careful selection, and Cancel has to be able to undo that. +""" +from __future__ import annotations + +from pathlib import Path + +from PyQt6.QtCore import Qt, pyqtSignal +from PyQt6.QtGui import QImage, QPixmap +from PyQt6.QtWidgets import ( + QDialog, QDialogButtonBox, QFileDialog, QFormLayout, QGroupBox, + QHBoxLayout, QLabel, QLineEdit, QListWidget, QPushButton, QSplitter, + QStackedWidget, QVBoxLayout, QWidget, +) + +from lintunes.cassette import probe as probe_mod +from lintunes.cassette import syncthing_api as st + +AVATAR_PX = 256 +AVATAR_PREVIEW_PX = 72 +AVATAR_FILE = "avatar.png" + + +def load_avatar_image(path: Path) -> QImage | None: + """``path`` decoded and cropped/scaled to a square AVATAR_PX, or None if + it isn't an image Qt can read.""" + image = QImage(str(path)) + if image.isNull(): + return None + side = min(image.width(), image.height()) + image = image.copy((image.width() - side) // 2, + (image.height() - side) // 2, side, side) + return image.scaled(AVATAR_PX, AVATAR_PX, + Qt.AspectRatioMode.IgnoreAspectRatio, + Qt.TransformationMode.SmoothTransformation) + + +class AvatarSquare(QLabel): + """Fixed square preview; gray with a hint when empty.""" + + def __init__(self, parent=None): + super().__init__(parent) + self.setFixedSize(AVATAR_PREVIEW_PX, AVATAR_PREVIEW_PX) + self.setAlignment(Qt.AlignmentFlag.AlignCenter) + self.setStyleSheet("QLabel { border: 1px solid palette(mid);" + " border-radius: 6px; background: palette(mid); }") + self.set_image(None) + + def set_image(self, image: QImage | None): + if image is None or image.isNull(): + self.setPixmap(QPixmap()) + self.setText("none") + return + self.setText("") + self.setPixmap(QPixmap.fromImage(image).scaled( + AVATAR_PREVIEW_PX - 2, AVATAR_PREVIEW_PX - 2, + Qt.AspectRatioMode.KeepAspectRatio, + Qt.TransformationMode.SmoothTransformation)) + + +def status_text(result) -> tuple[str, str]: + """(headline, next step) for a ProbeResult — plain language, always with + something to do.""" + if result is None: + return "Checking Syncthing…", "" + if result.state == probe_mod.OK: + return f"Syncthing {result.version} is running.", "" + if result.state == probe_mod.NOT_RUNNING: + return ("Syncthing isn't running.", + "Start it (for example: systemctl --user start syncthing, or " + "open the Syncthing app), then press Retry.") + return ("Can't reach Syncthing's API.", + "LinTunes needs Syncthing's API key. It's in Syncthing's web page " + "under Actions ▸ Settings ▸ General ▸ API Key, or LinTunes can read " + "it from Syncthing's config file for you.") + + +class GeneralPage(QWidget): + retry_requested = pyqtSignal() + + def __init__(self, state, root: Path, config: dict, parent=None): + super().__init__(parent) + self._state = state + self._root = root + self._config = config + self.new_avatar: QImage | None = None # staged until OK + self.avatar_removed = False + + layout = QVBoxLayout(self) + + box = QGroupBox("Syncthing") + form = QVBoxLayout(box) + self._status = QLabel() + self._status.setWordWrap(True) + self._hint = QLabel() + self._hint.setWordWrap(True) + self._hint.setTextInteractionFlags( + Qt.TextInteractionFlag.TextSelectableByMouse) + form.addWidget(self._status) + form.addWidget(self._hint) + self._key_row = QWidget() + key_layout = QHBoxLayout(self._key_row) + key_layout.setContentsMargins(0, 0, 0, 0) + self.api_key = QLineEdit(config.get(st.API_KEY_KEY, "")) + # Empty = use whatever Syncthing's config says (follows a key change). + self.api_key.setPlaceholderText("API key (blank: read from Syncthing)") + self.api_key.setEchoMode(QLineEdit.EchoMode.Password) + read = QPushButton("Read from Syncthing") + read.clicked.connect(self._read_key) + key_layout.addWidget(self.api_key, stretch=1) + key_layout.addWidget(read) + form.addWidget(self._key_row) + self._retry = QPushButton("Retry") + self._retry.clicked.connect(self.retry_requested) + form.addWidget(self._retry, alignment=Qt.AlignmentFlag.AlignLeft) + layout.addWidget(box) + + you = QGroupBox("You") + you_form = QFormLayout(you) + self.display_name = QLineEdit(state.display_name) + self.display_name.setToolTip("What friends see") + you_form.addRow("Your display name:", self.display_name) + avatar_row = QHBoxLayout() + self._avatar = AvatarSquare() + existing = root / state.avatar if state.avatar else None + if existing is not None and existing.exists(): + self._avatar.set_image(QImage(str(existing))) + choose = QPushButton("Choose…") + choose.clicked.connect(self._choose_avatar) + remove = QPushButton("Remove") + remove.clicked.connect(self._remove_avatar) + buttons = QVBoxLayout() + buttons.addWidget(choose) + buttons.addWidget(remove) + buttons.addStretch(1) + avatar_row.addWidget(self._avatar) + avatar_row.addLayout(buttons) + avatar_row.addStretch(1) + you_form.addRow("Avatar:", avatar_row) + layout.addWidget(you) + layout.addStretch(1) + + self.show_probe(None) + + def show_probe(self, result): + headline, hint = status_text(result) + self._status.setText(headline) + self._hint.setText(hint) + self._hint.setVisible(bool(hint)) + self._retry.setVisible(result is not None and not result.ok) + self._key_row.setVisible(result is not None + and result.state == probe_mod.NO_API) + + def _read_key(self): + found = st.read_gui_config() + if found is None: + self._hint.setText("Couldn't find Syncthing's config file. Copy the " + "key from Syncthing's web page instead.") + return + # Blank means "use Syncthing's own", so a key Syncthing later rotates + # is followed rather than pinned in config.json. + self.api_key.clear() + self.retry_requested.emit() + + def _choose_avatar(self): + path, _ = QFileDialog.getOpenFileName( + self, "Choose an Avatar", str(Path.home()), + "Images (*.png *.jpg *.jpeg *.gif *.bmp *.webp)") + if not path: + return + image = load_avatar_image(Path(path)) + if image is None: + self._hint.setText("That file isn't an image LinTunes can read.") + self._hint.setVisible(True) + return + self.new_avatar = image + self.avatar_removed = False + self._avatar.set_image(image) + + def _remove_avatar(self): + self.new_avatar = None + self.avatar_removed = True + self._avatar.set_image(None) + + +class SyncSettingsDialog(QDialog): + """Edits ``state`` (a copy — the caller's is untouched until accepted). + + After ``exec()`` returns Accepted, read ``state`` and ``config_changes`` + and call ``apply()`` — the caller owns saving so it can regenerate what + the change affects.""" + + def __init__(self, state, root: Path, config: dict, parent=None, + probe=None): + super().__init__(parent) + self.setWindowTitle("Sync Settings") + self.resize(820, 560) + self.state = state.copy() + self._root = root + self._config = config + + splitter = QSplitter(Qt.Orientation.Horizontal) + self.pages_list = QListWidget() + self.pages = QStackedWidget() + splitter.addWidget(self.pages_list) + splitter.addWidget(self.pages) + splitter.setSizes([190, 630]) + splitter.setCollapsible(1, False) + + self.general = GeneralPage(self.state, root, config) + self._add_page("General", self.general) + self.friend_pages = {} + for friend in sorted(self.state.friends.values(), + key=lambda f: f.name.casefold()): + page = self._friend_page(friend) + self.friend_pages[friend.token] = page + self._add_page(friend.name, page) + self.pages_list.currentRowChanged.connect(self.pages.setCurrentIndex) + self.pages_list.setCurrentRow(0) + + buttons = QDialogButtonBox(QDialogButtonBox.StandardButton.Ok + | QDialogButtonBox.StandardButton.Cancel) + buttons.accepted.connect(self._accept) + buttons.rejected.connect(self.reject) + + layout = QVBoxLayout(self) + layout.addWidget(splitter, stretch=1) + layout.addWidget(buttons) + + self._probe = probe or probe_mod.SyncthingProbe(self) + self._probe.finished.connect(self.general.show_probe) + self.general.retry_requested.connect(self._retry) + self._probe.start(self._probe_config()) + + def _add_page(self, label: str, page: QWidget): + self.pages_list.addItem(label) + self.pages.addWidget(page) + + def _friend_page(self, friend) -> QWidget: + page = QWidget() + layout = QVBoxLayout(page) + title = QLabel(friend.name) + font = title.font() + font.setPointSize(font.pointSize() + 3) + font.setBold(True) + title.setFont(font) + layout.addWidget(title) + layout.addStretch(1) + return page + + def _probe_config(self) -> dict: + config = dict(self._config) + key = self.general.api_key.text().strip() + if key: + config[st.API_KEY_KEY] = key + return config + + def _retry(self): + self.general.show_probe(None) + self._probe.start(self._probe_config()) + + def _accept(self): + name = self.general.display_name.text().strip() + self.state.display_name = name or self.state.display_name + self.accept() + + def config_changes(self) -> dict: + """config.json keys the user changed here.""" + key = self.general.api_key.text().strip() + if key != (self._config.get(st.API_KEY_KEY) or ""): + return {st.API_KEY_KEY: key} + return {} + + def apply(self): + """Write the staged avatar into the cassette root and point the state + at it. Called by the owner after an accepted exec().""" + if self.general.new_avatar is not None: + self._root.mkdir(parents=True, exist_ok=True) + tmp = self._root / (AVATAR_FILE + ".tmp.png") + if self.general.new_avatar.save(str(tmp), "PNG"): + tmp.replace(self._root / AVATAR_FILE) + self.state.avatar = AVATAR_FILE + elif self.general.avatar_removed: + self.state.avatar = "" diff --git a/lintunes/preferences.py b/lintunes/preferences.py index c0ad1b6..fecad6b 100644 --- a/lintunes/preferences.py +++ b/lintunes/preferences.py @@ -41,6 +41,9 @@ DEFAULTS = { "playlist_ids": [], "copy_art": True, }, + # Which machine runs friend sharing (Cassette): {machine_id, hostname}. + # Synced so the other machine can name it; see lintunes/cassette/host.py. + "cassette_host": None, } diff --git a/tests/test_round64.py b/tests/test_round64.py new file mode 100644 index 0000000..7a2c751 --- /dev/null +++ b/tests/test_round64.py @@ -0,0 +1,365 @@ +"""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 = """ + +
{address}
+ {key} + +""" + + +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()