Unity 8
process_helpers.py
1 # -*- Mode: Python; coding: utf-8; indent-tabs-mode: nil; tab-width: 4 -*-
2 #
3 # Unity Autopilot Utilities
4 # Copyright (C) 2013, 2014, 2015 Canonical
5 #
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.
10 #
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.
15 #
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/>.
18 #
19 
20 import logging
21 import subprocess
22 
23 import ubuntuuitoolkit
24 from autopilot.introspection import get_proxy_object_for_existing_process
25 
26 from unity8 import greeter
27 
28 
29 logger = logging.getLogger(__name__)
30 
31 
32 class JobError(Exception):
33  pass
34 
35 
36 class CannotAccessUnity(Exception):
37  pass
38 
39 
40 def unlock_unity():
41  """Helper function that attempts to unlock the unity greeter.
42 
43  """
44  greeter.wait_for_greeter()
45  greeter.hide_greeter_with_dbus()
46 
47 
48 def lock_unity():
49  """Helper function that attempts to lock unity greeter.
50 
51  """
52  greeter.show_greeter_with_dbus()
53  greeter.wait_for_greeter()
54 
55 
56 def restart_unity_with_testability(*args):
57  """Restarts (or starts) unity with testability enabled.
58 
59  Passes *args arguments to the launched process.
60 
61  """
62  args += ("QT_LOAD_TESTABILITY=1",)
63  return restart_unity(*args)
64 
65 
66 def restart_unity(*args):
67  """Restarts (or starts) unity8 using the provided arguments.
68 
69  Passes *args arguments to the launched process.
70 
71  :raises subprocess.CalledProcessError: if unable to stop or start the
72  unity8 upstart job.
73 
74  """
75  status = _get_unity_status()
76  if "start/" in status:
77  stop_job('unity8')
78 
79  pid = start_job('unity8', *args)
80  return _get_unity_proxy_object(pid)
81 
82 
83 def start_job(name, *args):
84  """Start a job.
85 
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.
90 
91  """
92  logger.info('Starting job {} with arguments {}.'.format(name, args))
93  command = ['/sbin/initctl', 'start', name] + list(args)
94  try:
95  output = subprocess.check_output(
96  command,
97  stderr=subprocess.STDOUT,
98  universal_newlines=True,
99  )
100  logger.info(output)
101  pid = get_job_pid(name)
102  except subprocess.CalledProcessError as e:
103  e.args += ('Failed to start {}: {}.'.format(name, e.output),)
104  raise
105  else:
106  return pid
107 
108 
109 def get_job_pid(name):
110  """Return the process id of a running job.
111 
112  :param str name: The name of the job.
113  :raises JobError: if the job is not running.
114 
115  """
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])
120 
121 
122 def get_job_status(name):
123  """Return the status of a job.
124 
125  :param str name: The name of the job.
126  :raises JobError: if it's not possible to get the status of the job.
127 
128  """
129  try:
130  return subprocess.check_output([
131  '/sbin/initctl',
132  'status',
133  name
134  ], universal_newlines=True)
135  except subprocess.CalledProcessError as error:
136  raise JobError(
137  "Unable to get {}'s status: {}".format(name, error)
138  )
139 
140 
141 def stop_job(name):
142  """Stop a job.
143 
144  :param str name: The name of the job.
145  :raises CalledProcessError: if the job failed to stop.
146 
147  """
148  logger.info('Stoping job {}.'.format(name))
149  command = ['/sbin/initctl', 'stop', name]
150  try:
151  output = subprocess.check_output(
152  command,
153  stderr=subprocess.STDOUT,
154  universal_newlines=True,
155  )
156  logger.info(output)
157  except subprocess.CalledProcessError as e:
158  e.args += ('Failed to stop {}: {}.'.format(name, e.output),)
159  raise
160 
161 
162 def is_job_running(name):
163  """Return True if the job is running. Otherwise, False.
164 
165  :param str name: The name of the job.
166  :raises JobError: if it's not possible to get the status of the job.
167 
168  """
169  return 'start/' in get_job_status(name)
170 
171 
172 def _get_unity_status():
173  try:
174  return get_job_status('unity8')
175  except JobError as error:
176  raise CannotAccessUnity(str(error))
177 
178 
179 def _get_unity_pid():
180  try:
181  return get_job_pid('unity8')
182  except JobError as error:
183  raise CannotAccessUnity(str(error))
184 
185 
186 def _get_unity_proxy_object(pid):
187  return get_proxy_object_for_existing_process(
188  pid=pid,
189  emulator_base=ubuntuuitoolkit.UbuntuUIToolkitCustomProxyObjectBase
190  )