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