tests for asynciotimeouts py import unittest import time import asyncio from test testasyncio utils import awaitwithouttask def teardownmodule asyncio seteventlooppolicynone class timeouttestsunittest isolatedasynciotestcase async def testtimeoutbasicself with self assertraisestimeouterror async with asyncio timeout0 01 as cm await asyncio sleep10 self asserttruecm expired async def testtimeoutatbasicself loop asyncio getrunningloop with self assertraisestimeouterror deadline loop time 0 01 async with asyncio timeoutatdeadline as cm await asyncio sleep10 self asserttruecm expired self assertequaldeadline cm when async def testnestedtimeoutsself loop asyncio getrunningloop cancelled false with self assertraisestimeouterror deadline loop time 0 01 async with asyncio timeoutatdeadline as cm1 only the topmost context manager should raise timeouterror try async with asyncio timeoutatdeadline as cm2 await asyncio sleep10 except asyncio cancellederror cancelled true raise self asserttruecancelled self asserttruecm1 expired self asserttruecm2 expired async def testwaitercancelledself cancelled false with self assertraisestimeouterror async with asyncio timeout0 01 try await asyncio sleep10 except asyncio cancellederror cancelled true raise self asserttruecancelled async def testtimeoutnotcalledself loop asyncio getrunningloop async with asyncio timeout10 as cm await asyncio sleep0 01 t1 loop time self assertfalsecm expired self assertgreatercm when t1 async def testtimeoutdisabledself async with asyncio timeoutnone as cm await asyncio sleep0 01 self assertfalsecm expired self assertisnonecm when async def testtimeoutatdisabledself async with asyncio timeoutatnone as cm await asyncio sleep0 01 self assertfalsecm expired self assertisnonecm when async def testtimeoutzeroself loop asyncio getrunningloop t0 loop time with self assertraisestimeouterror async with asyncio timeout0 as cm await asyncio sleep10 t1 loop time self asserttruecm expired self asserttruet0 cm when t1 async def testtimeoutzerosleepzeroself loop asyncio getrunningloop t0 loop time with self assertraisestimeouterror async with asyncio timeout0 as cm await asyncio sleep0 t1 loop time self asserttruecm expired self asserttruet0 cm when t1 async def testtimeoutinthepastsleepzeroself loop asyncio getrunningloop t0 loop time with self assertraisestimeouterror async with asyncio timeout11 as cm await asyncio sleep0 t1 loop time self asserttruecm expired self asserttruet0 cm when t1 async def testforeignexceptionpassedself with self assertraiseskeyerror async with asyncio timeout0 01 as cm raise keyerror self assertfalsecm expired async def testforeignexceptionontimeoutself async def crash try await asyncio sleep1 finally 10 with self assertraiseszerodivisionerror async with asyncio timeout0 01 await crash async def testforeigncanceldoesnttimeoutifnotexpiredself with self assertraisesasyncio cancellederror async with asyncio timeout10 as cm asyncio currenttask cancel await asyncio sleep10 self assertfalsecm expired async def testoutertaskisnotcancelledself async def outer none with self assertraisestimeouterror async with asyncio timeout0 001 await asyncio sleep10 task asyncio createtaskouter await task self assertfalsetask cancelled self asserttruetask done async def testnestedtimeoutsconcurrentself with self assertraisestimeouterror async with asyncio timeout0 002 with self assertraisestimeouterror async with asyncio timeout0 1 pretend we crunch some numbers time sleep0 01 await asyncio sleep1 async def testnestedtimeoutsloopbusyself after the inner timeout is an expensive operation which should be stopped by the outer timeout loop asyncio getrunningloop disable a message about long running task loop slowcallbackduration 10 t0 loop time with self assertraisestimeouterror async with asyncio timeout0 1 1 with self assertraisestimeouterror async with asyncio timeout0 01 2 pretend the loop is busy for a while time sleep0 1 await asyncio sleep1 timeouterror was cought by 2 await asyncio sleep10 this sleep should be interrupted by 1 t1 loop time self asserttruet0 t1 t0 1 async def testrescheduleself loop asyncio getrunningloop fut loop createfuture deadline1 loop time 10 deadline2 deadline1 20 async def f async with asyncio timeoutatdeadline1 as cm fut setresultcm await asyncio sleep50 task asyncio createtaskf cm await fut self assertequalcm when deadline1 cm rescheduledeadline2 self assertequalcm when deadline2 cm reschedulenone self assertisnonecm when task cancel with self assertraisesasyncio cancellederror await task self assertfalsecm expired async def testrepractiveself async with asyncio timeout10 as cm self assertregexreprcm rtimeout active whend d async def testreprexpiredself with self assertraisestimeouterror async with asyncio timeout0 01 as cm await asyncio sleep10 self assertequalreprcm timeout expired async def testreprfinishedself async with asyncio timeout10 as cm await asyncio sleep0 self assertequalreprcm timeout finished async def testreprdisabledself async with asyncio timeoutnone as cm self assertequalreprcm rtimeout active whennone async def testnestedtimeoutinfinallyself with self assertraisestimeouterror async with asyncio timeout0 01 try await asyncio sleep1 finally with self assertraisestimeouterror async with asyncio timeout0 01 await asyncio sleep10 async def testtimeoutaftercancellationself try asyncio currenttask cancel await asyncio sleep1 work which will be cancelled except asyncio cancellederror pass finally with self assertraisestimeouterror async with asyncio timeout0 0 await asyncio sleep1 some cleanup async def testcancelintimeoutaftercancellationself try asyncio currenttask cancel await asyncio sleep1 work which will be cancelled except asyncio cancellederror pass finally with self assertraisesasyncio cancellederror async with asyncio timeout1 0 asyncio currenttask cancel await asyncio sleep2 some cleanup async def testtimeoutexceptioncause self with self assertraisesasyncio timeouterror as exc async with asyncio timeout0 await asyncio sleep1 cause exc exception cause assert isinstancecause asyncio cancellederror async def testtimeoutalreadyenteredself async with asyncio timeout0 01 as cm with self assertraisesregexruntimeerror has already been entered async with cm pass async def testtimeoutdoubleenterself async with asyncio timeout0 01 as cm pass with self assertraisesregexruntimeerror has already been entered async with cm pass async def testtimeoutfinishedself async with asyncio timeout0 01 as cm pass with self assertraisesregexruntimeerror finished cm reschedule0 02 async def testtimeoutexpiredself with self assertraisestimeouterror async with asyncio timeout0 01 as cm await asyncio sleep1 with self assertraisesregexruntimeerror expired cm reschedule0 02 async def testtimeoutexpiringself async with asyncio timeout0 01 as cm with self assertraisesasyncio cancellederror await asyncio sleep1 with self assertraisesregexruntimeerror expiring cm reschedule0 02 async def testtimeoutnotenteredself cm asyncio timeout0 01 with self assertraisesregexruntimeerror has not been entered cm reschedule0 02 async def testtimeoutwithouttaskself cm asyncio timeout0 01 with self assertraisesregexruntimeerror task await awaitwithouttaskcm aenter with self assertraisesregexruntimeerror has not been entered cm reschedule0 02 if name main unittest main only the topmost context manager should raise timeouterror pretend we crunch some numbers after the inner timeout is an expensive operation which should be stopped by the outer timeout disable a message about long running task 1 2 pretend the loop is busy for a while timeouterror was cought by 2 this sleep should be interrupted by 1 work which will be cancelled some cleanup work which will be cancelled some cleanup
import unittest
import time

import asyncio

from test.test_asyncio.utils import await_without_task


def tearDownModule():
    asyncio.set_event_loop_policy(None)

class TimeoutTests(unittest.IsolatedAsyncioTestCase):

    async def test_timeout_basic(self):
        with self.assertRaises(TimeoutError):
            async with asyncio.timeout(0.01) as cm:
                await asyncio.sleep(10)
        self.assertTrue(cm.expired())

    async def test_timeout_at_basic(self):
        loop = asyncio.get_running_loop()

        with self.assertRaises(TimeoutError):
            deadline = loop.time() + 0.01
            async with asyncio.timeout_at(deadline) as cm:
                await asyncio.sleep(10)
        self.assertTrue(cm.expired())
        self.assertEqual(deadline, cm.when())

    async def test_nested_timeouts(self):
        loop = asyncio.get_running_loop()
        cancelled = False
        with self.assertRaises(TimeoutError):
            deadline = loop.time() + 0.01
            async with asyncio.timeout_at(deadline) as cm1:
                
                try:
                    async with asyncio.timeout_at(deadline) as cm2:
                        await asyncio.sleep(10)
                except asyncio.CancelledError:
                    cancelled = True
                    raise
        self.assertTrue(cancelled)
        self.assertTrue(cm1.expired())
        self.assertTrue(cm2.expired())

    async def test_waiter_cancelled(self):
        cancelled = False
        with self.assertRaises(TimeoutError):
            async with asyncio.timeout(0.01):
                try:
                    await asyncio.sleep(10)
                except asyncio.CancelledError:
                    cancelled = True
                    raise
        self.assertTrue(cancelled)

    async def test_timeout_not_called(self):
        loop = asyncio.get_running_loop()
        async with asyncio.timeout(10) as cm:
            await asyncio.sleep(0.01)
        t1 = loop.time()

        self.assertFalse(cm.expired())
        self.assertGreater(cm.when(), t1)

    async def test_timeout_disabled(self):
        async with asyncio.timeout(None) as cm:
            await asyncio.sleep(0.01)

        self.assertFalse(cm.expired())
        self.assertIsNone(cm.when())

    async def test_timeout_at_disabled(self):
        async with asyncio.timeout_at(None) as cm:
            await asyncio.sleep(0.01)

        self.assertFalse(cm.expired())
        self.assertIsNone(cm.when())

    async def test_timeout_zero(self):
        loop = asyncio.get_running_loop()
        t0 = loop.time()
        with self.assertRaises(TimeoutError):
            async with asyncio.timeout(0) as cm:
                await asyncio.sleep(10)
        t1 = loop.time()
        self.assertTrue(cm.expired())
        self.assertTrue(t0 <= cm.when() <= t1)

    async def test_timeout_zero_sleep_zero(self):
        loop = asyncio.get_running_loop()
        t0 = loop.time()
        with self.assertRaises(TimeoutError):
            async with asyncio.timeout(0) as cm:
                await asyncio.sleep(0)
        t1 = loop.time()
        self.assertTrue(cm.expired())
        self.assertTrue(t0 <= cm.when() <= t1)

    async def test_timeout_in_the_past_sleep_zero(self):
        loop = asyncio.get_running_loop()
        t0 = loop.time()
        with self.assertRaises(TimeoutError):
            async with asyncio.timeout(-11) as cm:
                await asyncio.sleep(0)
        t1 = loop.time()
        self.assertTrue(cm.expired())
        self.assertTrue(t0 >= cm.when() <= t1)

    async def test_foreign_exception_passed(self):
        with self.assertRaises(KeyError):
            async with asyncio.timeout(0.01) as cm:
                raise KeyError
        self.assertFalse(cm.expired())

    async def test_foreign_exception_on_timeout(self):
        async def crash():
            try:
                await asyncio.sleep(1)
            finally:
                1/0
        with self.assertRaises(ZeroDivisionError):
            async with asyncio.timeout(0.01):
                await crash()

    async def test_foreign_cancel_doesnt_timeout_if_not_expired(self):
        with self.assertRaises(asyncio.CancelledError):
            async with asyncio.timeout(10) as cm:
                asyncio.current_task().cancel()
                await asyncio.sleep(10)
        self.assertFalse(cm.expired())

    async def test_outer_task_is_not_cancelled(self):
        async def outer() -> None:
            with self.assertRaises(TimeoutError):
                async with asyncio.timeout(0.001):
                    await asyncio.sleep(10)

        task = asyncio.create_task(outer())
        await task
        self.assertFalse(task.cancelled())
        self.assertTrue(task.done())

    async def test_nested_timeouts_concurrent(self):
        with self.assertRaises(TimeoutError):
            async with asyncio.timeout(0.002):
                with self.assertRaises(TimeoutError):
                    async with asyncio.timeout(0.1):
                        
                        time.sleep(0.01)
                        await asyncio.sleep(1)

    async def test_nested_timeouts_loop_busy(self):
        
        
        loop = asyncio.get_running_loop()
        
        loop.slow_callback_duration = 10
        t0 = loop.time()
        with self.assertRaises(TimeoutError):
            async with asyncio.timeout(0.1):  
                with self.assertRaises(TimeoutError):
                    async with asyncio.timeout(0.01):  
                        
                        time.sleep(0.1)
                        await asyncio.sleep(1)
                
                await asyncio.sleep(10) 
        t1 = loop.time()
        self.assertTrue(t0 <= t1 <= t0 + 1)

    async def test_reschedule(self):
        loop = asyncio.get_running_loop()
        fut = loop.create_future()
        deadline1 = loop.time() + 10
        deadline2 = deadline1 + 20

        async def f():
            async with asyncio.timeout_at(deadline1) as cm:
                fut.set_result(cm)
                await asyncio.sleep(50)

        task = asyncio.create_task(f())
        cm = await fut

        self.assertEqual(cm.when(), deadline1)
        cm.reschedule(deadline2)
        self.assertEqual(cm.when(), deadline2)
        cm.reschedule(None)
        self.assertIsNone(cm.when())

        task.cancel()

        with self.assertRaises(asyncio.CancelledError):
            await task
        self.assertFalse(cm.expired())

    async def test_repr_active(self):
        async with asyncio.timeout(10) as cm:
            self.assertRegex(repr(cm), r"<Timeout \[active\] when=\d+\.\d*>")

    async def test_repr_expired(self):
        with self.assertRaises(TimeoutError):
            async with asyncio.timeout(0.01) as cm:
                await asyncio.sleep(10)
        self.assertEqual(repr(cm), "<Timeout [expired]>")

    async def test_repr_finished(self):
        async with asyncio.timeout(10) as cm:
            await asyncio.sleep(0)

        self.assertEqual(repr(cm), "<Timeout [finished]>")

    async def test_repr_disabled(self):
        async with asyncio.timeout(None) as cm:
            self.assertEqual(repr(cm), r"<Timeout [active] when=None>")

    async def test_nested_timeout_in_finally(self):
        with self.assertRaises(TimeoutError):
            async with asyncio.timeout(0.01):
                try:
                    await asyncio.sleep(1)
                finally:
                    with self.assertRaises(TimeoutError):
                        async with asyncio.timeout(0.01):
                            await asyncio.sleep(10)

    async def test_timeout_after_cancellation(self):
        try:
            asyncio.current_task().cancel()
            await asyncio.sleep(1)  
        except asyncio.CancelledError:
            pass
        finally:
            with self.assertRaises(TimeoutError):
                async with asyncio.timeout(0.0):
                    await asyncio.sleep(1)  

    async def test_cancel_in_timeout_after_cancellation(self):
        try:
            asyncio.current_task().cancel()
            await asyncio.sleep(1)  
        except asyncio.CancelledError:
            pass
        finally:
            with self.assertRaises(asyncio.CancelledError):
                async with asyncio.timeout(1.0):
                    asyncio.current_task().cancel()
                    await asyncio.sleep(2)  

    async def test_timeout_exception_cause (self):
        with self.assertRaises(asyncio.TimeoutError) as exc:
            async with asyncio.timeout(0):
                await asyncio.sleep(1)
        cause = exc.exception.__cause__
        assert isinstance(cause, asyncio.CancelledError)

    async def test_timeout_already_entered(self):
        async with asyncio.timeout(0.01) as cm:
            with self.assertRaisesRegex(RuntimeError, "has already been entered"):
                async with cm:
                    pass

    async def test_timeout_double_enter(self):
        async with asyncio.timeout(0.01) as cm:
            pass
        with self.assertRaisesRegex(RuntimeError, "has already been entered"):
            async with cm:
                pass

    async def test_timeout_finished(self):
        async with asyncio.timeout(0.01) as cm:
            pass
        with self.assertRaisesRegex(RuntimeError, "finished"):
            cm.reschedule(0.02)

    async def test_timeout_expired(self):
        with self.assertRaises(TimeoutError):
            async with asyncio.timeout(0.01) as cm:
                await asyncio.sleep(1)
        with self.assertRaisesRegex(RuntimeError, "expired"):
            cm.reschedule(0.02)

    async def test_timeout_expiring(self):
        async with asyncio.timeout(0.01) as cm:
            with self.assertRaises(asyncio.CancelledError):
                await asyncio.sleep(1)
            with self.assertRaisesRegex(RuntimeError, "expiring"):
                cm.reschedule(0.02)

    async def test_timeout_not_entered(self):
        cm = asyncio.timeout(0.01)
        with self.assertRaisesRegex(RuntimeError, "has not been entered"):
            cm.reschedule(0.02)

    async def test_timeout_without_task(self):
        cm = asyncio.timeout(0.01)
        with self.assertRaisesRegex(RuntimeError, "task"):
            await await_without_task(cm.__aenter__())
        with self.assertRaisesRegex(RuntimeError, "has not been entered"):
            cm.reschedule(0.02)


if __name__ == '__main__':
    unittest.main()