Coverage for src/robotide/contrib/testrunner/testrunnerplugin.py: 43%

1# -*- encoding: utf-8 -*- 

2# Copyright 2010 Orbitz WorldWide 

3# 

4# Licensed under the Apache License, Version 2.0 (the "License"); 

5# you may not use this file except in compliance with the License. 

6# You may obtain a copy of the License at 

7# 

8# http://www.apache.org/licenses/LICENSE-2.0 

9# 

10# Unless required by applicable law or agreed to in writing, software 

11# distributed under the License is distributed on an "AS IS" BASIS, 

12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

13# See the License for the specific language governing permissions and 

14# limitations under the License. 

15 

16# Modified by NSN 

17# Copyright 2010-2012 Nokia Solutions and Networks 

18# Copyright 2013-2015 Nokia Networks 

19# Copyright 2016- Robot Framework Foundation 

20# 

21# Licensed under the Apache License, Version 2.0 (the "License"); 

22# you may not use this file except in compliance with the License. 

23# You may obtain a copy of the License at 

24# 

25# http://www.apache.org/licenses/LICENSE-2.0 

26# 

27# Unless required by applicable law or agreed to in writing, software 

28# distributed under the License is distributed on an "AS IS" BASIS, 

29# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

30# See the License for the specific language governing permissions and 

31# limitations under the License. 

32 

33"""A plugin for running tests from within RIDE 

34 

35Some icons courtesy Mark James and provided under a creative commons 

36license. See http://www.famfamfam.com/lab/icons/silk 

37 

38Note: this plugin creates a temporary directory for use while a test 

39is running. This directory is normally removed when RIDE exits. If 

40RIDE is shut down abnormally this directory may not get removed. The 

41directories that are created match the pattern RIDE*.d and are in a 

42temporary directory appropriate for the platform (for example, on 

43linux it's /tmp). 

44 

45You can safely manually remove these directories, except for the one 

46being used for a currently running test. 

47""" 

48import atexit (empty)

49import builtins (empty)

50import datetime (empty)

51import shutil (empty)

52import subprocess (empty)

53import tempfile (empty)

54import threading (empty)

55import time (empty)

56import os (empty)

57import psutil (empty)

58import re (empty)

59import wx (empty)

60import wx.stc (empty)

61from functools import reduce (empty)

62from queue import Queue (empty)

63from wx.lib.embeddedimage import PyEmbeddedImage (empty)

64 

65from robotide.action.shortcut import localize_shortcuts (empty)

66from robotide.context import IS_WINDOWS, IS_MAC (empty)

67from robotide.contrib.testrunner import TestRunner (empty)

68from robotide.contrib.testrunner import runprofiles (empty)

69from robotide.contrib.testrunner.ArgsParser import ArgsParser (empty)

70from robotide.contrib.testrunner.CommandArgs import CommandArgs (empty)

71from robotide.contrib.testrunner.Command import Command (empty)

72from robotide.contrib.testrunner.FileWriter import FileWriter (empty)

73from robotide.contrib.testrunner.SettingsParser import SettingsParser (empty)

74from robotide.controller.macrocontrollers import TestCaseController (empty)

75from robotide.controller.filecontrollers import start_filemanager (empty)

76from robotide.publish import RideSettingsChanged, PUBLISHER (empty)

77from robotide.publish.messages import RideTestSelectedForRunningChanged (empty)

78from robotide.pluginapi import Plugin, ActionInfo (empty)

79from robotide.ui.notebook import NoteBook (empty)

80from robotide.widgets import Label, ImageProvider, ButtonWithHandler, RIDEDialog (empty)

81from robotide.robotapi import LOG_LEVELS (empty)

82from robotide.utils import robottime (empty)

83from robotide.preferences import PreferenceEditor (empty)

84from robotide.preferences.editors import read_fonts (empty)

85from sys import getfilesystemencoding, platform (empty)

86from robotide.lib.robot.utils.encodingsniffer import (get_console_encoding, (empty)

87 get_system_encoding) 

88 

89_ = wx.GetTranslation # To keep linter/code analyser happy (empty)

90builtins.__dict__['_'] = wx.GetTranslation (empty)

91 

92CONSOLE_ENCODING = get_console_encoding() (empty)

93SYSTEM_ENCODING = get_system_encoding() (empty)

94OUTPUT_ENCODING = getfilesystemencoding() (empty)

95encoding = {'CONSOLE': CONSOLE_ENCODING, (empty)

96 'SYSTEM': SYSTEM_ENCODING, 

97 'OUTPUT': OUTPUT_ENCODING} 

98 

99FONT_FACE = 'font face' (empty)

100STOP_RUNNING_TEST = _('Stop a running test') (empty)

101STEP_OVER = _('Step over') (empty)

102ID_RUN = wx.NewIdRef() (empty)

103ID_RUNDEBUG = wx.NewIdRef() (empty)

104ID_STOP = wx.NewIdRef() (empty)

105ID_PAUSE = wx.NewIdRef() (empty)

106ID_CONTINUE = wx.NewIdRef() (empty)

107ID_STEP_NEXT = wx.NewIdRef() (empty)

108ID_STEP_OVER = wx.NewIdRef() (empty)

109ID_OPEN_LOGS_DIR = wx.NewId() (empty)

110ID_SHOW_REPORT = wx.NewIdRef() (empty)

111ID_SHOW_LOG = wx.NewIdRef() (empty)

112ID_AUTOSAVE = wx.NewIdRef() (empty)

113ID_PAUSE_ON_FAILURE = wx.NewIdRef() (empty)

114ID_SHOW_MESSAGE_LOG = wx.NewIdRef() (empty)

115STYLE_DEFAULT = 0 (empty)

116STYLE_STDERR = 2 (empty)

117STYLE_PASS = 1 (empty)

118STYLE_SKIP = 3 (empty)

119STYLE_FAIL = 4 (empty)

120STYLE_LINK = 5 (empty)

121FILE_MANAGER = 'file manager' (empty)

122 

123ATEXIT_LOCK = threading.RLock() (empty)

124 

125 

126def _run_profile(name, run_prefix): (empty)

127 return type('Profile', (runprofiles.PybotProfile,), (empty)

128 {'name': name, 'get_command': lambda self: run_prefix}) 

129 

130 

131class TestRunnerPlugin(Plugin): (empty)

132 __doc__ = _("""A plugin for running tests from within RIDE""") (empty)

133 defaults = {"auto_save": False, (empty)

134 "confirm run": True, 

135 "profile_name": "robot", 

136 "show_console_log": True, 

137 "show_message_log": True, 

138 "active_status_bar": True, 

139 "sash_position": 200, 

140 "run_profiles": 

141 [('jybot', 'jybot' + ('.bat' if os.name == 'nt' else '')), 

142 ('pybot', 'pybot' + ('.bat' if os.name == 'nt' else '')), 

143 ('robot 3.1', 'robot')], 

144 "font size": 10, 

145 "font face": 'Courier New', 

146 "foreground": 'black', 

147 "background": 'white', 

148 "error": 'red', 

149 "use colors": False, 

150 "fail color": '#FF8E8E', 

151 "pass color": '#9FCC9F', 

152 "skip color": 'yellow' 

153 } 

154 

155 report_regex = re.compile(r"^Report: {2}(.*\.html)$", re.MULTILINE) (empty)

156 log_regex = re.compile(r"^Log: {5}(.*\.html)$", re.MULTILINE) (empty)

157 title = 'Run' (empty)

158 

159 def __init__(self, application=None): (empty)

160 self.app = application (empty)

161 self.title = _('Run') (empty)

162 Plugin.__init__(self, application, initially_enabled=True, (empty)

163 default_settings=self.defaults) 

164 self.version = "3.1" (empty)

165 self.metadata = { (empty)

166 "url": 

167 "https://github.com/robotframework/RIDE/wiki/Test-Runner-Plugin"} 

168 self._reload_timer = None (empty)

169 self._frame = application.frame (empty)

170 self._report_file = None (empty)

171 self._log_file = None (empty)

172 self._controls = {} (empty)

173 self._running = False (empty)

174 self._currently_executing_keyword = None (empty)

175 self._test_runner = TestRunner(application.model) (empty)

176 self._register_shortcuts() (empty)

177 self._min_log_level_number = LOG_LEVELS['INFO'] (empty)

178 self._pause_on_failure = False (empty)

179 self._selected_tests: {TestCaseController} = set() (empty)

180 self._process = psutil.Process() (empty)

181 self._initmemory = None (empty)

182 self._limitmemory = None # This will be +80% (empty)

183 self._maxmemmsg = None (empty)

184 self.active_status_bar = self.__getattr__('active_status_bar') (empty)

185 self.use_colors = self.__getattr__('use colors') (empty)

186 self.fail_color = self.__getattr__('fail color') (empty)

187 self.pass_color = self.__getattr__('pass color') (empty)

188 self.skip_color = self.__getattr__('skip color') (empty)

189 

190 @property (empty)

191 def _names_to_run(self): (empty)

192 return list( 

193 map(lambda ctrl: (ctrl.datafile_controller.longname, ctrl.longname), 

194 self._selected_tests)) 

195 

196 def _register_shortcuts(self): (empty)

197 self.register_shortcut('CtrlCmd-C', self._copy_from_log_ctrls) (empty)

198 self.register_shortcut('CtrlCmd-L', self.on_show_log) (empty)

199 self.register_shortcut('CtrlCmd-R', self.on_show_report) (empty)

200 if IS_WINDOWS or IS_MAC: 200 ↛ exitline 200 didn't return from function '_register_shortcuts' because the condition on line 200 was always true(empty)

201 self.register_shortcut('Del', self._delete_pressed) (empty)

202 

203 def _delete_pressed(self, event): (empty)

204 __ = event 

205 if self.notebook.current_page_title != self.title: 

206 return 

207 self.get_current_profile().delete_pressed() 

208 

209 def _copy_from_log_ctrls(self, event): (empty)

210 __ = event 

211 if self.notebook.current_page_title != self.title: 

212 return 

213 if self._console_log_ctrl.GetSTCFocus(): 

214 self._console_log_ctrl.Copy() 

215 return 

216 if self._message_log_ctrl.GetSTCFocus(): 

217 self._message_log_ctrl.Copy() 

218 

219 def enable(self): (empty)

220 self.tree.set_checkboxes_for_tests() (empty)

221 self._read_run_profiles() (empty)

222 self._register_actions() (empty)

223 self._add_tab_to_notebook() (empty)

224 self._init_profile_choice(self.profile_name) (empty)

225 self._subscribe_to_events() (empty)

226 self._test_runner.enable(self._test_runner_events_handler) (empty)

227 self._set_stopped() (empty)

228 self._create_temporary_directory() (empty)

229 

230 def _register_actions(self): (empty)

231 run_action_info = ActionInfo(_("Tools"), _("Run Tests"), self.on_run, None, (empty)

232 "F8", ImageProvider().TOOLBAR_PLAY, 

233 _("Run the selected tests"), position=10) 

234 self._run_action = self.register_action(run_action_info) (empty)

235 run_action_debug = ActionInfo(_("Tools"), _("Run Tests with Debug"), (empty)

236 self.on_run_debug, None, 

237 "F9", getBugIconBitmap(), 

238 _("Run the selected tests with Debug"), 

239 position=8) 

240 self._run_action = self.register_action(run_action_debug) (empty)

241 stop_action_info = ActionInfo(_("Tools"), _("Stop Test Run"), self.on_stop, (empty)

242 None, "CtrlCmd-F8", 

243 ImageProvider().TOOLBAR_STOP, 

244 STOP_RUNNING_TEST, position=11) 

245 self._stop_action = self.register_action(stop_action_info) (empty)

246 

247 def _read_run_profiles(self): (empty)

248 self._read_run_profiles_from_config() (empty)

249 self._read_run_profiles_from_classes() (empty)

250 

251 def _read_run_profiles_from_config(self): (empty)

252 # Have to keep reference so that these classes are not garbage collected 

253 self._profile_classes_from_config = [_run_profile(name, run_prefix) (empty)

254 for name, run_prefix in 

255 self.run_profiles] 

256 

257 def _read_run_profiles_from_classes(self): (empty)

258 for profile in self._get_all_subclasses(runprofiles.BaseProfile): (empty)

259 self._test_runner.add_profile(profile.name, profile(plugin=self)) (empty)

260 

261 def _get_all_subclasses(self, class_): (empty)

262 classes = [] (empty)

263 for sub_class in class_.__subclasses__(): (empty)

264 classes += [sub_class] + self._get_all_subclasses(sub_class) (empty)

265 return classes (empty)

266 

267 def _subscribe_to_events(self): (empty)

268 self.subscribe(self.on_test_selected_for_running_changed, (empty)

269 RideTestSelectedForRunningChanged) 

270 self.subscribe(self.on_settings_changed, RideSettingsChanged) (empty)

271 

272 def on_settings_changed(self, message): (empty)

273 """Updates settings""" 

274 section, setting = message.keys (empty)

275 # print("DEBUG: enter OnSettingsChanged section %s" % (section)) 

276 if section == 'Test Run': # DEBUG temporarily we have two sections 276 ↛ 277line 276 didn't jump to line 277 because the condition on line 276 was never true(empty)

277 self.defaults.setdefault(setting, message.new) 

278 self.save_setting(setting, message.new) 

279 

280 def on_test_selected_for_running_changed(self, message): (empty)

281 self._selected_tests = message.tests 

282 

283 def disable(self): (empty)

284 self._remove_from_notebook() 

285 self._test_runner.clear_server() 

286 self.unsubscribe_all() 

287 self.unregister_actions() 

288 

289 def _create_temporary_directory(self): (empty)

290 self._default_output_dir = tempfile.mkdtemp(".d", "RIDE") (empty)

291 atexit.register(self._remove_temporary_directory) (empty)

292 # this plugin creates a temporary directory which _should_ 

293 # get reaped at exit. Sometimes things happen which might 

294 # cause it to not get deleted. Maybe this would be a good 

295 # place to check for temporary directories that match the 

296 # signature and delete them if they are more than a few 

297 # days old... 

298 

299 def _remove_temporary_directory(self): (empty)

300 with ATEXIT_LOCK: 

301 if os.path.exists(self._default_output_dir): 

302 shutil.rmtree(self._default_output_dir) 

303 

304 def on_close(self, event): (empty)

305 """Shut down the running services and processes""" 

306 self._test_runner.kill_process() 

307 if self._process_timer: 

308 self._process_timer.Stop() 

309 self._test_runner.shutdown_server() 

310 event.Skip() 

311 

312 def _reset_memory_calc(self): (empty)

313 self._initmemory = self._process.memory_info()[0] 

314 self._limitmemory = self._initmemory * 1.80 

315 self._maxmemmsg = None 

316 

317 def on_stop(self, event): (empty)

318 """Called when the user clicks the "Stop" button 

319 

320 This sends a SIGINT to the running process, with the 

321 same effect as typing control-c when running from the 

322 command line.""" 

323 __ = event 

324 self._reset_memory_calc() 

325 self._append_to_console_log(_('[ SENDING STOP SIGNAL ]\n'), 

326 source='stderr') 

327 self._test_runner.send_stop_signal() 

328 if self.active_status_bar: 

329 self.statusbar_message(_('[ SENDING STOP SIGNAL ]\n'), 5000) 

330 

331 def on_pause(self, event): (empty)

332 __ = event 

333 self._reset_memory_calc() 

334 self._append_to_console_log(_('[ SENDING PAUSE SIGNAL ]\n')) 

335 self._test_runner.send_pause_signal() 

336 if self.active_status_bar: 

337 self.statusbar_message(_('[ SENDING PAUSE SIGNAL ]\n'), 5000) 

338 

339 def on_continue(self, event): (empty)

340 __ = event 

341 self._reset_memory_calc() 

342 self._append_to_console_log(_('[ SENDING CONTINUE SIGNAL ]\n')) 

343 self._test_runner.send_continue_signal() 

344 if self.active_status_bar: 

345 self.statusbar_message(_('[ SENDING CONTINUE SIGNAL ]\n'), 5000) 

346 

347 def on_step_next(self, event): (empty)

348 __ = event 

349 self._reset_memory_calc() 

350 self._append_to_console_log(_('[ SENDING STEP NEXT SIGNAL ]\n')) 

351 self._test_runner.send_step_next_signal() 

352 if self.active_status_bar: 

353 self.statusbar_message(_('[ SENDING STEP NEXT SIGNAL ]\n'), 5000) 

354 

355 def on_step_over(self, event): (empty)

356 __ = event 

357 self._reset_memory_calc() 

358 self._append_to_console_log(_('[ SENDING STEP OVER SIGNAL ]\n')) 

359 self._test_runner.send_step_over_signal() 

360 if self.active_status_bar: 

361 self.statusbar_message(_('[ SENDING STEP OVER SIGNAL ]\n'), 5000) 

362 

363 def on_run(self, event): (empty)

364 """ Called when the user clicks or presses the F8, Run Tests """ 

365 __ = event 

366 self._run_tests() 

367 

368 def on_run_debug(self, event): (empty)

369 """ Called when the user clicks or presses the F9, Run Tests with Debug 

370 It can still be overwritten in RIDE Arguments line 

371 """ 

372 __ = event 

373 self._run_tests("DEBUG") 

374 

375 def _run_tests(self, log_level='INFO'): (empty)

376 if not self._can_start_running_tests(): 

377 return 

378 if self.__getattr__('confirm run') \ 

379 and not self._tests_selected() \ 

380 and not self._ask_user_to_run_anyway(): 

381 # In Linux NO runs dialog 4 times 

382 return 

383 self._reset_memory_calc() 

384 profile = self.get_current_profile() 

385 self.use_colors = self.__getattr__('use colors') 

386 command_args = self._create_command_args(profile.get_command_args(), log_level, self.use_colors) 

387 args_file = self._save_command_args_in_file(command_args) 

388 # print(f"DEBUG: testrunnerplugin _run_tests AFTER _save_command_args_in_file") 

389 command = self._create_command(profile.get_command(), args_file) 

390 self._initialize_variables_for_running(profile.get_settings(), command_args) 

391 self._initialize_ui_for_running() 

392 # DEBUG on Py3 it not shows correct if tags with latin chars 

393 self._append_to_console_log(_("command: %s\n") % command) 

394 try: 

395 self._test_runner.run_command(command, self._get_current_working_dir(profile)) 

396 self._process_timer.Start(41) # roughly 24fps 

397 self._set_running() 

398 self._progress_bar.Start() 

399 except Exception as e: 

400 self._set_stopped() 

401 error, log_message = self.get_current_profile().format_error(str(e), None) 

402 self._append_to_console_log(error, source='stderr') 

403 if log_message: 

404 log_message.publish() 

405 

406 def _create_command_args(self, profile_command_args, log_level='INFO', use_colors=False): (empty)

407 return CommandArgs().with_existing_args(profile_command_args) \ 

408 .with_log_level(log_level) \ 

409 .with_output_directory(self._default_output_dir) \ 

410 .with_python_path(self.global_settings.get('pythonpath', None)) \ 

411 .with_console_width(self._get_console_width()) \ 

412 .without_console_color(not use_colors) \ 

413 .with_runnable_tests(self._names_to_run) \ 

414 .build() 

415 

416 def _save_command_args_in_file(self, args): (empty)

417 arg_file = os.path.join(self._default_output_dir, 'argfile.txt') 

418 FileWriter.write(arg_file, args, 'wb') 

419 return arg_file 

420 

421 def _create_command(self, profile_command, args_file): (empty)

422 return Command().with_prefix(profile_command) \ 

423 .with_args_file(args_file) \ 

424 .with_listener(self._test_runner.get_listener_port(), 

425 self._pause_on_failure) \ 

426 .with_tests_suite_file(self.model.suite.source) \ 

427 .build() 

428 

429 def _initialize_variables_for_running(self, profile_settings, args): (empty)

430 self._report_file = self._log_file = None 

431 self._log_message_queue = Queue() 

432 

433 self._min_log_level_number = \ 

434 ArgsParser.get_message_log_level(args) 

435 

436 self._logs_directory = \ 

437 ArgsParser.get_output_directory(args, self._default_output_dir) 

438 

439 console_log_name = \ 

440 SettingsParser.get_console_log_name(profile_settings) 

441 self._console_log = '' if not console_log_name \ 

442 else os.path.join(self._logs_directory, console_log_name) 

443 

444 def _get_current_working_dir(self, profile): (empty)

445 if profile.name == runprofiles.CustomScriptProfile.name: 

446 return profile.get_cwd() 

447 if os.path.isdir(self.model.suite.source): 

448 return self.model.suite.source 

449 return os.path.dirname(self.model.suite.source) 

450 

451 def _can_start_running_tests(self): (empty)

452 if self._running or self.model.suite is None: 

453 return False 

454 if not self.is_unsaved_changes(): 

455 return True 

456 if self.auto_save or self._ask_user_to_save_before_running(): 

457 self.save_all_unsaved_changes() 

458 return True 

459 return False 

460 

461 @staticmethod (empty)

462 def _ask_user_to_save_before_running(): (empty)

463 message_box = RIDEDialog(title=_("Unsaved Modifications"), 

464 message=_("""There are unsaved modifications. 

465 Do you want to save all changes and run the tests?"""), style=wx.ICON_QUESTION | wx.YES_NO) 

466 ret = message_box.ShowModal() 

467 return ret == wx.ID_YES 

468 

469 def _tests_selected(self): (empty)

470 return len(self._selected_tests) != 0 

471 

472 @staticmethod (empty)

473 def _ask_user_to_run_anyway(): (empty)

474 message_box = RIDEDialog(title=_('No tests selected'), message=_('No tests selected. \nContinue anyway?'), 

475 style=wx.ICON_QUESTION | wx.YES_NO) 

476 ret = message_box.ShowModal() 

477 return ret == wx.ID_YES 

478 

479 def _initialize_ui_for_running(self): (empty)

480 self._show_notebook_tab() 

481 self._clear_log_ctrls() 

482 self._local_toolbar.EnableTool(ID_OPEN_LOGS_DIR, False) 

483 self._local_toolbar.EnableTool(ID_SHOW_REPORT, False) 

484 self._local_toolbar.EnableTool(ID_SHOW_LOG, False) 

485 self._report_file = self._log_file = None 

486 self._log_message_queue = Queue() 

487 

488 def _clear_log_ctrls(self): (empty)

489 self._clear_text_ctrl(self._console_log_ctrl) 

490 self._clear_text_ctrl(self._message_log_ctrl) 

491 

492 @staticmethod (empty)

493 def _clear_text_ctrl(text_ctrl): (empty)

494 text_ctrl.SetReadOnly(False) 

495 text_ctrl.ClearAll() 

496 text_ctrl.SetReadOnly(True) 

497 

498 def on_open_logs_directory(self, event): (empty)

499 """Called when the user clicks on the "Open Logs Directory" button""" 

500 __ = event 

501 # Determine explorer defined tool 

502 try: 

503 tool = self.global_settings['General'][FILE_MANAGER] 

504 except KeyError: 

505 tool = None 

506 if os.path.exists(self._logs_directory): 

507 start_filemanager(self._logs_directory, tool) 

508 else: 

509 self._notify_user_no_logs_directory() 

510 

511 def on_show_report(self, event): (empty)

512 """Called when the user clicks on the "Report" button""" 

513 __ = event 

514 if self._report_file: 

515 # wx.LaunchDefaultBrowser("file:%s" % os.path.abspath(self._report_file)) 

516 url = f"file:{os.path.abspath(self._report_file)}" 

517 self.open_browser(url) 

518 

519 def on_show_log(self, event): (empty)

520 """Called when the user clicks on the "Log" button""" 

521 __ = event 

522 if self._log_file: 

523 # wx.LaunchDefaultBrowser("file:%s" % os.path.abspath(self._log_file)) 

524 url = f"file:{os.path.abspath(self._log_file)}" 

525 self.open_browser(url) 

526 

527 def open_browser(self, url): (empty)

528 # Determine custom browser definition 

529 try: 

530 browser = self.global_settings['Plugins']['Test Runner']['browser'] 

531 except KeyError: 

532 browser = None 

533 if browser: 

534 subprocess.Popen([browser, url]) 

535 else: 

536 wx.LaunchDefaultBrowser(url) 

537 

538 def on_process_ended(self, event): (empty)

539 __ = event 

540 output, errors, log_message = self._test_runner.get_output_and_errors( 

541 self.get_current_profile()) 

542 self._append_to_console_log(output) 

543 self._read_report_and_log_from_stdout_if_needed() 

544 if len(errors) > 0: 

545 self._append_to_console_log(errors, source="stderr") 

546 if self._process_timer: 

547 self._process_timer.Stop() 

548 self._set_stopped() 

549 self._progress_bar.Stop() 

550 now = datetime.datetime.now().timetuple() 

551 self._append_to_console_log(_("\nTest finished {}").format(robottime.format_time(now))) 

552 self._test_runner.command_ended() 

553 if log_message: 

554 log_message.publish() 

555 self._local_toolbar.EnableTool(ID_OPEN_LOGS_DIR, True) 

556 

557 def _read_report_and_log_from_stdout_if_needed(self): (empty)

558 output = self._console_log_ctrl.GetText() 

559 if not self._report_file: 

560 self._report_file = \ 

561 self._get_report_or_log(output, self.report_regex) 

562 if self._report_file: 

563 self._local_toolbar.EnableTool(ID_SHOW_REPORT, True) 

564 if not self._log_file: 

565 self._log_file = self._get_report_or_log(output, self.log_regex) 

566 if self._log_file: 

567 self._local_toolbar.EnableTool(ID_SHOW_LOG, True) 

568 

569 @staticmethod (empty)

570 def _get_report_or_log(output, regex): (empty)

571 res = regex.search(output) 

572 return res.group(1) if res and os.path.isfile(res.group(1)) else None 

573 

574 def on_timer(self, event): (empty)

575 """Get process output""" 

576 __ = event 

577 if not self._log_message_queue.empty(): 

578 if self._process.memory_info()[0] <= self._limitmemory: 

579 texts = [] 

580 while not self._log_message_queue.empty(): 

581 texts += [self._log_message_queue.get()] 

582 self._append_to_message_log('\n' + '\n'.join(texts)) 

583 else: 

584 if not self._maxmemmsg: 

585 self._maxmemmsg = '\n' + _("Messages log exceeded 80% of process memory, stopping for now...") 

586 self._append_to_message_log(self._maxmemmsg, "stderr") 

587 if not self._test_runner.is_running(): 

588 self.on_process_ended(None) 

589 return 

590 out_buffer, err_buffer, __ = self._test_runner.get_output_and_errors(self.get_current_profile()) 

591 if len(out_buffer) > 0: 

592 self._append_to_console_log(out_buffer, source="stdout") 

593 if len(err_buffer) > 0: 

594 if self._get_last_output_char() != "\n": 

595 # Robot prints partial lines to stdout to make the 

596 # interactive experience better. It all goes to 

597 # heck in a handbasket if something shows up on 

598 # stderr. So, to fix that we'll add a newline if 

599 # the previous character isn't a newline. 

600 self._append_to_console_log("\n") 

601 self._append_to_console_log(err_buffer, source="stderr") 

602 

603 def _get_last_output_char(self): (empty)

604 """Return the last character in the output window""" 

605 pos = self._console_log_ctrl.PositionBefore( 

606 self._console_log_ctrl.GetLength()) 

607 char = self._console_log_ctrl.GetCharAt(pos) 

608 return chr(char) 

609 

610 def _show_notebook_tab(self): (empty)

611 """Show the Run notebook tab""" 

612 if not self.panel: 

613 self._add_tab_to_notebook() 

614 self._reload_model() 

615 self.show_tab(self.panel) 

616 

617 def _append_to_message_log(self, text, source="stdout"): (empty)

618 self._append_text(self._message_log_ctrl, text, source) 

619 

620 def _append_to_console_log(self, text, source="stdout"): (empty)

621 """Put output to the text control""" 

622 self._append_text(self._console_log_ctrl, text, source) 

623 if self._console_log: 

624 FileWriter.write(self._console_log, [text], "ab", "a") 

625 

626 def _append_text(self, text_ctrl, text, source="stdout"): (empty)

627 # text could be bytes or str 

628 if not self.panel or not text_ctrl: 

629 return 

630 self._color_map = list() 

631 if self.use_colors: 

632 text = self.parse_colors(text) 

633 text_ctrl.update_scroll_width(text) 

634 # we need this information to decide whether to autoscroll or not 

635 new_text_start = text_ctrl.GetLength() 

636 line_count = text_ctrl.GetLineCount() 

637 last_visible_line = \ 

638 text_ctrl.GetFirstVisibleLine() + text_ctrl.LinesOnScreen() - 1 

639 

640 text_ctrl.SetReadOnly(False) 

641 pos = text_ctrl.GetLastPosition() # DEBUG: Process \r with Replace 

642 # print(f"DEBUG: _append_text lastposition={pos}") 

643 text_ctrl.AppendText(text) 

644 new_text_end = text_ctrl.GetLength() 

645 

646 if wx.VERSION < (4, 1, 0): 

647 text_ctrl.StartStyling(new_text_start, 0x1f) 

648 else: 

649 text_ctrl.StartStyling(new_text_start) 

650 text_ctrl.SetStyling(0, STYLE_DEFAULT) 

651 if source == "stderr" and not self.use_colors: 

652 text_ctrl.SetStyling(new_text_end - new_text_start, STYLE_STDERR) 

653 

654 if self.use_colors and self._color_map: 

655 style = None 

656 previous_start = None 

657 for item in self._color_map: 

658 previous_style = style 

659 if item[1] == 'RED': 

660 style = STYLE_FAIL 

661 elif item[1] == 'GREEN': 

662 style = STYLE_PASS 

663 elif item[1] == 'YELLOW': 

664 style = STYLE_SKIP 

665 elif item[1] == 'BLUE': 

666 style = STYLE_LINK 

667 elif item[1] is None: 

668 style = STYLE_DEFAULT 

669 if style: 

670 if wx.VERSION < (4, 1, 0): 

671 text_ctrl.StartStyling(pos + item[0], 0x1f) 

672 else: 

673 text_ctrl.StartStyling(pos + item[0]) 

674 previous_start = item[0] 

675 if previous_style and style == STYLE_DEFAULT: 

676 text_ctrl.SetStyling(item[0] - previous_start, previous_style) 

677 if previous_start and style: 

678 text_ctrl.SetStyling(item[0] - previous_start, STYLE_DEFAULT) 

679 

680 text_ctrl.SetReadOnly(True) 

681 if last_visible_line >= line_count - 4: 

682 line_count = text_ctrl.GetLineCount() 

683 text_ctrl.ScrollToLine(line_count) 

684 

685 def parse_colors(self, txt): (empty)

686 # print(f"DEBUG: enter parse_colors {txt}") 

687 idx = 0 

688 t_size = len(txt) 

689 while idx < t_size: 

690 # print(f"{str(txt[idx])}") 

691 if txt[idx] == 27: # .startswith('\033[32m'): 

692 color = False 

693 if txt[idx + 1:idx+5] == b']8;;' and txt[idx+6] != 27: # New file:// URI in RF 7.1, Start 

694 color = True 

695 self.store_color(idx, 'BLUE') 

696 if txt[idx + 1:idx+5] == b'[34m': 

697 color = True 

698 self.store_color(idx, 'BLUE') 

699 # print(f"DEBUG: parse_colors got BLUE") 

700 if txt[idx + 1:idx+5] == b'[33m': 

701 color = True 

702 self.store_color(idx, 'YELLOW') 

703 # print(f"DEBUG: parse_colors got YELLOW") 

704 if txt[idx+1:idx+5] == b'[32m': 

705 color = True 

706 self.store_color(idx, 'GREEN') 

707 # print(f"DEBUG: parse_colors got GREEN") 

708 if txt[idx+1:idx+5] == b'[31m': 

709 color = True 

710 self.store_color(idx, 'RED') 

711 # print(f"DEBUG: parse_colors got RED") 

712 if txt[idx+1:idx+4] == b'[0m': 

713 # print(f"DEBUG: parse_colors reset to NORMAL") 

714 self.store_color(idx, None) 

715 txt = txt[:idx] + txt[idx+4:] 

716 elif txt[idx + 1] == 92: 

717 self.store_color(idx, None) 

718 txt = txt[:idx] + txt[idx+2:] 

719 elif txt[idx + 1:idx + 7] == b']8;;\x1b\\': # New file:// URI in RF 7.1, End 

720 self.store_color(idx, None) 

721 txt = txt[:idx] + txt[idx + 7:] 

722 elif color: 

723 txt = txt[:idx] + txt[idx+5:] 

724 if idx >= len(txt): 

725 break 

726 t_size = len(txt) 

727 if idx < t_size: 

728 idx += 1 

729 return txt 

730 

731 def store_color(self, idx, color): (empty)