/usr/share/pyshared/wimpiggy/test.py is in python-wimpiggy 0.0.7.36+dfsg-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 | # This file is part of Parti.
# Copyright (C) 2008, 2009 Nathaniel Smith <njs@pobox.com>
# Parti is released under the terms of the GNU GPL v2, or, at your option, any
# later version. See the file COPYING for details.
import subprocess
import sys
import os
import traceback
import atexit
import errno
import gobject
import gtk.gdk
from wimpiggy.util import one_arg_signal
# Skip contents of this file when looking for tests
__test__ = False
def assert_raises(exc_class, f, *args, **kwargs):
# exc_class can be a tuple.
try:
value = f(*args, **kwargs)
except exc_class:
pass
except:
(cls, e, tb) = sys.exc_info()
raise AssertionError, (("unexpected exception: %s: %s\n"
+ "Original traceback:\n%s")
% (cls, e, traceback.format_exc()))
else:
raise AssertionError, \
"wanted exception, got normal return (%r)" % (value,)
def assert_emits(f, obj, signal, slot=None):
"""Call f(obj), and assert that 'signal' is emitted. Optionally, also
passes signaled data to 'slot', which may make its own assertions."""
backchannel = {
"signal_was_emitted": False,
"slot_exc": None,
}
def real_slot(*args, **kwargs):
backchannel["signal_was_emitted"] = True
if slot is not None:
try:
slot(*args, **kwargs)
except:
backchannel["slot_exc"] = sys.exc_info()
connection = obj.connect(signal, real_slot)
try:
f(obj)
finally:
obj.disconnect(connection)
assert backchannel["signal_was_emitted"]
if backchannel["slot_exc"] is not None:
exc = backchannel["slot_exc"]
raise exc[0], exc[1], exc[2]
def assert_mainloop_emits(obj, signal, slot=None):
"""Runs the mainloop and asserts that 'signal' is emitted. Optionally,
also passes signaled data to 'slot', which may make its own assertions."""
def real_slot(*args, **kwargs):
gtk.main_quit()
if slot is not None:
slot(*args, **kwargs)
assert_emits(lambda x: gtk.main(), obj, signal, real_slot)
class Session(object):
def __init__(self, display_name):
self._my_process = os.getpid()
self.display_name = display_name
self._x11 = None
self._dbus = None
self._dbus_address = None
def _alive(self, pid):
# Just in case it's a zombie, reap it; otherwise, do nothing.
try:
os.waitpid(pid, os.WNOHANG)
except OSError, e:
pass
# Then use the old SIG 0 trick.
try:
os.kill(pid, 0)
except OSError, e:
if e.errno == errno.ESRCH:
return False
return True
def _x_really_running(self):
try:
d = gtk.gdk.Display(self.display_name)
d.close()
except RuntimeError:
return False
return True
def validate(self):
assert os.getpid() == self._my_process
# FIXME: add some sort of check in here that X has actually reset
# since the last time -- e.g., set a prop on the root window and make
# sure it isn't here anymore. This is to make sure that other
# connections got properly disconnected, etc.
if (self._x11 is None
or self._dbus is None
or not self._alive(self._x11.pid)
or not self._alive(self._dbus.pid)
or not self._x_really_running()):
self.destroy()
self._x11 = subprocess.Popen(["Xvfb-for-wimpiggy", self.display_name,
"-ac",
#"-audit", "10",
"+extension", "Composite",
# Need to set the depth like this to
# get non-paletted visuals:
"-screen", "0", "1024x768x24+32"],
executable="Xvfb",
stderr=open("/dev/null", "w"))
self._dbus = subprocess.Popen(["dbus-daemon-for-wimpiggy", "--session",
"--nofork", "--print-address"],
executable="dbus-daemon",
stdout=subprocess.PIPE)
self._dbus_address = self._dbus.stdout.readline().strip()
def destroy(self):
if os.getpid() != self._my_process:
return
if self._x11 is not None:
try:
os.kill(self._x11.pid, 15)
self._x11.wait()
except OSError:
pass
self._x11 = None
if self._dbus is not None:
try:
os.kill(self._dbus.pid, 15)
self._dbus.wait()
except OSError:
pass
self._dbus = None
self._dbus_address = None
_the_session = None
class TestWithSession(object):
"A test that runs with its own isolated X11 and D-Bus session."
display_name = ":13"
display = None
@classmethod
def preForkClassSetUp(cls):
global _the_session
if _the_session is None:
_the_session = Session(cls.display_name)
atexit.register(_the_session.destroy)
_the_session.validate()
def setUp(self):
# This is not a race condition, nor do we need to sleep here, because
# gtk.gdk.Display.__init__ is smart enough to silently block until the
# X server comes up. By using 127.0.0.1 explicitly we can force it to
# use TCP over loopback and that means wireshark can work.
self.display = gtk.gdk.Display("127.0.0.1" + self.display_name)
default_display = gtk.gdk.display_manager_get().get_default_display()
if default_display is not None:
default_display.close()
# This line is critical, because many gtk functions (even
# _for_display/_for_screen functions) actually use the default
# display, even if only temporarily. For instance,
# gtk_clipboard_for_display creates a GtkInvisible, which
# unconditionally sets its colormap (using the default display) before
# gtk_clipboard_for_display gets a chance to switch it to the proper
# display. So the end result is that we always need a valid default
# display of some sort:
gtk.gdk.display_manager_get().set_default_display(self.display)
print("Opened new display %r" % (self.display,))
os.environ["DBUS_SESSION_BUS_ADDRESS"] = _the_session._dbus_address
def tearDown(self):
# Could do cleanup here (close X11 connections, unset
# os.environ["DBUS_SESSION_BUS_ADDRESS"], etc.), but our test runner
# runs us in a forked off process that will exit immediately after
# this, so who cares?
pass
def clone_display(self):
clone = gtk.gdk.Display(self.display.get_name())
print("Cloned new display %r" % (clone,))
return clone
class MockEventReceiver(gobject.GObject):
__gsignals__ = {
"child-map-request-event": one_arg_signal,
"child-configure-request-event": one_arg_signal,
"wimpiggy-focus-in-event": one_arg_signal,
"wimpiggy-focus-out-event": one_arg_signal,
"wimpiggy-client-message-event": one_arg_signal,
"wimpiggy-map-event": one_arg_signal,
"wimpiggy-unmap-event": one_arg_signal,
"wimpiggy-child-map-event": one_arg_signal,
}
def do_child_map_request_event(self, event):
print("do_child_map_request_event")
assert False
def do_child_configure_request_event(self, event):
print("do_child_configure_request_event")
assert False
def do_wimpiggy_focus_in_event(self, event):
print("do_wimpiggy_focus_in_event")
assert False
def do_wimpiggy_focus_out_event(self, event):
print("do_wimpiggy_focus_out_event")
assert False
def do_wimpiggy_client_message_event(self, event):
print("do_wimpiggy_client_message_event")
assert False
def do_wimpiggy_map_event(self, event):
print("do_wimpiggy_map_event")
assert False
def do_wimpiggy_child_map_event(self, event):
print("do_wimpiggy_child_map_event")
assert False
def do_wimpiggy_unmap_event(self, event):
print("do_wimpiggy_unmap_event")
gobject.type_register(MockEventReceiver)
|