(self, context_message, timeout=None)
| 204 | self.MESSAGES_TIMEOUT = timeout |
| 205 | |
| 206 | def get_next_message(self, context_message, timeout=None): |
| 207 | if timeout is None: |
| 208 | timeout = self.MESSAGES_TIMEOUT |
| 209 | try: |
| 210 | msg = self._queue.get(block=True, timeout=timeout) |
| 211 | self.on_message_found(msg) |
| 212 | except: |
| 213 | raise TimeoutError( |
| 214 | "No message was written in %s seconds. Error message:\n%s" |
| 215 | % ( |
| 216 | timeout, |
| 217 | context_message, |
| 218 | ) |
| 219 | ) |
| 220 | else: |
| 221 | frame = sys._getframe().f_back.f_back |
| 222 | frame_info = "" |
| 223 | while frame: |
| 224 | if not frame.f_code.co_name.startswith("test_"): |
| 225 | frame = frame.f_back |
| 226 | continue |
| 227 | |
| 228 | if frame.f_code.co_filename.endswith("debugger_unittest.py"): |
| 229 | frame = frame.f_back |
| 230 | continue |
| 231 | |
| 232 | stack_msg = ' -- File "%s", line %s, in %s\n' % (frame.f_code.co_filename, frame.f_lineno, frame.f_code.co_name) |
| 233 | if "run" == frame.f_code.co_name: |
| 234 | frame_info = stack_msg # Ok, found the writer thread 'run' method (show only that). |
| 235 | break |
| 236 | frame_info += stack_msg |
| 237 | frame = frame.f_back |
| 238 | # Just print the first which is not debugger_unittest.py |
| 239 | break |
| 240 | |
| 241 | frame = None |
| 242 | sys.stdout.write( |
| 243 | "Message returned in get_next_message(): %s -- ctx: %s, asked at:\n%s\n" |
| 244 | % (unquote_plus(unquote_plus(msg)), context_message, frame_info) |
| 245 | ) |
| 246 | |
| 247 | if not self.accept_xml_messages: |
| 248 | if "<xml" in msg: |
| 249 | raise AssertionError("Xml messages disabled. Received: %s" % (msg,)) |
| 250 | return msg |
| 251 | |
| 252 | def _read(self, size): |
| 253 | while True: |
no test coverage detected