# -*- coding: utf-8 -*- # Copyright (C) 2010, 2011, 2012, 2013 Sebastian Wiesner # This library is free software; you can redistribute it and/or modify it # under the terms of the GNU Lesser General Public License as published by the # Free Software Foundation; either version 2.1 of the License, or (at your # option) any later version. # This library 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 Lesser General Public License # for more details. # You should have received a copy of the GNU Lesser General Public License # along with this library; if not, write to the Free Software Foundation, # Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA # isort: STDLIB import random # isort: THIRDPARTY import pytest # isort: FIRSTPARTY from tests.utils.udev import DeviceDatabase # isort: LOCAL from pyudev import Devices, Monitor try: # isort: STDLIB from unittest import mock except ImportError: # isort: THIRDPARTY import mock @pytest.fixture def monitor(request): return Monitor.from_netlink(request.getfixturevalue("context")) @pytest.fixture def fake_monitor_device(request): context = request.getfixturevalue("context") device = random.choice(list(DeviceDatabase.db())) return Devices.from_path(context, device.device_path) def test_fake_monitor(fake_monitor, fake_monitor_device): """ Test the fake monitor just to make sure, that it works. """ assert fake_monitor.poll(timeout=0) is None fake_monitor.trigger_event() device = fake_monitor.poll() assert device == fake_monitor_device class ObserverTestBase: def setup_method(self, method): self.observer = None self.no_emitted_signals = 0 self.setup() def teardown_method(self, method): if self.observer is not None: self.destroy_observer() self.teardown() def setup(self): pass def teardown(self): pass def destroy_observer(self): self.observer.enabled = False def create_observer(self, monitor): raise NotImplementedError() def create_event_loop(self, self_stop_timeout=5000): raise NotImplementedError() def start_event_loop(self, start_callback): raise NotImplementedError() def stop_event_loop(self): raise NotImplementedError() def connect_signal(self, callback, action=None): raise NotImplementedError() def prepare_test(self, monitor): self.create_event_loop(self_stop_timeout=5000) self.create_observer(monitor) def test_monitor(self, fake_monitor): self.prepare_test(fake_monitor) # test that the monitor attribute is correct assert self.observer.monitor is fake_monitor def test_events_fake_monitor(self, fake_monitor, fake_monitor_device): self.prepare_test(fake_monitor) event_callback = mock.Mock(side_effect=lambda *args: self.stop_event_loop()) self.connect_signal(event_callback) self.start_event_loop(fake_monitor.trigger_event) event_callback.assert_called_with(fake_monitor_device) class QtObserverTestBase(ObserverTestBase): def setup(self): self.qtcore = pytest.importorskip("{0}.QtCore".format(self.BINDING_NAME)) def create_observer(self, monitor): name = self.BINDING_NAME.lower() mod = __import__("pyudev.{0}".format(name), None, None, [name]) self.observer = mod.MonitorObserver(monitor) def connect_signal(self, callback): self.observer.deviceEvent.connect(callback) def create_event_loop(self, self_stop_timeout=5000): self.app = self.qtcore.QCoreApplication.instance() if not self.app: self.app = self.qtcore.QCoreApplication([]) self.qtcore.QTimer.singleShot(self_stop_timeout, self.stop_event_loop) def start_event_loop(self, start_callback): self.qtcore.QTimer.singleShot(0, start_callback) self.app.exec_() def stop_event_loop(self): self.app.quit() class TestPysideObserver(QtObserverTestBase): BINDING_NAME = "PySide" class TestPyQt4Observer(QtObserverTestBase): BINDING_NAME = "PyQt4" class TestPyQt5Observer(QtObserverTestBase): BINDING_NAME = "PyQt5" class TestGlibObserver(ObserverTestBase): def setup(self): self.event_sources = [] pytest.importorskip("gi") self.glib = pytest.importorskip("gi.repository.GLib") def teardown(self): for source in self.event_sources: self.glib.source_remove(source) def create_observer(self, monitor): # isort: LOCAL from pyudev.glib import MonitorObserver self.observer = MonitorObserver(monitor) def connect_signal(self, callback): # drop the sender argument from glib signal connections def _wrapper(obj, *args, **kwargs): return callback(*args, **kwargs) self.observer.connect("device-event", _wrapper) def create_event_loop(self, self_stop_timeout=5000): self.mainloop = self.glib.MainLoop() self.event_sources.append( self.glib.timeout_add(self_stop_timeout, self.stop_event_loop) ) def start_event_loop(self, start_callback): def _wrapper(*args, **kwargs): start_callback(*args, **kwargs) return False self.event_sources.append(self.glib.timeout_add(0, _wrapper)) self.mainloop.run() def stop_event_loop(self): self.mainloop.quit() return False @pytest.mark.skipif( str('"DISPLAY" not in os.environ'), reason="Display required for wxPython" ) class TestWxObserver(ObserverTestBase): def setup(self): self.wx = pytest.importorskip("wx") def create_observer(self, monitor): # isort: LOCAL from pyudev import wx self.observer = wx.MonitorObserver(monitor) def connect_signal(self, callback): # isort: LOCAL from pyudev.wx import EVT_DEVICE_EVENT def _wrapper(event): return callback(event.device) self.observer.Bind(EVT_DEVICE_EVENT, _wrapper) def create_event_loop(self, self_stop_timeout=5000): self.app = self.wx.App(False) # need to create a dummy Frame to get the mainloop running. Praise the # broken wx API… self.app.frame = self.wx.Frame(None) timer = self.wx.PyTimer(self.stop_event_loop) timer.Start(self_stop_timeout, True) def start_event_loop(self, start_callback): timer = self.wx.PyTimer(start_callback) timer.Start(0, True) self.app.MainLoop() def stop_event_loop(self): self.app.Exit()