Coverage for src/robotide/contrib/testrunner/testrunner.py: 37%

1# Copyright 2010 Orbitz WorldWide 

2# 

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

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

5# You may obtain a copy of the License at 

6# 

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

8# 

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

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

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

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

13# limitations under the License. 

14 

15# Modified by NSN 

16# Copyright 2010-2012 Nokia Solutions and Networks 

17# Copyright 2013-2015 Nokia Networks 

18# Copyright 2016- Robot Framework Foundation 

19# 

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

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

22# You may obtain a copy of the License at 

23# 

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

25# 

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

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

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

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

30# limitations under the License. 

31 

32import socketserver as SocketServer (empty)

33import threading (empty)

34 

35from robotide.contrib.testrunner.Process import Process (empty)

36from robotide.contrib.testrunner.TestRunnerAgent import StreamHandler (empty)

37from robotide.controller.testexecutionresults import TestExecutionResults (empty)

38 

39 

40# Solution from https://stackoverflow.com/questions/10009753/ 

41# python-dealing-with-mixed-encoding-files 

42def mixed_decoder(unicode_error): (empty)

43 err_str = unicode_error[1] 

44 err_len = unicode_error.end - unicode_error.start 

45 next_position = unicode_error.start + err_len 

46 err_hex = err_str[unicode_error.start:unicode_error.end].encode('hex') 

47 # Alternative, return u'?', next_position 

48 return u'%s' % err_hex, next_position # Comment this line out to get a ? 

49 

50# codecs.register_error("mixed", mixed_decoder) 

51 

52 

53class TestRunner(object): (empty)

54 

55 def __init__(self, project): (empty)

56 self._process = None (empty)

57 self._server = None (empty)

58 self._server_thread = None (empty)

59 self._pause_on_failure = False (empty)

60 self._pid_to_kill = None (empty)

61 self._results = TestExecutionResults() (empty)

62 self._port = None (empty)

63 self._project = project (empty)

64 self.profiles = {} (empty)

65 self._pause_longname = None (empty)

66 self._pause_testname = None (empty)

67 

68 def enable(self, result_handler): (empty)

69 self._start_listener_server(result_handler) (empty)

70 

71 def add_profile(self, name, item): (empty)

72 self.profiles[name] = item (empty)

73 

74 def get_profile(self, name): (empty)

75 return self.profiles[name] (empty)

76 

77 def get_profile_names(self): (empty)

78 return sorted(self.profiles.keys()) (empty)

79 

80 def _start_listener_server(self, result_handler): (empty)

81 def handle(*args): (empty)

82 self._result_handler(*args) 

83 result_handler(*args) 

84 

85 self._server = RideListenerServer(RideListenerHandler, handle) (empty)

86 self._server_thread = threading.Thread( (empty)

87 target=self._server.serve_forever) 

88 # DEPRECATED: self._server_thread.setDaemon(True) 

89 self._server_thread.daemon = True (empty)

90 self._server_thread.start() (empty)

91 self._port = self._server.server_address[1] (empty)

92 

93 def _result_handler(self, event, *args): (empty)

94 if event == 'pid': 

95 self._pid_to_kill = int(args[0]) 

96 if event == 'port' and self._process: 

97 self._process.set_port(args[0]) 

98 if event == 'start_test': 

99 longname = args[1]['longname'] 

100 testname = args[0] 

101 self._results.set_running(self._get_test_controller(longname, 

102 testname)) 

103 self._pause_longname = longname 

104 self._pause_testname = testname 

105 

106 if event == 'continue': 

107 self._results.set_running(self._get_test_controller( 

108 self._pause_longname, self._pause_testname)) 

109 

110 if event == 'paused': 

111 self._results.set_paused(self._get_test_controller( 

112 self._pause_longname, self._pause_testname)) 

113 if event == 'end_test': 

114 longname = args[1]['longname'] 

115 testname = args[0] 

116 if args[1]['status'] == 'PASS': 

117 self._results.set_passed(self._get_test_controller(longname, 

118 testname)) 

119 elif args[1]['status'] == 'SKIP': 

120 self._results.set_skipped(self._get_test_controller(longname, 

121 testname)) 

122 else: 

123 self._results.set_failed(self._get_test_controller(longname, 

124 testname)) 

125 

126 def _get_test_controller(self, longname, testname=None): (empty)

127 ret = self._project.find_controller_by_longname(longname, testname) 

128 return ret 

129 

130 def clear_server(self): (empty)

131 self._server = None 

132 

133 def shutdown_server(self): (empty)

134 if self._server: 

135 self._server.shutdown() 

136 

137 def test_execution_started(self): (empty)

138 self._results.test_execution_started() 

139 

140 def kill_process(self): (empty)

141 if self._process: 

142 self._process.kill(force=True) 

143 

144 def send_pause_on_failure(self, pause): (empty)

145 if self._process: 

146 self._process.pause_on_failure(pause) 

147 

148 def send_stop_signal(self): (empty)

149 if self._process: 

150 self._process.kill(killer_pid=self._pid_to_kill) 

151 

152 def send_pause_signal(self): (empty)

153 if self._process: 

154 self._process.pause() 

155 

156 def send_continue_signal(self): (empty)

157 if self._process: 

158 self._process.resume() 

159 

160 def send_step_next_signal(self): (empty)

161 if self._process: 

162 self._process.step_next() 

163 

164 def send_step_over_signal(self): (empty)

165 if self._process: 

166 self._process.step_over() 

167 

168 def run_command(self, command, cwd): (empty)

169 self._pid_to_kill = None 

170 self._process = Process(cwd) 

171 self._process.run_command(command) 

172 

173 def get_output_and_errors(self, profile): (empty)

174 stdout, stderr, returncode = self._process.get_output(), \ 

175 self._process.get_errors(), \ 

176 self._process.get_returncode() 

177 error, log_message = profile.format_error(stderr, returncode) 

178 return stdout, error, log_message 

179 

180 def get_listener_port(self): (empty)

181 return self._port 

182 

183 def is_running(self): (empty)

184 return self._process and self._process.is_alive() 

185 

186 def command_ended(self): (empty)

187 self._results.set_stopped(None) 

188 self._process = None 

189 

190 

191# The following two classes implement a small line-buffered socket 

192# server. It is designed to run in a separate thread, read data 

193# from the given port and update the UI -- hopefully all in a 

194# thread-safe manner. 

195class RideListenerServer(SocketServer.TCPServer): (empty)

196 """Implements a simple line-buffered socket server""" 

197 allow_reuse_address = True (empty)

198 

199 def __init__(self, request_handler_class, callback): (empty)

200 SocketServer.TCPServer.__init__(self, ("", 0), request_handler_class) (empty)

201 self.callback = callback (empty)

202 

203 

204class RideListenerHandler(SocketServer.StreamRequestHandler): (empty)

205 def handle(self): (empty)

206 decoder = StreamHandler(self.request.makefile('r')) 

207 while True: 

208 try: 

209 (name, args) = decoder.load() 

210 self.server.callback(name, *args) 

211 except (EOFError, IOError): 

212 # I should log this... 

213 break