23 import ubuntuuitoolkit
24 from autopilot.introspection
import get_proxy_object_for_existing_process
26 from unity8
import greeter
29 logger = logging.getLogger(__name__)
32 class JobError(Exception):
36 class CannotAccessUnity(Exception):
41 """Helper function that attempts to unlock the unity greeter.
44 greeter.wait_for_greeter()
45 greeter.hide_greeter_with_dbus()
49 """Helper function that attempts to lock unity greeter.
52 greeter.show_greeter_with_dbus()
53 greeter.wait_for_greeter()
56 def restart_unity_with_testability(*args):
57 """Restarts (or starts) unity with testability enabled.
59 Passes *args arguments to the launched process.
62 args += (
"QT_LOAD_TESTABILITY=1",)
63 return restart_unity(*args)
66 def restart_unity(*args):
67 """Restarts (or starts) unity8 using the provided arguments.
69 Passes *args arguments to the launched process.
71 :raises subprocess.CalledProcessError: if unable to stop or start the
75 status = _get_unity_status()
76 if "start/" in status:
79 pid = start_job(
'unity8', *args)
80 return _get_unity_proxy_object(pid)
83 def start_job(name, *args):
86 :param str name: The name of the job.
87 :param args: The arguments to be used when starting the job.
88 :return: The process id of the started job.
89 :raises CalledProcessError: if the job failed to start.
92 logger.info(
'Starting job {} with arguments {}.'.format(name, args))
93 command = [
'/sbin/initctl',
'start', name] + list(args)
95 output = subprocess.check_output(
97 stderr=subprocess.STDOUT,
98 universal_newlines=
True,
101 pid = get_job_pid(name)
102 except subprocess.CalledProcessError
as e:
103 e.args += (
'Failed to start {}: {}.'.format(name, e.output),)
109 def get_job_pid(name):
110 """Return the process id of a running job.
112 :param str name: The name of the job.
113 :raises JobError: if the job is not running.
116 status = get_job_status(name)
117 if "start/" not in status:
118 raise JobError(
'{} is not in the running state.'.format(name))
119 return int(status.split()[-1])
122 def get_job_status(name):
123 """Return the status of a job.
125 :param str name: The name of the job.
126 :raises JobError: if it's not possible to get the status of the job.
130 return subprocess.check_output([
134 ], universal_newlines=
True)
135 except subprocess.CalledProcessError
as error:
137 "Unable to get {}'s status: {}".format(name, error)
144 :param str name: The name of the job.
145 :raises CalledProcessError: if the job failed to stop.
148 logger.info(
'Stoping job {}.'.format(name))
149 command = [
'/sbin/initctl',
'stop', name]
151 output = subprocess.check_output(
153 stderr=subprocess.STDOUT,
154 universal_newlines=
True,
157 except subprocess.CalledProcessError
as e:
158 e.args += (
'Failed to stop {}: {}.'.format(name, e.output),)
162 def is_job_running(name):
163 """Return True if the job is running. Otherwise, False.
165 :param str name: The name of the job.
166 :raises JobError: if it's not possible to get the status of the job.
169 return 'start/' in get_job_status(name)
172 def _get_unity_status():
174 return get_job_status(
'unity8')
175 except JobError
as error:
176 raise CannotAccessUnity(str(error))
179 def _get_unity_pid():
181 return get_job_pid(
'unity8')
182 except JobError
as error:
183 raise CannotAccessUnity(str(error))
186 def _get_unity_proxy_object(pid):
187 return get_proxy_object_for_existing_process(
189 emulator_base=ubuntuuitoolkit.UbuntuUIToolkitCustomProxyObjectBase