| 466 | |
| 467 | @unittest.skipUnless(hasattr(os, "pipe"), "requires os.pipe()") |
| 468 | def test_optional_abilities(self): |
| 469 | # Test for OSError when optional APIs are not supported |
| 470 | # The purpose of this test is to try fileno(), reading, writing and |
| 471 | # seeking operations with various objects that indicate they do not |
| 472 | # support these operations. |
| 473 | |
| 474 | def pipe_reader(): |
| 475 | [r, w] = os.pipe() |
| 476 | os.close(w) # So that read() is harmless |
| 477 | return self.FileIO(r, "r") |
| 478 | |
| 479 | def pipe_writer(): |
| 480 | [r, w] = os.pipe() |
| 481 | self.addCleanup(os.close, r) |
| 482 | # Guarantee that we can write into the pipe without blocking |
| 483 | thread = threading.Thread(target=os.read, args=(r, 100)) |
| 484 | thread.start() |
| 485 | self.addCleanup(thread.join) |
| 486 | return self.FileIO(w, "w") |
| 487 | |
| 488 | def buffered_reader(): |
| 489 | return self.BufferedReader(self.MockUnseekableIO()) |
| 490 | |
| 491 | def buffered_writer(): |
| 492 | return self.BufferedWriter(self.MockUnseekableIO()) |
| 493 | |
| 494 | def buffered_random(): |
| 495 | return self.BufferedRandom(self.BytesIO()) |
| 496 | |
| 497 | def buffered_rw_pair(): |
| 498 | return self.BufferedRWPair(self.MockUnseekableIO(), |
| 499 | self.MockUnseekableIO()) |
| 500 | |
| 501 | def text_reader(): |
| 502 | class UnseekableReader(self.MockUnseekableIO): |
| 503 | writable = self.BufferedIOBase.writable |
| 504 | write = self.BufferedIOBase.write |
| 505 | return self.TextIOWrapper(UnseekableReader(), "ascii") |
| 506 | |
| 507 | def text_writer(): |
| 508 | class UnseekableWriter(self.MockUnseekableIO): |
| 509 | readable = self.BufferedIOBase.readable |
| 510 | read = self.BufferedIOBase.read |
| 511 | return self.TextIOWrapper(UnseekableWriter(), "ascii") |
| 512 | |
| 513 | tests = ( |
| 514 | (pipe_reader, "fr"), (pipe_writer, "fw"), |
| 515 | (buffered_reader, "r"), (buffered_writer, "w"), |
| 516 | (buffered_random, "rws"), (buffered_rw_pair, "rw"), |
| 517 | (text_reader, "r"), (text_writer, "w"), |
| 518 | (self.BytesIO, "rws"), (self.StringIO, "rws"), |
| 519 | ) |
| 520 | |
| 521 | def do_test(test, obj, abilities): |
| 522 | readable = "r" in abilities |
| 523 | self.assertEqual(obj.readable(), readable) |
| 524 | writable = "w" in abilities |
| 525 | self.assertEqual(obj.writable(), writable) |