1 # -*- coding: utf-8 -*- 2 3 """ 4 Test suite for PEP 380 implementation 5 6 adapted from original tests written by Greg Ewing 7 see <http://www.cosc.canterbury.ac.nz/greg.ewing/python/yield-from/YieldFrom-Python3.1.2-rev5.zip> 8 """ 9 10 import unittest 11 import io 12 import sys 13 import inspect 14 import parser 15 16 from test.support import captured_stderr, disable_gc, gc_collect 17 18 class TestPEP380Operation(unittest.TestCase): 19 """ 20 Test semantics. 21 """ 22 23 def test_delegation_of_initial_next_to_subgenerator(self): 24 """ 25 Test delegation of initial next() call to subgenerator 26 """ 27 trace = [] 28 def g1(): 29 trace.append("Starting g1") 30 yield from g2() 31 trace.append("Finishing g1") 32 def g2(): 33 trace.append("Starting g2") 34 yield 42 35 trace.append("Finishing g2") 36 for x in g1(): 37 trace.append("Yielded %s" % (x,)) 38 self.assertEqual(trace,[ 39 "Starting g1", 40 "Starting g2", 41 "Yielded 42", 42 "Finishing g2", 43 "Finishing g1", 44 ]) 45 46 def test_raising_exception_in_initial_next_call(self): 47 """ 48 Test raising exception in initial next() call 49 """ 50 trace = [] 51 def g1(): 52 try: 53 trace.append("Starting g1") 54 yield from g2() 55 finally: 56 trace.append("Finishing g1") 57 def g2(): 58 try: 59 trace.append("Starting g2") 60 raise ValueError("spanish inquisition occurred") 61 finally: 62 trace.append("Finishing g2") 63 try: 64 for x in g1(): 65 trace.append("Yielded %s" % (x,)) 66 except ValueError as e: 67 self.assertEqual(e.args[0], "spanish inquisition occurred") 68 else: 69 self.fail("subgenerator failed to raise ValueError") 70 self.assertEqual(trace,[ 71 "Starting g1", 72 "Starting g2", 73 "Finishing g2", 74 "Finishing g1", 75 ]) 76 77 def test_delegation_of_next_call_to_subgenerator(self): 78 """ 79 Test delegation of next() call to subgenerator 80 """ 81 trace = [] 82 def g1(): 83 trace.append("Starting g1") 84 yield "g1 ham" 85 yield from g2() 86 yield "g1 eggs" 87 trace.append("Finishing g1") 88 def g2(): 89 trace.append("Starting g2") 90 yield "g2 spam" 91 yield "g2 more spam" 92 trace.append("Finishing g2") 93 for x in g1(): 94 trace.append("Yielded %s" % (x,)) 95 self.assertEqual(trace,[ 96 "Starting g1", 97 "Yielded g1 ham", 98 "Starting g2", 99 "Yielded g2 spam", 100 "Yielded g2 more spam", 101 "Finishing g2", 102 "Yielded g1 eggs", 103 "Finishing g1", 104 ]) 105 106 def test_raising_exception_in_delegated_next_call(self): 107 """ 108 Test raising exception in delegated next() call 109 """ 110 trace = [] 111 def g1(): 112 try: 113 trace.append("Starting g1") 114 yield "g1 ham" 115 yield from g2() 116 yield "g1 eggs" 117 finally: 118 trace.append("Finishing g1") 119 def g2(): 120 try: 121 trace.append("Starting g2") 122 yield "g2 spam" 123 raise ValueError("hovercraft is full of eels") 124 yield "g2 more spam" 125 finally: 126 trace.append("Finishing g2") 127 try: 128 for x in g1(): 129 trace.append("Yielded %s" % (x,)) 130 except ValueError as e: 131 self.assertEqual(e.args[0], "hovercraft is full of eels") 132 else: 133 self.fail("subgenerator failed to raise ValueError") 134 self.assertEqual(trace,[ 135 "Starting g1", 136 "Yielded g1 ham", 137 "Starting g2", 138 "Yielded g2 spam", 139 "Finishing g2", 140 "Finishing g1", 141 ]) 142 143 def test_delegation_of_send(self): 144 """ 145 Test delegation of send() 146 """ 147 trace = [] 148 def g1(): 149 trace.append("Starting g1") 150 x = yield "g1 ham" 151 trace.append("g1 received %s" % (x,)) 152 yield from g2() 153 x = yield "g1 eggs" 154 trace.append("g1 received %s" % (x,)) 155 trace.append("Finishing g1") 156 def g2(): 157 trace.append("Starting g2") 158 x = yield "g2 spam" 159 trace.append("g2 received %s" % (x,)) 160 x = yield "g2 more spam" 161 trace.append("g2 received %s" % (x,)) 162 trace.append("Finishing g2") 163 g = g1() 164 y = next(g) 165 x = 1 166 try: 167 while 1: 168 y = g.send(x) 169 trace.append("Yielded %s" % (y,)) 170 x += 1 171 except StopIteration: 172 pass 173 self.assertEqual(trace,[ 174 "Starting g1", 175 "g1 received 1", 176 "Starting g2", 177 "Yielded g2 spam", 178 "g2 received 2", 179 "Yielded g2 more spam", 180 "g2 received 3", 181 "Finishing g2", 182 "Yielded g1 eggs", 183 "g1 received 4", 184 "Finishing g1", 185 ]) 186 187 def test_handling_exception_while_delegating_send(self): 188 """ 189 Test handling exception while delegating 'send' 190 """ 191 trace = [] 192 def g1(): 193 trace.append("Starting g1") 194 x = yield "g1 ham" 195 trace.append("g1 received %s" % (x,)) 196 yield from g2() 197 x = yield "g1 eggs" 198 trace.append("g1 received %s" % (x,)) 199 trace.append("Finishing g1") 200 def g2(): 201 trace.append("Starting g2") 202 x = yield "g2 spam" 203 trace.append("g2 received %s" % (x,)) 204 raise ValueError("hovercraft is full of eels") 205 x = yield "g2 more spam" 206 trace.append("g2 received %s" % (x,)) 207 trace.append("Finishing g2") 208 def run(): 209 g = g1() 210 y = next(g) 211 x = 1 212 try: 213 while 1: 214 y = g.send(x) 215 trace.append("Yielded %s" % (y,)) 216 x += 1 217 except StopIteration: 218 trace.append("StopIteration") 219 self.assertRaises(ValueError,run) 220 self.assertEqual(trace,[ 221 "Starting g1", 222 "g1 received 1", 223 "Starting g2", 224 "Yielded g2 spam", 225 "g2 received 2", 226 ]) 227 228 def test_delegating_close(self): 229 """ 230 Test delegating 'close' 231 """ 232 trace = [] 233 def g1(): 234 try: 235 trace.append("Starting g1") 236 yield "g1 ham" 237 yield from g2() 238 yield "g1 eggs" 239 finally: 240 trace.append("Finishing g1") 241 def g2(): 242 try: 243 trace.append("Starting g2") 244 yield "g2 spam" 245 yield "g2 more spam" 246 finally: 247 trace.append("Finishing g2") 248 g = g1() 249 for i in range(2): 250 x = next(g) 251 trace.append("Yielded %s" % (x,)) 252 g.close() 253 self.assertEqual(trace,[ 254 "Starting g1", 255 "Yielded g1 ham", 256 "Starting g2", 257 "Yielded g2 spam", 258 "Finishing g2", 259 "Finishing g1" 260 ]) 261 262 def test_handing_exception_while_delegating_close(self): 263 """ 264 Test handling exception while delegating 'close' 265 """ 266 trace = [] 267 def g1(): 268 try: 269 trace.append("Starting g1") 270 yield "g1 ham" 271 yield from g2() 272 yield "g1 eggs" 273 finally: 274 trace.append("Finishing g1") 275 def g2(): 276 try: 277 trace.append("Starting g2") 278 yield "g2 spam" 279 yield "g2 more spam" 280 finally: 281 trace.append("Finishing g2") 282 raise ValueError("nybbles have exploded with delight") 283 try: 284 g = g1() 285 for i in range(2): 286 x = next(g) 287 trace.append("Yielded %s" % (x,)) 288 g.close() 289 except ValueError as e: 290 self.assertEqual(e.args[0], "nybbles have exploded with delight") 291 self.assertIsInstance(e.__context__, GeneratorExit) 292 else: 293 self.fail("subgenerator failed to raise ValueError") 294 self.assertEqual(trace,[ 295 "Starting g1", 296 "Yielded g1 ham", 297 "Starting g2", 298 "Yielded g2 spam", 299 "Finishing g2", 300 "Finishing g1", 301 ]) 302 303 def test_delegating_throw(self): 304 """ 305 Test delegating 'throw' 306 """ 307 trace = [] 308 def g1(): 309 try: 310 trace.append("Starting g1") 311 yield "g1 ham" 312 yield from g2() 313 yield "g1 eggs" 314 finally: 315 trace.append("Finishing g1") 316 def g2(): 317 try: 318 trace.append("Starting g2") 319 yield "g2 spam" 320 yield "g2 more spam" 321 finally: 322 trace.append("Finishing g2") 323 try: 324 g = g1() 325 for i in range(2): 326 x = next(g) 327 trace.append("Yielded %s" % (x,)) 328 e = ValueError("tomato ejected") 329 g.throw(e) 330 except ValueError as e: 331 self.assertEqual(e.args[0], "tomato ejected") 332 else: 333 self.fail("subgenerator failed to raise ValueError") 334 self.assertEqual(trace,[ 335 "Starting g1", 336 "Yielded g1 ham", 337 "Starting g2", 338 "Yielded g2 spam", 339 "Finishing g2", 340 "Finishing g1", 341 ]) 342 343 def test_value_attribute_of_StopIteration_exception(self): 344 """ 345 Test 'value' attribute of StopIteration exception 346 """ 347 trace = [] 348 def pex(e): 349 trace.append("%s: %s" % (e.__class__.__name__, e)) 350 trace.append("value = %s" % (e.value,)) 351 e = StopIteration() 352 pex(e) 353 e = StopIteration("spam") 354 pex(e) 355 e.value = "eggs" 356 pex(e) 357 self.assertEqual(trace,[ 358 "StopIteration: ", 359 "value = None", 360 "StopIteration: spam", 361 "value = spam", 362 "StopIteration: spam", 363 "value = eggs", 364 ]) 365 366 367 def test_exception_value_crash(self): 368 # There used to be a refcount error when the return value 369 # stored in the StopIteration has a refcount of 1. 370 def g1(): 371 yield from g2() 372 def g2(): 373 yield "g2" 374 return [42] 375 self.assertEqual(list(g1()), ["g2"]) 376 377 378 def test_generator_return_value(self): 379 """ 380 Test generator return value 381 """ 382 trace = [] 383 def g1(): 384 trace.append("Starting g1") 385 yield "g1 ham" 386 ret = yield from g2() 387 trace.append("g2 returned %r" % (ret,)) 388 for v in 1, (2,), StopIteration(3): 389 ret = yield from g2(v) 390 trace.append("g2 returned %r" % (ret,)) 391 yield "g1 eggs" 392 trace.append("Finishing g1") 393 def g2(v = None): 394 trace.append("Starting g2") 395 yield "g2 spam" 396 yield "g2 more spam" 397 trace.append("Finishing g2") 398 if v: 399 return v 400 for x in g1(): 401 trace.append("Yielded %s" % (x,)) 402 self.assertEqual(trace,[ 403 "Starting g1", 404 "Yielded g1 ham", 405 "Starting g2", 406 "Yielded g2 spam", 407 "Yielded g2 more spam", 408 "Finishing g2", 409 "g2 returned None", 410 "Starting g2", 411 "Yielded g2 spam", 412 "Yielded g2 more spam", 413 "Finishing g2", 414 "g2 returned 1", 415 "Starting g2", 416 "Yielded g2 spam", 417 "Yielded g2 more spam", 418 "Finishing g2", 419 "g2 returned (2,)", 420 "Starting g2", 421 "Yielded g2 spam", 422 "Yielded g2 more spam", 423 "Finishing g2", 424 "g2 returned StopIteration(3,)", 425 "Yielded g1 eggs", 426 "Finishing g1", 427 ]) 428 429 def test_delegation_of_next_to_non_generator(self): 430 """ 431 Test delegation of next() to non-generator 432 """ 433 trace = [] 434 def g(): 435 yield from range(3) 436 for x in g(): 437 trace.append("Yielded %s" % (x,)) 438 self.assertEqual(trace,[ 439 "Yielded 0", 440 "Yielded 1", 441 "Yielded 2", 442 ]) 443 444 445 def test_conversion_of_sendNone_to_next(self): 446 """ 447 Test conversion of send(None) to next() 448 """ 449 trace = [] 450 def g(): 451 yield from range(3) 452 gi = g() 453 for x in range(3): 454 y = gi.send(None) 455 trace.append("Yielded: %s" % (y,)) 456 self.assertEqual(trace,[ 457 "Yielded: 0", 458 "Yielded: 1", 459 "Yielded: 2", 460 ]) 461 462 def test_delegation_of_close_to_non_generator(self): 463 """ 464 Test delegation of close() to non-generator 465 """ 466 trace = [] 467 def g(): 468 try: 469 trace.append("starting g") 470 yield from range(3) 471 trace.append("g should not be here") 472 finally: 473 trace.append("finishing g") 474 gi = g() 475 next(gi) 476 with captured_stderr() as output: 477 gi.close() 478 self.assertEqual(output.getvalue(), '') 479 self.assertEqual(trace,[ 480 "starting g", 481 "finishing g", 482 ]) 483 484 def test_delegating_throw_to_non_generator(self): 485 """ 486 Test delegating 'throw' to non-generator 487 """ 488 trace = [] 489 def g(): 490 try: 491 trace.append("Starting g") 492 yield from range(10) 493 finally: 494 trace.append("Finishing g") 495 try: 496 gi = g() 497 for i in range(5): 498 x = next(gi) 499 trace.append("Yielded %s" % (x,)) 500 e = ValueError("tomato ejected") 501 gi.throw(e) 502 except ValueError as e: 503 self.assertEqual(e.args[0],"tomato ejected") 504 else: 505 self.fail("subgenerator failed to raise ValueError") 506 self.assertEqual(trace,[ 507 "Starting g", 508 "Yielded 0", 509 "Yielded 1", 510 "Yielded 2", 511 "Yielded 3", 512 "Yielded 4", 513 "Finishing g", 514 ]) 515 516 def test_attempting_to_send_to_non_generator(self): 517 """ 518 Test attempting to send to non-generator 519 """ 520 trace = [] 521 def g(): 522 try: 523 trace.append("starting g") 524 yield from range(3) 525 trace.append("g should not be here") 526 finally: 527 trace.append("finishing g") 528 try: 529 gi = g() 530 next(gi) 531 for x in range(3): 532 y = gi.send(42) 533 trace.append("Should not have yielded: %s" % (y,)) 534 except AttributeError as e: 535 self.assertIn("send", e.args[0]) 536 else: 537 self.fail("was able to send into non-generator") 538 self.assertEqual(trace,[ 539 "starting g", 540 "finishing g", 541 ]) 542 543 def test_broken_getattr_handling(self): 544 """ 545 Test subiterator with a broken getattr implementation 546 """ 547 class Broken: 548 def __iter__(self): 549 return self 550 def __next__(self): 551 return 1 552 def __getattr__(self, attr): 553 1/0 554 555 def g(): 556 yield from Broken() 557 558 with self.assertRaises(ZeroDivisionError): 559 gi = g() 560 self.assertEqual(next(gi), 1) 561 gi.send(1) 562 563 with self.assertRaises(ZeroDivisionError): 564 gi = g() 565 self.assertEqual(next(gi), 1) 566 gi.throw(AttributeError) 567 568 with captured_stderr() as output: 569 gi = g() 570 self.assertEqual(next(gi), 1) 571 gi.close() 572 self.assertIn('ZeroDivisionError', output.getvalue()) 573 574 def test_exception_in_initial_next_call(self): 575 """ 576 Test exception in initial next() call 577 """ 578 trace = [] 579 def g1(): 580 trace.append("g1 about to yield from g2") 581 yield from g2() 582 trace.append("g1 should not be here") 583 def g2(): 584 yield 1/0 585 def run(): 586 gi = g1() 587 next(gi) 588 self.assertRaises(ZeroDivisionError,run) 589 self.assertEqual(trace,[ 590 "g1 about to yield from g2" 591 ]) 592 593 def test_attempted_yield_from_loop(self): 594 """ 595 Test attempted yield-from loop 596 """ 597 trace = [] 598 def g1(): 599 trace.append("g1: starting") 600 yield "y1" 601 trace.append("g1: about to yield from g2") 602 yield from g2() 603 trace.append("g1 should not be here") 604 605 def g2(): 606 trace.append("g2: starting") 607 yield "y2" 608 trace.append("g2: about to yield from g1") 609 yield from gi 610 trace.append("g2 should not be here") 611 try: 612 gi = g1() 613 for y in gi: 614 trace.append("Yielded: %s" % (y,)) 615 except ValueError as e: 616 self.assertEqual(e.args[0],"generator already executing") 617 else: 618 self.fail("subgenerator didn't raise ValueError") 619 self.assertEqual(trace,[ 620 "g1: starting", 621 "Yielded: y1", 622 "g1: about to yield from g2", 623 "g2: starting", 624 "Yielded: y2", 625 "g2: about to yield from g1", 626 ]) 627 628 def test_returning_value_from_delegated_throw(self): 629 """ 630 Test returning value from delegated 'throw' 631 """ 632 trace = [] 633 def g1(): 634 try: 635 trace.append("Starting g1") 636 yield "g1 ham" 637 yield from g2() 638 yield "g1 eggs" 639 finally: 640 trace.append("Finishing g1") 641 def g2(): 642 try: 643 trace.append("Starting g2") 644 yield "g2 spam" 645 yield "g2 more spam" 646 except LunchError: 647 trace.append("Caught LunchError in g2") 648 yield "g2 lunch saved" 649 yield "g2 yet more spam" 650 class LunchError(Exception): 651 pass 652 g = g1() 653 for i in range(2): 654 x = next(g) 655 trace.append("Yielded %s" % (x,)) 656 e = LunchError("tomato ejected") 657 g.throw(e) 658 for x in g: 659 trace.append("Yielded %s" % (x,)) 660 self.assertEqual(trace,[ 661 "Starting g1", 662 "Yielded g1 ham", 663 "Starting g2", 664 "Yielded g2 spam", 665 "Caught LunchError in g2", 666 "Yielded g2 yet more spam", 667 "Yielded g1 eggs", 668 "Finishing g1", 669 ]) 670 671 def test_next_and_return_with_value(self): 672 """ 673 Test next and return with value 674 """ 675 trace = [] 676 def f(r): 677 gi = g(r) 678 next(gi) 679 try: 680 trace.append("f resuming g") 681 next(gi) 682 trace.append("f SHOULD NOT BE HERE") 683 except StopIteration as e: 684 trace.append("f caught %r" % (e,)) 685 def g(r): 686 trace.append("g starting") 687 yield 688 trace.append("g returning %r" % (r,)) 689 return r 690 f(None) 691 f(1) 692 f((2,)) 693 f(StopIteration(3)) 694 self.assertEqual(trace,[ 695 "g starting", 696 "f resuming g", 697 "g returning None", 698 "f caught StopIteration()", 699 "g starting", 700 "f resuming g", 701 "g returning 1", 702 "f caught StopIteration(1,)", 703 "g starting", 704 "f resuming g", 705 "g returning (2,)", 706 "f caught StopIteration((2,),)", 707 "g starting", 708 "f resuming g", 709 "g returning StopIteration(3,)", 710 "f caught StopIteration(StopIteration(3,),)", 711 ]) 712 713 def test_send_and_return_with_value(self): 714 """ 715 Test send and return with value 716 """ 717 trace = [] 718 def f(r): 719 gi = g(r) 720 next(gi) 721 try: 722 trace.append("f sending spam to g") 723 gi.send("spam") 724 trace.append("f SHOULD NOT BE HERE") 725 except StopIteration as e: 726 trace.append("f caught %r" % (e,)) 727 def g(r): 728 trace.append("g starting") 729 x = yield 730 trace.append("g received %r" % (x,)) 731 trace.append("g returning %r" % (r,)) 732 return r 733 f(None) 734 f(1) 735 f((2,)) 736 f(StopIteration(3)) 737 self.assertEqual(trace, [ 738 "g starting", 739 "f sending spam to g", 740 "g received 'spam'", 741 "g returning None", 742 "f caught StopIteration()", 743 "g starting", 744 "f sending spam to g", 745 "g received 'spam'", 746 "g returning 1", 747 'f caught StopIteration(1,)', 748 'g starting', 749 'f sending spam to g', 750 "g received 'spam'", 751 'g returning (2,)', 752 'f caught StopIteration((2,),)', 753 'g starting', 754 'f sending spam to g', 755 "g received 'spam'", 756 'g returning StopIteration(3,)', 757 'f caught StopIteration(StopIteration(3,),)' 758 ]) 759 760 def test_catching_exception_from_subgen_and_returning(self): 761 """ 762 Test catching an exception thrown into a 763 subgenerator and returning a value 764 """ 765 def inner(): 766 try: 767 yield 1 768 except ValueError: 769 trace.append("inner caught ValueError") 770 return value 771 772 def outer(): 773 v = yield from inner() 774 trace.append("inner returned %r to outer" % (v,)) 775 yield v 776 777 for value in 2, (2,), StopIteration(2): 778 trace = [] 779 g = outer() 780 trace.append(next(g)) 781 trace.append(repr(g.throw(ValueError))) 782 self.assertEqual(trace, [ 783 1, 784 "inner caught ValueError", 785 "inner returned %r to outer" % (value,), 786 repr(value), 787 ]) 788 789 def test_throwing_GeneratorExit_into_subgen_that_returns(self): 790 """ 791 Test throwing GeneratorExit into a subgenerator that 792 catches it and returns normally. 793 """ 794 trace = [] 795 def f(): 796 try: 797 trace.append("Enter f") 798 yield 799 trace.append("Exit f") 800 except GeneratorExit: 801 return 802 def g(): 803 trace.append("Enter g") 804 yield from f() 805 trace.append("Exit g") 806 try: 807 gi = g() 808 next(gi) 809 gi.throw(GeneratorExit) 810 except GeneratorExit: 811 pass 812 else: 813 self.fail("subgenerator failed to raise GeneratorExit") 814 self.assertEqual(trace,[ 815 "Enter g", 816 "Enter f", 817 ]) 818 819 def test_throwing_GeneratorExit_into_subgenerator_that_yields(self): 820 """ 821 Test throwing GeneratorExit into a subgenerator that 822 catches it and yields. 823 """ 824 trace = [] 825 def f(): 826 try: 827 trace.append("Enter f") 828 yield 829 trace.append("Exit f") 830 except GeneratorExit: 831 yield 832 def g(): 833 trace.append("Enter g") 834 yield from f() 835 trace.append("Exit g") 836 try: 837 gi = g() 838 next(gi) 839 gi.throw(GeneratorExit) 840 except RuntimeError as e: 841 self.assertEqual(e.args[0], "generator ignored GeneratorExit") 842 else: 843 self.fail("subgenerator failed to raise GeneratorExit") 844 self.assertEqual(trace,[ 845 "Enter g", 846 "Enter f", 847 ]) 848 849 def test_throwing_GeneratorExit_into_subgen_that_raises(self): 850 """ 851 Test throwing GeneratorExit into a subgenerator that 852 catches it and raises a different exception. 853 """ 854 trace = [] 855 def f(): 856 try: 857 trace.append("Enter f") 858 yield 859 trace.append("Exit f") 860 except GeneratorExit: 861 raise ValueError("Vorpal bunny encountered") 862 def g(): 863 trace.append("Enter g") 864 yield from f() 865 trace.append("Exit g") 866 try: 867 gi = g() 868 next(gi) 869 gi.throw(GeneratorExit) 870 except ValueError as e: 871 self.assertEqual(e.args[0], "Vorpal bunny encountered") 872 self.assertIsInstance(e.__context__, GeneratorExit) 873 else: 874 self.fail("subgenerator failed to raise ValueError") 875 self.assertEqual(trace,[ 876 "Enter g", 877 "Enter f", 878 ]) 879 880 def test_yield_from_empty(self): 881 def g(): 882 yield from () 883 self.assertRaises(StopIteration, next, g()) 884 885 def test_delegating_generators_claim_to_be_running(self): 886 # Check with basic iteration 887 def one(): 888 yield 0 889 yield from two() 890 yield 3 891 def two(): 892 yield 1 893 try: 894 yield from g1 895 except ValueError: 896 pass 897 yield 2 898 g1 = one() 899 self.assertEqual(list(g1), [0, 1, 2, 3]) 900 # Check with send 901 g1 = one() 902 res = [next(g1)] 903 try: 904 while True: 905 res.append(g1.send(42)) 906 except StopIteration: 907 pass 908 self.assertEqual(res, [0, 1, 2, 3]) 909 # Check with throw 910 class MyErr(Exception): 911 pass 912 def one(): 913 try: 914 yield 0 915 except MyErr: 916 pass 917 yield from two() 918 try: 919 yield 3 920 except MyErr: 921 pass 922 def two(): 923 try: 924 yield 1 925 except MyErr: 926 pass 927 try: 928 yield from g1 929 except ValueError: 930 pass 931 try: 932 yield 2 933 except MyErr: 934 pass 935 g1 = one() 936 res = [next(g1)] 937 try: 938 while True: 939 res.append(g1.throw(MyErr)) 940 except StopIteration: 941 pass 942 # Check with close 943 class MyIt(object): 944 def __iter__(self): 945 return self 946 def __next__(self): 947 return 42 948 def close(self_): 949 self.assertTrue(g1.gi_running) 950 self.assertRaises(ValueError, next, g1) 951 def one(): 952 yield from MyIt() 953 g1 = one() 954 next(g1) 955 g1.close() 956 957 def test_delegator_is_visible_to_debugger(self): 958 def call_stack(): 959 return [f[3] for f in inspect.stack()] 960 961 def gen(): 962 yield call_stack() 963 yield call_stack() 964 yield call_stack() 965 966 def spam(g): 967 yield from g 968 969 def eggs(g): 970 yield from g 971 972 for stack in spam(gen()): 973 self.assertTrue('spam' in stack) 974 975 for stack in spam(eggs(gen())): 976 self.assertTrue('spam' in stack and 'eggs' in stack) 977 978 def test_custom_iterator_return(self): 979 # See issue #15568 980 class MyIter: 981 def __iter__(self): 982 return self 983 def __next__(self): 984 raise StopIteration(42) 985 def gen(): 986 nonlocal ret 987 ret = yield from MyIter() 988 ret = None 989 list(gen()) 990 self.assertEqual(ret, 42) 991 992 def test_close_with_cleared_frame(self): 993 # See issue #17669. 994 # 995 # Create a stack of generators: outer() delegating to inner() 996 # delegating to innermost(). The key point is that the instance of 997 # inner is created first: this ensures that its frame appears before 998 # the instance of outer in the GC linked list. 999 # 1000 # At the gc.collect call: 1001 # - frame_clear is called on the inner_gen frame. 1002 # - gen_dealloc is called on the outer_gen generator (the only 1003 # reference is in the frame's locals). 1004 # - gen_close is called on the outer_gen generator. 1005 # - gen_close_iter is called to close the inner_gen generator, which 1006 # in turn calls gen_close, and gen_yf. 1007 # 1008 # Previously, gen_yf would crash since inner_gen's frame had been 1009 # cleared (and in particular f_stacktop was NULL). 1010 1011 def innermost(): 1012 yield 1013 def inner(): 1014 outer_gen = yield 1015 yield from innermost() 1016 def outer(): 1017 inner_gen = yield 1018 yield from inner_gen 1019 1020 with disable_gc(): 1021 inner_gen = inner() 1022 outer_gen = outer() 1023 outer_gen.send(None) 1024 outer_gen.send(inner_gen) 1025 outer_gen.send(outer_gen) 1026 1027 del outer_gen 1028 del inner_gen 1029 gc_collect() 1030 1031 def test_send_tuple_with_custom_generator(self): 1032 # See issue #21209. 1033 class MyGen: 1034 def __iter__(self): 1035 return self 1036 def __next__(self): 1037 return 42 1038 def send(self, what): 1039 nonlocal v 1040 v = what 1041 return None 1042 def outer(): 1043 v = yield from MyGen() 1044 g = outer() 1045 next(g) 1046 v = None 1047 g.send((1, 2, 3, 4)) 1048 self.assertEqual(v, (1, 2, 3, 4)) 1049 1050 1051 if __name__ == '__main__': 1052 unittest.main() 1053
