1# Copyright (c) Twisted Matrix Laboratories. 2# See LICENSE for details. 3 4""" 5GI/GTK3 reactor tests. 6""" 7 8from __future__ import division, absolute_import 9 10import sys, os 11try: 12 from twisted.internet import gireactor 13 from gi.repository import Gio 14except ImportError: 15 gireactor = None 16 gtk3reactor = None 17else: 18 # gtk3reactor may be unavailable even if gireactor is available; in 19 # particular in pygobject 3.4/gtk 3.6, when no X11 DISPLAY is found. 20 try: 21 from twisted.internet import gtk3reactor 22 except ImportError: 23 gtk3reactor = None 24 else: 25 from gi.repository import Gtk 26 27from twisted.python.filepath import FilePath 28from twisted.python.runtime import platform 29from twisted.internet.defer import Deferred 30from twisted.internet.error import ReactorAlreadyRunning 31from twisted.internet.protocol import ProcessProtocol 32from twisted.trial.unittest import TestCase, SkipTest 33from twisted.internet.test.reactormixins import ReactorBuilder 34from twisted.test.test_twisted import SetAsideModule 35from twisted.internet.interfaces import IReactorProcess 36 37# Skip all tests if gi is unavailable: 38if gireactor is None: 39 skip = "gtk3/gi not importable" 40 41 42class GApplicationRegistration(ReactorBuilder, TestCase): 43 """ 44 GtkApplication and GApplication are supported by 45 L{twisted.internet.gtk3reactor} and L{twisted.internet.gireactor}. 46 47 We inherit from L{ReactorBuilder} in order to use some of its 48 reactor-running infrastructure, but don't need its test-creation 49 functionality. 50 """ 51 def runReactor(self, app, reactor): 52 """ 53 Register the app, run the reactor, make sure app was activated, and 54 that reactor was running, and that reactor can be stopped. 55 """ 56 if not hasattr(app, "quit"): 57 raise SkipTest("Version of PyGObject is too old.") 58 59 result = [] 60 def stop(): 61 result.append("stopped") 62 reactor.stop() 63 def activate(widget): 64 result.append("activated") 65 reactor.callLater(0, stop) 66 app.connect('activate', activate) 67 68 # We want reactor.stop() to *always* stop the event loop, even if 69 # someone has called hold() on the application and never done the 70 # corresponding release() -- for more details see 71 # http://developer.gnome.org/gio/unstable/GApplication.html. 72 app.hold() 73 74 reactor.registerGApplication(app) 75 ReactorBuilder.runReactor(self, reactor) 76 self.assertEqual(result, ["activated", "stopped"]) 77 78 79 def test_gApplicationActivate(self): 80 """ 81 L{Gio.Application} instances can be registered with a gireactor. 82 """ 83 reactor = gireactor.GIReactor(useGtk=False) 84 self.addCleanup(self.unbuildReactor, reactor) 85 app = Gio.Application( 86 application_id='com.twistedmatrix.trial.gireactor', 87 flags=Gio.ApplicationFlags.FLAGS_NONE) 88 89 self.runReactor(app, reactor) 90 91 92 def test_gtkApplicationActivate(self): 93 """ 94 L{Gtk.Application} instances can be registered with a gtk3reactor. 95 """ 96 reactor = gtk3reactor.Gtk3Reactor() 97 self.addCleanup(self.unbuildReactor, reactor) 98 app = Gtk.Application( 99 application_id='com.twistedmatrix.trial.gtk3reactor', 100 flags=Gio.ApplicationFlags.FLAGS_NONE) 101 102 self.runReactor(app, reactor) 103 104 if gtk3reactor is None: 105 test_gtkApplicationActivate.skip = ( 106 "Gtk unavailable (may require running with X11 DISPLAY env set)") 107 108 109 def test_portable(self): 110 """ 111 L{gireactor.PortableGIReactor} doesn't support application 112 registration at this time. 113 """ 114 reactor = gireactor.PortableGIReactor() 115 self.addCleanup(self.unbuildReactor, reactor) 116 app = Gio.Application( 117 application_id='com.twistedmatrix.trial.gireactor', 118 flags=Gio.ApplicationFlags.FLAGS_NONE) 119 self.assertRaises(NotImplementedError, 120 reactor.registerGApplication, app) 121 122 123 def test_noQuit(self): 124 """ 125 Older versions of PyGObject lack C{Application.quit}, and so won't 126 allow registration. 127 """ 128 reactor = gireactor.GIReactor(useGtk=False) 129 self.addCleanup(self.unbuildReactor, reactor) 130 # An app with no "quit" method: 131 app = object() 132 exc = self.assertRaises(RuntimeError, reactor.registerGApplication, app) 133 self.assertTrue(exc.args[0].startswith( 134 "Application registration is not")) 135 136 137 def test_cantRegisterAfterRun(self): 138 """ 139 It is not possible to register a C{Application} after the reactor has 140 already started. 141 """ 142 reactor = gireactor.GIReactor(useGtk=False) 143 self.addCleanup(self.unbuildReactor, reactor) 144 app = Gio.Application( 145 application_id='com.twistedmatrix.trial.gireactor', 146 flags=Gio.ApplicationFlags.FLAGS_NONE) 147 148 def tryRegister(): 149 exc = self.assertRaises(ReactorAlreadyRunning, 150 reactor.registerGApplication, app) 151 self.assertEqual(exc.args[0], 152 "Can't register application after reactor was started.") 153 reactor.stop() 154 reactor.callLater(0, tryRegister) 155 ReactorBuilder.runReactor(self, reactor) 156 157 158 def test_cantRegisterTwice(self): 159 """ 160 It is not possible to register more than one C{Application}. 161 """ 162 reactor = gireactor.GIReactor(useGtk=False) 163 self.addCleanup(self.unbuildReactor, reactor) 164 app = Gio.Application( 165 application_id='com.twistedmatrix.trial.gireactor', 166 flags=Gio.ApplicationFlags.FLAGS_NONE) 167 reactor.registerGApplication(app) 168 app2 = Gio.Application( 169 application_id='com.twistedmatrix.trial.gireactor2', 170 flags=Gio.ApplicationFlags.FLAGS_NONE) 171 exc = self.assertRaises(RuntimeError, 172 reactor.registerGApplication, app2) 173 self.assertEqual(exc.args[0], 174 "Can't register more than one application instance.") 175 176 177 178class PygtkCompatibilityTests(TestCase): 179 """ 180 pygtk imports are either prevented, or a compatiblity layer is used if 181 possible. 182 """ 183 184 def test_noCompatibilityLayer(self): 185 """ 186 If no compatiblity layer is present, imports of gobject and friends 187 are disallowed. 188 189 We do this by running a process where we make sure gi.pygtkcompat 190 isn't present. 191 """ 192 from twisted.internet import reactor 193 if not IReactorProcess.providedBy(reactor): 194 raise SkipTest("No process support available in this reactor.") 195 196 result = Deferred() 197 class Stdout(ProcessProtocol): 198 data = b"" 199 200 def errReceived(self, err): 201 print(err) 202 203 def outReceived(self, data): 204 self.data += data 205 206 def processExited(self, reason): 207 result.callback(self.data) 208 209 path = FilePath(__file__.encode("utf-8")).sibling( 210 b"process_gireactornocompat.py").path 211 reactor.spawnProcess(Stdout(), sys.executable, [sys.executable, path], 212 env=os.environ) 213 result.addCallback(self.assertEqual, b"success") 214 return result 215 216 217 def test_compatibilityLayer(self): 218 """ 219 If compatiblity layer is present, importing gobject uses the gi 220 compatibility layer. 221 """ 222 if "gi.pygtkcompat" not in sys.modules: 223 raise SkipTest("This version of gi doesn't include pygtkcompat.") 224 import gobject 225 self.assertTrue(gobject.__name__.startswith("gi.")) 226 227 228 229class Gtk3ReactorTests(TestCase): 230 """ 231 Tests for L{gtk3reactor}. 232 """ 233 234 def test_requiresDISPLAY(self): 235 """ 236 On X11, L{gtk3reactor} is unimportable if the C{DISPLAY} environment 237 variable is not set. 238 """ 239 display = os.environ.get("DISPLAY", None) 240 if display is not None: 241 self.addCleanup(os.environ.__setitem__, "DISPLAY", display) 242 del os.environ["DISPLAY"] 243 with SetAsideModule("twisted.internet.gtk3reactor"): 244 exc = self.assertRaises(ImportError, 245 __import__, "twisted.internet.gtk3reactor") 246 self.assertEqual( 247 exc.args[0], 248 "Gtk3 requires X11, and no DISPLAY environment variable is set") 249 250 if platform.getType() != "posix" or platform.isMacOSX(): 251 test_requiresDISPLAY.skip = "This test is only relevant when using X11" 252