Possible bug fix for clipboard sync and handling.

This commit is contained in:
Storm Dragon committed 2026-10-10 18:08:50 -04:00
1 parent 1638f66a93
commit c018f5fd29
10 files changed
+225 -26

No files matched your search

+1
View File
@@ -271,6 +271,7 @@ clipboard_export_path=/tmp/fenrirClipboard
# Keep Fenrir's clipboard history and the X clipboard synchronized. # Keep Fenrir's clipboard history and the X clipboard synchronized.
# In fenrir -x, an empty sync_display uses the current DISPLAY. # In fenrir -x, an empty sync_display uses the current DISPLAY.
# In console/TTY mode, set sync_display explicitly, for example :0. # In console/TTY mode, set sync_display explicitly, for example :0.
# Manual graphical clipboard import/export also use sync_display when set.
sync_enabled=False sync_enabled=False
sync_display= sync_display=
sync_interval=0.5 sync_interval=0.5
@@ -9,9 +9,9 @@ from fenrirscreenreader.core.i18n import _
from fenrirscreenreader.utils import x_clipboard from fenrirscreenreader.utils import x_clipboard
def write_clipboard(clipboard): def write_clipboard(clipboard, display):
"""Write the graphical clipboard without accessing Fenrir state.""" """Write the graphical clipboard without accessing Fenrir state."""
return x_clipboard.write_text(clipboard, scan_displays=True) return x_clipboard.write_text(clipboard, display, scan_displays=True)
class command: class command:
@@ -47,6 +47,9 @@ class command:
clipboard = self.env["runtime"][ clipboard = self.env["runtime"][
"MemoryManager" "MemoryManager"
].get_index_list_element("clipboardHistory") ].get_index_list_element("clipboardHistory")
display = self.env["runtime"]["SettingsManager"].get_setting(
"clipboard", "sync_display"
)
task_id = self.env["runtime"][ task_id = self.env["runtime"][
"BackgroundTaskManager" "BackgroundTaskManager"
@@ -54,6 +57,7 @@ class command:
write_clipboard, write_clipboard,
lambda result: self._handle_result(clipboard, result), lambda result: self._handle_result(clipboard, result),
clipboard, clipboard,
display,
) )
if task_id is not None: if task_id is not None:
self._task_ids.add(task_id) self._task_ids.add(task_id)
@@ -8,9 +8,9 @@ from fenrirscreenreader.core.i18n import _
from fenrirscreenreader.utils import x_clipboard from fenrirscreenreader.utils import x_clipboard
def read_clipboard(): def read_clipboard(display):
"""Read the graphical clipboard without accessing Fenrir state.""" """Read the graphical clipboard without accessing Fenrir state."""
return x_clipboard.read_text(scan_displays=True) return x_clipboard.read_text(display, scan_displays=True)
class command: class command:
@@ -32,8 +32,11 @@ class command:
return _("imports the graphical clipboard to Fenrir's clipboard") return _("imports the graphical clipboard to Fenrir's clipboard")
def run(self): def run(self):
display = self.env["runtime"]["SettingsManager"].get_setting(
"clipboard", "sync_display"
)
task_id = self.env["runtime"]["BackgroundTaskManager"].submit_task( task_id = self.env["runtime"]["BackgroundTaskManager"].submit_task(
read_clipboard, self._handle_result read_clipboard, self._handle_result, display
) )
if task_id is not None: if task_id is not None:
self._task_ids.add(task_id) self._task_ids.add(task_id)
@@ -58,6 +61,9 @@ class command:
self.env["runtime"]["MemoryManager"].add_value_to_first_index( self.env["runtime"]["MemoryManager"].add_value_to_first_index(
"clipboardHistory", clipboard_content "clipboardHistory", clipboard_content
) )
sync_manager = self.env["runtime"].get("ClipboardSyncManager")
if sync_manager:
sync_manager.mark_imported_from_x(clipboard_content)
self.env["runtime"]["OutputManager"].present_text( self.env["runtime"]["OutputManager"].present_text(
_("Import to Clipboard"), _("Import to Clipboard"),
sound_icon="CopyToClipboard", sound_icon="CopyToClipboard",
@@ -43,8 +43,8 @@ def synchronize_clipboards(
result["error"] = ( result["error"] = (
"ClipboardSyncManager copy failed: " + str(error) "ClipboardSyncManager copy failed: " + str(error)
) )
result["last_observed_fenrir"] = fenrir_text
if written: if written:
result["last_observed_fenrir"] = fenrir_text
result["last_written_to_x"] = fenrir_text result["last_written_to_x"] = fenrir_text
result["last_observed_x"] = fenrir_text result["last_observed_x"] = fenrir_text
return result return result
@@ -72,6 +72,7 @@ class ClipboardSyncManager:
self.interval = 0.5 self.interval = 0.5
self.running = False self.running = False
self._task_id = None self._task_id = None
self._transfer_revision = 0
self._scheduler_stop = threading.Event() self._scheduler_stop = threading.Event()
self._sync_event_pending = threading.Event() self._sync_event_pending = threading.Event()
self._scheduler_thread = None self._scheduler_thread = None
@@ -106,10 +107,19 @@ class ClipboardSyncManager:
def mark_written_to_x(self, text): def mark_written_to_x(self, text):
if not isinstance(text, str) or not text: if not isinstance(text, str) or not text:
return return
self._transfer_revision += 1
self.last_written_to_x = text self.last_written_to_x = text
self.last_observed_fenrir = text self.last_observed_fenrir = text
self.last_observed_x = text self.last_observed_x = text
def mark_imported_from_x(self, text):
if not isinstance(text, str) or not text:
return
self._transfer_revision += 1
self.last_imported_from_x = text
self.last_observed_fenrir = text
self.last_observed_x = text
def _resolve_startup(self): def _resolve_startup(self):
settings_manager = self.env["runtime"]["SettingsManager"] settings_manager = self.env["runtime"]["SettingsManager"]
if not settings_manager.get_setting_as_bool( if not settings_manager.get_setting_as_bool(
@@ -173,11 +183,14 @@ class ClipboardSyncManager:
return return
fenrir_text = self._get_fenrir_clipboard_text() fenrir_text = self._get_fenrir_clipboard_text()
transfer_revision = self._transfer_revision
self._task_id = self.env["runtime"][ self._task_id = self.env["runtime"][
"BackgroundTaskManager" "BackgroundTaskManager"
].submit_task( ].submit_task(
synchronize_clipboards, synchronize_clipboards,
lambda result: self._handle_poll_result(fenrir_text, result), lambda result: self._handle_poll_result(
fenrir_text, transfer_revision, result
),
self.display, self.display,
fenrir_text, fenrir_text,
self.last_observed_fenrir, self.last_observed_fenrir,
@@ -186,7 +199,9 @@ class ClipboardSyncManager:
if self._task_id is None: if self._task_id is None:
self._sync_event_pending.clear() self._sync_event_pending.clear()
def _handle_poll_result(self, fenrir_snapshot, task_result): def _handle_poll_result(
self, fenrir_snapshot, transfer_revision, task_result
):
self._task_id = None self._task_id = None
self._sync_event_pending.clear() self._sync_event_pending.clear()
if not self.running: if not self.running:
@@ -201,10 +216,13 @@ class ClipboardSyncManager:
result = task_result.get("value") or {} result = task_result.get("value") or {}
if result.get("error"): if result.get("error"):
self._debug(result["error"]) self._debug(result["error"])
if (
self._get_fenrir_clipboard_text() != fenrir_snapshot
or transfer_revision != self._transfer_revision
):
return
import_text = result.get("import_text") import_text = result.get("import_text")
if import_text: if import_text:
if self._get_fenrir_clipboard_text() != fenrir_snapshot:
return
self.env["runtime"]["MemoryManager"].add_value_to_first_index( self.env["runtime"]["MemoryManager"].add_value_to_first_index(
"clipboardHistory", import_text "clipboardHistory", import_text
) )
@@ -239,7 +257,7 @@ class ClipboardSyncManager:
if self._write_x_clipboard_text(fenrir_text): if self._write_x_clipboard_text(fenrir_text):
self.last_written_to_x = fenrir_text self.last_written_to_x = fenrir_text
self.last_observed_x = fenrir_text self.last_observed_x = fenrir_text
self.last_observed_fenrir = fenrir_text self.last_observed_fenrir = fenrir_text
return return
if x_changed: if x_changed:
+1 -1
View File
@@ -4,5 +4,5 @@
# Fenrir TTY screen reader # Fenrir TTY screen reader
# By Chrys, Storm Dragon, and contributors. # By Chrys, Storm Dragon, and contributors.
version = "2026.10.05" version = "2026.10.10"
code_name = "testing" code_name = "testing"
+11 -8
View File
@@ -30,9 +30,9 @@ def _display_candidates(display="", scan_displays=False):
if display: if display:
return [display] return [display]
current_display = os.environ.get("DISPLAY", "") current_display = os.environ.get("DISPLAY", "")
candidates = []
if current_display: if current_display:
candidates.append(current_display) return [current_display]
candidates = []
if scan_displays: if scan_displays:
candidates.extend(f":{index}" for index in range(10)) candidates.extend(f":{index}" for index in range(10))
if not candidates: if not candidates:
@@ -98,7 +98,6 @@ def _command_error(command, result):
def read_text(display="", scan_displays=False): def read_text(display="", scan_displays=False):
last_error = None last_error = None
command_succeeded = False
with _display_lock: with _display_lock:
for candidate in _display_candidates(display, scan_displays): for candidate in _display_candidates(display, scan_displays):
commands = _read_commands(candidate) commands = _read_commands(candidate)
@@ -108,17 +107,21 @@ def read_text(display="", scan_displays=False):
try: try:
result = _run_command(command, candidate) result = _run_command(command, candidate)
if result.returncode != 0: if result.returncode != 0:
# xclip reports a missing text target for both an
# unowned selection and a non-text clipboard.
stderr = (result.stderr or b"").strip()
if command[0] == "xclip" and stderr in (
b"Error: target STRING not available",
b"Error: target UTF8_STRING not available",
):
return None
last_error = _command_error(command, result) last_error = _command_error(command, result)
continue continue
command_succeeded = True
text = result.stdout.decode(_ENCODING) text = result.stdout.decode(_ENCODING)
if text: return text or None
return text
except Exception as error: except Exception as error:
last_error = error last_error = error
if last_error: if last_error:
if command_succeeded:
return None
raise last_error raise last_error
return None return None
@@ -28,12 +28,15 @@ def test_clipboard_import_changes_history_only_after_result_event():
"BackgroundTaskManager": task_manager, "BackgroundTaskManager": task_manager,
"MemoryManager": memory_manager, "MemoryManager": memory_manager,
"OutputManager": Mock(), "OutputManager": Mock(),
"SettingsManager": Mock(get_setting=Mock(return_value=":2")),
"ClipboardSyncManager": Mock(),
} }
} }
command = import_clipboard_from_x.command() command = import_clipboard_from_x.command()
command.initialize(environment) command.initialize(environment)
command.run() command.run()
assert task_manager.args == (":2",)
memory_manager.add_value_to_first_index.assert_not_called() memory_manager.add_value_to_first_index.assert_not_called()
task_manager.callback( task_manager.callback(
@@ -42,6 +45,8 @@ def test_clipboard_import_changes_history_only_after_result_event():
memory_manager.add_value_to_first_index.assert_called_once_with( memory_manager.add_value_to_first_index.assert_called_once_with(
"clipboardHistory", "copied text" "clipboardHistory", "copied text"
) )
sync_manager = environment["runtime"]["ClipboardSyncManager"]
sync_manager.mark_imported_from_x.assert_called_once_with("copied text")
def test_clipboard_export_updates_sync_state_only_after_result_event(): def test_clipboard_export_updates_sync_state_only_after_result_event():
@@ -58,15 +63,29 @@ def test_clipboard_export_updates_sync_state_only_after_result_event():
"DebugManager": Mock(), "DebugManager": Mock(),
"MemoryManager": memory_manager, "MemoryManager": memory_manager,
"OutputManager": Mock(), "OutputManager": Mock(),
"SettingsManager": Mock(get_setting=Mock(return_value=":2")),
} }
} }
command = export_clipboard_to_x.command() command = export_clipboard_to_x.command()
command.initialize(environment) command.initialize(environment)
command.run() command.run()
assert task_manager.args == ("copied text", ":2")
sync_manager.mark_written_to_x.assert_not_called() sync_manager.mark_written_to_x.assert_not_called()
task_manager.callback( task_manager.callback(
{"task_id": 7, "succeeded": True, "value": True} {"task_id": 7, "succeeded": True, "value": True}
) )
sync_manager.mark_written_to_x.assert_called_once_with("copied text") sync_manager.mark_written_to_x.assert_called_once_with("copied text")
def test_manual_clipboard_workers_use_configured_display(monkeypatch):
read_text = Mock(return_value="text")
write_text = Mock(return_value=True)
monkeypatch.setattr(import_clipboard_from_x.x_clipboard, "read_text", read_text)
monkeypatch.setattr(export_clipboard_to_x.x_clipboard, "write_text", write_text)
assert import_clipboard_from_x.read_clipboard(":2") == "text"
assert export_clipboard_to_x.write_clipboard("text", ":2") is True
read_text.assert_called_once_with(":2", scan_displays=True)
write_text.assert_called_once_with("text", ":2", scan_displays=True)
+96
View File
@@ -355,3 +355,99 @@ def test_sync_worker_returns_changes_without_manager_access(monkeypatch):
assert result["import_text"] == "from x" assert result["import_text"] == "from x"
assert result["last_observed_x"] == "from x" assert result["last_observed_x"] == "from x"
def run_sync_task(manager):
task_manager = manager.env['runtime']['BackgroundTaskManager']
manager.handle_sync_event()
function, callback, *args = task_manager.submit_task.call_args.args
callback({'task_id': 12, 'succeeded': True, 'value': function(*args)})
@pytest.mark.unit
@pytest.mark.parametrize('failure', [False, RuntimeError('temporary failure')])
def test_runtime_sync_retries_failed_exports(monkeypatch, failure):
manager = ClipboardSyncManager()
manager.env = build_env(fenrir_text='new copy')
manager.env['runtime']['BackgroundTaskManager'] = Mock(
submit_task=Mock(return_value=12),
)
manager.running = True
manager.display = ':1'
manager.last_observed_fenrir = 'previous copy'
manager.last_observed_x = 'previous copy'
write_text = Mock(side_effect=[failure, True])
monkeypatch.setattr(
'fenrirscreenreader.core.clipboardSyncManager.x_clipboard.read_text',
Mock(return_value='previous copy'),
)
monkeypatch.setattr(
'fenrirscreenreader.core.clipboardSyncManager.x_clipboard.write_text',
write_text,
)
run_sync_task(manager)
assert manager.last_observed_fenrir == 'previous copy'
run_sync_task(manager)
assert write_text.call_count == 2
assert manager.last_observed_fenrir == 'new copy'
assert manager.last_observed_x == 'new copy'
manager.env['runtime']['MemoryManager'].add_value_to_first_index.assert_not_called()
@pytest.mark.unit
def test_poll_once_retries_failed_export(monkeypatch):
manager = ClipboardSyncManager()
manager.env = build_env(fenrir_text='new copy')
manager.display = ':1'
write_text = Mock(side_effect=[False, True])
monkeypatch.setattr(
'fenrirscreenreader.core.clipboardSyncManager.x_clipboard.read_text',
Mock(return_value=None),
)
monkeypatch.setattr(
'fenrirscreenreader.core.clipboardSyncManager.x_clipboard.write_text',
write_text,
)
manager.poll_once()
assert manager.last_observed_fenrir is None
manager.poll_once()
assert write_text.call_count == 2
assert manager.last_written_to_x == 'new copy'
@pytest.mark.unit
@pytest.mark.parametrize('manual_action', ['export', 'import'])
def test_runtime_sync_discards_result_superseded_by_manual_transfer(
manual_action,
):
manager = ClipboardSyncManager()
manager.env = build_env(fenrir_text='current copy')
task_manager = Mock(submit_task=Mock(return_value=12))
manager.env['runtime']['BackgroundTaskManager'] = task_manager
manager.running = True
manager.last_observed_fenrir = 'current copy'
manager.last_observed_x = 'current copy'
manager.handle_sync_event()
if manual_action == 'export':
manager.mark_written_to_x('current copy')
else:
manager.mark_imported_from_x('current copy')
callback = task_manager.submit_task.call_args.args[1]
callback({
'task_id': 12, 'succeeded': True,
'value': {
'import_text': 'old x copy',
'last_imported_from_x': 'old x copy',
'last_observed_fenrir': 'old x copy',
'last_observed_x': 'old x copy',
},
})
manager.env['runtime']['MemoryManager'].add_value_to_first_index.assert_not_called()
assert manager.last_observed_x == 'current copy'
assert manager._task_id is None
assert not manager._sync_event_pending.is_set()
+52
View File
@@ -107,3 +107,55 @@ def test_write_text_propagates_clipboard_timeout(monkeypatch):
with pytest.raises(subprocess.TimeoutExpired): with pytest.raises(subprocess.TimeoutExpired):
x_clipboard.write_text("clipboard text", ":3") x_clipboard.write_text("clipboard text", ":3")
@pytest.mark.unit
@pytest.mark.parametrize("operation", ["read", "write"])
def test_current_display_failure_does_not_scan_other_sessions(
monkeypatch, operation,
):
monkeypatch.setenv("DISPLAY", ":0")
monkeypatch.setattr(x_clipboard.shutil, "which", command_exists)
run_command = Mock(return_value=subprocess.CompletedProcess(
["xclip"], 1, stdout=b"", stderr=b"Can't open display: :0",
))
monkeypatch.setattr(x_clipboard.subprocess, "run", run_command)
with pytest.raises(RuntimeError):
if operation == "read":
x_clipboard.read_text(scan_displays=True)
else:
x_clipboard.write_text("text", scan_displays=True)
assert run_command.call_count == 1
assert run_command.call_args.kwargs["env"]["DISPLAY"] == ":0"
@pytest.mark.unit
def test_empty_clipboard_stops_display_scan(monkeypatch):
monkeypatch.delenv("DISPLAY", raising=False)
monkeypatch.setattr(x_clipboard.shutil, "which", command_exists)
run_command = Mock(return_value=subprocess.CompletedProcess(
["xclip"], 0, stdout=b"", stderr=b"",
))
monkeypatch.setattr(x_clipboard.subprocess, "run", run_command)
assert x_clipboard.read_text(scan_displays=True) is None
assert run_command.call_count == 1
@pytest.mark.unit
@pytest.mark.parametrize("target", ["STRING", "UTF8_STRING"])
def test_unavailable_text_selection_is_an_empty_clipboard(
monkeypatch, target,
):
monkeypatch.setenv("DISPLAY", ":0")
monkeypatch.setattr(x_clipboard.shutil, "which", command_exists)
run_command = Mock(return_value=subprocess.CompletedProcess(
["xclip"], 1, stdout=b"",
stderr=f"Error: target {target} not available\n".encode(),
))
monkeypatch.setattr(x_clipboard.subprocess, "run", run_command)
assert x_clipboard.read_text(scan_displays=True) is None
assert run_command.call_count == 1
+6 -6
View File
@@ -8,7 +8,7 @@ msgid ""
msgstr "" msgstr ""
"Project-Id-Version: fenrir\n" "Project-Id-Version: fenrir\n"
"Report-Msgid-Bugs-To: \n" "Report-Msgid-Bugs-To: \n"
"POT-Creation-Date: 2026-10-05 12:36-0400\n" "POT-Creation-Date: 2026-10-10 18:03-0400\n"
"PO-Revision-Date: YEAR-MO-DA HO:MI+ZONE\n" "PO-Revision-Date: YEAR-MO-DA HO:MI+ZONE\n"
"Last-Translator: FULL NAME <EMAIL@ADDRESS>\n" "Last-Translator: FULL NAME <EMAIL@ADDRESS>\n"
"Language-Team: LANGUAGE <LL@li.org>\n" "Language-Team: LANGUAGE <LL@li.org>\n"
@@ -456,11 +456,11 @@ msgstr ""
msgid "Export current fenrir clipboard to X or GUI clipboard" msgid "Export current fenrir clipboard to X or GUI clipboard"
msgstr "" msgstr ""
#: src/fenrirscreenreader/commands/commands/export_clipboard_to_x.py:73 #: src/fenrirscreenreader/commands/commands/export_clipboard_to_x.py:77
msgid "exported to the X session." msgid "exported to the X session."
msgstr "" msgstr ""
#: src/fenrirscreenreader/commands/commands/export_clipboard_to_x.py:84 #: src/fenrirscreenreader/commands/commands/export_clipboard_to_x.py:88
msgid "failed to export to X clipboard. No available display found." msgid "failed to export to X clipboard. No available display found."
msgstr "" msgstr ""
@@ -485,7 +485,7 @@ msgid "File does not exist"
msgstr "" msgstr ""
#: src/fenrirscreenreader/commands/commands/import_clipboard_from_file.py:52 #: src/fenrirscreenreader/commands/commands/import_clipboard_from_file.py:52
#: src/fenrirscreenreader/commands/commands/import_clipboard_from_x.py:62 #: src/fenrirscreenreader/commands/commands/import_clipboard_from_x.py:68
msgid "Import to Clipboard" msgid "Import to Clipboard"
msgstr "" msgstr ""
@@ -493,11 +493,11 @@ msgstr ""
msgid "imports the graphical clipboard to Fenrir's clipboard" msgid "imports the graphical clipboard to Fenrir's clipboard"
msgstr "" msgstr ""
#: src/fenrirscreenreader/commands/commands/import_clipboard_from_x.py:46 #: src/fenrirscreenreader/commands/commands/import_clipboard_from_x.py:49
msgid "Clipboard import failed" msgid "Clipboard import failed"
msgstr "" msgstr ""
#: src/fenrirscreenreader/commands/commands/import_clipboard_from_x.py:53 #: src/fenrirscreenreader/commands/commands/import_clipboard_from_x.py:56
msgid "No text found in clipboard or no accessible display" msgid "No text found in clipboard or no accessible display"
msgstr "" msgstr ""