(console_setup)
| 133 | |
| 134 | |
| 135 | def test_console_debugger_connected(console_setup): |
| 136 | class _DebuggerWriterThread(AbstractWriterThread): |
| 137 | FORCE_KILL_PROCESS_WHEN_FINISHED_OK = True |
| 138 | |
| 139 | def __init__(self): |
| 140 | AbstractWriterThread.__init__(self) |
| 141 | socket_name = get_socket_name(close=True) |
| 142 | server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| 143 | server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) |
| 144 | server_socket.bind(socket_name) |
| 145 | server_socket.listen(1) |
| 146 | self.port = socket_name[1] |
| 147 | self.__server_socket = server_socket |
| 148 | |
| 149 | def run(self): |
| 150 | print("waiting for second process") |
| 151 | self.sock, addr = self.__server_socket.accept() |
| 152 | print("accepted second process") |
| 153 | |
| 154 | from tests_python.debugger_unittest import ReaderThread |
| 155 | |
| 156 | self.reader_thread = ReaderThread(self.sock) |
| 157 | self.reader_thread.start() |
| 158 | |
| 159 | self._sequence = -1 |
| 160 | # initial command is always the version |
| 161 | self.write_version() |
| 162 | self.log.append("start_socket") |
| 163 | self.write_make_initial_run() |
| 164 | time.sleep(1) |
| 165 | |
| 166 | seq = self.write_list_threads() |
| 167 | msg = self.wait_for_list_threads(seq) |
| 168 | assert msg.thread["name"] == "MainThread" |
| 169 | assert msg.thread["id"] == "console_main" |
| 170 | |
| 171 | self.write_get_frame("console_main", "1") |
| 172 | self.wait_for_vars( |
| 173 | [ |
| 174 | [ |
| 175 | '<var name="a" type="int" qualifier="%s" value="int: 10"' % (builtin_qualifier,), |
| 176 | '<var name="a" type="int" value="int', # jython |
| 177 | ], |
| 178 | ] |
| 179 | ) |
| 180 | |
| 181 | self.finished_ok = True |
| 182 | |
| 183 | with console_setup.check_console() as writer: |
| 184 | writer.execute_line("a = 10") |
| 185 | |
| 186 | debugger_writer_thread = _DebuggerWriterThread() |
| 187 | debugger_writer_thread.start() |
| 188 | writer.connect_to_debugger(debugger_writer_thread.port) |
| 189 | |
| 190 | wait_for_condition(lambda: debugger_writer_thread.finished_ok) |
| 191 | writer.execute_line('print("TEST SUCEEDED")') |
| 192 |
nothing calls this directly
no test coverage detected