"""Round 65: Cassette invites and the handshake. * An invite code is one pasteable line carrying the inviter's device id, display name and a one-time token. The token is embedded in both folder ids. * Syncthing hides an *unknown* device's folders (verified against v1.30: a stranger shows only as a pending device), so while an invite is open the inviter **probes** a new knock — adds it with nothing shared — to see which folder it offers. The right ``-b`` completes the friendship and uses the invite; anything else is removed again and never probed twice. * Every friendship and invite has a plain-language status with a next step. The ``syncthing``-marked test runs the whole thing between three real Syncthing instances on loopback (``--syncthing`` to opt in). """ import time from pathlib import Path import pytest from lintunes.cassette import invite as inv from lintunes.cassette import publish from lintunes.cassette import service as svc_mod from lintunes.cassette import syncthing_api as st from lintunes.cassette.service import CassetteService, friend_status from lintunes.cassette.state import ( CANCELLED, INVITEE, INVITER, OPEN, USED, Friend, in_dir, out_dir, ) DEV_A = "AAAAAAA-BBBBBBB-CCCCCCC-DDDDDDD-EEEEEEE-FFFFFFF-GGGGGGG-HHHHHHH" DEV_B = "BBBBBBB-BBBBBBB-CCCCCCC-DDDDDDD-EEEEEEE-FFFFFFF-GGGGGGG-HHHHHHH" DEV_C = "CCCCCCC-BBBBBBB-CCCCCCC-DDDDDDD-EEEEEEE-FFFFFFF-GGGGGGG-HHHHHHH" # ---- invite codes ---- class TestInviteCode: def test_round_trip(self): code = inv.encode(DEV_A, "trav ✨", "0123456789abcdef") assert code.startswith("LTC1.") assert "=" not in code parsed = inv.decode(code) assert parsed == inv.InviteCode(DEV_A, "trav ✨", "0123456789abcdef") def test_survives_chat_app_line_breaks(self): code = inv.encode(DEV_A, "trav", "0123456789abcdef") mangled = " " + code[:20] + "\n" + code[20:40] + " \r\n" + code[40:] + "\n" assert inv.decode(mangled).device_id == DEV_A @pytest.mark.parametrize("text", [ "", "hello", "LTC1.", "LTC1.!!!notbase64", "LTC1.bm90IGpzb24", ]) def test_garbage_is_not_valid(self, text): with pytest.raises(inv.InviteError) as info: inv.decode(text) assert "whole code" in str(info.value) def test_truncated_code_is_not_valid(self): code = inv.encode(DEV_A, "trav", "0123456789abcdef") with pytest.raises(inv.InviteError): inv.decode(code[:-12]) def test_folder_ids(self): token = "0123456789abcdef" assert inv.parse_folder_id(inv.inviter_folder(token)) == (token, "a") assert inv.parse_folder_id(inv.invitee_folder(token)) == (token, "b") assert inv.parse_folder_id("music") is None assert inv.parse_folder_id("lintunes-cassette-xyz-b") is None # ---- statuses ---- class TestStatus: def _friend(self, **kw): return Friend(token="t", device_id=DEV_B, name="Sam", **kw) def test_waiting_for_computer_names_the_fixes(self): status = friend_status(self._friend(), connected=False, in_accepted=False, remote_state="", up_to_date=False, last_seen="") assert status.code == svc_mod.WAITING_COMPUTER assert status.text == "Waiting for Sam's computer" assert "global discovery" in status.hint and "relaying" in status.hint def test_waiting_for_lintunes(self): status = friend_status(self._friend(connected_at=""), connected=True, in_accepted=True, remote_state="notSharing", up_to_date=False, last_seen="") assert status.text == "Waiting for Sam to open LinTunes" def test_connected_and_offline(self): friend = self._friend(connected_at="2026-09-01T00:00:00", last_synced="2026-09-02T00:00:00") up = friend_status(friend, connected=True, in_accepted=True, remote_state="valid", up_to_date=True, last_seen="x") assert (up.code, up.up_to_date, up.last_synced) == \ (svc_mod.CONNECTED, True, "2026-09-02T00:00:00") down = friend_status(friend, connected=False, in_accepted=True, remote_state="", up_to_date=False, last_seen="x") assert down.code == svc_mod.OFFLINE # ---- an in-memory Syncthing ---- class FakeSyncthing: """Enough of Syncthing's model for the handshake: devices, folders and what's pending. ``offer`` plays a remote device knocking.""" def __init__(self, my_id): self.id = my_id self.devices = {} self.folders = {} self.pending_dev = {} self.offers = {} # folder id -> {device id: {...}} self.connected = set() def version(self): return "v1.30.0" def my_id(self): return self.id def device(self, d): return self.devices.get(d) def add_device(self, d, name, addresses=("dynamic",)): self.devices[d] = {"deviceID": d, "name": name} self.pending_dev.pop(d, None) def remove_device(self, d, name=""): self.devices.pop(d, None) def folder(self, f): return self.folders.get(f) def add_folder(self, f, label, path, kind, device_ids=(), step=""): self.folders[f] = {"id": f, "type": kind, "path": str(path), "devices": [{"deviceID": d} for d in device_ids]} self.offers.pop(f, None) def share_folder(self, f, d, step=""): devices = self.folders[f]["devices"] if not any(x["deviceID"] == d for x in devices): devices.append({"deviceID": d}) def remove_folder(self, f, step=""): self.folders.pop(f, None) def pending_devices(self): return {d: dict(v) for d, v in self.pending_dev.items()} def pending_folders(self, device_id=None): # Syncthing only reports folders from devices it knows. return {f: {"offeredBy": {d: {} for d in by if d in self.devices}} for f, by in self.offers.items() if any(d in self.devices for d in by)} def connections(self): return {d: {"connected": d in self.connected} for d in self.devices} def device_stats(self): return {} def completion(self, folder, device=None): return {"completion": 100, "remoteState": "valid"} def knock(self, device_id, name="stranger-laptop", offers=()): self.pending_dev[device_id] = {"name": name, "time": svc_mod.now_iso() + "Z"} for folder_id in offers: self.offers.setdefault(folder_id, {})[device_id] = {} def _service(tmp_path, fake, name="trav"): service = CassetteService(root=tmp_path / name, config_loader=lambda: {}) service.client = lambda: fake service._state.display_name = name return service class TestHandshakeLogic: def test_create_invite_makes_a_sendonly_folder_and_intro(self, qapp, tmp_path): fake = FakeSyncthing(DEV_A) service = _service(tmp_path, fake) code = service.create_invite() parsed = inv.decode(code) assert parsed.device_id == DEV_A and parsed.name == "trav" folder = fake.folders[inv.inviter_folder(parsed.token)] assert folder["type"] == "sendonly" and folder["devices"] == [] intro = publish.read_intro(out_dir(service.root, parsed.token)) assert intro["name"] == "trav" assert (out_dir(service.root, parsed.token) / ".stignore").exists() assert service.state().invites[parsed.token].status == OPEN def test_matching_knock_completes_and_uses_the_invite(self, qapp, tmp_path): fake = FakeSyncthing(DEV_A) service = _service(tmp_path, fake) token = inv.decode(service.create_invite()).token fake.knock(DEV_B, offers=[inv.invitee_folder(token)]) service.reconcile(fake) # probe: add with nothing shared assert DEV_B in fake.devices assert DEV_B in service.state().probing service.reconcile(fake) # now its folder is visible: complete state = service.state() assert state.invites[token].status == USED friend = state.friends[token] assert friend.role == INVITER and friend.device_id == DEV_B assert fake.folders[inv.invitee_folder(token)]["type"] == "receiveonly" assert {"deviceID": DEV_B} in fake.folders[inv.inviter_folder(token)]["devices"] assert not state.probing def test_stranger_is_let_go_and_left_pending(self, qapp, tmp_path): fake = FakeSyncthing(DEV_A) service = _service(tmp_path, fake) service.create_invite() fake.knock(DEV_C, offers=["some-other-folder"]) service.reconcile(fake) service.reconcile(fake) assert DEV_C not in fake.devices assert DEV_C in service.state().rejected fake.knock(DEV_C) # Syncthing lists it again… service.reconcile(fake) assert DEV_C not in fake.devices # …and it isn't probed again assert not service.state().friends def test_used_code_from_a_second_device_is_refused(self, qapp, tmp_path): fake = FakeSyncthing(DEV_A) service = _service(tmp_path, fake) token = inv.decode(service.create_invite()).token fake.knock(DEV_B, offers=[inv.invitee_folder(token)]) service.reconcile(fake) service.reconcile(fake) fake.knock(DEV_C, offers=[inv.invitee_folder(token)]) service.reconcile(fake) service.reconcile(fake) assert DEV_C not in fake.devices assert [f.device_id for f in service.state().friends.values()] == [DEV_B] def test_no_open_invite_means_no_probing(self, qapp, tmp_path): fake = FakeSyncthing(DEV_A) service = _service(tmp_path, fake) fake.knock(DEV_B) service.reconcile(fake) assert DEV_B not in fake.devices def test_paste_side(self, qapp, tmp_path): fake = FakeSyncthing(DEV_B) service = _service(tmp_path, fake, name="sam") code = inv.encode(DEV_A, "trav", "0123456789abcdef") friend = service.accept_code(code) assert friend.role == INVITEE and friend.name == "trav" assert fake.devices[DEV_A]["name"] == "trav" out = fake.folders[inv.invitee_folder("0123456789abcdef")] assert out["type"] == "sendonly" and out["devices"] == [{"deviceID": DEV_A}] with pytest.raises(inv.InviteError) as info: service.accept_code(code) assert str(info.value) == svc_mod.ALREADY_USED # The inviter shares -a back: accepted receive-only. fake.offers[inv.inviter_folder("0123456789abcdef")] = {DEV_A: {}} service.reconcile(fake) assert fake.folders[inv.inviter_folder("0123456789abcdef")]["type"] \ == "receiveonly" def test_own_code_is_refused(self, qapp, tmp_path): fake = FakeSyncthing(DEV_A) service = _service(tmp_path, fake) code = service.create_invite() with pytest.raises(inv.InviteError) as info: service.accept_code(code) assert str(info.value) == svc_mod.OWN_CODE def test_cancel_and_remove(self, qapp, tmp_path): fake = FakeSyncthing(DEV_A) service = _service(tmp_path, fake) token = inv.decode(service.create_invite()).token service.cancel_invite(token) assert service.state().invites[token].status == CANCELLED assert inv.inviter_folder(token) not in fake.folders token = inv.decode(service.create_invite()).token fake.knock(DEV_B, offers=[inv.invitee_folder(token)]) service.reconcile(fake) service.reconcile(fake) (in_dir(service.root, token) / "x").write_text("x") service.remove_friend(token) assert not service.state().friends assert DEV_B not in fake.devices assert not fake.folders.get(inv.invitee_folder(token)) assert not in_dir(service.root, token).exists() def test_connected_notice_once(self, qapp, tmp_path): fake = FakeSyncthing(DEV_A) service = _service(tmp_path, fake) notices = [] service.notice.connect(notices.append) token = inv.decode(service.create_invite()).token fake.knock(DEV_B, offers=[inv.invitee_folder(token)]) service.reconcile(fake) service.reconcile(fake) fake.connected.add(DEV_B) # Their intro arrives: the friend's display name replaces the hostname. publish.publish_intro(in_dir(service.root, token), "Sam", None) service.reconcile(fake) service.reconcile(fake) assert notices == ["Sam is now connected."] assert service.status(token).code == svc_mod.CONNECTED # ---- the real thing ---- def _wait(condition, services, timeout=120): deadline = time.time() + timeout while time.time() < deadline: for service, client in services: service.reconcile(client) if condition(): return True time.sleep(1) return False @pytest.mark.syncthing def test_real_handshake(qapp, tmp_path, syncthing_pair): alice, bob, carol = syncthing_pair listen = {i.device_id: [i.listen_address] for i in syncthing_pair} def make(instance): config = {st.API_KEY_KEY: instance.api_key, st.ADDRESS_KEY: instance.address} service = CassetteService(root=tmp_path / instance.name, config_loader=lambda: config, addresses_for=lambda d: listen[d]) service._state.display_name = instance.name.title() return service, service.client() a, a_client = make(alice) b, b_client = make(bob) c, c_client = make(carol) code = a.create_invite() b.accept_code(code) token = inv.decode(code).token def both_connected(): sa, sb = a.status(token), b.status(token) return (sa and sb and sa.code == svc_mod.CONNECTED and sb.code == svc_mod.CONNECTED) assert _wait(both_connected, [(a, a_client), (b, b_client)]), \ (a.status(token), b.status(token)) assert _wait(lambda: a.state().friends[token].name == "Bob" and b.state().friends[token].name == "Alice", [(a, a_client), (b, b_client)], timeout=60) assert a.state().invites[token].status == USED # Carol got hold of the same code. Alice has no open invite left, so # Carol's knock stays pending in Alice's Syncthing, untouched, and Carol # is told it may have been used. c.accept_code(code) assert _wait(lambda: carol.device_id in a_client.pending_devices(), [(a, a_client), (c, c_client)], timeout=90) for _ in range(3): a.reconcile(a_client) assert a_client.device(carol.device_id) is None assert len(a.state().friends) == 1 carol_status = c.status(token) assert carol_status.code == svc_mod.WAITING_COMPUTER assert "already have been used" in carol_status.hint # ---- the dialogs ---- class TestDialogs: def test_failure_text_names_step_and_next_move(self): from lintunes.gui.cassette_dialogs import failure_text down = failure_text(st.SyncthingError("Adding Sam's computer", "x", st.NOT_RUNNING)) assert down.startswith("Adding Sam's computer failed") and "Start" in down said = failure_text(st.SyncthingError("Sharing", "invalid device ID")) assert said == "Sharing failed. Syncthing said: invalid device ID" assert failure_text(inv.InviteError(inv.NOT_VALID)) == inv.NOT_VALID def test_add_friend_shows_bad_code_inline(self, qapp, tmp_path): from lintunes.gui.cassette_dialogs import AddFriendDialog service = _service(tmp_path, FakeSyncthing(DEV_B)) dialog = AddFriendDialog(service) dialog.field.setPlainText("hello there") dialog._connect() assert "whole code" in dialog.message.text() assert dialog.friend is None def test_add_friend_connects(self, qapp, tmp_path): from lintunes.gui.cassette_dialogs import AddFriendDialog service = _service(tmp_path, FakeSyncthing(DEV_B)) dialog = AddFriendDialog(service) dialog.field.setPlainText(inv.encode(DEV_A, "trav", "0123456789abcdef")) dialog._connect() assert dialog.friend.name == "trav" def test_share_dialog_copies(self, qapp): from PyQt6.QtGui import QGuiApplication from lintunes.gui.cassette_dialogs import ShareInviteDialog dialog = ShareInviteDialog("LTC1.abc") dialog._copy() assert QGuiApplication.clipboard().text() == "LTC1.abc" def test_settings_stage_cancel_and_remove_until_ok(self, qapp, tmp_path, monkeypatch): from PyQt6.QtWidgets import QMessageBox from lintunes.gui.sync_settings_dialog import SyncSettingsDialog from tests.test_round64 import _NoProbe fake = FakeSyncthing(DEV_A) service = _service(tmp_path, fake) kept = inv.decode(service.create_invite()).token friend_token = inv.decode(service.create_invite()).token fake.knock(DEV_B, offers=[inv.invitee_folder(friend_token)]) service.reconcile(fake) service.reconcile(fake) dialog = SyncSettingsDialog(service.state(), service.root, {}, probe=_NoProbe(), service=service) dialog.general._cancel(kept, service) monkeypatch.setattr(QMessageBox, "question", lambda *a, **k: QMessageBox.StandardButton.Yes) dialog.friend_pages[friend_token]._remove() # Nothing has happened yet… assert service.state().invites[kept].status == OPEN assert friend_token in service.state().friends # …until OK. service.apply_settings(dialog.state, dialog.cancelled_invites, dialog.removed_friends) assert service.state().invites[kept].status == CANCELLED assert friend_token not in service.state().friends def test_format_when(self): from lintunes.gui.sync_settings_dialog import format_when assert format_when("") == "never" assert format_when("0001-01-01T00:00:00Z") == "never" assert format_when(svc_mod.now_iso()).startswith("today ") assert "2020" in format_when("2020-03-04T05:06:07Z")