| import asyncio | |
| import decimal | |
| import unittest | |
| def tearDownModule(): | |
| asyncio.set_event_loop(None) | |
| @unittest.skipUnless(decimal.HAVE_CONTEXTVAR, "decimal is built with a thread-local context") | |
| class DecimalContextTest(unittest.TestCase): | |
| def test_asyncio_task_decimal_context(self): | |
| async def fractions(t, precision, x, y): | |
| with decimal.localcontext() as ctx: | |
| ctx.prec = precision | |
| a = decimal.Decimal(x) / decimal.Decimal(y) | |
| await asyncio.sleep(t) | |
| b = decimal.Decimal(x) / decimal.Decimal(y ** 2) | |
| return a, b | |
| async def main(): | |
| r1, r2 = await asyncio.gather( | |
| fractions(0.1, 3, 1, 3), fractions(0.2, 6, 1, 3)) | |
| return r1, r2 | |
| r1, r2 = asyncio.run(main()) | |
| self.assertEqual(str(r1[0]), '0.333') | |
| self.assertEqual(str(r1[1]), '0.111') | |
| self.assertEqual(str(r2[0]), '0.333333') | |
| self.assertEqual(str(r2[1]), '0.111111') | |
| if __name__ == '__main__': | |
| unittest.main() |