|
1 |
| -"""Additional signal safety tests for "with" and "async with" |
| 1 | +"""Additional signal safety tests for "with" statements |
2 | 2 | """
|
3 | 3 |
|
4 | 4 | from test.support import cpython_only, verbose
|
@@ -84,41 +84,5 @@ def traced_function():
|
84 | 84 | self.fail(f"Exception wasn't raised @{target_offset}")
|
85 | 85 |
|
86 | 86 |
|
87 |
| - def _test_asynchronous_cm(self): |
88 |
| - # NOTE: this can't work, since asyncio is written in Python, and hence |
89 |
| - # will always process pending calls at some point during the evaluation |
90 |
| - # of __aenter__ and __aexit__ |
91 |
| - # |
92 |
| - # So to handle that case, we need to some way to tell the event loop |
93 |
| - # to convert pending call processing into calls to |
94 |
| - # asyncio.get_event_loop().call_soon() instead of processing them |
95 |
| - # immediately |
96 |
| - class AsyncTrackingCM(): |
97 |
| - def __init__(self): |
98 |
| - self.enter_without_exit = None |
99 |
| - async def __aenter__(self): |
100 |
| - self.enter_without_exit = True |
101 |
| - async def __aexit__(self, *args): |
102 |
| - self.enter_without_exit = False |
103 |
| - tracking_cm = AsyncTrackingCM() |
104 |
| - async def traced_coroutine(): |
105 |
| - async with tracking_cm: |
106 |
| - 1 + 1 |
107 |
| - return |
108 |
| - target_offset = -1 |
109 |
| - max_offset = len(traced_coroutine.__code__.co_code) - 2 |
110 |
| - loop = asyncio.get_event_loop() |
111 |
| - while target_offset < max_offset: |
112 |
| - target_offset += 1 |
113 |
| - raise_after_offset(traced_coroutine, target_offset) |
114 |
| - try: |
115 |
| - loop.run_until_complete(traced_coroutine()) |
116 |
| - except InjectedException: |
117 |
| - # key invariant: if we entered the CM, we exited it |
118 |
| - self.assertFalse(tracking_cm.enter_without_exit) |
119 |
| - else: |
120 |
| - self.fail(f"Exception wasn't raised @{target_offset}") |
121 |
| - |
122 |
| - |
123 | 87 | if __name__ == '__main__':
|
124 | 88 | unittest.main()
|
0 commit comments