1
1
# -*- Mode: Python; coding: utf-8; indent-tabs-mode: nil; tab-width: 4 -*-
2
# Copyright 2012 Canonical
3
# Author: Thomi Richards
5
# This program is free software: you can redistribute it and/or modify it
6
# under the terms of the GNU General Public License version 3, as published
7
# by the Free Software Foundation.
10
"""Package for introspection support."""
3
# Autopilot Functional Test Tool
4
# Copyright (C) 2012-2013 Canonical
6
# This program is free software: you can redistribute it and/or modify
7
# it under the terms of the GNU General Public License as published by
8
# the Free Software Foundation, either version 3 of the License, or
9
# (at your option) any later version.
11
# This program is distributed in the hope that it will be useful,
12
# but WITHOUT ANY WARRANTY; without even the implied warranty of
13
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14
# GNU General Public License for more details.
16
# You should have received a copy of the GNU General Public License
17
# along with this program. If not, see <http://www.gnu.org/licenses/>.
21
"""Package for introspection support.
23
This package contains the internal implementation of the autopilot introspection
24
mechanism, and probably isn't useful to most test authors.
27
from __future__ import absolute_import
13
from gi.repository import Gio
16
from testtools.content import text_content
17
32
from time import sleep
36
from autopilot.introspection.backends import DBusAddress
22
37
from autopilot.introspection.constants import (
24
39
QT_AUTOPILOT_IFACE,
25
40
AP_INTROSPECTION_IFACE,
26
DBUS_INTROSPECTION_IFACE,
28
42
from autopilot.introspection.dbus import (
29
clear_object_registry,
30
44
DBusIntrospectionObject,
31
object_passes_filters,
45
get_classname_from_path,
47
from autopilot.dbus_handler import get_session_bus
34
48
from autopilot.utilities import get_debug_logger
37
51
logger = logging.getLogger(__name__)
40
class ApplicationIntrospectionTestMixin(object):
41
"""A mix-in class to make launching applications for introsection easier.
43
.. important:: You should not instantiate this class directly. Instead, use
44
one of the derived classes.
48
def launch_test_application(self, application, *arguments, **kwargs):
49
"""Launch *application* and retrieve a proxy object for the application.
51
Use this method to launch a supported application and start testing it.
52
The application can be specified as:
54
* A Desktop file, either with or without a path component.
55
* An executable file, either with a path, or one that is in the $PATH.
57
This method supports the following keyword arguments:
59
* *launch_dir*. If set to a directory that exists the process will be
60
launched from that directory.
62
* *capture_output*. If set to True (the default), the process output
63
will be captured and attached to the test as test detail.
65
:raises: **ValueError** if unknown keyword arguments are passed.
66
:return: A proxy object that represents the application. Introspection
67
data is retrievable via this object.
70
if not isinstance(application, basestring):
71
raise TypeError("'application' parameter must be a string.")
72
cwd = kwargs.pop('launch_dir', None)
73
capture_output = kwargs.pop('capture_output', True)
75
raise ValueError("Unknown keyword arguments: %s." %
76
(', '.join( repr(k) for k in kwargs.keys())))
78
if application.endswith('.desktop'):
79
proc = Gio.DesktopAppInfo.new(application)
80
application = proc.get_executable()
82
path, args = self.prepare_environment(application, list(arguments))
84
process = launch_autopilot_enabled_process(path,
88
self.addCleanup(self._kill_process_and_attach_logs, process)
89
return get_autopilot_proxy_object_for_process(process)
54
def get_application_launcher(app_path):
55
"""Return an instance of :class:`ApplicationLauncher` that knows how to launch
56
the application at 'app_path'.
58
# TODO: this is a teeny bit hacky - we call ldd to check whether this application
59
# links to certain library. We're assuming that linking to libQt* or libGtk*
60
# means the application is introspectable. This excludes any non-dynamically
61
# linked executables, which we may need to fix further down the line.
63
ldd_output = subprocess.check_output(["ldd", app_path]).strip().lower()
64
except subprocess.CalledProcessError as e:
66
if 'libqtcore' in ldd_output or 'libqt5core' in ldd_output:
67
from autopilot.introspection.qt import QtApplicationLauncher
68
return QtApplicationLauncher()
69
elif 'libgtk' in ldd_output:
70
from autopilot.introspection.gtk import GtkApplicationLauncher
71
return GtkApplicationLauncher()
75
def get_application_launcher_from_string_hint(hint):
76
"""Return in instance of :class:`ApplicationLauncher` given a string hint."""
77
from autopilot.introspection.qt import QtApplicationLauncher
78
from autopilot.introspection.gtk import GtkApplicationLauncher
82
return QtApplicationLauncher()
84
return GtkApplicationLauncher()
88
def launch_application(launcher, application, *arguments, **kwargs):
89
"""Launch an application, and return a process object.
91
:param launcher: An instance of the :class:`ApplicationLauncher` class to
92
prepare the environment before launching the application itself.
95
if not isinstance(application, basestring):
96
raise TypeError("'application' parameter must be a string.")
97
cwd = kwargs.pop('launch_dir', None)
98
capture_output = kwargs.pop('capture_output', True)
100
raise ValueError("Unknown keyword arguments: %s." %
101
(', '.join( repr(k) for k in kwargs.keys())))
103
path, args = launcher.prepare_environment(application, list(arguments))
105
process = launch_process(path,
113
class ApplicationLauncher(object):
114
"""A class that knows how to launch an application with a certain type of
115
introspection enabled.
91
119
def prepare_environment(self, app_path, arguments):
92
120
"""Prepare the application, or environment to launch with autopilot-support.
100
128
raise NotImplementedError("Sub-classes must implement this method.")
102
def _kill_process_and_attach_logs(self, process):
104
logger.info("waiting for process to exit.")
106
if process.returncode is not None:
109
logger.info("Terminating process group, since it hasn't exited after 10 seconds.")
110
os.killpg(process.pid, signal.SIGTERM)
112
stdout, stderr = process.communicate()
113
self.addDetail('process-stdout', text_content(stdout))
114
self.addDetail('process-stderr', text_content(stderr))
117
def launch_autopilot_enabled_process(application, args, capture_output, **kwargs):
118
"""Launch an autopilot-enabled process and return the proxy object."""
132
def launch_process(application, args, capture_output, **kwargs):
133
"""Launch an autopilot-enabled process and return the process object."""
119
134
commandline = [application]
120
135
commandline.extend(args)
121
136
logger.info("Launching process: %r", commandline)
199
188
raise RuntimeError("Unable to find Autopilot interface.")
202
def make_proxy_object_from_service_name(service_name, obj_path):
203
"""Returns a root proxy object given a DBus service name."""
191
def get_child_pids(pid):
192
"""Get a list of all child process Ids, for the given parent.
195
def get_children(pid):
196
command = ['ps', '-o', 'pid', '--ppid', str(pid), '--noheaders']
198
raw_output = subprocess.check_output(command)
199
except subprocess.CalledProcessError:
201
return [int(p) for p in raw_output.split()]
204
data = get_children(pid)
208
data.extend(get_children(pid))
213
def get_dbus_address_object(service_name, obj_path, dbus_addr=None):
204
214
# parameters can sometimes be dbus.String instances, sometimes QString instances.
205
215
# it's easier to convert them here than at the calling sites.
206
216
service_name = str(service_name)
207
217
obj_path = str(obj_path)
209
proxy_bases = get_proxy_object_base_clases(service_name, obj_path)
210
cls_name, cls_state = get_proxy_object_class_name_and_state(service_name, obj_path)
219
if dbus_addr is not None:
220
be = DBusAddress.CustomBus(dbus_addr, service_name, obj_path)
222
be = DBusAddress.SessionBus(service_name, obj_path)
226
def make_proxy_object(data_source, emulator_base):
227
"""Returns a root proxy object given a DBus service name."""
229
proxy_bases = get_proxy_object_base_clases(data_source)
230
if emulator_base is None:
231
emulator_base = type('DefaultEmulatorBase', (CustomEmulatorBase,), {})
232
proxy_bases = proxy_bases + (emulator_base, )
233
cls_name, cls_state = get_proxy_object_class_name_and_state(data_source)
212
235
clsobj = type(str(cls_name),
214
dict(DBUS_SERVICE=service_name,
237
dict(_Backend = data_source
218
241
proxy = clsobj.get_root_instance()
222
def get_proxy_object_base_clases(service_name, obj_path):
245
def get_proxy_object_base_clases(backend):
223
246
"""Return tuple of the base classes to use when creating a proxy object
224
247
for the given service name & path.
242
263
return tuple(bases)
245
def get_proxy_object_class_name_and_state(service_name, obj_path):
266
def get_proxy_object_class_name_and_state(backend):
246
267
"""Return the class name and root state dictionary."""
247
dbus_object = get_session_bus().get_object(service_name, obj_path)
248
dbus_iface = dbus.Interface(dbus_object, AP_INTROSPECTION_IFACE)
249
return dbus_iface.GetState("/")[0]
252
class ApplicationProxyObect(DBusIntrospectionObject):
268
object_path, object_state = backend.introspection_iface.GetState("/")[0]
269
return get_classname_from_path(object_path), object_state
272
class ApplicationProxyObject(DBusIntrospectionObject):
253
273
"""A class that better supports query data from an application."""
255
def __init__(self, state, path_info=None):
256
super(ApplicationProxyObect, self).__init__(state, path_info)
275
def __init__(self, state, path):
276
super(ApplicationProxyObject, self).__init__(state, path)
257
277
self._process = None
259
def select_single(self, type_name='*', **kwargs):
260
"""Get a single node from the introspection tree, with type equal to
261
*type_name* and (optionally) matching the keyword filters present in
266
>>> app.select_single('QPushButton', objectName='clickme')
267
... returns a QPushButton whose 'objectName' property is 'clickme'.
269
If nothing is returned from the query, this method returns None.
271
:raises: **ValueError** if the query returns more than one item. *If you
272
want more than one item, use select_many instead*.
275
instances = self.select_many(type_name, **kwargs)
276
if len(instances) > 1:
277
raise ValueError("More than one item was returned for query")
282
def select_many(self, type_name='*', **kwargs):
283
"""Get a list of nodes from the introspection tree, with type equal to
284
*type_name* and (optionally) matching the keyword filters present in
289
>>> app.select_many('QPushButton', enabled=True)
290
... returns a list of QPushButtons that are enabled.
292
If you only want to get one item, use select_single instead.
295
logger.debug("Selecting objects of %s with attributes: %r",
296
'any type' if type_name == '*' else 'type ' + type_name,
299
path = "//%s" % type_name
300
state_dicts = self.get_state_by_path(path)
301
instances = [self.make_introspection_object(i) for i in state_dicts]
302
return filter(lambda i: object_passes_filters(i, **kwargs), instances)
304
279
def set_process(self, process):
305
280
"""Set the subprocess.Popen object of the process that this is a proxy for.
added
removed
autopilot/introspection/__init__.py
