# -*- coding: utf-8 -*-
# Copyright: SCLE SFE
# Contributor: Julien Pagès <j.parkouss@gmail.com>
#
# This software is a computer program whose purpose is to test graphical
# applications written with the QT framework (http://qt.digia.com/).
#
# This software is governed by the CeCILL v2.1 license under French law and
# abiding by the rules of distribution of free software. You can use,
# modify and/ or redistribute the software under the terms of the CeCILL
# license as circulated by CEA, CNRS and INRIA at the following URL
# "http://www.cecill.info".
#
# As a counterpart to the access to the source code and rights to copy,
# modify and redistribute granted by the license, users are provided only
# with a limited warranty and the software's author, the holder of the
# economic rights, and the successive licensors have only limited
# liability.
#
# In this respect, the user's attention is drawn to the risks associated
# with loading, using, modifying and/or developing or reproducing the
# software by the user in light of its specific status of free software,
# that may mean that it is complicated to manipulate, and that also
# therefore means that it is reserved for developers and experienced
# professionals having in-depth computer knowledge. Users are therefore
# encouraged to load and test the software's suitability as regards their
# requirements in conditions enabling the security of their systems and/or
# data to be ensured and, more generally, to use and operate it in the
# same conditions as regards security.
#
# The fact that you are presently reading this means that you have had
# knowledge of the CeCILL v2.1 license and that you accept its terms.
"""
This module allow to communicate with a libFunq server with
:class:`FunqClient`.
"""
import socket
import json
import errno
import os
import shlex
import subprocess
import base64
from collections import defaultdict
import logging
from funq.aliases import HooqAliases
from funq.tools import wait_for
from funq.models import Widget
from funq.errors import FunqError, TimeOutError
LOG = logging.getLogger('funq.client')
[docs]class FunqClient(object):
"""
Allow to communicate with a libFunq server.
This is the main class used to manipulate tested application.
"""
DEFAULT_HOST = 'localhost'
DEFAULT_PORT = 9999
def __init__(self, host=None, port=None, aliases=None,
timeout_connection=10):
if host is None:
host = self.DEFAULT_HOST
if port is None:
port = self.DEFAULT_PORT
if aliases is None:
aliases = HooqAliases()
elif isinstance(aliases, basestring):
aliases = HooqAliases.from_file(aliases)
elif not isinstance(aliases, HooqAliases):
raise TypeError("aliases must be None or str or an"
" instance of HooqAliases")
self.aliases = aliases
self._socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
def connect():
""" try to connect """
try:
self._socket.connect((host, port))
return True
except socket.error, e:
if e.errno != errno.ECONNREFUSED:
raise
return e
wait_for(connect, timeout_connection, 0.2)
self._fsocket = self._socket.makefile(mode="rw")
[docs] def duplicate(self):
"""
Allow to manipulate the application in another thread.
Returns a new instance of :class:`FunqClient` with a new socket.
Example::
# `client_copy` may be used in concurrence with `client`.
client_copy = client.duplicate()
"""
host, port = self._socket.getpeername()
return FunqClient(host=host, port=port, aliases=self.aliases)
def close(self):
"""
Close the libFunq socket.
The instance become useless after this method is called. This
method is automatically called on the object destruction.
"""
self._socket.close()
def __del__(self):
self.close()
def _raw_send(self, action, kwargs):
"""
Send a message without waiting for an answer.
"""
kwargs['action'] = action
rawdata = json.dumps(kwargs)
message = '%s\n%s' % (len(rawdata), rawdata)
f = self._fsocket
f.write(message)
f.flush()
def send_command(self, action, **kwargs):
"""
Send a message to the libFunq server and returns the decoded
answer.
:raises: :class:`funq.errors.FunqError` on error
"""
self._raw_send(action, kwargs)
f = self._fsocket
header = f.readline()
if not header:
raise FunqError("NoResponseFromApplication",
u"Pas de réponse de l'application testée -"
u" probablement un crash.")
to_read = int(header)
response = json.loads(f.read(to_read))
if response.get('success') is False:
raise FunqError(response["errName"], response["errDesc"])
return response
def quit(self):
"""
Ask the tested application to quit by calling qApp->quit().
"""
self._raw_send('quit', {})
[docs] def take_screenshot(self, stream='screenshot.png', format_='PNG'):
"""
Take a screenshot of the active desktop.
"""
data = self.send_command('desktop_screenshot', format=format_)
if isinstance(stream, basestring):
stream = open(stream, 'wb')
raw = base64.standard_b64decode(data['data'])
stream.write(raw) # pylint: disable=E1103
[docs] def keyclick(self, text):
"""
Simulate keyboard entry by sending keypress and keyrelease events
for each character of the given text.
"""
self.send_command('widget_keyclick', text=text)
[docs] def shortcut(self, key_sequence):
"""
Send a shortcut defined with a text sequence. The format of this
text sequence is defined with QKeySequence::fromString (see QT
documentation for more details).
Example::
client.shortcut('F2')
"""
self.send_command('shortcut', keysequence=key_sequence)
[docs] def drag_n_drop(self, src_widget, src_pos=None,
dest_widget=None, dest_pos=None):
"""
Do a drag and drop.
:param src_widget: source widget
:param src_pos: starting position for the drag. If None, the center
of `src_widget` will be used, else it must be a
tuple (x, y) in widget coordinates.
:param dest_widget: destination widget. If None, src_widget will
be used.
:param dest_pos: ending position for the drop. If None, the center
of `dest_widget` will be used, else it must be a
tuple (x, y) in widget coordinates.
"""
if dest_widget is None:
dest_widget = src_widget
if src_pos is not None:
src_pos = ','.join(map(str, src_pos))
if dest_pos is not None:
dest_pos = ','.join(map(str, dest_pos))
self.send_command("drag_n_drop",
srcoid=src_widget.oid,
destoid=dest_widget.oid,
srcpos=src_pos,
destpos=dest_pos)
class ApplicationContext(object): # pylint: disable=R0903
"""
This is the context of a tested application.
Instanciate this class may launch the tested application with funq
(if appconfig.executable does not starts with "socket://").
Then it will try to connect to the libFunq server with a
:class:`FunqClient` accessible from the member **funq**.
When the instance is garbage collected, :meth:`terminate` is
automatically called to close the **funq** member and terminate
the tested application process.
"""
def __init__(self, appconfig, client_class=FunqClient):
self._process, self.funq = None, None
if not appconfig.executable.startswith('socket://'):
self._start_test_process(appconfig)
host = None # means localhost
else:
host = appconfig.executable[9:]
self.funq = client_class(
host=host,
port=appconfig.funq_port,
aliases=appconfig.create_aliases(),
timeout_connection=appconfig.timeout_connection
)
def _start_test_process(self, appconfig):
"""
Start the process of the tested application.
"""
env = appconfig.env
cmd = []
funq_port = appconfig.funq_port
stdout = appconfig.executable_stdout
stderr = appconfig.executable_stderr
if stderr:
if stderr == stdout:
stderr = subprocess.STDOUT
else:
stderr = open(stderr, 'a')
if stdout:
stdout = open(stdout, 'a')
if not appconfig.attach:
# libFunq is compiled inside the tested application binary
if env is None:
env = os.environ
# copy env
env = dict(env.items())
env['FUNQ_ACTIVATION'] = '1'
if funq_port:
env['FUNQ_PORT'] = str(funq_port)
else:
# inject libFunq with funq executable
if not appconfig.global_options.funq_attach_exe:
raise RuntimeError("To use funq, you have to specify the"
" nose option --funq-attach-exe"
" or put the funq executable in PATH")
cmd = [appconfig.global_options.funq_attach_exe]
if funq_port:
cmd.append('--port')
cmd.append(str(funq_port))
if appconfig.with_valgrind:
cmd.append('valgrind')
cmd.extend(appconfig.valgrind_args)
cmd.append(appconfig.executable)
cmd.extend(appconfig.args)
LOG.info("The tested application will be launched in the"
" directory %r with the command %r", appconfig.cwd, cmd)
self._process = subprocess.Popen(cmd,
cwd=appconfig.cwd,
stdout=stdout,
stderr=stderr,
env=env)
LOG.info("Launching tested application [%s].", self._process.pid)
def _kill_process(self):
"""
Kill the application tested process
"""
if self._process:
# wait for a nice exit
try:
wait_for(lambda: self._process.poll() is not None, 10, 0.05)
except TimeOutError:
pass
if self._process.returncode is None:
# application seems blocked ! try to terminate it ...
LOG.warn("The tested application [%s] can not be stopped"
" nicely.", self._process.pid)
self._process.terminate()
self._process.wait()
self._process = None
def terminate(self):
"""
Try to kill the process and close the **funq** object.
"""
if self.funq:
if self._process is not None:
# the process may be already dead
try:
wait_for(lambda: self._process.poll() is not None,
0.05,
0.01)
except TimeOutError:
pass
if self._process.returncode is not None:
# process terminated unexpectedly (-11: SegFault)
LOG.critical("The tested application [%s] has terminated"
" unexpectedly (return code: %s)",
self._process.pid, self._process.returncode)
self._process = None
else:
# try to exit nicely the tested application process
# with a call to qApp->quit().
LOG.info("Closing tested application [%s].",
self._process.pid)
try:
self.funq.quit()
except socket.error:
pass
try:
self.funq.close()
except socket.error:
pass
self.funq = None
self._kill_process()
def __del__(self):
self.terminate()
[docs]class ApplicationConfig(object): # pylint: disable=R0902
"""
This object hold the configuration of the application to test, mostly
retrieved from the funq configuration file.
Each parameter is accessible on the instance, allowing to retrieve
the tested application path for example with *config.executable*,
or its exeution path with *config.cwd*.
:param executable: complete path to the tested application
:param args: executable arguments
:param funq_port: socket port number for the libFunq connection
:param cwd: execution path for the tested application. If None, the
value will be the directory of executable.
:param env: dict environment variables. If None, os.environ will be
used.
:param timeout_connection: timeout to try to connect to libFunq.
:param aliases: path to the aliases file
:param executable_stdout: file path to redirect stdout or None.
:param executable_stderr: file path to redirect stderr or None.
:param attach: Indicate if the process is attached or if it is a
distant connection.
:param screenshot_on_error: Indicate if screenshots must be taken
on errors.
:param with_valgrind: indicate if valgrind must be used.
:param valgrind_args: valgrind arguments
:param global_options: options from the funq nose plugin.
"""
def __init__(self, executable, # pylint: disable=R0913
args=(),
funq_port=None,
cwd=None,
env=None,
timeout_connection=10,
aliases=None,
executable_stdout=None,
executable_stderr=None,
attach=True,
screenshot_on_error=False,
with_valgrind=False,
valgrind_args=('--leak-check=full',
'--show-reachable=yes'),
global_options=None):
self.executable = executable
self.args = args
self.funq_port = funq_port
self.cwd = cwd or os.path.dirname(executable) or os.getcwd()
self.env = env
self.timeout_connection = timeout_connection
self.aliases = aliases
self.executable_stdout = executable_stdout
self.executable_stderr = executable_stderr
self.attach = attach
self.screenshot_on_error = screenshot_on_error
self.with_valgrind = with_valgrind
self.valgrind_args = valgrind_args
self.global_options = global_options
def create_aliases(self):
"""
Create and returns and aliases object.
"""
if not self.aliases:
return None
return HooqAliases.from_file(self.aliases,
self.global_options.funq_gkit_file,
self.global_options.funq_gkit)
@classmethod
def from_conf(cls, conf, section, global_options):
"""
Create an instance of :class:`ApplicationConfig` from a
funq configuration section.
"""
basedir = os.path.dirname(global_options.funq_conf)
executable = conf.get(section, 'executable')
if not executable.startswith('socket://') and (
basedir and not os.path.isabs(executable)):
executable = os.path.join(basedir, executable)
kwargs = {'global_options': global_options}
if conf.has_option(section, 'args'):
kwargs['args'] = shlex.split(conf.get(section, 'args'))
if conf.has_option(section, 'funq_port'):
kwargs['funq_port'] = conf.getint(section, 'funq_port')
if kwargs['funq_port'] == 0 and \
not executable.startswith('socket://'):
# take an available port
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind(('', 0))
kwargs['funq_port'] = sock.getsockname()[1]
sock.close()
del sock
if conf.has_option(section, 'timeout_connection'):
kwargs['timeout_connection'] = conf.getint(section,
'timeout_connection')
if conf.has_option(section, 'attach'):
kwargs["attach"] = conf.getboolean(section, 'attach')
for optname in ('cwd', 'aliases', 'executable_stdout',
'executable_stderr'):
if conf.has_option(section, optname):
kwargs[optname] = conf.get(section, optname)
if basedir and not os.path.isabs(kwargs[optname]):
kwargs[optname] = os.path.join(basedir, kwargs[optname])
# devnull if NULL specified in the config file
for optname in ('executable_stdout', 'executable_stderr'):
if conf.has_option(section, optname) and \
conf.get(section, optname) == 'NULL':
kwargs[optname] = os.devnull
if conf.has_option(section, 'with_valgrind'):
kwargs["with_valgrind"] = \
conf.getboolean(section, 'with_valgrind')
if conf.has_option(section, 'valgrind_args'):
kwargs['valgrind_args'] = \
shlex.split(conf.get(section, 'valgrind_args'))
if conf.has_option(section, 'screenshot_on_error'):
kwargs["screenshot_on_error"] = \
conf.getboolean(section, 'screenshot_on_error')
return cls(executable, **kwargs)
class ApplicationRegistry(object):
"""
Handle multiple :class:`ApplicationConfig`. A global instance is
used in :mod:`funq.noseplugin` to keep every configuration defined
in the funq configuration file.
"""
def __init__(self):
self.confs = defaultdict(dict)
def register_from_conf(self, conf, global_options):
"""
Save configurations given a funq config.
"""
for section in conf.sections():
if ':' in section:
app, mode = section.split(':', 1)
else:
app = section
appconf = ApplicationConfig.from_conf(
conf, section, global_options)
self.register_config(app, appconf)
def register_config(self, name, conf):
""" Save the config *name* """
self.confs[name] = conf
def config(self, name):
"""
Returns the :class:`ApplicationConfig` associated to *name*.
:param name: name of the configuration
"""
return self.confs[name]