From fc2878f81ec935ed15bc8dd9e52940689627ea42 Mon Sep 17 00:00:00 2001 From: SinghCod3r Date: Thu, 8 Oct 2026 17:45:34 +0530 Subject: [PATCH 1/6] Move driverless PPD validation off GTK thread --- authconn.py | 71 +++++++++++++++++++++++++++++++++++++++++++- newprinter.py | 16 ++++++++++ test_newprinter.py | 74 ++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 160 insertions(+), 1 deletion(-) diff --git a/authconn.py b/authconn.py index 777c16a11..13de2b948 100644 --- a/authconn.py +++ b/authconn.py @@ -18,6 +18,7 @@ ## Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. import threading +import queue import config import cups import cupspk @@ -231,6 +232,71 @@ def _using_polkit (self): def _make_binding (self, fname, fn): return lambda *args, **kwds: self._authloop (fname, fn, *args, **kwds) + def _execute_in_worker(self, fname, *args, **kwds): + if not hasattr(self, '_worker_queue'): + self._worker_queue = queue.Queue() + self._worker_thread = threading.Thread(target=self._worker_loop, daemon=True) + self._worker_thread.start() + + box = [] + self._worker_queue.put((fname, args, kwds, box)) + + loop = GLib.MainLoop() + source_id = [None] + def poll(): + if box: + if loop.is_running(): + loop.quit() + source_id[0] = None + return False + return True + + source_id[0] = GLib.timeout_add(20, poll) + loop.run() + if source_id[0] is not None: + GLib.source_remove(source_id[0]) + + ok, res, exc = box[0] + if not ok: + raise exc + return res + + def _worker_loop(self): + def worker_password_cb(prompt): + debugprint("Got worker password callback") + if getattr(self, '_cancel', False) or getattr(self, '_auth_called', False): + return '' + self._auth_called = True + self._prompt = prompt + return getattr(self, '_use_password', '') + + cups.setPasswordCB(worker_password_cb) + conn = None + + while True: + item = self._worker_queue.get() + if item is None: + break + fname, args, kwds, box = item + + try: + if conn is None: + conn = cups.Connection(host=self._server, + port=self._port, + encryption=self._encryption) + + cups.setUser(self._use_user) + fn = getattr(conn, fname) + res = fn(*args, **kwds) + + box.append((True, res, None)) + except Exception as e: + box.append((False, None, e)) + finally: + self._worker_queue.task_done() + + cups.setPasswordCB(None) + def _authloop (self, fname, fn, *args, **kwds): self._passes = 0 # remove signature if dbus is not being used and signature is provided @@ -251,7 +317,10 @@ def _authloop (self, fname, fn, *args, **kwds): cups.setUser (self._use_user) - result = fn.__call__ (*args, **kwds) + if fname == 'getServerPPD': + result = self._execute_in_worker(fname, *args, **kwds) + else: + result = fn.__call__ (*args, **kwds) if fname == 'adminGetServerSettings': # Special case for a rubbish bit of API. diff --git a/newprinter.py b/newprinter.py index 6b6b35236..fbb675bf3 100644 --- a/newprinter.py +++ b/newprinter.py @@ -1978,6 +1978,14 @@ def _validateDriverlessPPD(self, ppdname): self.cups._begin_operation(_("validating driverless PPD")) try: f = self.cups.getServerPPD(ppdname) + + if getattr(self, 'device', None) is None: + try: + os.unlink(f) + except OSError: + pass + return None + try: ppd = cups.PPD(f) return ppd @@ -2088,6 +2096,10 @@ def _installPrinterFromDeviceID (self, devid, page_nr, step): if getattr(self.device, 'driverless', False): ppdname = "driverless:%s" % self.device.uri validated_ppd = self._validateDriverlessPPD(ppdname) + + if getattr(self, 'device', None) is None: + return + if validated_ppd is None: debugprint("Driverless PPD validation failed; abandoning driverless mode completely") self.device.driverless = False @@ -2195,6 +2207,10 @@ def _installPrinterFromDeviceID (self, devid, page_nr, step): else: while ppdnamelist and isinstance(ppdnamelist[0], str) and ppdnamelist[0].startswith("driverless:"): validated_ppd = self._validateDriverlessPPD(ppdnamelist[0]) + + if getattr(self, 'device', None) is None: + return + if validated_ppd is None: debugprint("Driverless PPD validation failed; abandoning driverless mode completely") self.device.driverless = False diff --git a/test_newprinter.py b/test_newprinter.py index b249dab12..288eb2561 100644 --- a/test_newprinter.py +++ b/test_newprinter.py @@ -2869,3 +2869,77 @@ def test_duplicate_state_two_uses_existing_ppd_loader(): np._loadPPDsForDevice.assert_called_once() assert result == newprinter.NewPrinterGUI.INSTALL_RESULT_OPS_PENDING + +def test_regression_getServerPPD_cancellation_real_v2_race(monkeypatch): + """ + Test the V2 worker + nested GLib + on_NPCancel() production path. + Verifies that real cancellation respects the worker thread lifecycle + and does not trigger AttributeError when the late worker completes. + """ + import authconn + import threading + from gi.repository import GLib + + np = get_dummy_gui() + np.device = cupshelpers.Device( + "ipp://test", + **{ + "device-info": "driverless", + "device-id": "MFG:Test;MDL:Printer;", + "device-make-and-model": "Test Printer", + } + ) + np.device.driverless = True + + np.NewPrinterWindow = MagicMock() + np.emit = MagicMock() + + np.cups = authconn.Connection(host='127.0.0.1', port=631, encryption=cups.HTTP_ENCRYPT_NEVER) + np.cups._begin_operation = MagicMock() + np.cups._end_operation = MagicMock() + + worker_blocked = threading.Event() + cancel_completed = threading.Event() + + main_thread_id = threading.get_ident() + worker_thread_id = [None] + + class SlowSpyConnection: + def __init__(self, *args, **kwargs): + pass + + def getServerPPD(self, ppdname): + worker_thread_id[0] = threading.get_ident() + worker_blocked.set() + + if not cancel_completed.wait(timeout=5.0): + raise RuntimeError("Test timeout: GLib callback did not complete cancellation") + + return "/tmp/fake_ppd" + + def setClientName(self, name): + pass + + monkeypatch.setattr(cups, "Connection", SlowSpyConnection) + monkeypatch.setattr(cups, "PPD", MagicMock()) + + def fire_cancel(): + if not worker_blocked.is_set(): + return True + + real_cancel = newprinter.NewPrinterGUI.on_NPCancel.__get__(np) + real_cancel(None) + + cancel_completed.set() + return False + + GLib.idle_add(fire_cancel) + + np._validateDriverlessPPD = newprinter.NewPrinterGUI._validateDriverlessPPD.__get__(np) + real_install = newprinter.NewPrinterGUI._installPrinterFromDeviceID.__get__(np) + + real_install(None, 0, 1) + + assert worker_thread_id[0] is not None, "Worker never executed" + assert worker_thread_id[0] != main_thread_id, "Worker executed on the main thread" + assert np.device is None, "on_NPCancel failed to clear device state" From 87c4519c34395e55a209f32b2fb6123c7e3aa9b6 Mon Sep 17 00:00:00 2001 From: SinghCod3r Date: Sat, 10 Oct 2026 12:27:49 +0530 Subject: [PATCH 2/6] Handle driverless PPD worker timeouts --- authconn.py | 80 ++++++++++++++++-- test_newprinter.py | 206 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 278 insertions(+), 8 deletions(-) diff --git a/authconn.py b/authconn.py index 13de2b948..7c7cf2fbb 100644 --- a/authconn.py +++ b/authconn.py @@ -19,6 +19,8 @@ import threading import queue +import time + import config import cups import cupspk @@ -34,6 +36,11 @@ import gettext gettext.install(domain=config.PACKAGE, localedir=config.localedir) N_ = lambda x: x +class WorkerTimeoutError(RuntimeError): + pass + +class WorkerUnavailableError(RuntimeError): + pass cups.require("1.9.60") class AuthDialog(Gtk.Dialog): @@ -233,18 +240,49 @@ def _make_binding (self, fname, fn): return lambda *args, **kwds: self._authloop (fname, fn, *args, **kwds) def _execute_in_worker(self, fname, *args, **kwds): + timeout_val = kwds.pop('_timeout', 30.0) + if not hasattr(self, '_worker_queue'): + self._worker_lock = threading.Lock() + self._worker_state = 'IDLE' self._worker_queue = queue.Queue() self._worker_thread = threading.Thread(target=self._worker_loop, daemon=True) self._worker_thread.start() + with self._worker_lock: + if getattr(self, '_worker_state', 'IDLE') != 'IDLE': + raise WorkerUnavailableError("CUPS worker is unavailable or busy") + self._worker_state = 'BUSY' + box = [] - self._worker_queue.put((fname, args, kwds, box)) + req_state = {'completed': False, 'timed_out': False} + self._worker_queue.put((fname, args, kwds, box, req_state)) loop = GLib.MainLoop() source_id = [None] + start_time = time.monotonic() + def poll(): - if box: + with self._worker_lock: + is_completed = req_state['completed'] + + if is_completed: + if loop.is_running(): + loop.quit() + source_id[0] = None + return False + + if time.monotonic() - start_time > timeout_val: + with self._worker_lock: + if req_state['completed']: + if loop.is_running(): + loop.quit() + source_id[0] = None + return False + + self._worker_state = 'UNAVAILABLE' + req_state['timed_out'] = True + if loop.is_running(): loop.quit() source_id[0] = None @@ -252,9 +290,16 @@ def poll(): return True source_id[0] = GLib.timeout_add(20, poll) - loop.run() - if source_id[0] is not None: - GLib.source_remove(source_id[0]) + + try: + loop.run() + finally: + if source_id[0] is not None: + GLib.source_remove(source_id[0]) + source_id[0] = None + + if req_state['timed_out']: + raise WorkerTimeoutError("CUPS worker timeout") ok, res, exc = box[0] if not ok: @@ -277,7 +322,7 @@ def worker_password_cb(prompt): item = self._worker_queue.get() if item is None: break - fname, args, kwds, box = item + fname, args, kwds, box, req_state = item try: if conn is None: @@ -289,10 +334,29 @@ def worker_password_cb(prompt): fn = getattr(conn, fname) res = fn(*args, **kwds) - box.append((True, res, None)) + res_tuple = (True, res, None) except Exception as e: - box.append((False, None, e)) + res_tuple = (False, None, e) finally: + with self._worker_lock: + if req_state['timed_out']: + if self._worker_state == 'UNAVAILABLE': + conn = None + else: + req_state['completed'] = True + box.append(res_tuple) + self._worker_state = 'IDLE' + + # Clear references to avoid retaining connection or data + fn = None + res = None + res_tuple = None + item = None + args = None + kwds = None + box = None + req_state = None + self._worker_queue.task_done() cups.setPasswordCB(None) diff --git a/test_newprinter.py b/test_newprinter.py index 288eb2561..ebb519947 100644 --- a/test_newprinter.py +++ b/test_newprinter.py @@ -2943,3 +2943,209 @@ def fire_cancel(): assert worker_thread_id[0] is not None, "Worker never executed" assert worker_thread_id[0] != main_thread_id, "Worker executed on the main thread" assert np.device is None, "on_NPCancel failed to clear device state" +def test_regression_worker_timeout_races(monkeypatch): + import authconn + import threading + import time + from gi.repository import GLib + + np = get_dummy_gui() + np.device = cupshelpers.Device( + "ipp://test", + **{ + "device-info": "driverless", + "device-id": "MFG:Test;MDL:Printer;", + "device-make-and-model": "Test Printer", + } + ) + np.device.driverless = True + + np.NewPrinterWindow = MagicMock() + np.emit = MagicMock() + + np.cups = authconn.Connection(host='127.0.0.1', port=631, encryption=cups.HTTP_ENCRYPT_NEVER) + np.cups._begin_operation = MagicMock() + np.cups._end_operation = MagicMock() + np.cups._cancel = False + + worker_block_event = threading.Event() + worker_can_finish = threading.Event() + conn_created = [0] + + class MasterSpyConnection: + def __init__(self, *args, **kwargs): + conn_created[0] += 1 + + def getServerPPD(self, ppdname): + if ppdname == "driverless:timeout": + worker_block_event.set() + worker_can_finish.wait() + return "/tmp/fake_ppd" + elif ppdname == "driverless:valueerror": + raise ValueError("Intentional worker exception") + elif ppdname == "driverless:authfail": + raise cups.IPPError(cups.IPP_NOT_AUTHORIZED, 'auth') + elif ppdname == "driverless:cancel": + worker_block_event.set() + worker_can_finish.wait() + return "/tmp/fake_ppd" + elif ppdname == "driverless:race": + worker_block_event.set() + worker_can_finish.wait() + return "/tmp/fake_ppd" + else: + return "/tmp/fake_ppd" + + def setClientName(self, name): + pass + + monkeypatch.setattr(cups, "Connection", MasterSpyConnection) + monkeypatch.setattr(cups, "PPD", MagicMock()) + np._validateDriverlessPPD = newprinter.NewPrinterGUI._validateDriverlessPPD.__get__(np) + + original_execute = np.cups._execute_in_worker + def execute_with_timeout(fname, *args, **kwds): + kwds.setdefault('_timeout', 0.1) + return original_execute(fname, *args, **kwds) + np.cups._execute_in_worker = execute_with_timeout + + # Initialize worker by doing a normal request first + res = np.cups._execute_in_worker('getServerPPD', 'driverless:normal', _timeout=2.0) + assert res == "/tmp/fake_ppd" + + worker_finished = threading.Event() + original_task_done = np.cups._worker_queue.task_done + + def mock_task_done(): + original_task_done() + worker_finished.set() + + # Wrap the test body in try/finally to ensure background threads don't hang the suite on failure + try: + np.cups._worker_queue.task_done = mock_task_done + + + + # 2. Blocked worker & Fail-fast + worker_block_event.clear() + worker_can_finish.clear() + worker_finished.clear() + try: + np.cups._execute_in_worker('getServerPPD', 'driverless:timeout', _timeout=0.1) + assert False, "Should have timed out" + except authconn.WorkerTimeoutError: + pass + + assert worker_block_event.is_set(), "Worker should be blocked" + + try: + np.cups._execute_in_worker('getServerPPD', 'driverless:normal') + assert False, "Should have raised WorkerUnavailableError immediately" + except authconn.WorkerUnavailableError as e: + assert "unavailable or busy" in str(e) + + # Wait for IDLE deterministically + worker_can_finish.set() # Let it recover + worker_finished.wait(2.0) + + with np.cups._worker_lock: + assert np.cups._worker_state == 'IDLE', "Did not recover to IDLE" + + # 3. Late result isolation (Worker finishes after timeout has already triggered) + worker_block_event.clear() + worker_can_finish.clear() + worker_finished.clear() + + try: + np.cups._execute_in_worker('getServerPPD', 'driverless:timeout', _timeout=0.1) + assert False, "Should have timed out" + except authconn.WorkerTimeoutError: + pass + + # Main thread has exited timeout loop. Unblock worker. + worker_can_finish.set() + worker_finished.wait(2.0) + + assert np.cups._worker_state == 'IDLE', "Worker must be IDLE after late completion" + + # 4. Recovery safe + conn_count_before = conn_created[0] + res = np.cups._execute_in_worker('getServerPPD', 'driverless:normal', _timeout=2.0) + assert res == "/tmp/fake_ppd" + assert conn_created[0] > conn_count_before, "Connection must have been recreated" + + # 5. Auth retry behavior through _authloop() + auth_called = [False] + def mock_perform_auth(*args, **kwds): + auth_called[0] = True + np.cups._cancel = True + np.cups._cannot_auth = False + return -1 # Cancel further retry + np.cups._perform_authentication = mock_perform_auth + + try: + np.cups.getServerPPD('driverless:authfail') + except cups.IPPError: + pass + assert auth_called[0], "Authloop should have triggered _perform_authentication" + + # 6. Cancellation during GLib loop + worker_block_event.clear() + worker_can_finish.clear() + def fire_cancel(): + if worker_block_event.wait(0.5): + np.on_NPCancel(None) + worker_can_finish.set() + return False + GLib.idle_add(fire_cancel) + + res = np._validateDriverlessPPD('driverless:cancel') + assert res is None, "Should discard late result due to cancellation" + assert np.device is None + assert np.cups._worker_state == 'IDLE' + + # 7. Unavoidable Race Window (Worker finishes while main thread is evaluating timeout) + worker_block_event.clear() + worker_can_finish.clear() + worker_finished.clear() + + original_monotonic = time.monotonic + def mock_monotonic(): + if worker_block_event.is_set() and not worker_can_finish.is_set(): + # We have entered the poll() evaluation and the worker is blocked on the CUPS call. + # Unblock the worker and wait for it to finish publishing the result. + worker_can_finish.set() + worker_finished.wait(2.0) + # Advance the clock to guarantee the timeout condition evaluates to True. + return original_monotonic() + 100.0 + return original_monotonic() + + monkeypatch.setattr(authconn.time, "monotonic", mock_monotonic) + + res_race = np.cups._execute_in_worker('getServerPPD', 'driverless:race', _timeout=0.1) + assert res_race == "/tmp/fake_ppd", "Must succeed because worker finished during the poll evaluation" + assert np.cups._worker_state == 'IDLE' + + # 8. Transient BUSY state + res1 = np.cups._execute_in_worker('getServerPPD', 'driverless:normal', _timeout=2.0) + assert res1 == "/tmp/fake_ppd" + res2 = np.cups._execute_in_worker('getServerPPD', 'driverless:normal', _timeout=2.0) + assert res2 == "/tmp/fake_ppd" + + # 9. Exception propagation + try: + np.cups._execute_in_worker('getServerPPD', 'driverless:valueerror', _timeout=2.0) + assert False, "Should have raised ValueError" + except ValueError as e: + assert "Intentional worker exception" in str(e) + + assert np.cups._worker_state == 'IDLE', "Worker must be IDLE after exception" + + res_after_err = np.cups._execute_in_worker('getServerPPD', 'driverless:normal', _timeout=2.0) + assert res_after_err == "/tmp/fake_ppd" + + finally: + worker_can_finish.set() + worker_block_event.set() + worker_finished.set() + np.cups._worker_queue.task_done = original_task_done From 9b688f349592452f0307e3cb44742d96f82b9ed1 Mon Sep 17 00:00:00 2001 From: SinghCod3r Date: Sun, 11 Oct 2026 12:38:04 +0530 Subject: [PATCH 3/6] refactor: prevent GTK stalls during printer operations --- authconn.py | 27 +- newprinter.py | 71 +++-- ppdsloader.py | 137 +++++++--- system-config-printer.py | 163 +++++++++--- test_newprinter.py | 556 ++++++++++++++++++++++++++++++++++++++- 5 files changed, 848 insertions(+), 106 deletions(-) diff --git a/authconn.py b/authconn.py index 7c7cf2fbb..fdd2b010e 100644 --- a/authconn.py +++ b/authconn.py @@ -239,6 +239,20 @@ def _using_polkit (self): def _make_binding (self, fname, fn): return lambda *args, **kwds: self._authloop (fname, fn, *args, **kwds) + def run_on_worker (self, op): + """ + Returns a callable wrapper that executes the specified CUPS operation + on the dedicated background worker thread while pumping the GLib + main loop. + + :param op: Either a method name string (e.g. 'getServerPPD', 'getPPD') + or a callable accepting the worker-owned cups.Connection + as its first parameter: op(conn, *args, **kwds). + """ + def wrapper (*args, **kwds): + return self._execute_in_worker (op, *args, **kwds) + return wrapper + def _execute_in_worker(self, fname, *args, **kwds): timeout_val = kwds.pop('_timeout', 30.0) @@ -331,8 +345,12 @@ def worker_password_cb(prompt): encryption=self._encryption) cups.setUser(self._use_user) - fn = getattr(conn, fname) - res = fn(*args, **kwds) + if callable(fname): + fn = None + res = fname(conn, *args, **kwds) + else: + fn = getattr(conn, fname) + res = fn(*args, **kwds) res_tuple = (True, res, None) except Exception as e: @@ -356,6 +374,7 @@ def worker_password_cb(prompt): kwds = None box = None req_state = None + fname = None self._worker_queue.task_done() @@ -381,8 +400,8 @@ def _authloop (self, fname, fn, *args, **kwds): cups.setUser (self._use_user) - if fname == 'getServerPPD': - result = self._execute_in_worker(fname, *args, **kwds) + if fname in ('getServerPPD', 'getPPD'): + result = self.run_on_worker(fname)(*args, **kwds) else: result = fn.__call__ (*args, **kwds) diff --git a/newprinter.py b/newprinter.py index fbb675bf3..a1725de33 100644 --- a/newprinter.py +++ b/newprinter.py @@ -898,10 +898,35 @@ def _initialiseDuplicateMode (self): self._initialiseWidgetsForMode ("duplicate") if getattr(self, 'discovered_device', None) is not None: self.device = self.discovered_device - if "driverless" in self.device.info: + if getattr(self.device, 'make_and_model', '') and "driverless" in self.device.make_and_model: + self.device.driverless = True + elif "driverless" in self.device.info: self.device.driverless = True else: self._initialiseDeviceFromURI () + + # Fallback to reconstructing device metadata from the existing PPD + # so duplicate mode can match drivers as accurately as add printer mode + if self.orig_ppd and getattr(self, 'device', None): + if not getattr(self.device, 'id', ''): + attr = self.orig_ppd.findAttr("1284DeviceID") + if attr and attr.value: + self.device.id = attr.value + self.device.id_dict = cupshelpers.parseDeviceID(self.device.id) + + if not getattr(self.device, 'make_and_model', ''): + attr = self.orig_ppd.findAttr("ModelName") + if attr and attr.value: + self.device.make_and_model = attr.value + + if not getattr(self.device, 'driverless', False): + # A driverless PPD might have "driverless" in its NickName or ShortNickName + for attr_name in ("ShortNickName", "NickName", "ModelName"): + attr = self.orig_ppd.findAttr(attr_name) + if attr and attr.value and "driverless" in attr.value.lower(): + self.device.driverless = True + break + if self.orig_ppd: self.ppd = self.orig_ppd self.entNPDriver.set_text ( @@ -1049,29 +1074,31 @@ def on_ppdsloader_finished_next (self, ppdsloader): and dialog_mode is one of the printer creation modes. """ - self._getPPDs_reply (ppdsloader) - if not self.ppds: - self._set_driver_ppd_loading(False) - self.setNPButtons() - return - - if ppdsloader._jockey_has_answered: - self.searchedfordriverpackages = True + try: + self._getPPDs_reply (ppdsloader) + if not self.ppds: + self._set_driver_ppd_loading(False) + self.setNPButtons() + return - debugprint ("Loaded PPDs this time; try nextNPTab again...") - self.nextnptab_rerun = True - if self.ntbkNewPrinter.get_current_page () in (self.PAGE_DESCRIBE_PRINTER, - self.PAGE_SELECT_INSTALL_METHOD): - self.ntbkNewPrinter.set_current_page (self.PAGE_SELECT_INSTALL_METHOD) - self.nextNPTab (step = 0) - else: - self.nextNPTab () - self._set_driver_ppd_loading(False) + if ppdsloader._jockey_has_answered: + self.searchedfordriverpackages = True + + debugprint ("Loaded PPDs this time; try nextNPTab again...") + self.nextnptab_rerun = True + if self.ntbkNewPrinter.get_current_page () in (self.PAGE_DESCRIBE_PRINTER, + self.PAGE_SELECT_INSTALL_METHOD): + self.ntbkNewPrinter.set_current_page (self.PAGE_SELECT_INSTALL_METHOD) + self.nextNPTab (step = 0) + else: + self.nextNPTab () + self._set_driver_ppd_loading(False) + finally: + self._hide_searching_spinner() # get PPDs def _getPPDs_reply (self, ppdsloader): - self._hide_searching_spinner() exc = ppdsloader.get_error () if exc: ppdsloader.destroy () @@ -2235,7 +2262,7 @@ def _installPrinterFromDeviceID (self, devid, page_nr, step): else: ppdname = None status = None - elif (self.dialog_mode == "ppd" and self.orig_ppd): + elif (self.dialog_mode in ("ppd", "duplicate") and self.orig_ppd): attr = self.orig_ppd.findAttr("NickName") if not attr: attr = self.orig_ppd.findAttr("ModelName") @@ -4252,7 +4279,9 @@ def found_network_printer_callback (self, new_device): ### def getDeviceURI(self): - if self.dialog_mode in ['printer_with_uri', 'ppd']: + if self.dialog_mode in ['printer_with_uri', 'ppd'] or ( + self.dialog_mode == "duplicate" and + getattr(self, 'duplicate_driver_selection', False)): return self.device.uri type = self.device.type diff --git a/ppdsloader.py b/ppdsloader.py index 2cb41aebf..04cfed694 100644 --- a/ppdsloader.py +++ b/ppdsloader.py @@ -61,6 +61,10 @@ def __init__ (self, device_id=None, parent=None, device_uri=None, GObject.GObject.__init__ (self) debugprint ("+%s" % self) self._device_id = device_id + if self._device_id: + self._devid_dict = cupshelpers.parseDeviceID (self._device_id) + else: + self._devid_dict = None self._device_uri = device_uri self._device_make_and_model = device_make_and_model self._parent = parent @@ -120,16 +124,26 @@ def destroy (self): def get_installed_files (self): return self._installed_files + def _wait_for_processing (self): + thread = getattr (self, '_processing_thread', None) + if thread is not None: + thread.join () + self._processing_thread = None + def get_ppds (self): + self._wait_for_processing () return self._ppds def get_driverless_ppds (self): + self._wait_for_processing () return self._driverless_ppds def get_ppdsmatch_result (self): + self._wait_for_processing () return self._ppdsmatch_result def get_error (self): + self._wait_for_processing () debugprint ("%s: stored error is %s" % (self, repr (self._exc))) return self._exc @@ -156,53 +170,92 @@ def _cups_reply (self, conn, result): if self._destroyed: return - driverless_ppds = {} - normal_ppds = {} - if result: - for name, ppd in result.items (): - if name.startswith ("driverless:") or name.startswith ("driverless-fax:"): - driverless_ppds[name] = ppd + def process_ppds (): + try: + driverless_ppds = {} + normal_ppds = {} + if result: + for name, ppd in result.items (): + if name.startswith ("driverless:") or name.startswith ("driverless-fax:"): + driverless_ppds[name] = ppd + else: + normal_ppds[name] = ppd + + self._driverless_ppds = driverless_ppds + ppds = cupshelpers.ppds.PPDs (normal_ppds, language=self._language) + self._ppds = ppds + self._need_requery_cups = False + # Pre-index makes on background thread to avoid GTK stall during fillMakeList + try: + ppds.getMakes () + except Exception: + pass + if self._device_id: + fit = ppds.\ + getPPDNamesFromDeviceID (self._devid_dict["MFG"], + self._devid_dict["MDL"], + self._devid_dict["DES"], + self._devid_dict["CMD"], + self._device_uri, + self._device_make_and_model) + + ppdnamelist = ppds.\ + orderPPDNamesByPreference (list(fit.keys ()), + self._installed_files, + devid=self._devid_dict, + fit=fit) + self._ppdsmatch_result = (fit, ppdnamelist) else: - normal_ppds[name] = ppd + self._ppdsmatch_result = None - self._driverless_ppds = driverless_ppds - ppds = cupshelpers.ppds.PPDs (normal_ppds, language=self._language) - self._ppds = ppds - self._need_requery_cups = False - if self._device_id: - fit = ppds.\ - getPPDNamesFromDeviceID (self._devid_dict["MFG"], - self._devid_dict["MDL"], - self._devid_dict["DES"], - self._devid_dict["CMD"], - self._device_uri, - self._device_make_and_model) - - ppdnamelist = ppds.\ - orderPPDNamesByPreference (list(fit.keys ()), - self._installed_files, - devid=self._devid_dict, - fit=fit) - self._ppdsmatch_result = (fit, ppdnamelist) - - ppdname = ppdnamelist[0] - if (self._bus and - not fit[ppdname].startswith ("exact") and - not self._jockey_queried and - self._devid_dict["MFG"] and - self._devid_dict["MDL"] and - self._local_cups): - # Try to install packages using jockey if - # - there's no appropriate driver (PPD) locally available - # - we are configuring local CUPS server - self._jockey_queried = True - self._query_jockey () - return + GLib.idle_add (self._on_ppds_processed, conn, None) + except Exception as e: + GLib.idle_add (self._on_ppds_processed, conn, e) - conn.destroy () - self._conn = None + self._processing_thread = threading.Thread (target=process_ppds, daemon=True) + self._processing_thread.start () + + def _on_ppds_processed (self, conn, exc): + self._wait_for_processing () + if self._destroyed: + if conn: + conn.destroy () + self._conn = None + return False + + if exc is not None: + if conn: + conn.destroy () + self._conn = None + self._ppds = None + self._driverless_ppds = {} + self._exc = exc + self.emit ('finished') + return False + + if self._device_id and self._ppdsmatch_result: + fit, ppdnamelist = self._ppdsmatch_result + if ppdnamelist: + ppdname = ppdnamelist[0] + if (self._bus and + not fit[ppdname].startswith ("exact") and + not self._jockey_queried and + self._devid_dict["MFG"] and + self._devid_dict["MDL"] and + self._local_cups): + # Try to install packages using jockey if + # - there's no appropriate driver (PPD) locally available + # - we are configuring local CUPS server + self._jockey_queried = True + self._query_jockey () + return False + + if conn: + conn.destroy () + self._conn = None self.emit ('finished') + return False def _cups_error (self, conn, exc): if self._destroyed: diff --git a/system-config-printer.py b/system-config-printer.py index ed42dcea2..a71cd0dbc 100755 --- a/system-config-printer.py +++ b/system-config-printer.py @@ -98,9 +98,11 @@ def show_help(): import installpackage import jobviewer import authconn +import asyncconn import monitor import errordialogs from errordialogs import * +from vectorspinner import VectorSpinner import userdefault from serversettings import ServerSettings from ToolbarSearchEntry import * @@ -1667,20 +1669,80 @@ def duplicate_printer (self, new_name): def _get_discovered_device_for_uri(self, uri): """Recover the normal Add Printer Device identity for a queue URI.""" + def match_device(devices): + normalized_uri = newprinter._normalize_uri(uri).rstrip('/') + for device in devices.values(): + if newprinter._normalize_uri(device.uri).rstrip('/') == normalized_uri: + return device + return None + + # For unit tests where self.cups is mocked or not a production authconn: + if not isinstance(getattr(self, 'cups', None), authconn.Connection): + try: + devices = cupshelpers.getDevices(self.cups) + return match_device(devices) + except Exception: + return None + + # In production, use asyncconn to preserve privileged PolicyKit semantics + # while pumping the GLib main loop to prevent GTK freezes. + matched_device = [None] + loop = GLib.MainLoop() + timeout_id = [None] + conn = None + + def on_reply(c, devices): + matched_device[0] = match_device(devices) + if loop.is_running(): + loop.quit() + + def on_error(c, exc): + if loop.is_running(): + loop.quit() + + def on_timeout(): + timeout_id[0] = None + if loop.is_running(): + loop.quit() + return False + try: - devices = cupshelpers.getDevices(self.cups) + conn = asyncconn.Connection(parent=self.PrintersWindow, + host=self.connect_server, + encryption=self.connect_encrypt) + scheme = str(uri.split(':', 1)[0]) if ':' in uri else '' + schemes = [scheme] if scheme else [] + if scheme in ['socket', 'lpd', 'ipp', 'ipps', 'dnssd']: + schemes.extend(['dnssd', 'snmp']) + kwds = {'reply_handler': on_reply, 'error_handler': on_error} + if schemes: + kwds['include_schemes'] = schemes + + cupshelpers.getDevices(conn, **kwds) + timeout_id[0] = GLib.timeout_add(3000, on_timeout) + loop.run() except Exception: - return None + pass + finally: + if timeout_id[0] is not None: + GLib.source_remove(timeout_id[0]) + if conn is not None: + try: + conn.destroy() + except Exception: + pass - normalized_uri = newprinter._normalize_uri(uri).rstrip('/') - for device in devices.values(): - if newprinter._normalize_uri(device.uri).rstrip('/') == normalized_uri: - return device - return None + return matched_device[0] def on_duplicate_activate(self, *UNUSED): + if getattr(self, '_duplicating', False): + return + iconview = self.dests_iconview paths = iconview.get_selected_items () + if not paths: + return + model = self.dests_iconview.get_model () iter = model.get_iter (paths[0]) name = model.get_value (iter, 2) @@ -1717,38 +1779,65 @@ def on_duplicate_activate(self, *UNUSED): return # For normal printers, use the New Printer GUI in duplicate mode - ppd_filename = None - try: - ppd_filename = self.cups.getPPD(name) - ppd_obj = cups.PPD(ppd_filename) - except (cups.IPPError, RuntimeError): - ppd_obj = None - finally: - if ppd_filename: - try: - os.unlink(ppd_filename) - except OSError: - pass - - discovered_device = self._get_discovered_device_for_uri( - printer.device_uri) - + self._duplicating = True busy (self.PrintersWindow) self.desensitise_new_printer_widgets () - if not self.newPrinterGUI.init("duplicate", - device_uri=printer.device_uri, - name=name, - ppd=ppd_obj, - host=self.connect_server, - encryption=self.connect_encrypt, - parent=self.PrintersWindow, - description=printer.info, - location=printer.location, - is_shared=printer.is_shared, - discovered_device=discovered_device): - self.sensitise_new_printer_widgets () - self.monitor.update () - ready (self.PrintersWindow) + + loading_dialog = Gtk.Dialog(title="", transient_for=self.PrintersWindow, modal=True, destroy_with_parent=True) + loading_dialog.set_decorated(False) + loading_dialog.set_resizable(False) + + hbox = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=15) + hbox.set_margin_start(20) + hbox.set_margin_end(20) + hbox.set_margin_top(20) + hbox.set_margin_bottom(20) + + spinner = VectorSpinner() + spinner.start() + hbox.pack_start(spinner, False, False, 0) + + label = Gtk.Label(label=_("Preparing Duplicate Printer…")) + hbox.pack_start(label, True, True, 0) + + loading_dialog.get_content_area().pack_start(hbox, True, True, 0) + loading_dialog.show_all() + + try: + ppd_filename = None + try: + ppd_filename = self.cups.getPPD(name) + ppd_obj = cups.PPD(ppd_filename) + except (cups.IPPError, RuntimeError, authconn.WorkerTimeoutError, authconn.WorkerUnavailableError): + ppd_obj = None + finally: + if ppd_filename: + try: + os.unlink(ppd_filename) + except OSError: + pass + + discovered_device = self._get_discovered_device_for_uri( + printer.device_uri) + + if not self.newPrinterGUI.init("duplicate", + device_uri=printer.device_uri, + name=name, + ppd=ppd_obj, + host=self.connect_server, + encryption=self.connect_encrypt, + parent=self.PrintersWindow, + description=printer.info, + location=printer.location, + is_shared=printer.is_shared, + discovered_device=discovered_device): + self.sensitise_new_printer_widgets () + self.monitor.update () + finally: + spinner.stop() + loading_dialog.destroy() + self._duplicating = False + ready (self.PrintersWindow) def on_entDuplicateName_changed(self, widget): # restrict diff --git a/test_newprinter.py b/test_newprinter.py index ebb519947..e0a90695f 100644 --- a/test_newprinter.py +++ b/test_newprinter.py @@ -21,13 +21,16 @@ import gi import tempfile +import threading +import time import pytest import newprinter gi.require_version('Gtk', '3.0') -from gi.repository import Gtk +from gi.repository import Gtk, GLib from unittest.mock import MagicMock import cupshelpers import cups +import authconn class MockPrinter: def __init__(self, uri): @@ -301,6 +304,40 @@ def test_driverless_installable_options_reached(monkeypatch): np._loadPPDsForDevice.assert_not_called() +def test_searching_spinner_visible_during_driverless_validation(): + """Verify that the searching spinner remains visible while nextNPTab executes.""" + np = get_dummy_gui() + + loader = MagicMock() + loader._jockey_has_answered = False + loader.get_error.return_value = None + loader.get_ppds.return_value = {"driverless:foo": {}} + loader.get_ppdsmatch_result.return_value = {} + + # We want to intercept nextNPTab and check if the spinner has been hidden yet. + spinner_states = [] + + def mock_hide(): + spinner_states.append("hidden") + np._hide_searching_spinner = mock_hide + + def mock_nextNPTab(step=1): + if "hidden" not in spinner_states: + spinner_states.append("visible") + else: + spinner_states.append("already_hidden") + + np.nextNPTab = mock_nextNPTab + np.ntbkNewPrinter = MagicMock() + np.ntbkNewPrinter.get_current_page.return_value = np.PAGE_DESCRIBE_PRINTER + np._set_driver_ppd_loading = MagicMock() + + np.on_ppdsloader_finished_next(loader) + + assert "visible" in spinner_states, "Spinner was hidden before nextNPTab finished" + assert "hidden" in spinner_states, "Spinner was not hidden eventually" + assert spinner_states == ["visible", "hidden"], "Unexpected sequence of events" + def test_broken_driverless_getnpppd_returns_none_on_runtime_error(monkeypatch): """When cups.PPD() raises RuntimeError for a driverless PPD, getNPPPD() must return None (not the stale string).""" @@ -3110,8 +3147,10 @@ def fire_cancel(): worker_finished.clear() original_monotonic = time.monotonic + mock_calls = [0] def mock_monotonic(): - if worker_block_event.is_set() and not worker_can_finish.is_set(): + mock_calls[0] += 1 + if mock_calls[0] > 1 and worker_block_event.is_set() and not worker_can_finish.is_set(): # We have entered the poll() evaluation and the worker is blocked on the CUPS call. # Unblock the worker and wait for it to finish publishing the result. worker_can_finish.set() @@ -3149,3 +3188,516 @@ def mock_monotonic(): worker_block_event.set() worker_finished.set() np.cups._worker_queue.task_done = original_task_done + + +def test_generic_worker_wrapper_api(monkeypatch): + """Verify run_on_worker executes off the main thread, pumps GLib events, supports both + string method names and worker-bound callables, and executes sequentially without leaks.""" + c = authconn.Connection(host='127.0.0.1', port=631, encryption=cups.HTTP_ENCRYPT_NEVER) + main_thread = threading.current_thread() + worker_threads = [] + glib_pumped = [] + + class SpyConn: + def __init__(self, *args, **kwargs): + pass + + def getPPD(self, name): + worker_threads.append(threading.current_thread()) + time.sleep(0.04) + if name == "broken": + raise cups.IPPError(cups.IPP_INTERNAL_ERROR, "failed") + return f"/tmp/ppd_{name}" + + monkeypatch.setattr(authconn.cups, "Connection", SpyConn) + + # Schedule a GLib idle callback with standard priority to verify main loop pumps during worker wait + def on_idle(): + glib_pumped.append(True) + return False + + GLib.idle_add(on_idle, priority=GLib.PRIORITY_DEFAULT) + + # 1. String method name wrapper + res = c.run_on_worker('getPPD')("myprinter") + assert res == "/tmp/ppd_myprinter" + assert len(worker_threads) == 1 + assert worker_threads[0] != main_thread + assert len(glib_pumped) > 0, "GLib main loop must pump events while worker is executing" + + # 2. Worker-bound callable wrapper + def custom_worker_op(conn, x, y): + worker_threads.append(threading.current_thread()) + assert isinstance(conn, SpyConn) + return x + y + + res2 = c.run_on_worker(custom_worker_op)(10, 20) + assert res2 == 30 + assert len(worker_threads) == 2 + assert worker_threads[1] != main_thread + + # 3. Sequential execution immediately after previous completion + res3 = c.run_on_worker('getPPD')("secondprinter") + assert res3 == "/tmp/ppd_secondprinter" + assert c._worker_state == 'IDLE' + + # 4. Exception propagation out of wrapper + try: + c.run_on_worker('getPPD')("broken") + assert False, "Should have raised IPPError" + except cups.IPPError as e: + assert e.args[0] == cups.IPP_INTERNAL_ERROR + + assert c._worker_state == 'IDLE', "Worker must be IDLE after exception" + + +def test_authloop_getPPD_offloading_and_retry(monkeypatch): + """Verify getPPD offloads through _authloop and handles authentication challenge and retry.""" + c = authconn.Connection(host='127.0.0.1', port=631, encryption=cups.HTTP_ENCRYPT_NEVER) + c._prompt_allowed = True + c._cancel = False + + auth_calls = [0] + ppd_calls = [] + + class SpyConn: + def __init__(self, *args, **kwargs): + pass + + def getPPD(self, name): + ppd_calls.append(len(ppd_calls)) + if len(ppd_calls) == 1: + raise cups.IPPError(cups.IPP_NOT_AUTHORIZED, "unauthorized") + return f"/tmp/ppd_{name}" + + monkeypatch.setattr(authconn.cups, "Connection", SpyConn) + + def mock_perform_authentication(): + auth_calls[0] += 1 + c._has_failed = False + return 1 if len(ppd_calls) < 2 else 0 + + monkeypatch.setattr(c, "_perform_authentication", mock_perform_authentication) + + res = c.getPPD("secure_printer") + assert res == "/tmp/ppd_secure_printer" + assert auth_calls[0] >= 2, "Authentication callback must be invoked on challenge" + assert len(ppd_calls) == 2, "Worker must retry operation after authentication" + assert c._worker_state == 'IDLE' + + +def test_ppdsloader_background_worker_responsiveness(): + """Verify PPDsLoader processes CPU-bound PPD indexing in background thread without freezing GTK.""" + import ppdsloader + loader = ppdsloader.PPDsLoader(device_id="MFG:Test;MDL:Printer;", device_uri="ipp://test/printer") + mock_conn = MagicMock() + + cups_result = { + "foomatic:HP-LaserJet.ppd": { + "ppd-make-and-model": ["HP LaserJet"], + "ppd-natural-language": ["en"], + "ppd-make": ["HP"], + }, + "driverless:ipps://TestPrinter._ipps._tcp.local/": { + "ppd-make-and-model": ["Test Printer, driverless"], + "ppd-natural-language": ["en"], + "ppd-make": ["Test"], + "ppd-device-id": ["MFG:Test;MDL:Printer;"], + } + } + + finished_fired = [] + glib_pumped_while_loading = [] + loop = GLib.MainLoop() + + def on_idle(): + glib_pumped_while_loading.append(True) + return False + + GLib.idle_add(on_idle, priority=GLib.PRIORITY_DEFAULT) + + def on_finished(p): + finished_fired.append(True) + loop.quit() + + loader.connect("finished", on_finished) + + # Start processing via _cups_reply + loader._cups_reply(mock_conn, cups_result) + assert loader._processing_thread is not None, "Processing thread must be spawned" + assert loader._processing_thread != threading.current_thread(), "PPD processing must run outside GTK main thread" + + # Run loop until finished signal fires from background worker completion + loop.run() + + assert len(finished_fired) == 1 + assert len(glib_pumped_while_loading) > 0, "GTK main loop must pump events while PPDs are parsed" + ppds_obj = loader.get_ppds() + assert "foomatic:HP-LaserJet.ppd" in ppds_obj.ppds + assert hasattr(ppds_obj, "makes") and bool(ppds_obj.makes), "Makes index must be pre-indexed on worker thread" + assert "driverless:ipps://TestPrinter._ipps._tcp.local/" in loader.get_driverless_ppds() + assert loader.get_ppdsmatch_result() is not None + + +def test_ppdsloader_destroy_cancels_emission(): + """Verify calling destroy() on PPDsLoader prevents finished signal emission.""" + import ppdsloader + loader = ppdsloader.PPDsLoader(device_id="MFG:Test;MDL:Printer;", device_uri="ipp://test/printer") + mock_conn = MagicMock() + + finished_fired = [] + loader.connect("finished", lambda p: finished_fired.append(True)) + + loader.destroy() + loader._cups_reply(mock_conn, {}) + + if getattr(loader, '_processing_thread', None) is not None: + loader._processing_thread.join() + + # Flush main loop + ctx = GLib.MainContext.default() + while ctx.pending(): + ctx.iteration(False) + + assert len(finished_fired) == 0, "Finished signal must not be emitted when destroyed" + + +def test_choose_driver_responsiveness_and_spinner_lifecycle(): + """Verify on_btnNPChooseDriver_clicked sets loading spinner, resets driverless flag, + and advances tab responsively.""" + np = newprinter.NewPrinterGUI.__new__(newprinter.NewPrinterGUI) + np.dialog_mode = 'printer' + np.device = MagicMock() + np.device.driverless = True + np.exactdrivermatch = True + np.searchedfordriverpackages = False + spinner_states = [] + np._set_driver_ppd_loading = lambda state: spinner_states.append(state) + np.rbtnNPFoomatic = MagicMock() + np.on_rbtnNPFoomatic_toggled = MagicMock() + np.ntbkNewPrinter = MagicMock() + np.PAGE_SELECT_INSTALL_METHOD = 1 + next_tab_calls = [] + np.nextNPTab = lambda step=None: next_tab_calls.append(step) + np.ppds = None + np.ppdsloader = None + + np.on_btnNPChooseDriver_clicked(MagicMock()) + + assert spinner_states[0] is True, "Driver loading spinner must be started immediately" + assert np.device.driverless is False + assert np.exactdrivermatch is False + assert np.searchedfordriverpackages is True + assert next_tab_calls == [0] + + +def test_duplicate_device_lookup_preserves_policykit_and_keeps_gtk_responsive(monkeypatch): + """Verify _get_discovered_device_for_uri uses asyncconn (PolicyKit) when cups is authconn.Connection, + pumps the GLib main loop while pending, limits schemes, cleans up connection, and spawns no threads.""" + import importlib + import authconn + import asyncconn + scp_module = importlib.import_module("system-config-printer") + + class DummySCP: + _get_discovered_device_for_uri = scp_module.GUI._get_discovered_device_for_uri + + gui = DummySCP() + gui.PrintersWindow = MagicMock() + gui.connect_server = "local" + gui.connect_encrypt = 0 + # Simulate production authconn + gui.cups = authconn.Connection.__new__(authconn.Connection) + + events_pumped = [] + conn_instances = [] + get_devices_calls = [] + + class FakeAsyncConn: + def __init__(self, parent=None, host=None, encryption=None): + self.parent = parent + self.host = host + self.encryption = encryption + self.destroyed = False + conn_instances.append(self) + + def destroy(self): + self.destroyed = True + + def fake_get_devices(conn, **kwds): + get_devices_calls.append((conn, kwds)) + reply_handler = kwds.get("reply_handler") + + def deliver_reply(): + dev_dict = { + "device-id": "MFG:Xerox;MDL:B235;", + "device-info": "Xerox B235 (driverless)", + "device-make-and-model": "Xerox B235", + } + fake_device = cupshelpers.Device("ipps://Xerox%20B235._ipps._tcp.local", **dev_dict) + reply_handler(conn, {fake_device.uri: fake_device}) + return False + + def on_idle(): + events_pumped.append("idle_ran") + GLib.idle_add(deliver_reply) + return False + + GLib.idle_add(on_idle) + + monkeypatch.setattr(asyncconn, "Connection", FakeAsyncConn) + monkeypatch.setattr(cupshelpers, "getDevices", fake_get_devices) + + thread_count_before = threading.active_count() + device = gui._get_discovered_device_for_uri("ipps://Xerox%20B235._ipps._tcp.local") + thread_count_after = threading.active_count() + + assert thread_count_after == thread_count_before, "No extra threads must be created" + assert len(events_pumped) > 0, "GLib main loop must actively pump events while device lookup is pending" + assert device is not None + assert device.uri == "ipps://Xerox%20B235._ipps._tcp.local" + assert device.id_dict["MFG"] == "Xerox" + assert len(conn_instances) == 1 + assert conn_instances[0].parent is gui.PrintersWindow + assert conn_instances[0].host == "local" + assert conn_instances[0].destroyed is True, "Connection must be destroyed after lookup" + + conn_arg, kwds_arg = get_devices_calls[0] + assert "include_schemes" in kwds_arg + assert "ipps" in kwds_arg["include_schemes"] + assert "dnssd" in kwds_arg["include_schemes"] + + +def test_duplicate_device_lookup_error_and_timeout_resilience(monkeypatch): + """Verify _get_discovered_device_for_uri exits cleanly on timeout or D-Bus error without freezing or raising.""" + import importlib + import authconn + import asyncconn + scp_module = importlib.import_module("system-config-printer") + + class DummySCP: + _get_discovered_device_for_uri = scp_module.GUI._get_discovered_device_for_uri + + gui = DummySCP() + gui.PrintersWindow = MagicMock() + gui.connect_server = None + gui.connect_encrypt = None + gui.cups = authconn.Connection.__new__(authconn.Connection) + + # 1. Error callback path + class ErrorAsyncConn: + def __init__(self, *a, **k): + self.destroyed = False + + def destroy(self): + self.destroyed = True + + def error_get_devices(conn, **kwds): + err_handler = kwds.get("error_handler") + GLib.idle_add(lambda: err_handler(conn, RuntimeError("PolicyKit error")) and False) + + monkeypatch.setattr(asyncconn, "Connection", ErrorAsyncConn) + monkeypatch.setattr(cupshelpers, "getDevices", error_get_devices) + + res_err = gui._get_discovered_device_for_uri("ipp://test/printer") + assert res_err is None, "Must return None on discovery error" + + # 2. Timeout path & late reply isolation + captured_reply_handler = [None] + class TimeoutAsyncConn: + def __init__(self, *a, **k): + self.destroyed = False + + def destroy(self): + self.destroyed = True + + def timeout_get_devices(conn, **kwds): + captured_reply_handler[0] = kwds.get("reply_handler") + + monkeypatch.setattr(asyncconn, "Connection", TimeoutAsyncConn) + monkeypatch.setattr(cupshelpers, "getDevices", timeout_get_devices) + orig_timeout_add = GLib.timeout_add + def fast_timeout_add(interval, callback): + return orig_timeout_add(10, callback) + monkeypatch.setattr(GLib, "timeout_add", fast_timeout_add) + + res_timeout = gui._get_discovered_device_for_uri("ipp://test/printer") + assert res_timeout is None, "Must return None on discovery timeout" + + # 3. Late reply after timeout does not raise or corrupt state + if captured_reply_handler[0]: + fake_dev = cupshelpers.Device("ipp://test/printer") + # Invoking late reply handler on destroyed/abandoned connection must be safe + captured_reply_handler[0](TimeoutAsyncConn(), {fake_dev.uri: fake_dev}) + + +def test_duplicate_printer_loading_ui_responsiveness(monkeypatch): + import importlib + import gi + gi.require_version('Gtk', '3.0') + from gi.repository import Gtk, GLib + + scp_module = importlib.import_module("system-config-printer") + + gui = scp_module.GUI.__new__(scp_module.GUI) + gui.PrintersWindow = Gtk.Window() + gui.dests_iconview = MagicMock() + gui.printers = {} + gui._duplicating = False + gui.desensitise_new_printer_widgets = MagicMock() + gui.sensitise_new_printer_widgets = MagicMock() + gui.monitor = MagicMock() + gui.newPrinterGUI = MagicMock() + gui.connect_server = None + gui.connect_encrypt = None + + gui.cups = MagicMock() + + # Set up selected item + gui.dests_iconview.get_selected_items.return_value = ["path"] + model_mock = MagicMock() + model_mock.get_value.return_value = "TestPrinter" + gui.dests_iconview.get_model.return_value = model_mock + + printer_mock = MagicMock() + printer_mock.is_class = False + printer_mock.device_uri = "test://uri" + gui.printers["TestPrinter"] = printer_mock + + # We will track dialog visibility by patching Gtk.Dialog.show_all and destroy + dialog_state = {"visible": False, "created": False} + original_show_all = Gtk.Dialog.show_all + original_destroy = Gtk.Dialog.destroy + + def mock_show_all(self): + dialog_state["visible"] = True + dialog_state["created"] = True + original_show_all(self) + + def mock_destroy(self): + dialog_state["visible"] = False + original_destroy(self) + + monkeypatch.setattr(Gtk.Dialog, "show_all", mock_show_all) + monkeypatch.setattr(Gtk.Dialog, "destroy", mock_destroy) + + events_pumped = [] + + def fake_getPPD(name): + assert dialog_state["visible"], "Loading dialog must be visible during retrieval" + + # simulate user repeatedly clicking duplicate + gui.on_duplicate_activate() + assert gui.cups.getPPD.call_count == 1, "Repeated click should not launch concurrent operation" + + def on_idle(): + events_pumped.append("idle_ran") + loop.quit() + return False + + loop = GLib.MainLoop() + GLib.idle_add(on_idle) + loop.run() + + return "fake_ppd.ppd" + + gui.cups.getPPD = MagicMock(side_effect=fake_getPPD) + + gui._get_discovered_device_for_uri = MagicMock(return_value=None) + + gui.on_duplicate_activate() + + assert "idle_ran" in events_pumped, "GLib events were not processed" + assert not dialog_state["visible"], "Dialog was not cleaned up on success" + assert not gui._duplicating, "Duplicating flag was not reset on success" + + # Now test error path + gui.cups.getPPD = MagicMock(side_effect=RuntimeError("timeout")) + dialog_state["visible"] = False + dialog_state["created"] = False + gui.on_duplicate_activate() + + assert not dialog_state["visible"], "Dialog was not cleaned up on error" + assert not gui._duplicating, "Duplicating flag was not reset on error" +def test_duplicate_choose_different_driver_preserves_uri(): + """Verify that in duplicate mode, getDeviceURI returns the original device URI, and does not parse unvisited UI widgets.""" + np = get_dummy_gui() + np.getDeviceURI = newprinter.NewPrinterGUI.getDeviceURI.__get__(np) + np.dialog_mode = "duplicate" + np.duplicate_driver_selection = True + + np.device = MagicMock() + np.device.uri = "socket://192.168.1.50" + np.device.type = "socket" + + uri = np.getDeviceURI() + assert uri == "socket://192.168.1.50", "getDeviceURI must preserve the device URI in duplicate driver selection." + + +def test_duplicate_driver_selection_fallback_to_orig_ppd(): + """Verify that if the device is not discovered on the network, duplicate mode falls back to matching the original PPD's Make/Model.""" + np = get_dummy_gui() + np._installPrinterFromDeviceID = newprinter.NewPrinterGUI._installPrinterFromDeviceID.__get__(np) + np.dialog_mode = "duplicate" + np.duplicate_driver_selection = True + + np.device = MagicMock() + np.device.id = "" + np.device.make_and_model = "" + np.device.driverless = False + np.devid = "" + np.device._driverless_failed = False + + np.orig_ppd = MockCupsPPD({"NickName": "Original Make Model Printer"}) + + np.ppds = MagicMock() + np.ppds.getInfoFromModel.return_value = {"original-driver": {"ppd-make-and-model": "Original Make Model Printer"}} + + np._installPrinterOrSearchForDriver = MagicMock(return_value=True) + + newprinter.nonfatalException = lambda *args: None + + np._installPrinterFromDeviceID("", 2, 1) + + np._installPrinterOrSearchForDriver.assert_called_with("", "original-driver", "exact", 2, 1) + + +def test_duplicate_printer_driverless_ppd_selection_fallback(): + import os + os.environ["SYSTEM_CONFIG_PRINTER_UI"] = "ui" + import gi + gi.require_version('Gtk', '3.0') + from gi.repository import Gtk, GLib + import newprinter + import cupshelpers + + gui = newprinter.NewPrinterGUI() + gui.printers = {} + gui.cups = MagicMock() + + class FakePPD: + def findAttr(self, name): + class Attr: + pass + a = Attr() + if name == "1284DeviceID": + a.value = "MFG:Xerox;MDL:B235;" + return a + elif name == "ModelName": + a.value = "Xerox B235" + return a + elif name == "NickName": + a.value = "Xerox B235, driverless, cups-filters 1.28.15" + return a + return None + + orig_ppd = FakePPD() + + gui.init("duplicate", device_uri="socket://192.168.1.100", name="Xerox", ppd=orig_ppd, discovered_device=None) + + assert gui.device.id == "MFG:Xerox;MDL:B235;" + assert gui.device.make_and_model == "Xerox B235" + assert gui.device.driverless is True + + gui.on_btnNPChooseDriver_clicked(None) + assert gui.device.driverless is True From b0839a940371774529dd2c14e64da18c84b17654 Mon Sep 17 00:00:00 2001 From: SinghCod3r Date: Mon, 12 Oct 2026 15:22:46 +0530 Subject: [PATCH 4/6] refactor: add generic worker execution for blocking CUPS operations --- authconn.py | 241 ++++++++++++--- newprinter.py | 15 +- system-config-printer.py | 7 +- test_newprinter.py | 15 +- test_worker_architecture.py | 571 ++++++++++++++++++++++++++++++++++++ vectorspinner.py | 51 ++++ 6 files changed, 851 insertions(+), 49 deletions(-) create mode 100644 test_worker_architecture.py diff --git a/authconn.py b/authconn.py index fdd2b010e..230736e58 100644 --- a/authconn.py +++ b/authconn.py @@ -36,6 +36,9 @@ import gettext gettext.install(domain=config.PACKAGE, localedir=config.localedir) N_ = lambda x: x +class OperationCanceled(RuntimeError): + pass + class WorkerTimeoutError(RuntimeError): pass @@ -162,6 +165,26 @@ def remove_auth_info (self, host=None, port=None): global_authinfocache = _AuthInfoCache () +def _dispatch_worker_op(op, conn, args, kwds): + """ + Execute a worker operation against the worker-owned cups.Connection + or as a standalone callable. + """ + if isinstance(op, str): + fn = getattr(conn, op) + return fn(*args, **kwds) + + if hasattr(op, '_op_name') and getattr(op, '_target_conn', None) is not None: + fn = getattr(conn, op._op_name) + return fn(*args, **kwds) + + if callable(op): + if getattr(op, '_with_connection', False): + return op(conn, *args, **kwds) + return op(*args, **kwds) + + raise TypeError(f"Operation {op!r} is not callable or a valid method name") + class Connection: def __init__ (self, parent=None, try_as_root=True, lock=False, host=None, port=None, encryption=None): @@ -183,9 +206,25 @@ def __init__ (self, parent=None, try_as_root=True, lock=False, self._operation_stack = [] self._lock = lock self._gui_event = threading.Event () + self._auth_lock = threading.Lock () + self._cancel = False + self._cannot_auth = False + self._worker_lock = threading.Lock() + self._pending_requests = [] + self._active_req_state = None self._connect () + def cancel (self): + """Signal cooperative cancellation to any in-flight worker operation.""" + self._cancel = True + if hasattr(self, '_worker_lock'): + with self._worker_lock: + if getattr(self, '_active_req_state', None) is not None: + self._active_req_state['cancelled'] = True + elif getattr(self, '_pending_requests', None): + self._pending_requests[0]['cancelled'] = True + def _begin_operation (self, operation): debugprint ("%s: Operation += %s" % (self, operation)) self._operation_stack.append (operation) @@ -237,46 +276,116 @@ def _using_polkit (self): return isinstance (self._connection, cupspk.Connection) def _make_binding (self, fname, fn): - return lambda *args, **kwds: self._authloop (fname, fn, *args, **kwds) + binding = lambda *args, **kwds: self._authloop (fname, fn, *args, **kwds) + binding._op_name = fname + binding._target_conn = self + return binding - def run_on_worker (self, op): + def run_on_worker (self, op, feedback=None, spinner=None, delay_ms=250, cancel_check=None, with_connection=False): """ Returns a callable wrapper that executes the specified CUPS operation on the dedicated background worker thread while pumping the GLib main loop. - :param op: Either a method name string (e.g. 'getServerPPD', 'getPPD') - or a callable accepting the worker-owned cups.Connection - as its first parameter: op(conn, *args, **kwds). + :param op: Either a method name string (e.g. 'getServerPPD', 'getPPD'), + a bound Connection method (self.cups.getServerPPD), + or a callable: op(*args, **kwds) or op(conn, *args, **kwds) if with_connection. + :param feedback: Optional DelayedSpinner or WorkerFeedback instance. + :param spinner: Optional VectorSpinner or Gtk.Spinner widget. + :param delay_ms: Milliseconds to delay spinner appearance (default 250ms). + :param cancel_check: Optional callable returning True if caller was cancelled. + :param with_connection: If True, pass the worker cups.Connection to op. """ + if feedback is None and spinner is not None: + from vectorspinner import DelayedSpinner + feedback = DelayedSpinner(spinner, delay_ms=delay_ms) + + target_op = getattr(op, '_op_name', op) + if with_connection and callable(target_op): + try: + target_op._with_connection = True + except AttributeError: + pass + + is_conn_method = False + if isinstance(target_op, str) and hasattr(self._connection, target_op): + is_conn_method = True + elif hasattr(op, '_op_name') and getattr(op, '_target_conn', None) is self: + is_conn_method = True + def wrapper (*args, **kwds): - return self._execute_in_worker (op, *args, **kwds) + # Worker-thread re-entry: execute directly on worker connection to avoid deadlock + if hasattr(self, '_worker_thread') and threading.current_thread() == self._worker_thread: + conn = getattr(self, '_worker_conn', None) + if conn is None: + with self._auth_lock: + user = self._use_user + cups.setUser(user) + conn = cups.Connection(host=self._server, + port=self._port, + encryption=self._encryption) + self._worker_conn = conn + return _dispatch_worker_op(target_op, conn, args, kwds) + + with getattr(self, '_worker_lock', threading.Lock()): + has_active = bool(getattr(self, '_pending_requests', None)) or (getattr(self, '_active_req_state', None) is not None) + if not has_active: + self._cancel = False + + # Execution on GTK / main thread: + if is_conn_method: + fn = getattr(self._connection, target_op) + return self._authloop(target_op, fn, *args, _on_worker=True, + _feedback=feedback, _cancel_check=cancel_check, **kwds) + else: + return self._execute_in_worker(target_op, *args, _feedback=feedback, + _cancel_check=cancel_check, **kwds) + return wrapper def _execute_in_worker(self, fname, *args, **kwds): timeout_val = kwds.pop('_timeout', 30.0) + feedback = kwds.pop('_feedback', None) + cancel_check = kwds.pop('_cancel_check', None) if not hasattr(self, '_worker_queue'): - self._worker_lock = threading.Lock() + self._worker_lock = getattr(self, '_worker_lock', threading.Lock()) self._worker_state = 'IDLE' + self._worker_conn = None + self._pending_requests = getattr(self, '_pending_requests', []) + self._active_req_state = getattr(self, '_active_req_state', None) self._worker_queue = queue.Queue() self._worker_thread = threading.Thread(target=self._worker_loop, daemon=True) self._worker_thread.start() with self._worker_lock: - if getattr(self, '_worker_state', 'IDLE') != 'IDLE': - raise WorkerUnavailableError("CUPS worker is unavailable or busy") - self._worker_state = 'BUSY' + if getattr(self, '_worker_unresponsive', False): + raise WorkerUnavailableError("Worker is unavailable or busy") box = [] - req_state = {'completed': False, 'timed_out': False} + req_state = {'completed': False, 'timed_out': False, 'cancelled': False} + with self._worker_lock: + if not hasattr(self, '_pending_requests'): + self._pending_requests = [] + self._pending_requests.append(req_state) self._worker_queue.put((fname, args, kwds, box, req_state)) loop = GLib.MainLoop() source_id = [None] start_time = time.monotonic() + if feedback is not None: + feedback.start() + def poll(): + if (cancel_check and cancel_check()) or req_state['cancelled']: + with self._worker_lock: + req_state['cancelled'] = True + if loop.is_running(): + loop.quit() + source_id[0] = None + return False + with self._worker_lock: is_completed = req_state['completed'] @@ -294,8 +403,8 @@ def poll(): source_id[0] = None return False - self._worker_state = 'UNAVAILABLE' req_state['timed_out'] = True + self._worker_unresponsive = True if loop.is_running(): loop.quit() @@ -308,29 +417,32 @@ def poll(): try: loop.run() finally: + with self._worker_lock: + if hasattr(self, '_pending_requests') and req_state in self._pending_requests: + self._pending_requests.remove(req_state) if source_id[0] is not None: GLib.source_remove(source_id[0]) source_id[0] = None + if feedback is not None: + feedback.stop() + + if req_state['cancelled']: + raise OperationCanceled(_("Operation canceled")) if req_state['timed_out']: raise WorkerTimeoutError("CUPS worker timeout") + if not box: + raise WorkerUnavailableError("CUPS worker returned no result") + ok, res, exc = box[0] if not ok: raise exc return res def _worker_loop(self): - def worker_password_cb(prompt): - debugprint("Got worker password callback") - if getattr(self, '_cancel', False) or getattr(self, '_auth_called', False): - return '' - self._auth_called = True - self._prompt = prompt - return getattr(self, '_use_password', '') - - cups.setPasswordCB(worker_password_cb) conn = None + conn_user = None while True: item = self._worker_queue.get() @@ -338,28 +450,48 @@ def worker_password_cb(prompt): break fname, args, kwds, box, req_state = item + with self._worker_lock: + if req_state['cancelled']: + self._worker_queue.task_done() + continue + self._worker_state = 'BUSY' + self._active_req_state = req_state + try: - if conn is None: + with self._auth_lock: + target_user = self._use_user + + if conn is None or conn_user != target_user: + cups.setUser(target_user) conn = cups.Connection(host=self._server, port=self._port, encryption=self._encryption) + conn_user = target_user + self._worker_conn = conn - cups.setUser(self._use_user) - if callable(fname): - fn = None - res = fname(conn, *args, **kwds) - else: - fn = getattr(conn, fname) - res = fn(*args, **kwds) + cups.setUser(target_user) + res = _dispatch_worker_op(fname, conn, args, kwds) res_tuple = (True, res, None) except Exception as e: res_tuple = (False, None, e) finally: with self._worker_lock: - if req_state['timed_out']: - if self._worker_state == 'UNAVAILABLE': + self._active_req_state = None + is_discarded = req_state['timed_out'] or req_state['cancelled'] + if is_discarded: + # Clean up temporary files if a file path was returned + if (res_tuple[0] and isinstance(res_tuple[1], str) and + os.path.isabs(res_tuple[1]) and os.path.exists(res_tuple[1])): + try: + os.unlink(res_tuple[1]) + except OSError: + pass + if req_state['timed_out']: conn = None + conn_user = None + self._worker_conn = None + self._worker_unresponsive = False else: req_state['completed'] = True box.append(res_tuple) @@ -378,9 +510,9 @@ def worker_password_cb(prompt): self._worker_queue.task_done() - cups.setPasswordCB(None) + self._worker_conn = None - def _authloop (self, fname, fn, *args, **kwds): + def _authloop (self, fname, fn, *args, _on_worker=False, _feedback=None, _cancel_check=None, **kwds): self._passes = 0 # remove signature if dbus is not being used and signature is provided if not self._using_polkit(): @@ -390,7 +522,12 @@ def _authloop (self, fname, fn, *args, **kwds): retry = False while True: try: - if self._perform_authentication () == 0: + auth_stat = self._perform_authentication () + if auth_stat == 0: + break + if auth_stat < 0 or self._cancel: + if self._cancel and not self._cannot_auth: + raise cups.IPPError (0, _("Operation canceled")) break if c != self._connection: @@ -400,9 +537,12 @@ def _authloop (self, fname, fn, *args, **kwds): cups.setUser (self._use_user) - if fname in ('getServerPPD', 'getPPD'): - result = self.run_on_worker(fname)(*args, **kwds) + if _on_worker: + result = self._execute_in_worker (fname, *args, _feedback=_feedback, + _cancel_check=_cancel_check, **kwds) else: + if fn is None: + fn = getattr (self._connection, fname) result = fn.__call__ (*args, **kwds) if fname == 'adminGetServerSettings': @@ -444,6 +584,8 @@ def _authloop (self, fname, fn, *args, **kwds): debugprint ("%s: %s" % (e, repr (m))) raise + except OperationCanceled: + raise cups.IPPError (0, _("Operation canceled")) except cups.HTTPError as e: (s,) = e.args if not self._cancel: @@ -492,7 +634,8 @@ def _password_callback (self, prompt): self._auth_called = True self._prompt = prompt - return self._use_password + with self._auth_lock: + return self._use_password def _perform_authentication (self): self._passes += 1 @@ -510,7 +653,15 @@ def _perform_authentication (self): self._has_failed = False self._forbidden = False self._auth_called = False - self._cancel = False + req_cancelled = False + if hasattr(self, '_worker_lock'): + with self._worker_lock: + if getattr(self, '_active_req_state', None) is not None: + req_cancelled = self._active_req_state.get('cancelled', False) + elif getattr(self, '_pending_requests', None): + req_cancelled = self._pending_requests[0].get('cancelled', False) + if not req_cancelled: + self._cancel = False self._cannot_auth = False self._dialog_shown = False cups.setPasswordCB (self._password_callback) @@ -644,9 +795,11 @@ def _perform_authentication_with_dialog (self): self._on_authentication_response (d, response) def _on_authentication_response (self, dialog, response): - (user, self._use_password) = dialog.get_auth_info () - if user != '': - self._use_user = user + (user, password) = dialog.get_auth_info () + with self._auth_lock: + if user != '': + self._use_user = user + self._use_password = password global_authinfocache.cache_auth_info ((self._use_user, self._use_password), host=self._server, @@ -656,6 +809,12 @@ def _on_authentication_response (self, dialog, response): if (response == Gtk.ResponseType.CANCEL or response == Gtk.ResponseType.DELETE_EVENT): self._cancel = True + if hasattr(self, '_worker_lock'): + with self._worker_lock: + if getattr(self, '_active_req_state', None) is not None: + self._active_req_state['cancelled'] = True + elif getattr(self, '_pending_requests', None): + self._pending_requests[0]['cancelled'] = True if self._lock: self._gui_event.set () diff --git a/newprinter.py b/newprinter.py index a1725de33..368c9aeea 100644 --- a/newprinter.py +++ b/newprinter.py @@ -2004,7 +2004,13 @@ def _validateDriverlessPPD(self, ppdname): self.cups._begin_operation(_("validating driverless PPD")) try: - f = self.cups.getServerPPD(ppdname) + if isinstance(getattr(self, 'cups', None), authconn.Connection): + f = self.cups.run_on_worker( + self.cups.getServerPPD, + spinner=getattr(self, 'spinner', None), + cancel_check=lambda: getattr(self, 'device', None) is None)(ppdname) + else: + f = self.cups.getServerPPD(ppdname) if getattr(self, 'device', None) is None: try: @@ -5045,7 +5051,12 @@ def getNPPPD(self): self.cups._begin_operation (_("fetching PPD")) try: if ppd != "raw": - f = self.cups.getServerPPD(ppd) + if isinstance(getattr(self, 'cups', None), authconn.Connection): + f = self.cups.run_on_worker( + self.cups.getServerPPD, + spinner=getattr(self, 'spinner', None))(ppd) + else: + f = self.cups.getServerPPD(ppd) ppd = cups.PPD(f) os.unlink(f) except RuntimeError: diff --git a/system-config-printer.py b/system-config-printer.py index a71cd0dbc..a625291d8 100755 --- a/system-config-printer.py +++ b/system-config-printer.py @@ -1806,7 +1806,12 @@ def on_duplicate_activate(self, *UNUSED): try: ppd_filename = None try: - ppd_filename = self.cups.getPPD(name) + if isinstance(getattr(self, 'cups', None), authconn.Connection): + ppd_filename = self.cups.run_on_worker( + self.cups.getPPD, + spinner=spinner)(name) + else: + ppd_filename = self.cups.getPPD(name) ppd_obj = cups.PPD(ppd_filename) except (cups.IPPError, RuntimeError, authconn.WorkerTimeoutError, authconn.WorkerUnavailableError): ppd_obj = None diff --git a/test_newprinter.py b/test_newprinter.py index e0a90695f..8e7e88556 100644 --- a/test_newprinter.py +++ b/test_newprinter.py @@ -3113,17 +3113,21 @@ def mock_task_done(): # 5. Auth retry behavior through _authloop() auth_called = [False] + orig_perform_auth = np.cups._perform_authentication def mock_perform_auth(*args, **kwds): auth_called[0] = True np.cups._cancel = True np.cups._cannot_auth = False return -1 # Cancel further retry - np.cups._perform_authentication = mock_perform_auth try: - np.cups.getServerPPD('driverless:authfail') - except cups.IPPError: - pass + np.cups._perform_authentication = mock_perform_auth + try: + np.cups.run_on_worker(np.cups.getServerPPD)('driverless:authfail') + except cups.IPPError: + pass + finally: + np.cups._perform_authentication = orig_perform_auth assert auth_called[0], "Authloop should have triggered _perform_authentication" # 6. Cancellation during GLib loop @@ -3230,6 +3234,7 @@ def custom_worker_op(conn, x, y): worker_threads.append(threading.current_thread()) assert isinstance(conn, SpyConn) return x + y + custom_worker_op._with_connection = True res2 = c.run_on_worker(custom_worker_op)(10, 20) assert res2 == 30 @@ -3279,7 +3284,7 @@ def mock_perform_authentication(): monkeypatch.setattr(c, "_perform_authentication", mock_perform_authentication) - res = c.getPPD("secure_printer") + res = c.run_on_worker(c.getPPD)("secure_printer") assert res == "/tmp/ppd_secure_printer" assert auth_calls[0] >= 2, "Authentication callback must be invoked on challenge" assert len(ppd_calls) == 2, "Worker must retry operation after authentication" diff --git a/test_worker_architecture.py b/test_worker_architecture.py new file mode 100644 index 000000000..18a970be8 --- /dev/null +++ b/test_worker_architecture.py @@ -0,0 +1,571 @@ +#!/usr/bin/python3 + +## Copyright (C) 2015 Red Hat, Inc. +## Authors: +## Alexander Pevzner +## Ayush Singh + +## This program is free software; you can redistribute it and/or modify +## it under the terms of the GNU General Public License as published by +## the Free Software Foundation; either version 2 of the License, or +## (at your option) any later version. + +## This program is distributed in the hope that it will be useful, +## but WITHOUT ANY WARRANTY; without even the implied warranty of +## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +## GNU General Public License for more details. + +## You should have received a copy of the GNU General Public License +## along with this program; if not, write to the Free Software +## Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. + +import os +import tempfile +import threading +import time +import pytest +import cups +import authconn +from vectorspinner import DelayedSpinner, VectorSpinner +from gi.repository import GLib, Gtk + + +class FakeCupsConnection: + """Mock cups.Connection for worker unit tests.""" + def __init__(self, *args, **kwargs): + self.host = kwargs.get('host', 'localhost') + self.port = kwargs.get('port', 631) + self.user = cups.getUser() + self.client_name = "test" + + def getPrinters(self): + return {} + + def addPrinter(self, *args, **kwargs): + pass + + def getServerPPD(self, name): + return f"/tmp/fake_server_{name}.ppd" + + def getPPD(self, name): + return f"/tmp/fake_local_{name}.ppd" + + def getDevices(self, *args, **kwargs): + return {"test://uri": {"device-info": "Test Printer"}} + + +@pytest.fixture(autouse=True) +def patch_cups_connection(monkeypatch): + monkeypatch.setattr(cups, "Connection", FakeCupsConnection) + monkeypatch.setattr(authconn.cups, "Connection", FakeCupsConnection) + + +def test_generic_worker_callable(): + conn = authconn.Connection(host="127.0.0.1", port=631, encryption=cups.HTTP_ENCRYPT_NEVER) + main_thread = threading.current_thread() + exec_threads = [] + + # Standalone function without connection parameter + def compute(x, y): + exec_threads.append(threading.current_thread()) + return x + y + + res1 = conn.run_on_worker(compute)(10, 25) + assert res1 == 35 + assert exec_threads[0] != main_thread + assert exec_threads[0] == conn._worker_thread + + # Function expecting worker connection + def probe_conn(c, prefix): + exec_threads.append(threading.current_thread()) + return f"{prefix}:{c.host}" + probe_conn._with_connection = True + + res2 = conn.run_on_worker(probe_conn)("server") + assert res2 == "server:127.0.0.1" + assert exec_threads[1] == conn._worker_thread + + +def test_bound_cups_method(): + conn = authconn.Connection(host="127.0.0.1", port=631, encryption=cups.HTTP_ENCRYPT_NEVER) + res = conn.run_on_worker(conn.getServerPPD)("my_printer") + assert res == "/tmp/fake_server_my_printer.ppd" + + +def test_string_operation(): + conn = authconn.Connection(host="127.0.0.1", port=631, encryption=cups.HTTP_ENCRYPT_NEVER) + res = conn.run_on_worker("getServerPPD")("string_printer") + assert res == "/tmp/fake_server_string_printer.ppd" + + +def test_fifo_sequential_execution(): + conn = authconn.Connection(host="127.0.0.1", port=631, encryption=cups.HTTP_ENCRYPT_NEVER) + execution_order = [] + + def op1(): + execution_order.append(1) + return "first" + + def op2(): + execution_order.append(2) + return "second" + + res1 = conn.run_on_worker(op1)() + res2 = conn.run_on_worker(op2)() + + assert res1 == "first" + assert res2 == "second" + assert execution_order == [1, 2] + + +def test_multiple_queued_operations(): + conn = authconn.Connection(host="127.0.0.1", port=631, encryption=cups.HTTP_ENCRYPT_NEVER) + results = [] + + def make_op(idx): + def op(): + results.append(idx) + return idx * 10 + return op + + out = [conn.run_on_worker(make_op(i))() for i in range(5)] + assert out == [0, 10, 20, 30, 40] + assert results == [0, 1, 2, 3, 4] + + +def test_worker_thread_reentrancy(): + conn = authconn.Connection(host="127.0.0.1", port=631, encryption=cups.HTTP_ENCRYPT_NEVER) + + def inner_op(n): + return n * 2 + + def outer_op(): + return conn.run_on_worker(inner_op)(21) + + result = conn.run_on_worker(outer_op)() + assert result == 42 + + +def test_clean_exception_propagation(): + conn = authconn.Connection(host="127.0.0.1", port=631, encryption=cups.HTTP_ENCRYPT_NEVER) + + def failing_op(): + raise ValueError("Intentional worker test failure") + + with pytest.raises(ValueError, match="Intentional worker test failure"): + conn.run_on_worker(failing_op)() + + def type_error_op(val): + return val + 123 # If val is str, raises TypeError + + with pytest.raises(TypeError): + conn.run_on_worker(type_error_op)("invalid_string_argument") + + +def test_worker_timeout(): + conn = authconn.Connection(host="127.0.0.1", port=631, encryption=cups.HTTP_ENCRYPT_NEVER) + finish_event = threading.Event() + + def slow_op(): + finish_event.wait(timeout=2.0) + return "late" + + start = time.monotonic() + with pytest.raises(authconn.WorkerTimeoutError): + + conn.run_on_worker(slow_op)(_timeout=0.1) + duration = time.monotonic() - start + assert duration < 1.0 + finish_event.set() + + +def test_worker_recovery_after_timeout(): + conn = authconn.Connection(host="127.0.0.1", port=631, encryption=cups.HTTP_ENCRYPT_NEVER) + stuck_gate = threading.Event() + + def stuck_op(): + stuck_gate.wait(timeout=2.0) + return "done_stuck" + + with pytest.raises(authconn.WorkerTimeoutError): + conn.run_on_worker(stuck_op)(_timeout=0.08) + + with pytest.raises(authconn.WorkerUnavailableError): + conn.run_on_worker(lambda: "fail_fast")() + + stuck_gate.set() + conn._worker_queue.join() + + res = conn.run_on_worker(lambda: "recovered")() + assert res == "recovered" + assert conn._worker_state == 'IDLE' + + +def test_cooperative_cancellation(): + conn = authconn.Connection(host="127.0.0.1", port=631, encryption=cups.HTTP_ENCRYPT_NEVER) + worker_started = threading.Event() + worker_finish = threading.Event() + cancelled_flag = [False] + + def cancellable_op(): + worker_started.set() + worker_finish.wait(timeout=2.0) + return "result" + + def trigger_cancel(): + worker_started.wait(timeout=2.0) + cancelled_flag[0] = True + + feeder = threading.Thread(target=trigger_cancel, daemon=True) + feeder.start() + + with pytest.raises(authconn.OperationCanceled): + conn.run_on_worker(cancellable_op, cancel_check=lambda: cancelled_flag[0])() + + worker_finish.set() + conn._worker_queue.join() + + +def test_cancellation_followed_by_fresh_operation(): + conn = authconn.Connection(host="127.0.0.1", port=631, encryption=cups.HTTP_ENCRYPT_NEVER) + worker_started = threading.Event() + worker_finish = threading.Event() + + def op1(): + worker_started.set() + worker_finish.wait(timeout=2.0) + return "op1" + + def cancel_thread(): + worker_started.wait(timeout=2.0) + conn.cancel() + + threading.Thread(target=cancel_thread, daemon=True).start() + + with pytest.raises(authconn.OperationCanceled): + conn.run_on_worker(op1)() + + worker_finish.set() + conn._worker_queue.join() + + res2 = conn.run_on_worker(lambda: "fresh_op_success")() + assert res2 == "fresh_op_success" + + +def test_authentication_retry(monkeypatch): + conn = authconn.Connection(host="127.0.0.1", port=631, encryption=cups.HTTP_ENCRYPT_NEVER) + attempts = [0] + auth_dialog_shown = [0] + + class AuthChallengedConnection(FakeCupsConnection): + def getPPD(self, name): + attempts[0] += 1 + if attempts[0] == 1: + raise cups.IPPError(cups.IPP_NOT_AUTHORIZED, "Not authorized") + return f"/tmp/authenticated_{name}.ppd" + + monkeypatch.setattr(cups, "Connection", AuthChallengedConnection) + monkeypatch.setattr(authconn.cups, "Connection", AuthChallengedConnection) + + def mock_perform_auth(): + auth_dialog_shown[0] += 1 + conn._has_failed = False + return 1 if attempts[0] < 2 else 0 + + monkeypatch.setattr(conn, "_perform_authentication", mock_perform_auth) + + res = conn.run_on_worker(conn.getPPD)("secure_queue") + assert res == "/tmp/authenticated_secure_queue.ppd" + assert attempts[0] == 2 + assert auth_dialog_shown[0] >= 2 + + +def test_authentication_cancellation(monkeypatch): + conn = authconn.Connection(host="127.0.0.1", port=631, encryption=cups.HTTP_ENCRYPT_NEVER) + attempts = [0] + auth_passes = [0] + + class AlwaysFailsConn(FakeCupsConnection): + def getPPD(self, name): + attempts[0] += 1 + raise cups.IPPError(cups.IPP_NOT_AUTHORIZED, "Not authorized") + + monkeypatch.setattr(cups, "Connection", AlwaysFailsConn) + monkeypatch.setattr(authconn.cups, "Connection", AlwaysFailsConn) + + orig_perform_auth = conn._perform_authentication + + def mock_cancel_auth(): + auth_passes[0] += 1 + if auth_passes[0] == 1: + return orig_perform_auth() + conn.cancel() + return -1 + + monkeypatch.setattr(conn, "_perform_authentication", mock_cancel_auth) + + with pytest.raises(cups.IPPError) as exc_info: + conn.run_on_worker(conn.getPPD)("secret_queue") + + err_code, err_msg = exc_info.value.args + assert err_code == 0 + assert "canceled" in str(err_msg).lower() + # Ensure cancellation remained set and did not enter an infinite retry loop + assert conn._cancel is True + assert attempts[0] == 1, "Must execute once, then abort on cancelled auth" + + monkeypatch.setattr(AlwaysFailsConn, "getPPD", lambda self, name: f"/tmp/next_{name}.ppd") + monkeypatch.setattr(conn, "_perform_authentication", orig_perform_auth) + + res_next = conn.run_on_worker(conn.getPPD)("next_queue") + assert res_next == "/tmp/next_next_queue.ppd" + assert conn._cancel is False + + +def test_worker_credential_change_recreates_connection(monkeypatch): + monkeypatch.setattr(authconn.global_authinfocache, "lookup_auth_info", lambda *a, **k: None) + conn = authconn.Connection(host="127.0.0.1", port=631, encryption=cups.HTTP_ENCRYPT_NEVER) + connections_created = [] + + class TrackedConnection(FakeCupsConnection): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + connections_created.append(cups.getUser()) + + def getServerPPD(self, name): + return f"/tmp/{cups.getUser()}_{name}.ppd" + + monkeypatch.setattr(cups, "Connection", TrackedConnection) + monkeypatch.setattr(authconn.cups, "Connection", TrackedConnection) + + conn._use_user = "alice" + res1 = conn.run_on_worker("getServerPPD")("test1") + assert "alice" in res1 + + conn._use_user = "root" + res2 = conn.run_on_worker("getServerPPD")("test2") + assert "root" in res2 + + assert len(connections_created) >= 2 + assert "alice" in connections_created + assert "root" in connections_created + + +def test_late_result_after_cancellation(): + conn = authconn.Connection(host="127.0.0.1", port=631, encryption=cups.HTTP_ENCRYPT_NEVER) + worker_started = threading.Event() + worker_finish = threading.Event() + + def slow_worker_op(): + worker_started.set() + worker_finish.wait(timeout=2.0) + return "late_unwanted_result" + + threading.Thread(target=lambda: (worker_started.wait(), conn.cancel()), daemon=True).start() + + with pytest.raises(authconn.OperationCanceled): + conn.run_on_worker(slow_worker_op)() + + worker_finish.set() + conn._worker_queue.join() + + res_next = conn.run_on_worker(lambda: "proper_next_result")() + assert res_next == "proper_next_result" + + +def test_cleanup_of_discarded_temporary_files(): + conn = authconn.Connection(host="127.0.0.1", port=631, encryption=cups.HTTP_ENCRYPT_NEVER) + worker_started = threading.Event() + worker_finish = threading.Event() + + with tempfile.NamedTemporaryFile(delete=False, suffix=".ppd") as tmp: + temp_ppd_path = tmp.name + + assert os.path.exists(temp_ppd_path) + + def blocking_ppd_op(): + worker_started.set() + worker_finish.wait(timeout=2.0) + return temp_ppd_path + + threading.Thread(target=lambda: (worker_started.wait(), conn.cancel()), daemon=True).start() + + with pytest.raises(authconn.OperationCanceled): + conn.run_on_worker(blocking_ppd_op)() + + worker_finish.set() + conn._worker_queue.join() + + assert not os.path.exists(temp_ppd_path), "Cancelled temporary file must be unlinked" + + +def test_gtk_responsiveness_during_worker_execution(): + conn = authconn.Connection(host="127.0.0.1", port=631, encryption=cups.HTTP_ENCRYPT_NEVER) + worker_started = threading.Event() + worker_can_proceed = threading.Event() + glib_event_ran = threading.Event() + + def slow_worker(): + worker_started.set() + assert worker_can_proceed.wait(timeout=5.0) + return "finished" + + def on_glib_idle(): + glib_event_ran.set() + worker_can_proceed.set() + return False + + def pump_glib(): + while not worker_started.is_set(): + time.sleep(0.01) + GLib.idle_add(on_glib_idle) + + threading.Thread(target=pump_glib, daemon=True).start() + + res = conn.run_on_worker(slow_worker)() + assert res == "finished" + assert glib_event_ran.is_set() + + +def test_delayed_spinner_fast_operation(): + spinner = VectorSpinner() + spinner.show = VectorSpinner.show + spinner.hide = VectorSpinner.hide + delayed = DelayedSpinner(spinner, delay_ms=250) + + assert not delayed.is_shown + + delayed.start() + time.sleep(0.02) + delayed.stop() + + assert not delayed.is_shown + assert delayed._timer_id is None + + +def test_delayed_spinner_slow_operation(): + spinner = VectorSpinner() + shown_events = [] + stopped_events = [] + + spinner.show = lambda: shown_events.append(True) + spinner.start = lambda: None + spinner.stop = lambda: stopped_events.append(True) + spinner.hide = lambda: None + + delayed = DelayedSpinner(spinner, delay_ms=50) + loop = GLib.MainLoop() + + delayed.start() + + def on_after_delay(): + assert delayed.is_shown + assert len(shown_events) == 1 + delayed.stop() + assert not delayed.is_shown + assert len(stopped_events) == 1 + loop.quit() + return False + + GLib.timeout_add(80, on_after_delay) + loop.run() + + assert len(shown_events) == 1 + assert len(stopped_events) == 1 + + +def test_cancelled_in_queue_skips_execution(): + conn = authconn.Connection(host="127.0.0.1", port=631, encryption=cups.HTTP_ENCRYPT_NEVER) + gate = threading.Event() + skipped_executed = [False] + + def blocking_op(): + gate.wait(timeout=2.0) + return "block_done" + + def skipped_op(): + skipped_executed[0] = True + return "should_not_run" + + t1 = threading.Thread(target=lambda: conn.run_on_worker(blocking_op)(), daemon=True) + t1.start() + + time.sleep(0.05) + + with pytest.raises(authconn.OperationCanceled): + conn.run_on_worker(skipped_op, cancel_check=lambda: True)() + + gate.set() + conn._worker_queue.join() + + assert not skipped_executed[0], "Queued request with req_state['cancelled'] must be skipped" + + +def test_cancellation_of_request_a_does_not_affect_request_b(): + conn = authconn.Connection(host="127.0.0.1", port=631, encryption=cups.HTTP_ENCRYPT_NEVER) + a_started = threading.Event() + a_release = threading.Event() + a_errors = [] + a_results = [] + b_errors = [] + b_results = [] + + def op_a(): + a_started.set() + a_release.wait(timeout=2.0) + return "result_a" + + def op_b(): + return "result_b" + + def run_a(): + try: + res = conn.run_on_worker(op_a)() + a_results.append(res) + except Exception as e: + a_errors.append(e) + + def run_b(): + try: + res = conn.run_on_worker(op_b)() + b_results.append(res) + except Exception as e: + b_errors.append(e) + + t_a = threading.Thread(target=run_a, daemon=True) + t_a.start() + assert a_started.wait(timeout=2.0), "Request A failed to start" + + t_b = threading.Thread(target=run_b, daemon=True) + t_b.start() + + deadline = time.monotonic() + 2.0 + req_b = None + while time.monotonic() < deadline: + with conn._worker_lock: + if len(conn._pending_requests) >= 2: + req_b = conn._pending_requests[1] + break + time.sleep(0.01) + assert req_b is not None, "Request B was not queued in pending requests within timeout" + + conn.cancel() + + t_a.join(timeout=2.0) + assert not t_a.is_alive(), "Thread A did not exit after cancellation" + assert len(a_errors) == 1, "Request A should have raised an exception" + assert isinstance(a_errors[0], authconn.OperationCanceled), "Request A should raise OperationCanceled" + + assert req_b['cancelled'] is False, "Request B's req_state must NOT be marked cancelled" + + a_release.set() + + t_b.join(timeout=2.0) + assert not t_b.is_alive(), "Thread B did not finish" + assert len(b_errors) == 0, f"Request B failed unexpectedly: {b_errors}" + assert b_results == ["result_b"], "Request B should return expected result" + + res_c = conn.run_on_worker(lambda: "result_c")() + assert res_c == "result_c", "Independent Request C should execute and return normally" + assert conn._cancel is False, "Connection cancellation flag should be reset for fresh operation" + diff --git a/vectorspinner.py b/vectorspinner.py index 01e6ca851..ddf055bcd 100644 --- a/vectorspinner.py +++ b/vectorspinner.py @@ -94,3 +94,54 @@ def _on_draw(self, widget, cr): return False + +class DelayedSpinner: + """A wrapper for GTK/VectorSpinner widgets that delays spinner appearance + by a specified threshold (e.g. 200-300ms) to prevent visual flicker + for fast operations. + """ + + def __init__(self, spinner, delay_ms=250): + self._spinner = spinner + self._delay_ms = delay_ms + self._timer_id = None + self._is_shown = False + + @property + def is_shown(self): + return self._is_shown + + def start(self): + """Schedule delayed appearance of the spinner.""" + if self._spinner is None or self._timer_id is not None or self._is_shown: + return + self._timer_id = GLib.timeout_add(self._delay_ms, self._on_delay_timer) + + def _on_delay_timer(self): + self._timer_id = None + self._is_shown = True + if self._spinner is not None: + try: + self._spinner.show() + if hasattr(self._spinner, 'start'): + self._spinner.start() + except (AttributeError, RuntimeError): + pass + return False + + def stop(self): + """Cancel delayed timer if pending, or stop/hide spinner if visible.""" + if self._timer_id is not None: + GLib.source_remove(self._timer_id) + self._timer_id = None + + if self._is_shown: + self._is_shown = False + if self._spinner is not None: + try: + if hasattr(self._spinner, 'stop'): + self._spinner.stop() + self._spinner.hide() + except (AttributeError, RuntimeError): + pass + From ded8b349f6d2c1ffbc77e194ac16eebc4d614a77 Mon Sep 17 00:00:00 2001 From: SinghCod3r Date: Tue, 13 Oct 2026 12:43:43 +0530 Subject: [PATCH 5/6] Fix duplicate spinner during driverless PPD validation --- newprinter.py | 1 - test_newprinter.py | 49 ++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 49 insertions(+), 1 deletion(-) diff --git a/newprinter.py b/newprinter.py index 368c9aeea..28e9a6710 100644 --- a/newprinter.py +++ b/newprinter.py @@ -2007,7 +2007,6 @@ def _validateDriverlessPPD(self, ppdname): if isinstance(getattr(self, 'cups', None), authconn.Connection): f = self.cups.run_on_worker( self.cups.getServerPPD, - spinner=getattr(self, 'spinner', None), cancel_check=lambda: getattr(self, 'device', None) is None)(ppdname) else: f = self.cups.getServerPPD(ppdname) diff --git a/test_newprinter.py b/test_newprinter.py index 8e7e88556..49d3c557d 100644 --- a/test_newprinter.py +++ b/test_newprinter.py @@ -3706,3 +3706,52 @@ class Attr: gui.on_btnNPChooseDriver_clicked(None) assert gui.device.driverless is True + + +def test_driverless_validation_does_not_pass_duplicate_spinner(monkeypatch): + """Verify driverless PPD validation runs on worker without passing a duplicate spinner, + preserving the 'Searching for drivers' UI spinner as the sole progress indicator.""" + np = get_dummy_gui() + np.device = cupshelpers.Device( + "ipp://test-printer.local", + **{ + "device-info": "Xerox Test Printer", + "device-id": "MFG:Xerox;MDL:B235;", + "device-make-and-model": "Xerox B235", + } + ) + np.device.driverless = True + np.spinner = MagicMock(name="bottom_left_spinner") + np.cups = authconn.Connection(host='127.0.0.1', port=631, encryption=cups.HTTP_ENCRYPT_NEVER) + np.cups._begin_operation = MagicMock() + np.cups._end_operation = MagicMock() + + captured_worker_calls = [] + + def spy_run_on_worker(op, **kwargs): + captured_worker_calls.append({ + "op": op, + "spinner": kwargs.get("spinner", None), + "feedback": kwargs.get("feedback", None), + "cancel_check": kwargs.get("cancel_check", None), + }) + def fake_worker_call(ppdname): + with tempfile.NamedTemporaryFile(delete=False, suffix=".ppd") as tmp: + tmp.write(b"*PPD-Adobe: \"4.3\"\n") + return tmp.name + return fake_worker_call + + np.cups.run_on_worker = spy_run_on_worker + monkeypatch.setattr(cups, "PPD", MagicMock()) + + real_validate = newprinter.NewPrinterGUI._validateDriverlessPPD.__get__(np) + result = real_validate("driverless:ipp://test-printer.local") + + assert len(captured_worker_calls) == 1, "Validation must run through run_on_worker" + call_info = captured_worker_calls[0] + assert call_info["op"] == np.cups.getServerPPD + assert call_info["spinner"] is None, "Must NOT pass self.spinner to run_on_worker to prevent duplicate spinner" + assert call_info["feedback"] is None, "Must NOT pass feedback to run_on_worker" + assert call_info["cancel_check"] is not None, "Cancel check callable must still be provided" + assert np.spinner.show.call_count == 0, "Bottom-left spinner must not be shown" + assert np.spinner.start.call_count == 0, "Bottom-left spinner must not be started" From d23af385a9d3daa2562ee626b615dd2f10066647 Mon Sep 17 00:00:00 2001 From: SinghCod3r Date: Sat, 17 Oct 2026 10:30:52 +0530 Subject: [PATCH 6/6] Move blocking CUPS requests to dedicated worker --- authconn.py | 88 ++---------- newprinter.py | 4 +- test_newprinter.py | 2 - test_worker_architecture.py | 266 ++++++++++++++++-------------------- 4 files changed, 129 insertions(+), 231 deletions(-) diff --git a/authconn.py b/authconn.py index 230736e58..fb4007026 100644 --- a/authconn.py +++ b/authconn.py @@ -36,9 +36,6 @@ import gettext gettext.install(domain=config.PACKAGE, localedir=config.localedir) N_ = lambda x: x -class OperationCanceled(RuntimeError): - pass - class WorkerTimeoutError(RuntimeError): pass @@ -210,20 +207,11 @@ def __init__ (self, parent=None, try_as_root=True, lock=False, self._cancel = False self._cannot_auth = False self._worker_lock = threading.Lock() - self._pending_requests = [] - self._active_req_state = None self._connect () def cancel (self): - """Signal cooperative cancellation to any in-flight worker operation.""" self._cancel = True - if hasattr(self, '_worker_lock'): - with self._worker_lock: - if getattr(self, '_active_req_state', None) is not None: - self._active_req_state['cancelled'] = True - elif getattr(self, '_pending_requests', None): - self._pending_requests[0]['cancelled'] = True def _begin_operation (self, operation): debugprint ("%s: Operation += %s" % (self, operation)) @@ -281,7 +269,7 @@ def _make_binding (self, fname, fn): binding._target_conn = self return binding - def run_on_worker (self, op, feedback=None, spinner=None, delay_ms=250, cancel_check=None, with_connection=False): + def run_on_worker (self, op, feedback=None, spinner=None, delay_ms=250, with_connection=False): """ Returns a callable wrapper that executes the specified CUPS operation on the dedicated background worker thread while pumping the GLib @@ -293,7 +281,6 @@ def run_on_worker (self, op, feedback=None, spinner=None, delay_ms=250, cancel_c :param feedback: Optional DelayedSpinner or WorkerFeedback instance. :param spinner: Optional VectorSpinner or Gtk.Spinner widget. :param delay_ms: Milliseconds to delay spinner appearance (default 250ms). - :param cancel_check: Optional callable returning True if caller was cancelled. :param with_connection: If True, pass the worker cups.Connection to op. """ if feedback is None and spinner is not None: @@ -327,33 +314,24 @@ def wrapper (*args, **kwds): self._worker_conn = conn return _dispatch_worker_op(target_op, conn, args, kwds) - with getattr(self, '_worker_lock', threading.Lock()): - has_active = bool(getattr(self, '_pending_requests', None)) or (getattr(self, '_active_req_state', None) is not None) - if not has_active: - self._cancel = False - # Execution on GTK / main thread: if is_conn_method: fn = getattr(self._connection, target_op) return self._authloop(target_op, fn, *args, _on_worker=True, - _feedback=feedback, _cancel_check=cancel_check, **kwds) + _feedback=feedback, **kwds) else: - return self._execute_in_worker(target_op, *args, _feedback=feedback, - _cancel_check=cancel_check, **kwds) + return self._execute_in_worker(target_op, *args, _feedback=feedback, **kwds) return wrapper def _execute_in_worker(self, fname, *args, **kwds): timeout_val = kwds.pop('_timeout', 30.0) feedback = kwds.pop('_feedback', None) - cancel_check = kwds.pop('_cancel_check', None) if not hasattr(self, '_worker_queue'): self._worker_lock = getattr(self, '_worker_lock', threading.Lock()) self._worker_state = 'IDLE' self._worker_conn = None - self._pending_requests = getattr(self, '_pending_requests', []) - self._active_req_state = getattr(self, '_active_req_state', None) self._worker_queue = queue.Queue() self._worker_thread = threading.Thread(target=self._worker_loop, daemon=True) self._worker_thread.start() @@ -363,11 +341,7 @@ def _execute_in_worker(self, fname, *args, **kwds): raise WorkerUnavailableError("Worker is unavailable or busy") box = [] - req_state = {'completed': False, 'timed_out': False, 'cancelled': False} - with self._worker_lock: - if not hasattr(self, '_pending_requests'): - self._pending_requests = [] - self._pending_requests.append(req_state) + req_state = {'completed': False, 'timed_out': False} self._worker_queue.put((fname, args, kwds, box, req_state)) loop = GLib.MainLoop() @@ -378,14 +352,6 @@ def _execute_in_worker(self, fname, *args, **kwds): feedback.start() def poll(): - if (cancel_check and cancel_check()) or req_state['cancelled']: - with self._worker_lock: - req_state['cancelled'] = True - if loop.is_running(): - loop.quit() - source_id[0] = None - return False - with self._worker_lock: is_completed = req_state['completed'] @@ -417,18 +383,12 @@ def poll(): try: loop.run() finally: - with self._worker_lock: - if hasattr(self, '_pending_requests') and req_state in self._pending_requests: - self._pending_requests.remove(req_state) if source_id[0] is not None: GLib.source_remove(source_id[0]) source_id[0] = None if feedback is not None: feedback.stop() - if req_state['cancelled']: - raise OperationCanceled(_("Operation canceled")) - if req_state['timed_out']: raise WorkerTimeoutError("CUPS worker timeout") @@ -451,11 +411,7 @@ def _worker_loop(self): fname, args, kwds, box, req_state = item with self._worker_lock: - if req_state['cancelled']: - self._worker_queue.task_done() - continue self._worker_state = 'BUSY' - self._active_req_state = req_state try: with self._auth_lock: @@ -477,9 +433,7 @@ def _worker_loop(self): res_tuple = (False, None, e) finally: with self._worker_lock: - self._active_req_state = None - is_discarded = req_state['timed_out'] or req_state['cancelled'] - if is_discarded: + if req_state['timed_out']: # Clean up temporary files if a file path was returned if (res_tuple[0] and isinstance(res_tuple[1], str) and os.path.isabs(res_tuple[1]) and os.path.exists(res_tuple[1])): @@ -487,11 +441,10 @@ def _worker_loop(self): os.unlink(res_tuple[1]) except OSError: pass - if req_state['timed_out']: - conn = None - conn_user = None - self._worker_conn = None - self._worker_unresponsive = False + conn = None + conn_user = None + self._worker_conn = None + self._worker_unresponsive = False else: req_state['completed'] = True box.append(res_tuple) @@ -512,7 +465,7 @@ def _worker_loop(self): self._worker_conn = None - def _authloop (self, fname, fn, *args, _on_worker=False, _feedback=None, _cancel_check=None, **kwds): + def _authloop (self, fname, fn, *args, _on_worker=False, _feedback=None, **kwds): self._passes = 0 # remove signature if dbus is not being used and signature is provided if not self._using_polkit(): @@ -538,8 +491,7 @@ def _authloop (self, fname, fn, *args, _on_worker=False, _feedback=None, _cancel cups.setUser (self._use_user) if _on_worker: - result = self._execute_in_worker (fname, *args, _feedback=_feedback, - _cancel_check=_cancel_check, **kwds) + result = self._execute_in_worker (fname, *args, _feedback=_feedback, **kwds) else: if fn is None: fn = getattr (self._connection, fname) @@ -584,8 +536,6 @@ def _authloop (self, fname, fn, *args, _on_worker=False, _feedback=None, _cancel debugprint ("%s: %s" % (e, repr (m))) raise - except OperationCanceled: - raise cups.IPPError (0, _("Operation canceled")) except cups.HTTPError as e: (s,) = e.args if not self._cancel: @@ -653,15 +603,7 @@ def _perform_authentication (self): self._has_failed = False self._forbidden = False self._auth_called = False - req_cancelled = False - if hasattr(self, '_worker_lock'): - with self._worker_lock: - if getattr(self, '_active_req_state', None) is not None: - req_cancelled = self._active_req_state.get('cancelled', False) - elif getattr(self, '_pending_requests', None): - req_cancelled = self._pending_requests[0].get('cancelled', False) - if not req_cancelled: - self._cancel = False + self._cancel = False self._cannot_auth = False self._dialog_shown = False cups.setPasswordCB (self._password_callback) @@ -809,12 +751,6 @@ def _on_authentication_response (self, dialog, response): if (response == Gtk.ResponseType.CANCEL or response == Gtk.ResponseType.DELETE_EVENT): self._cancel = True - if hasattr(self, '_worker_lock'): - with self._worker_lock: - if getattr(self, '_active_req_state', None) is not None: - self._active_req_state['cancelled'] = True - elif getattr(self, '_pending_requests', None): - self._pending_requests[0]['cancelled'] = True if self._lock: self._gui_event.set () diff --git a/newprinter.py b/newprinter.py index 28e9a6710..8e558f365 100644 --- a/newprinter.py +++ b/newprinter.py @@ -2005,9 +2005,7 @@ def _validateDriverlessPPD(self, ppdname): self.cups._begin_operation(_("validating driverless PPD")) try: if isinstance(getattr(self, 'cups', None), authconn.Connection): - f = self.cups.run_on_worker( - self.cups.getServerPPD, - cancel_check=lambda: getattr(self, 'device', None) is None)(ppdname) + f = self.cups.run_on_worker(self.cups.getServerPPD)(ppdname) else: f = self.cups.getServerPPD(ppdname) diff --git a/test_newprinter.py b/test_newprinter.py index 49d3c557d..f2e006e7f 100644 --- a/test_newprinter.py +++ b/test_newprinter.py @@ -3733,7 +3733,6 @@ def spy_run_on_worker(op, **kwargs): "op": op, "spinner": kwargs.get("spinner", None), "feedback": kwargs.get("feedback", None), - "cancel_check": kwargs.get("cancel_check", None), }) def fake_worker_call(ppdname): with tempfile.NamedTemporaryFile(delete=False, suffix=".ppd") as tmp: @@ -3752,6 +3751,5 @@ def fake_worker_call(ppdname): assert call_info["op"] == np.cups.getServerPPD assert call_info["spinner"] is None, "Must NOT pass self.spinner to run_on_worker to prevent duplicate spinner" assert call_info["feedback"] is None, "Must NOT pass feedback to run_on_worker" - assert call_info["cancel_check"] is not None, "Cancel check callable must still be provided" assert np.spinner.show.call_count == 0, "Bottom-left spinner must not be shown" assert np.spinner.start.call_count == 0, "Bottom-left spinner must not be started" diff --git a/test_worker_architecture.py b/test_worker_architecture.py index 18a970be8..597dd6a23 100644 --- a/test_worker_architecture.py +++ b/test_worker_architecture.py @@ -201,55 +201,89 @@ def stuck_op(): assert conn._worker_state == 'IDLE' -def test_cooperative_cancellation(): +def test_concurrency_caller_abandons_a_while_b_queued(): conn = authconn.Connection(host="127.0.0.1", port=631, encryption=cups.HTTP_ENCRYPT_NEVER) - worker_started = threading.Event() - worker_finish = threading.Event() - cancelled_flag = [False] + a_started = threading.Event() + a_release = threading.Event() + a_finished = threading.Event() - def cancellable_op(): - worker_started.set() - worker_finish.wait(timeout=2.0) - return "result" + with tempfile.NamedTemporaryFile(delete=False, suffix=".ppd") as tmp: + a_temp_file = tmp.name - def trigger_cancel(): - worker_started.wait(timeout=2.0) - cancelled_flag[0] = True + assert os.path.exists(a_temp_file) - feeder = threading.Thread(target=trigger_cancel, daemon=True) - feeder.start() + ui_state = {'alive': True, 'result': None} - with pytest.raises(authconn.OperationCanceled): - conn.run_on_worker(cancellable_op, cancel_check=lambda: cancelled_flag[0])() + def op_a(): + a_started.set() + a_release.wait(timeout=5.0) + a_finished.set() + return a_temp_file + + def caller_a_task(): + res = conn.run_on_worker(op_a)() + if ui_state['alive']: + ui_state['result'] = res + else: + # Emulate NewPrinterGUI lifecycle: discard result and unlink temporary file + try: + os.unlink(res) + except OSError: + pass + + t_a = threading.Thread(target=caller_a_task, daemon=True) + t_a.start() + assert a_started.wait(timeout=2.0), "Request A failed to start in worker" - worker_finish.set() - conn._worker_queue.join() + b_results = [] + def caller_b_task(): + res = conn.run_on_worker(lambda: "result_b")() + b_results.append(res) -def test_cancellation_followed_by_fresh_operation(): - conn = authconn.Connection(host="127.0.0.1", port=631, encryption=cups.HTTP_ENCRYPT_NEVER) - worker_started = threading.Event() - worker_finish = threading.Event() + t_b = threading.Thread(target=caller_b_task, daemon=True) + t_b.start() + time.sleep(0.05) - def op1(): - worker_started.set() - worker_finish.wait(timeout=2.0) - return "op1" + # Caller A abandons/cancels UI operation + ui_state['alive'] = False - def cancel_thread(): - worker_started.wait(timeout=2.0) - conn.cancel() + # Release A to finish + a_release.set() - threading.Thread(target=cancel_thread, daemon=True).start() + t_a.join(timeout=2.0) + assert not t_a.is_alive() + assert a_finished.is_set(), "A must run to completion in worker" + assert ui_state['result'] is None, "A's result must not be applied to abandoned caller" + assert not os.path.exists(a_temp_file), "A's temporary file must be unlinked when abandoned" - with pytest.raises(authconn.OperationCanceled): - conn.run_on_worker(op1)() + t_b.join(timeout=2.0) + assert not t_b.is_alive() + assert b_results == ["result_b"], "B must execute normally and return expected result" - worker_finish.set() - conn._worker_queue.join() + assert conn._worker_thread.is_alive(), "Worker thread must remain alive" + assert conn._worker_state == 'IDLE' - res2 = conn.run_on_worker(lambda: "fresh_op_success")() - assert res2 == "fresh_op_success" + res_c = conn.run_on_worker(lambda: "result_c")() + assert res_c == "result_c", "Worker must remain usable for subsequent request C" + assert conn._worker_state == 'IDLE' + + +def test_concurrency_a_raises_exception_b_and_c_still_run(): + conn = authconn.Connection(host="127.0.0.1", port=631, encryption=cups.HTTP_ENCRYPT_NEVER) + + def failing_op(): + raise cups.IPPError(cups.IPP_NOT_FOUND, "printer not found") + + with pytest.raises(cups.IPPError) as exc_info: + conn.run_on_worker(failing_op)() + assert exc_info.value.args[0] == cups.IPP_NOT_FOUND + + res_b = conn.run_on_worker(lambda: "result_b")() + assert res_b == "result_b" + + res_c = conn.run_on_worker(lambda: "result_c")() + assert res_c == "result_c" def test_authentication_retry(monkeypatch): @@ -351,26 +385,51 @@ def getServerPPD(self, name): assert "root" in connections_created -def test_late_result_after_cancellation(): +def test_concurrency_fifo_ordering_preserved(): conn = authconn.Connection(host="127.0.0.1", port=631, encryption=cups.HTTP_ENCRYPT_NEVER) - worker_started = threading.Event() - worker_finish = threading.Event() + a_started = threading.Event() + a_release = threading.Event() + execution_order = [] - def slow_worker_op(): - worker_started.set() - worker_finish.wait(timeout=2.0) - return "late_unwanted_result" + def op_a(): + a_started.set() + a_release.wait(timeout=5.0) + execution_order.append("A") + return "result_a" - threading.Thread(target=lambda: (worker_started.wait(), conn.cancel()), daemon=True).start() + def op_b(): + execution_order.append("B") + return "result_b" - with pytest.raises(authconn.OperationCanceled): - conn.run_on_worker(slow_worker_op)() + def op_c(): + execution_order.append("C") + return "result_c" - worker_finish.set() - conn._worker_queue.join() + results = {} + + def run_op(name, fn): + results[name] = conn.run_on_worker(fn)() + + t_a = threading.Thread(target=run_op, args=("A", op_a), daemon=True) + t_a.start() + assert a_started.wait(timeout=2.0) + + t_b = threading.Thread(target=run_op, args=("B", op_b), daemon=True) + t_b.start() + time.sleep(0.05) - res_next = conn.run_on_worker(lambda: "proper_next_result")() - assert res_next == "proper_next_result" + t_c = threading.Thread(target=run_op, args=("C", op_c), daemon=True) + t_c.start() + time.sleep(0.05) + + a_release.set() + + t_a.join(timeout=2.0) + t_b.join(timeout=2.0) + t_c.join(timeout=2.0) + + assert execution_order == ["A", "B", "C"], "FIFO execution order must be preserved" + assert results == {"A": "result_a", "B": "result_b", "C": "result_c"} def test_cleanup_of_discarded_temporary_files(): @@ -388,15 +447,19 @@ def blocking_ppd_op(): worker_finish.wait(timeout=2.0) return temp_ppd_path - threading.Thread(target=lambda: (worker_started.wait(), conn.cancel()), daemon=True).start() + with pytest.raises(authconn.WorkerTimeoutError): + conn.run_on_worker(blocking_ppd_op)(_timeout=0.08) - with pytest.raises(authconn.OperationCanceled): - conn.run_on_worker(blocking_ppd_op)() + assert os.path.exists(temp_ppd_path), "File still exists while worker is blocked" worker_finish.set() conn._worker_queue.join() - assert not os.path.exists(temp_ppd_path), "Cancelled temporary file must be unlinked" + assert not os.path.exists(temp_ppd_path), "Timed out temporary file must be unlinked by worker" + + res = conn.run_on_worker(lambda: "recovered")() + assert res == "recovered" + assert conn._worker_state == 'IDLE' def test_gtk_responsiveness_during_worker_execution(): @@ -472,100 +535,3 @@ def on_after_delay(): assert len(shown_events) == 1 assert len(stopped_events) == 1 - - -def test_cancelled_in_queue_skips_execution(): - conn = authconn.Connection(host="127.0.0.1", port=631, encryption=cups.HTTP_ENCRYPT_NEVER) - gate = threading.Event() - skipped_executed = [False] - - def blocking_op(): - gate.wait(timeout=2.0) - return "block_done" - - def skipped_op(): - skipped_executed[0] = True - return "should_not_run" - - t1 = threading.Thread(target=lambda: conn.run_on_worker(blocking_op)(), daemon=True) - t1.start() - - time.sleep(0.05) - - with pytest.raises(authconn.OperationCanceled): - conn.run_on_worker(skipped_op, cancel_check=lambda: True)() - - gate.set() - conn._worker_queue.join() - - assert not skipped_executed[0], "Queued request with req_state['cancelled'] must be skipped" - - -def test_cancellation_of_request_a_does_not_affect_request_b(): - conn = authconn.Connection(host="127.0.0.1", port=631, encryption=cups.HTTP_ENCRYPT_NEVER) - a_started = threading.Event() - a_release = threading.Event() - a_errors = [] - a_results = [] - b_errors = [] - b_results = [] - - def op_a(): - a_started.set() - a_release.wait(timeout=2.0) - return "result_a" - - def op_b(): - return "result_b" - - def run_a(): - try: - res = conn.run_on_worker(op_a)() - a_results.append(res) - except Exception as e: - a_errors.append(e) - - def run_b(): - try: - res = conn.run_on_worker(op_b)() - b_results.append(res) - except Exception as e: - b_errors.append(e) - - t_a = threading.Thread(target=run_a, daemon=True) - t_a.start() - assert a_started.wait(timeout=2.0), "Request A failed to start" - - t_b = threading.Thread(target=run_b, daemon=True) - t_b.start() - - deadline = time.monotonic() + 2.0 - req_b = None - while time.monotonic() < deadline: - with conn._worker_lock: - if len(conn._pending_requests) >= 2: - req_b = conn._pending_requests[1] - break - time.sleep(0.01) - assert req_b is not None, "Request B was not queued in pending requests within timeout" - - conn.cancel() - - t_a.join(timeout=2.0) - assert not t_a.is_alive(), "Thread A did not exit after cancellation" - assert len(a_errors) == 1, "Request A should have raised an exception" - assert isinstance(a_errors[0], authconn.OperationCanceled), "Request A should raise OperationCanceled" - - assert req_b['cancelled'] is False, "Request B's req_state must NOT be marked cancelled" - - a_release.set() - - t_b.join(timeout=2.0) - assert not t_b.is_alive(), "Thread B did not finish" - assert len(b_errors) == 0, f"Request B failed unexpectedly: {b_errors}" - assert b_results == ["result_b"], "Request B should return expected result" - - res_c = conn.run_on_worker(lambda: "result_c")() - assert res_c == "result_c", "Independent Request C should execute and return normally" - assert conn._cancel is False, "Connection cancellation flag should be reset for fresh operation" -