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  greeter.wait_for_greeter_gone()
47 
48 
49 def lock_unity():
50  """Helper function that attempts to lock unity greeter.
51 
52  """
53  greeter.show_greeter_with_dbus()
54  greeter.wait_for_greeter()
55 
56 
57 def restart_unity_with_testability(*args):
58  """Restarts (or starts) unity with testability enabled.
59 
60  Passes *args arguments to the launched process.
61 
62  """
63  args += ("QT_LOAD_TESTABILITY=1",)
64  return restart_unity(*args)
65 
66 
67 def restart_unity(*args):
68  """Restarts (or starts) unity8 using the provided arguments.
69 
70  Passes *args arguments to the launched process.
71 
72  :raises subprocess.CalledProcessError: if unable to stop or start the
73  unity8 upstart job.
74 
75  """
76  status = _get_unity_status()
77  if "start/" in status:
78  stop_job('unity8')
79 
80  pid = start_job('unity8', *args)
81  return _get_unity_proxy_object(pid)
82 
83 
84 def start_job(name, *args):
85  """Start a job.
86 
87  :param str name: The name of the job.
88  :param args: The arguments to be used when starting the job.
89  :return: The process id of the started job.
90  :raises CalledProcessError: if the job failed to start.
91 
92  """
93  logger.info('Starting job {} with arguments {}.'.format(name, args))
94  command = ['/sbin/initctl', 'start', name] + list(args)
95  try:
96  output = subprocess.check_output(
97  command,
98  stderr=subprocess.STDOUT,
99  universal_newlines=True,
100  )
101  logger.info(output)
102  pid = get_job_pid(name)
103  except subprocess.CalledProcessError as e:
104  e.args += ('Failed to start {}: {}.'.format(name, e.output),)
105  raise
106  else:
107  return pid
108 
109 
110 def get_job_pid(name):
111  """Return the process id of a running job.
112 
113  :param str name: The name of the job.
114  :raises JobError: if the job is not running.
115 
116  """
117  status = get_job_status(name)
118  if "start/" not in status:
119  raise JobError('{} is not in the running state.'.format(name))
120  return int(status.split()[-1])
121 
122 
123 def get_job_status(name):
124  """Return the status of a job.
125 
126  :param str name: The name of the job.
127  :raises JobError: if it's not possible to get the status of the job.
128 
129  """
130  try:
131  return subprocess.check_output([
132  '/sbin/initctl',
133  'status',
134  name
135  ], universal_newlines=True)
136  except subprocess.CalledProcessError as error:
137  raise JobError(
138  "Unable to get {}'s status: {}".format(name, error)
139  )
140 
141 
142 def stop_job(name):
143  """Stop a job.
144 
145  :param str name: The name of the job.
146  :raises CalledProcessError: if the job failed to stop.
147 
148  """
149  logger.info('Stopping job {}.'.format(name))
150  command = ['/sbin/initctl', 'stop', name]
151  try:
152  output = subprocess.check_output(
153  command,
154  stderr=subprocess.STDOUT,
155  universal_newlines=True,
156  )
157  logger.info(output)
158  except subprocess.CalledProcessError as e:
159  e.args += ('Failed to stop {}: {}.'.format(name, e.output),)
160  raise
161 
162 
163 def is_job_running(name):
164  """Return True if the job is running. Otherwise, False.
165 
166  :param str name: The name of the job.
167  :raises JobError: if it's not possible to get the status of the job.
168 
169  """
170  return 'start/' in get_job_status(name)
171 
172 
173 def _get_unity_status():
174  try:
175  return get_job_status('unity8')
176  except JobError as error:
177  raise CannotAccessUnity(str(error))
178 
179 
180 def _get_unity_pid():
181  try:
182  return get_job_pid('unity8')
183  except JobError as error:
184  raise CannotAccessUnity(str(error))
185 
186 
187 def _get_unity_proxy_object(pid):
188  return get_proxy_object_for_existing_process(
189  pid=pid,
190  emulator_base=ubuntuuitoolkit.UbuntuUIToolkitCustomProxyObjectBase
191  )