(self)
| 514 | self.assertEqual(['some text'], observer.data) |
| 515 | |
| 516 | def test_log_raw_single_buffer(self): |
| 517 | config = self._create_config() |
| 518 | self.create_and_start_executor(config) |
| 519 | |
| 520 | observer = _StoringObserver() |
| 521 | self.executor.get_raw_output_stream().subscribe(observer) |
| 522 | |
| 523 | self.write_process_output('some text') |
| 524 | self.write_process_output(' and continuation') |
| 525 | |
| 526 | wait_buffer_flush() |
| 527 | |
| 528 | self.assertEqual(['some text and continuation'], observer.data) |
| 529 | |
| 530 | def test_log_raw_multiple_buffers(self): |
| 531 | config = self._create_config() |
nothing calls this directly
no test coverage detected