1#!/usr/bin/env python3

2# -*- coding: utf-8 -*-

3#

4# window_test.py

5#

6# This file is part of Checkbox.

7#

8# Copyright 2012 Canonical Ltd.

9#

10# Authors: Alberto Milone <alberto.milone@canonical.com>

11#

12# Checkbox is free software: you can redistribute it and/or modify

13# it under the terms of the GNU General Public License version 3,

14# as published by the Free Software Foundation.

15

16#

17# Checkbox is distributed in the hope that it will be useful,

18# but WITHOUT ANY WARRANTY; without even the implied warranty of

19# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

20# GNU General Public License for more details.

21#

22# You should have received a copy of the GNU General Public License

23# along with Checkbox. If not, see <http://www.gnu.org/licenses/>.

24

25 import threading

26 import time

27 import os

28 import sys

29

30 from signal import SIGTSTP, SIGCONT, SIGTERM

31 from subprocess import check_call, check_output, Popen, PIPE

32 from argparse import ArgumentParser

33

34

35 class AppThread(threading.Thread):

36

37 def __init__(self, app_name):

38 self._appname = app_name

39 self.stdout = None

40 self.stderr = None

41 self.pid = None

42 threading.Thread.__init__(self)

43

44 def run(self):

45 proc = Popen(self._appname, stdout=PIPE, stderr=PIPE)

46 self.pid = proc.pid

47 print(' Starting "%s", PID: %d' % (self._appname, self.pid))

48 self.stdout, self.stderr = proc.communicate()

49

50

51 def open_close_process(app, timeout):

52 """Open and close a process after a timeout"""

53 status = 0

54 # Start the process in a separate thread

55 app_thread = AppThread(app)

56 app_thread.start()

57

58 # Wait until we have a pid

59 while app_thread.pid is None:

60 continue

61 pid = app_thread.pid

62

63 # Wait a bit and kill the process

64 time.sleep(timeout)

65 print(' Killing "%s", PID: %d' % (app, pid))

66 os.kill(pid, SIGTERM)

67

68 if app_thread.stderr:

69 print("Errors:\n%s" % app_thread.stderr, file=sys.stderr)

70 status = 1

71

72 time.sleep(timeout)

73

74 return status

75

76

77 def open_close_multi_process(app, timeout, apps_num):

78 """Open and close multiple processes after a timeout"""

79 status = 0

80 threads = []

81

82 for thread in range(apps_num):

83 app_thread = AppThread(app)

84 app_thread.start()

85 threads.append(app_thread)

86

87 for thread in threads:

88 # Wait until we have a pid

89 while thread.pid is None:

90 continue

91

92 # Wait a bit and kill the process

93 time.sleep(timeout)

94 for thread in threads:

95 print(' Killing "%s", PID: %d' % (app, thread.pid))

96 os.kill(thread.pid, SIGTERM)

97 if thread.stderr:

98 print("Errors:\n%s" % thread.stderr, file=sys.stderr)

99 status = 1

100

101 time.sleep(timeout)

102

103 return status

104

105

106 def open_suspend_close_process(app, timeout):

107 """Open, suspend and close a process after a timeout"""

108 status = 0

109 # Start the process in a separate thread

110 app_thread = AppThread(app)

111 app_thread.start()

112

113 # Wait until we have a pid

114 while app_thread.pid is None:

115 continue

116 pid = app_thread.pid

117

118 # Wait a bit and suspend the process

119 time.sleep(timeout)

120 print(' Suspending "%s", PID: %d' % (app, pid))

121 os.kill(pid, SIGTSTP)

122

123 # Wait a bit and resume the process

124 time.sleep(timeout)

125 print(' Resuming "%s", PID: %d' % (app, pid))

126 os.kill(pid, SIGCONT)

127

128 # Wait a bit and kill the process

129 time.sleep(timeout)

130 print(' Killing "%s", PID: %d' % (app, pid))

131 os.kill(pid, SIGTERM)

132

133 if app_thread.stderr:

134 print("Errors:\n%s" % app_thread.stderr, file=sys.stderr)

135 status = 1

136

137 time.sleep(timeout)

138

139 return status

140

141

142 def move_window(app, timeout):

143 status = 0

144

145 # Start the process in a separate thread

146 app_thread = AppThread(app)

147 app_thread.start()

148

149 while app_thread.pid is None:

150 continue

151

152 pid = app_thread.pid

153

154 time.sleep(3)

155

156 window_list = check_output(["wmctrl", "-l"], universal_newlines=True)

157 window_id = ""

158

159 for line in window_list.split("\n"):

160 if app in line:

161 window_id = line.split()[0]

162

163 if window_id:

164 # Get the screen information from GDK

165 from gi.repository import Gdk

166

167 screen = Gdk.Screen.get_default()

168 geom = screen.get_monitor_geometry(screen.get_primary_monitor())

169

170 # Find out the window information from xwininfo

171 win_x = ""

172 win_y = ""

173 win_width = ""

174 win_height = ""

175

176 for line in check_output(

177 ["xwininfo", "-name", app], universal_newlines=True

178 ).split("\n"):

179 if "Absolute upper-left X" in line:

180 win_x = line.split(": ")[-1].strip()

181 elif "Absolute upper-left Y" in line:

182 win_y = line.split(": ")[-1].strip()

183 elif "Width" in line:

184 win_width = line.split(": ")[-1].strip()

185 elif "Height" in line:

186 win_height = line.split(": ")[-1].strip()

187

188 move_line = ["0", win_x, win_y, win_width, win_height]

189

190 directions = {

191 "RIGHT": geom.width,

192 "DOWN": geom.height,

193 "LEFT": win_x,

194 "UP": win_y,

195 "STOP": None,

196 }

197 current = "RIGHT"

198

199 while current != "STOP":

200 if current == "RIGHT":

201 # Check if top right corner of window reached top right point

202 if int(move_line[1]) + int(win_width) != directions[current]:

203 new_x = int(move_line[1]) + 1

204 move_line[1] = str(new_x)

205 else:

206 current = "DOWN"

207 elif current == "DOWN":

208 if int(move_line[2]) + int(win_height) != directions[current]:

209 new_y = int(move_line[2]) + 1

210 move_line[2] = str(new_y)

211 else:

212 current = "LEFT"

213 elif current == "LEFT":

214 if int(move_line[1]) != int(directions[current]):

215 new_x = int(move_line[1]) - 1

216 move_line[1] = str(new_x)

217 else:

218 current = "UP"

219 elif current == "UP":

220 if int(move_line[2]) != int(directions[current]):

221 new_y = int(move_line[2]) - 1

222 move_line[2] = str(new_y)

223 else:

224 current = "STOP"

225

226 check_call(

227 ["wmctrl", "-i", "-r", window_id, "-e", ",".join(move_line)]

228 )

229

230 os.kill(pid, SIGTERM)

231 else:

232 print("Could not get window handle for %s" % app, file=sys.stderr)

233 status = 1

234

235 return status

236

237

238 def print_open_close(iterations, timeout, *args):

239 status = 0

240 print("Opening and closing a 3D window")

241 for it in range(iterations):

242 print("Iteration %d of %d:" % (it + 1, iterations))

243 exit_status = open_close_process("glxgears", timeout)

244 if exit_status != 0:

245 status = 1

246 print("")

247 return status

248

249

250 def print_suspend_resume(iterations, timeout, *args):

251 status = 0

252 print("Opening, suspending, resuming and closing a 3D window")

253 for it in range(iterations):

254 print("Iteration %d of %d:" % (it + 1, iterations))

255 exit_status = open_suspend_close_process("glxgears", timeout)

256 if exit_status != 0:

257 status = 1

258 print("")

259 return status

260

261

262 def print_open_close_multi(iterations, timeout, windows_number):

263 status = 0

264 print(

265 "Opening and closing %d 3D windows at the same time" % windows_number

266 )

267 for it in range(iterations):

268 print("Iteration %d of %d:" % (it + 1, iterations))

269 exit_status = open_close_multi_process(

270 "glxgears", timeout, windows_number

271 )

272 if exit_status != 0:

273 status = 1

274 print("")

275 return status

276

277

278 def print_move_window(iterations, timeout, *args):

279 status = 0

280 print("Moving a 3D window across the screen")

281

282 for it in range(iterations):

283 print("Iteration %d of %d:" % (it + 1, iterations))

284 status = move_window("glxgears", timeout)

285

286 print("")

287 return status

288

289

290 def main():

291 tests = {

292 "open-close": print_open_close,

293 "suspend-resume": print_suspend_resume,

294 "open-close-multi": print_open_close_multi,

295 "move": print_move_window,

296 }

297

298 parser = ArgumentParser(

299 description="Script that performs window operation"

300 )

301 parser.add_argument(

302 "-t",

303 "--test",

304 default="all",

305 help="The name of the test to run. \

306 Available tests: \

307 %s, all. \

308 Default is all"

309 % (", ".join(tests)),

310 )

311 parser.add_argument(

312 "-i",

313 "--iterations",

314 type=int,

315 default=1,

316 help="The number of times to run the test. \

317 Default is 1",

318 )

319 parser.add_argument(

320 "-a",

321 "--application",

322 default="glxgears",

323 help='The 3D application to launch. \

324 Default is "glxgears"',

325 )

326 parser.add_argument(

327 "-to",

328 "--timeout",

329 type=int,

330 default=3,

331 help="The time in seconds between each test. \

332 Default is 3",

333 )

334 parser.add_argument(

335 "-w",

336 "--windows-number",

337 type=int,

338 default=4,

339 help="The number of windows to open.",

340 )

341

342 args = parser.parse_args()

343

344 status = 0

345

346 test = tests.get(args.test)

347

348 if test:

349 status = test(args.iterations, args.timeout, args.windows_number)

350 else:

351 if args.test == "all":

352 for test in tests:

353 exit_status = tests[test](

354 args.iterations, args.timeout, args.windows_number

355 )

356 if exit_status != 0:

357 status = exit_status

358 else:

359 parser.error(

360 "-t or --test can only be used with one "

361 "of the following tests: "

362 "%s, all" % (", ".join(tests))

363 )

364

365 return status

366

367

368 if __name__ == "__main__":

369 exit(main())