(self)
| 933 | |
| 934 | @unittest.expectedFailure # TODO: RUSTPYTHON; AttributeError: 'coroutine' object has no attribute 'cr_suspended' |
| 935 | def test_cr_await(self): |
| 936 | @types.coroutine |
| 937 | def a(): |
| 938 | self.assertEqual(inspect.getcoroutinestate(coro_b), inspect.CORO_RUNNING) |
| 939 | self.assertIsNone(coro_b.cr_await) |
| 940 | yield |
| 941 | self.assertEqual(inspect.getcoroutinestate(coro_b), inspect.CORO_RUNNING) |
| 942 | self.assertIsNone(coro_b.cr_await) |
| 943 | |
| 944 | async def c(): |
| 945 | await a() |
| 946 | |
| 947 | async def b(): |
| 948 | self.assertIsNone(coro_b.cr_await) |
| 949 | await c() |
| 950 | self.assertIsNone(coro_b.cr_await) |
| 951 | |
| 952 | coro_b = b() |
| 953 | self.assertEqual(inspect.getcoroutinestate(coro_b), inspect.CORO_CREATED) |
| 954 | self.assertIsNone(coro_b.cr_await) |
| 955 | |
| 956 | coro_b.send(None) |
| 957 | self.assertEqual(inspect.getcoroutinestate(coro_b), inspect.CORO_SUSPENDED) |
| 958 | self.assertEqual(coro_b.cr_await.cr_await.gi_code.co_name, 'a') |
| 959 | |
| 960 | with self.assertRaises(StopIteration): |
| 961 | coro_b.send(None) # complete coroutine |
| 962 | self.assertEqual(inspect.getcoroutinestate(coro_b), inspect.CORO_CLOSED) |
| 963 | self.assertIsNone(coro_b.cr_await) |
| 964 | |
| 965 | @unittest.expectedFailure # TODO: RUSTPYTHON; TypeError: 'NoneType' object is not iterable |
| 966 | def test_corotype_1(self): |
nothing calls this directly
no test coverage detected