(self)
| 168 | @unittest.skipUnless(sys.platform in ('linux', 'android'), 'only works on Linux') |
| 169 | @unittest.skipUnless(hasattr(termios, 'TCXONC'), 'requires termios.TCXONC') |
| 170 | def test_ioctl_suspend_and_resume_output(self): |
| 171 | wfd = self.slave_fd |
| 172 | rfd = self.master_fd |
| 173 | write_suspended = threading.Event() |
| 174 | write_finished = threading.Event() |
| 175 | |
| 176 | def writer(): |
| 177 | os.write(wfd, b'abc') |
| 178 | self.assertTrue(write_suspended.wait(support.SHORT_TIMEOUT)) |
| 179 | os.write(wfd, b'def') |
| 180 | write_finished.set() |
| 181 | |
| 182 | with threading_helper.start_threads([threading.Thread(target=writer)]): |
| 183 | self.assertEqual(os.read(rfd, 3), b'abc') |
| 184 | try: |
| 185 | try: |
| 186 | fcntl.ioctl(wfd, termios.TCXONC, termios.TCOOFF) |
| 187 | finally: |
| 188 | write_suspended.set() |
| 189 | self.assertFalse(write_finished.wait(0.5), |
| 190 | 'output was not suspended') |
| 191 | finally: |
| 192 | fcntl.ioctl(wfd, termios.TCXONC, termios.TCOON) |
| 193 | self.assertTrue(write_finished.wait(support.SHORT_TIMEOUT), |
| 194 | 'output was not resumed') |
| 195 | self.assertEqual(os.read(rfd, 1024), b'def') |
| 196 | |
| 197 | def test_ioctl_set_window_size(self): |
| 198 | # (rows, columns, xpixel, ypixel) |
nothing calls this directly
no test coverage detected