| 1659 | @unittest.skipIf(not sys.flags.context_aware_warnings, |
| 1660 | "requires context aware warnings") |
| 1661 | def test_async_context(self): |
| 1662 | import asyncio |
| 1663 | |
| 1664 | # Events to force the execution interleaving we want. |
| 1665 | step_a1 = asyncio.Event() |
| 1666 | step_a2 = asyncio.Event() |
| 1667 | step_b1 = asyncio.Event() |
| 1668 | step_b2 = asyncio.Event() |
| 1669 | |
| 1670 | async def run_a(): |
| 1671 | with self.module.catch_warnings(record=True) as w: |
| 1672 | await step_a1.wait() |
| 1673 | # The warning emitted here should be caught be the enclosing |
| 1674 | # context manager. |
| 1675 | self.module.warn('run_a warning', UserWarning) |
| 1676 | step_b1.set() |
| 1677 | await step_a2.wait() |
| 1678 | self.assertEqual(len(w), 1) |
| 1679 | self.assertEqual(w[0].message.args[0], 'run_a warning') |
| 1680 | step_b2.set() |
| 1681 | |
| 1682 | async def run_b(): |
| 1683 | with self.module.catch_warnings(record=True) as w: |
| 1684 | step_a1.set() |
| 1685 | await step_b1.wait() |
| 1686 | # The warning emitted here should be caught be the enclosing |
| 1687 | # context manager. |
| 1688 | self.module.warn('run_b warning', UserWarning) |
| 1689 | step_a2.set() |
| 1690 | await step_b2.wait() |
| 1691 | self.assertEqual(len(w), 1) |
| 1692 | self.assertEqual(w[0].message.args[0], 'run_b warning') |
| 1693 | |
| 1694 | async def run_tasks(): |
| 1695 | await asyncio.gather(run_a(), run_b()) |
| 1696 | |
| 1697 | asyncio.run(run_tasks()) |
| 1698 | |
| 1699 | @unittest.skipIf(not sys.flags.context_aware_warnings, |
| 1700 | "requires context aware warnings") |