tests for locks py import unittest from unittest import mock import re import asyncio import collections strrgxrepr r pclass object at paddress r pextras r setunsetlockedunlockedfillingdrainingresettingbroken r value d r waiters d r waiters dd barrier r z rgxrepr re compilestrrgxrepr def teardownmodule asyncio seteventlooppolicynone class locktestsunittest isolatedasynciotestcase async def testreprself lock asyncio lock self asserttruereprlock endswith unlocked self asserttruergxrepr matchreprlock await lock acquire self asserttruereprlock endswith locked self asserttruergxrepr matchreprlock async def testlockself lock asyncio lock with self assertraisesregex typeerror object lock can t be used in await expression await lock self assertfalselock locked async def testlockdoesntacceptloopparameterself primitivescls asyncio lock asyncio condition asyncio event asyncio semaphore asyncio boundedsemaphore loop asyncio getrunningloop for cls in primitivescls with self assertraisesregex typeerror rfcls name init got an unexpected rfkeyword argument loop clslooploop async def testlockbywithstatementself primitives asyncio lock asyncio condition asyncio semaphore asyncio boundedsemaphore for lock in primitives await asyncio sleep0 01 self assertfalselock locked with self assertraisesregex typeerror robject w can t be used in await expression with await lock pass self assertfalselock locked async def testacquireself lock asyncio lock result self asserttrueawait lock acquire async def c1result if await lock acquire result append1 return true async def c2result if await lock acquire result append2 return true async def c3result if await lock acquire result append3 return true t1 asyncio createtaskc1result t2 asyncio createtaskc2result await asyncio sleep0 self assertequal result lock release await asyncio sleep0 self assertequal1 result await asyncio sleep0 self assertequal1 result t3 asyncio createtaskc3result lock release await asyncio sleep0 self assertequal1 2 result lock release await asyncio sleep0 self assertequal1 2 3 result self asserttruet1 done self asserttruet1 result self asserttruet2 done self asserttruet2 result self asserttruet3 done self asserttruet3 result async def testacquirecancelself lock asyncio lock self asserttrueawait lock acquire task asyncio createtasklock acquire asyncio getrunningloop callsoontask cancel with self assertraisesasyncio cancellederror await task self assertfalselock waiters async def testcancelraceself several tasks a acquires the lock b is blocked in acquire c is blocked in acquire now concurrently b is cancelled a releases the lock if b s waiter is marked cancelled but not yet removed from waiters a s release call will crash when trying to set b s waiter instead it should move on to c s waiter setup a has the lock b and c are waiting lock asyncio lock async def lockitname blocker await lock acquire try if blocker is not none await blocker finally lock release fa asyncio getrunningloop createfuture ta asyncio createtasklockit a fa await asyncio sleep0 self asserttruelock locked tb asyncio createtasklockit b none await asyncio sleep0 self assertequallenlock waiters 1 tc asyncio createtasklockit c none await asyncio sleep0 self assertequallenlock waiters 2 create the race and check without the fix this failed at the last assert fa setresultnone tb cancel self asserttruelock waiters0 cancelled await asyncio sleep0 self assertfalselock locked self asserttrueta done self asserttruetb cancelled await tc async def testcancelreleaseraceself issue 32734 acquire 4 locks cancel second release first and 2 locks are taken at once loop asyncio getrunningloop lock asyncio lock lockcount 0 callcount 0 async def lockit nonlocal lockcount nonlocal callcount callcount 1 await lock acquire lockcount 1 def trigger t1 cancel lock release await lock acquire t1 asyncio createtasklockit t2 asyncio createtasklockit t3 asyncio createtasklockit start scheduled tasks await asyncio sleep0 loop callsoontrigger with self assertraisesasyncio cancellederror wait for cancellation await t1 make sure only one lock was taken self assertequallockcount 1 while 3 calls were made to lockit self assertequalcallcount 3 self asserttruet1 cancelled and t2 done cleanup the task that is stuck on acquire t3 cancel await asyncio sleep0 self asserttruet3 cancelled async def testfinishedwaitercancelledself lock asyncio lock await lock acquire self asserttruelock locked tb asyncio createtasklock acquire await asyncio sleep0 self assertequallenlock waiters 1 create a second waiter wake up the first and cancel it without the fix the second was not woken up tc asyncio createtasklock acquire tb cancel lock release await asyncio sleep0 self asserttruelock locked self asserttruetb cancelled cleanup await tc async def testreleasenotacquiredself lock asyncio lock self assertraisesruntimeerror lock release async def testreleasenowaitersself lock asyncio lock await lock acquire self asserttruelock locked lock release self assertfalselock locked async def testcontextmanagerself lock asyncio lock self assertfalselock locked async with lock self asserttruelock locked self assertfalselock locked class eventtestsunittest isolatedasynciotestcase def testreprself ev asyncio event self asserttruereprev endswith unset match rgxrepr matchreprev self assertequalmatch group extras unset ev set self asserttruereprev endswith set self asserttruergxrepr matchreprev ev waiters appendmock mock self asserttrue waiters 1 in reprev self asserttruergxrepr matchreprev async def testwaitself ev asyncio event self assertfalseev isset result async def c1result if await ev wait result append1 async def c2result if await ev wait result append2 async def c3result if await ev wait result append3 t1 asyncio createtaskc1result t2 asyncio createtaskc2result await asyncio sleep0 self assertequal result t3 asyncio createtaskc3result ev set await asyncio sleep0 self assertequal3 1 2 result self asserttruet1 done self assertisnonet1 result self asserttruet2 done self assertisnonet2 result self asserttruet3 done self assertisnonet3 result async def testwaitonsetself ev asyncio event ev set res await ev wait self asserttrueres async def testwaitcancelself ev asyncio event wait asyncio createtaskev wait asyncio getrunningloop callsoonwait cancel with self assertraisesasyncio cancellederror await wait self assertfalseev waiters async def testclearself ev asyncio event self assertfalseev isset ev set self asserttrueev isset ev clear self assertfalseev isset async def testclearwithwaitersself ev asyncio event result async def c1result if await ev wait result append1 return true t asyncio createtaskc1result await asyncio sleep0 self assertequal result ev set ev clear self assertfalseev isset ev set ev set self assertequal1 lenev waiters await asyncio sleep0 self assertequal1 result self assertequal0 lenev waiters self asserttruet done self asserttruet result class conditiontestsunittest isolatedasynciotestcase async def testwaitself cond asyncio condition result async def c1result await cond acquire if await cond wait result append1 return true async def c2result await cond acquire if await cond wait result append2 return true async def c3result await cond acquire if await cond wait result append3 return true t1 asyncio createtaskc1result t2 asyncio createtaskc2result t3 asyncio createtaskc3result await asyncio sleep0 self assertequal result self assertfalsecond locked self asserttrueawait cond acquire cond notify await asyncio sleep0 self assertequal result self asserttruecond locked cond release await asyncio sleep0 self assertequal1 result self asserttruecond locked cond notify2 await asyncio sleep0 self assertequal1 result self asserttruecond locked cond release await asyncio sleep0 self assertequal1 2 result self asserttruecond locked cond release await asyncio sleep0 self assertequal1 2 3 result self asserttruecond locked self asserttruet1 done self asserttruet1 result self asserttruet2 done self asserttruet2 result self asserttruet3 done self asserttruet3 result async def testwaitcancelself cond asyncio condition await cond acquire wait asyncio createtaskcond wait asyncio getrunningloop callsoonwait cancel with self assertraisesasyncio cancellederror await wait self assertfalsecond waiters self asserttruecond locked async def testwaitcancelcontestedself cond asyncio condition await cond acquire self asserttruecond locked waittask asyncio createtaskcond wait await asyncio sleep0 self assertfalsecond locked notify but contest the lock before cancelling await cond acquire self asserttruecond locked cond notify asyncio getrunningloop callsoonwaittask cancel asyncio getrunningloop callsooncond release try await waittask except asyncio cancellederror should not happen since no cancellation points pass self asserttruecond locked async def testwaitcancelafternotifyself see bpo32841 waited false cond asyncio condition async def waitoncond nonlocal waited async with cond waited true make sure this area was reached await cond wait waiter asyncio createtaskwaitoncond await asyncio sleep0 start waiting await cond acquire cond notify await asyncio sleep0 get to acquire waiter cancel await asyncio sleep0 activate cancellation cond release await asyncio sleep0 cancellation should occur self asserttruewaiter cancelled self asserttruewaited async def testwaitunacquiredself cond asyncio condition with self assertraisesruntimeerror await cond wait async def testwaitforself cond asyncio condition presult false def predicate return presult result async def c1result await cond acquire if await cond waitforpredicate result append1 cond release return true t asyncio createtaskc1result await asyncio sleep0 self assertequal result await cond acquire cond notify cond release await asyncio sleep0 self assertequal result presult true await cond acquire cond notify cond release await asyncio sleep0 self assertequal1 result self asserttruet done self asserttruet result async def testwaitforunacquiredself cond asyncio condition predicate can return true immediately res await cond waitforlambda 1 2 3 self assertequal1 2 3 res with self assertraisesruntimeerror await cond waitforlambda false async def testnotifyself cond asyncio condition result async def c1result await cond acquire if await cond wait result append1 cond release return true async def c2result await cond acquire if await cond wait result append2 cond release return true async def c3result await cond acquire if await cond wait result append3 cond release return true t1 asyncio createtaskc1result t2 asyncio createtaskc2result t3 asyncio createtaskc3result await asyncio sleep0 self assertequal result await cond acquire cond notify1 cond release await asyncio sleep0 self assertequal1 result await cond acquire cond notify1 cond notify2048 cond release await asyncio sleep0 self assertequal1 2 3 result self asserttruet1 done self asserttruet1 result self asserttruet2 done self asserttruet2 result self asserttruet3 done self asserttruet3 result async def testnotifyallself cond asyncio condition result async def c1result await cond acquire if await cond wait result append1 cond release return true async def c2result await cond acquire if await cond wait result append2 cond release return true t1 asyncio createtaskc1result t2 asyncio createtaskc2result await asyncio sleep0 self assertequal result await cond acquire cond notifyall cond release await asyncio sleep0 self assertequal1 2 result self asserttruet1 done self asserttruet1 result self asserttruet2 done self asserttruet2 result def testnotifyunacquiredself cond asyncio condition self assertraisesruntimeerror cond notify def testnotifyallunacquiredself cond asyncio condition self assertraisesruntimeerror cond notifyall async def testreprself cond asyncio condition self asserttrue unlocked in reprcond self asserttruergxrepr matchreprcond await cond acquire self asserttrue locked in reprcond cond waiters appendmock mock self asserttrue waiters 1 in reprcond self asserttruergxrepr matchreprcond cond waiters appendmock mock self asserttrue waiters 2 in reprcond self asserttruergxrepr matchreprcond async def testcontextmanagerself cond asyncio condition self assertfalsecond locked async with cond self asserttruecond locked self assertfalsecond locked async def testexplicitlockself async def flocknone condnone if lock is none lock asyncio lock if cond is none cond asyncio conditionlock self assertiscond lock lock self assertfalselock locked self assertfalsecond locked async with cond self asserttruelock locked self asserttruecond locked self assertfalselock locked self assertfalsecond locked async with lock self asserttruelock locked self asserttruecond locked self assertfalselock locked self assertfalsecond locked all should work in the same way await f await fasyncio lock lock asyncio lock await flock asyncio conditionlock async def testambiguousloopsself loop asyncio neweventloop self addcleanuploop close async def wrongloopinlock with self assertraisestypeerror asyncio locklooploop actively disallowed since 3 10 lock asyncio lock lock loop loop use private api for testing async with lock acquired immediately via the fastpath without interaction with any event loop cond asyncio conditionlock cond acquire will trigger waiting on the lock and it will discover the event loop mismatch with self assertraisesregex runtimeerror is bound to a different event loop await cond acquire async def wrongloopincond same analogy here with the condition s loop lock asyncio lock async with lock with self assertraisestypeerror asyncio conditionlock looploop cond asyncio conditionlock cond loop loop with self assertraisesregex runtimeerror is bound to a different event loop await cond wait await wrongloopinlock await wrongloopincond async def testtimeoutinblockself condition asyncio condition async with condition with self assertraisesasyncio timeouterror await asyncio waitforcondition wait timeout0 5 class semaphoretestsunittest isolatedasynciotestcase def testinitialvaluezeroself sem asyncio semaphore0 self asserttruesem locked async def testreprself sem asyncio semaphore self asserttruereprsem endswith unlocked value 1 self asserttruergxrepr matchreprsem await sem acquire self asserttruereprsem endswith locked self asserttrue waiters not in reprsem self asserttruergxrepr matchreprsem if sem waiters is none sem waiters collections deque sem waiters appendmock mock self asserttrue waiters 1 in reprsem self asserttruergxrepr matchreprsem sem waiters appendmock mock self asserttrue waiters 2 in reprsem self asserttruergxrepr matchreprsem async def testsemaphoreself sem asyncio semaphore self assertequal1 sem value with self assertraisesregex typeerror object semaphore can t be used in await expression await sem self assertfalsesem locked self assertequal1 sem value def testsemaphorevalueself self assertraisesvalueerror asyncio semaphore 1 async def testacquireself sem asyncio semaphore3 result self asserttrueawait sem acquire self asserttrueawait sem acquire self assertfalsesem locked async def c1result await sem acquire result append1 return true async def c2result await sem acquire result append2 return true async def c3result await sem acquire result append3 return true async def c4result await sem acquire result append4 return true t1 asyncio createtaskc1result t2 asyncio createtaskc2result t3 asyncio createtaskc3result await asyncio sleep0 self assertequal1 result self asserttruesem locked self assertequal2 lensem waiters self assertequal0 sem value t4 asyncio createtaskc4result sem release sem release self assertequal0 sem value await asyncio sleep0 self assertequal0 sem value self assertequal3 lenresult self asserttruesem locked self assertequal1 lensem waiters self assertequal0 sem value self asserttruet1 done self asserttruet1 result racetasks t2 t3 t4 donetasks t for t in racetasks if t done and t result self assertequal2 lendonetasks cleanup locked semaphore sem release await asyncio gatherracetasks async def testacquirecancelself sem asyncio semaphore await sem acquire acquire asyncio createtasksem acquire asyncio getrunningloop callsoonacquire cancel with self assertraisesasyncio cancellederror await acquire self asserttruenot sem waiters or allwaiter done for waiter in sem waiters async def testacquirecancelbeforeawokenself sem asyncio semaphorevalue0 t1 asyncio createtasksem acquire t2 asyncio createtasksem acquire t3 asyncio createtasksem acquire t4 asyncio createtasksem acquire await asyncio sleep0 t1 cancel t2 cancel sem release await asyncio sleep0 await asyncio sleep0 numdone sumt done for t in t3 t4 self assertequalnumdone 1 self asserttruet3 done self assertfalset4 done t3 cancel t4 cancel await asyncio sleep0 async def testacquirehangself sem asyncio semaphorevalue0 t1 asyncio createtasksem acquire t2 asyncio createtasksem acquire await asyncio sleep0 t1 cancel sem release await asyncio sleep0 await asyncio sleep0 self asserttruesem locked self asserttruet2 done async def testacquirenohangself sem asyncio semaphore1 async def c1 async with sem await asyncio sleep0 t2 cancel async def c2 async with sem self assertfalsetrue t1 asyncio createtaskc1 t2 asyncio createtaskc2 r1 r2 await asyncio gathert1 t2 returnexceptionstrue self asserttruer1 is none self asserttrueisinstancer2 asyncio cancellederror await asyncio waitforsem acquire timeout1 0 def testreleasenotacquiredself sem asyncio boundedsemaphore self assertraisesvalueerror sem release async def testreleasenowaitersself sem asyncio semaphore await sem acquire self asserttruesem locked sem release self assertfalsesem locked async def testacquirefifoorderself sem asyncio semaphore1 result async def corotag await sem acquire result appendf tag1 await asyncio sleep0 01 sem release await sem acquire result appendf tag2 await asyncio sleep0 01 sem release async with asyncio taskgroup as tg tg createtaskcoro c1 tg createtaskcoro c2 tg createtaskcoro c3 self assertequal c11 c21 c31 c12 c22 c32 result async def testacquirefifoorder2self sem asyncio semaphore1 result async def c1result await sem acquire result append1 return true async def c2result await sem acquire result append2 sem release await sem acquire result append4 return true async def c3result await sem acquire result append3 return true t1 asyncio createtaskc1result t2 asyncio createtaskc2result t3 asyncio createtaskc3result await asyncio sleep0 sem release sem release tasks t1 t2 t3 await asyncio gathertasks self assertequal1 2 3 4 result async def testacquirefifoorder3self sem asyncio semaphore0 result async def c1result await sem acquire result append1 return true async def c2result await sem acquire result append2 return true async def c3result await sem acquire result append3 return true t1 asyncio createtaskc1result t2 asyncio createtaskc2result t3 asyncio createtaskc3result await asyncio sleep0 t1 cancel await asyncio sleep0 sem release sem release tasks t1 t2 t3 await asyncio gathertasks returnexceptionstrue self assertequal2 3 result class barriertestsunittest isolatedasynciotestcase async def asyncsetupself await super asyncsetup self n 5 def maketasksself n coro tasks asyncio createtaskcoro for in rangen return tasks async def gathertasksself n coro tasks self maketasksn coro res await asyncio gathertasks return res tasks async def testbarrierself barrier asyncio barrierself n self assertinfilling reprbarrier with self assertraisesregex typeerror object barrier can t be used in await expression await barrier self assertinfilling reprbarrier async def testreprself barrier asyncio barrierself n self asserttruergxrepr matchreprbarrier self assertinfilling reprbarrier waiters async def waitbarrier await barrier wait incr 2 for i in rangeincr waiters appendasyncio createtaskwaitbarrier await asyncio sleep0 self asserttruergxrepr matchreprbarrier self asserttruefwaiters incrself n in reprbarrier self assertinfilling reprbarrier create missing waiters for i in rangebarrier parties barrier nwaiting waiters appendasyncio createtaskwaitbarrier await asyncio sleep0 self asserttruergxrepr matchreprbarrier self assertindraining reprbarrier add a part of waiters for i in rangeincr waiters appendasyncio createtaskwaitbarrier await asyncio sleep0 and reset await barrier reset self asserttruergxrepr matchreprbarrier self assertinresetting reprbarrier add a part of waiters again for i in rangeincr waiters appendasyncio createtaskwaitbarrier await asyncio sleep0 and abort await barrier abort self asserttruergxrepr matchreprbarrier self assertinbroken reprbarrier self asserttruebarrier broken suppress unhandled exceptions await asyncio gatherwaiters returnexceptionstrue async def testbarrierpartiesself self assertraisesvalueerror lambda asyncio barrier0 self assertraisesvalueerror lambda asyncio barrier4 self assertisinstanceasyncio barrierself n asyncio barrier async def testcontextmanagerself self n 3 barrier asyncio barrierself n results async def coro async with barrier as i results appendi await self gathertasksself n coro self assertlistequalsortedresults listrangeself n self assertequalbarrier nwaiting 0 self assertfalsebarrier broken async def testfillingonetaskself barrier asyncio barrier1 async def f async with barrier as i return true ret await f self asserttrueret self assertequalbarrier nwaiting 0 self assertfalsebarrier broken async def testfillingonetasktwiceself barrier asyncio barrier1 t1 asyncio createtaskbarrier wait await asyncio sleep0 self assertequalbarrier nwaiting 0 t2 asyncio createtaskbarrier wait await asyncio sleep0 self assertequalt1 result t2 result self assertequalt1 done t2 done self assertequalbarrier nwaiting 0 self assertfalsebarrier broken async def testfillingtaskbytaskself self n 3 barrier asyncio barrierself n t1 asyncio createtaskbarrier wait await asyncio sleep0 self assertequalbarrier nwaiting 1 self assertinfilling reprbarrier t2 asyncio createtaskbarrier wait await asyncio sleep0 self assertequalbarrier nwaiting 2 self assertinfilling reprbarrier t3 asyncio createtaskbarrier wait await asyncio sleep0 await asyncio waitt1 t2 t3 self assertequalbarrier nwaiting 0 self assertfalsebarrier broken async def testfillingtaskswaittwiceself barrier asyncio barrierself n results async def coro async with barrier results appendtrue async with barrier results appendfalse await self gathertasksself n coro self assertequallenresults self n2 self assertequalresults counttrue self n self assertequalresults countfalse self n self assertequalbarrier nwaiting 0 self assertfalsebarrier broken async def testfillingtaskscheckreturnvalueself barrier asyncio barrierself n results1 results2 async def coro async with barrier results1 appendtrue async with barrier as i results2 appendtrue return i res await self gathertasksself n coro self assertequallenresults1 self n self asserttrueallresults1 self assertequallenresults2 self n self asserttrueallresults2 self assertlistequalsortedres listrangeself n self assertequalbarrier nwaiting 0 self assertfalsebarrier broken async def testdrainingstateself barrier asyncio barrierself n results async def coro async with barrier barrier state change to filling for the last task release results appenddraining in reprbarrier await self gathertasksself n coro self assertequallenresults self n self assertequalresults1 false self asserttrueallresults self n1 self assertequalbarrier nwaiting 0 self assertfalsebarrier broken async def testblockingtaskswhiledrainingself rewait 2 barrier asyncio barrierself n barriernowaiting asyncio barrierself n rewait results rewaitn rewait counter 0 async def coro nonlocal rewaitn first time waiting await barrier wait after wainting once for all tasks if rewaitn 0 rewaitn 1 wait again only for rewait tasks await barrier wait else wait for end of draining state await barriernowaiting wait wait for other waiting tasks await barrier wait a success means that barriernowaiting was waited for exactly nrewait3 times await self gathertasksself n coro async def testfillingtaskscanceloneself self n 3 barrier asyncio barrierself n results async def coro await barrier wait results appendtrue t1 asyncio createtaskcoro await asyncio sleep0 self assertequalbarrier nwaiting 1 t2 asyncio createtaskcoro await asyncio sleep0 self assertequalbarrier nwaiting 2 t1 cancel await asyncio sleep0 self assertequalbarrier nwaiting 1 with self assertraisesasyncio cancellederror await t1 self asserttruet1 cancelled t3 asyncio createtaskcoro await asyncio sleep0 self assertequalbarrier nwaiting 2 t4 asyncio createtaskcoro await asyncio gathert2 t3 t4 self assertequallenresults self n self asserttrueallresults self assertequalbarrier nwaiting 0 self assertfalsebarrier broken async def testresetbarrierself barrier asyncio barrier1 asyncio createtaskbarrier reset await asyncio sleep0 self assertequalbarrier nwaiting 0 self assertfalsebarrier broken async def testresetbarrierwhiletaskswaitingself barrier asyncio barrierself n results async def coro try await barrier wait except asyncio brokenbarriererror results appendtrue async def cororeset await barrier reset n1 tasks waiting on barrier with n parties tasks self maketasksself n1 coro await asyncio sleep0 reset the barrier asyncio createtaskcororeset await asyncio gathertasks self assertequallenresults self n1 self asserttrueallresults self assertequalbarrier nwaiting 0 self assertnotinresetting reprbarrier self assertfalsebarrier broken async def testresetbarrierwhentaskshalfdrainingself barrier asyncio barrierself n results1 restoftasks self n2 async def coro try await barrier wait except asyncio brokenbarriererror catch here waiting tasks results1 appendtrue else here drained task outside the barrier if restoftasks barrier count tasks outside the barrier await barrier reset await self gathertasksself n coro self assertequalresults1 truerestoftasks self assertequalbarrier nwaiting 0 self assertnotinresetting reprbarrier self assertfalsebarrier broken async def testresetbarrierwhentaskshalfdraininghalfblockingself barrier asyncio barrierself n results1 results2 blockingtasks self n2 count 0 async def coro nonlocal count try await barrier wait except asyncio brokenbarriererror here catch still waiting tasks results1 appendtrue so now waiting again to reach nbparties await barrier wait else count 1 if count blockingtasks reset now raise asyncio brokenbarriererror for waiting tasks await barrier reset so now waiting again to reach nbparties await barrier wait else try await barrier wait except asyncio brokenbarriererror here no catch blocked tasks go to wait results2 appendtrue await self gathertasksself n coro self assertequalresults1 trueblockingtasks self assertequalresults2 self assertequalbarrier nwaiting 0 self assertnotinresetting reprbarrier self assertfalsebarrier broken async def testresetbarrierwhiletaskswaitingandwaitingagainself barrier asyncio barrierself n results1 results2 async def coro1 try await barrier wait except asyncio brokenbarriererror results1 appendtrue finally await barrier wait results2 appendtrue async def coro2 async with barrier results2 appendtrue tasks self maketasksself n1 coro1 reset barrier n1 waiting tasks raise an brokenbarriererror asyncio createtaskbarrier reset await asyncio sleep0 complete waiting tasks in the finally asyncio createtaskcoro2 await asyncio gathertasks self assertfalsebarrier broken self assertequallenresults1 self n1 self asserttrueallresults1 self assertequallenresults2 self n self asserttrueallresults2 self assertequalbarrier nwaiting 0 async def testresetbarrierwhiletasksdrainingself barrier asyncio barrierself n results1 results2 results3 count 0 async def coro nonlocal count i await barrier wait count 1 if count self n last task exited from barrier await barrier reset wait here to reach the parties await barrier wait else try second waiting await barrier wait n1 tasks here results1 appendtrue except exception as e never goes here results2 appendtrue now pass the barrier again last wait must be completed k await barrier wait results3 appendtrue await self gathertasksself n coro self assertfalsebarrier broken self asserttrueallresults1 self assertequallenresults1 self n1 self assertequallenresults2 0 self assertequallenresults3 self n self asserttrueallresults3 self assertequalbarrier nwaiting 0 async def testabortbarrierself barrier asyncio barrier1 asyncio createtaskbarrier abort await asyncio sleep0 self assertequalbarrier nwaiting 0 self asserttruebarrier broken async def testabortbarrierwhentaskshalfdraininghalfblockingself barrier asyncio barrierself n results1 results2 blockingtasks self n2 count 0 async def coro nonlocal count try await barrier wait except asyncio brokenbarriererror here catch tasks waiting to drain results1 appendtrue else count 1 if count blockingtasks abort now raise asyncio brokenbarriererror for all tasks await barrier abort else try await barrier wait except asyncio brokenbarriererror here catch blocked tasks already drained results2 appendtrue await self gathertasksself n coro self asserttruebarrier broken self assertequalresults1 trueblockingtasks self assertequalresults2 trueself nblockingtasks1 self assertequalbarrier nwaiting 0 self assertnotinresetting reprbarrier async def testabortbarrierwhenexceptionself test from threading barrier see locktests testreset barrier asyncio barrierself n results1 results2 async def coro try async with barrier as i if i self n2 raise runtimeerror async with barrier results1 appendtrue except asyncio brokenbarriererror results2 appendtrue except runtimeerror await barrier abort await self gathertasksself n coro self asserttruebarrier broken self assertequallenresults1 0 self assertequallenresults2 self n1 self asserttrueallresults2 self assertequalbarrier nwaiting 0 async def testabortbarrierwhenexceptionthenresettingself test from threading barrier see locktests testabortandreset barrier1 asyncio barrierself n barrier2 asyncio barrierself n results1 results2 results3 async def coro try i await barrier1 wait if i self n2 raise runtimeerror await barrier1 wait results1 appendtrue except asyncio brokenbarriererror results2 appendtrue except runtimeerror await barrier1 abort synchronize and reset the barrier must synchronize first so that everyone has left it when we reset and after so that no one enters it before the reset i await barrier2 wait if i self n2 await barrier1 reset await barrier2 wait await barrier1 wait results3 appendtrue await self gathertasksself n coro self assertfalsebarrier1 broken self assertequallenresults1 0 self assertequallenresults2 self n1 self asserttrueallresults2 self assertequallenresults3 self n self asserttrueallresults3 self assertequalbarrier1 nwaiting 0 if name main unittest main barrier several tasks a acquires the lock b is blocked in acquire c is blocked in acquire now concurrently b is cancelled a releases the lock if b s waiter is marked cancelled but not yet removed from _waiters a s release call will crash when trying to set b s waiter instead it should move on to c s waiter setup a has the lock b and c are waiting create the race and check without the fix this failed at the last assert issue 32734 acquire 4 locks cancel second release first and 2 locks are taken at once start scheduled tasks wait for cancellation make sure only one lock was taken while 3 calls were made to lockit cleanup the task that is stuck on acquire create a second waiter wake up the first and cancel it without the fix the second was not woken up cleanup notify but contest the lock before cancelling should not happen since no cancellation points see bpo 32841 make sure this area was reached start waiting get to acquire activate cancellation cancellation should occur predicate can return true immediately all should work in the same way actively disallowed since 3 10 use private api for testing acquired immediately via the fast path without interaction with any event loop cond acquire will trigger waiting on the lock and it will discover the event loop mismatch same analogy here with the condition s loop cleanup locked semaphore create missing waiters add a part of waiters and reset add a part of waiters again and abort suppress unhandled exceptions barrier state change to filling for the last task release first time waiting after wainting once for all tasks wait again only for rewait tasks wait for end of draining state wait for other waiting tasks a success means that barrier_nowaiting was waited for exactly n rewait 3 times n 1 tasks waiting on barrier with n parties reset the barrier catch here waiting tasks here drained task outside the barrier tasks outside the barrier here catch still waiting tasks so now waiting again to reach nb_parties reset now raise asyncio brokenbarriererror for waiting tasks so now waiting again to reach nb_parties here no catch blocked tasks go to wait reset barrier n 1 waiting tasks raise an brokenbarriererror complete waiting tasks in the finally last task exited from barrier wait here to reach the parties second waiting n 1 tasks here never goes here now pass the barrier again last wait must be completed here catch tasks waiting to drain abort now raise asyncio brokenbarriererror for all tasks here catch blocked tasks already drained test from threading barrier see lock_tests test_reset test from threading barrier see lock_tests test_abort_and_reset synchronize and reset the barrier must synchronize first so that everyone has left it when we reset and after so that no one enters it before the reset
import unittest
from unittest import mock
import re

import asyncio
import collections

STR_RGX_REPR = (
    r'^<(?P<class>.*?) object at (?P<address>.*?)'
    r'\[(?P<extras>'
    r'(set|unset|locked|unlocked|filling|draining|resetting|broken)'
    r'(, value:\d)?'
    r'(, waiters:\d+)?'
    r'(, waiters:\d+\/\d+)?' 
    r')\]>\Z'
)
RGX_REPR = re.compile(STR_RGX_REPR)


def tearDownModule():
    asyncio.set_event_loop_policy(None)


class LockTests(unittest.IsolatedAsyncioTestCase):

    async def test_repr(self):
        lock = asyncio.Lock()
        self.assertTrue(repr(lock).endswith('[unlocked]>'))
        self.assertTrue(RGX_REPR.match(repr(lock)))

        await lock.acquire()
        self.assertTrue(repr(lock).endswith('[locked]>'))
        self.assertTrue(RGX_REPR.match(repr(lock)))

    async def test_lock(self):
        lock = asyncio.Lock()

        with self.assertRaisesRegex(
            TypeError,
            "object Lock can't be used in 'await' expression"
        ):
            await lock

        self.assertFalse(lock.locked())

    async def test_lock_doesnt_accept_loop_parameter(self):
        primitives_cls = [
            asyncio.Lock,
            asyncio.Condition,
            asyncio.Event,
            asyncio.Semaphore,
            asyncio.BoundedSemaphore,
        ]

        loop = asyncio.get_running_loop()

        for cls in primitives_cls:
            with self.assertRaisesRegex(
                TypeError,
                rf"{cls.__name__}\.__init__\(\) got an unexpected "
                rf"keyword argument 'loop'"
            ):
                cls(loop=loop)

    async def test_lock_by_with_statement(self):
        primitives = [
            asyncio.Lock(),
            asyncio.Condition(),
            asyncio.Semaphore(),
            asyncio.BoundedSemaphore(),
        ]

        for lock in primitives:
            await asyncio.sleep(0.01)
            self.assertFalse(lock.locked())
            with self.assertRaisesRegex(
                TypeError,
                r"object \w+ can't be used in 'await' expression"
            ):
                with await lock:
                    pass
            self.assertFalse(lock.locked())

    async def test_acquire(self):
        lock = asyncio.Lock()
        result = []

        self.assertTrue(await lock.acquire())

        async def c1(result):
            if await lock.acquire():
                result.append(1)
            return True

        async def c2(result):
            if await lock.acquire():
                result.append(2)
            return True

        async def c3(result):
            if await lock.acquire():
                result.append(3)
            return True

        t1 = asyncio.create_task(c1(result))
        t2 = asyncio.create_task(c2(result))

        await asyncio.sleep(0)
        self.assertEqual([], result)

        lock.release()
        await asyncio.sleep(0)
        self.assertEqual([1], result)

        await asyncio.sleep(0)
        self.assertEqual([1], result)

        t3 = asyncio.create_task(c3(result))

        lock.release()
        await asyncio.sleep(0)
        self.assertEqual([1, 2], result)

        lock.release()
        await asyncio.sleep(0)
        self.assertEqual([1, 2, 3], result)

        self.assertTrue(t1.done())
        self.assertTrue(t1.result())
        self.assertTrue(t2.done())
        self.assertTrue(t2.result())
        self.assertTrue(t3.done())
        self.assertTrue(t3.result())

    async def test_acquire_cancel(self):
        lock = asyncio.Lock()
        self.assertTrue(await lock.acquire())

        task = asyncio.create_task(lock.acquire())
        asyncio.get_running_loop().call_soon(task.cancel)
        with self.assertRaises(asyncio.CancelledError):
            await task
        self.assertFalse(lock._waiters)

    async def test_cancel_race(self):
        
        
        
        
        
        
        
        
        
        
        
        

        
        lock = asyncio.Lock()

        async def lockit(name, blocker):
            await lock.acquire()
            try:
                if blocker is not None:
                    await blocker
            finally:
                lock.release()

        fa = asyncio.get_running_loop().create_future()
        ta = asyncio.create_task(lockit('A', fa))
        await asyncio.sleep(0)
        self.assertTrue(lock.locked())
        tb = asyncio.create_task(lockit('B', None))
        await asyncio.sleep(0)
        self.assertEqual(len(lock._waiters), 1)
        tc = asyncio.create_task(lockit('C', None))
        await asyncio.sleep(0)
        self.assertEqual(len(lock._waiters), 2)

        
        
        fa.set_result(None)
        tb.cancel()
        self.assertTrue(lock._waiters[0].cancelled())
        await asyncio.sleep(0)
        self.assertFalse(lock.locked())
        self.assertTrue(ta.done())
        self.assertTrue(tb.cancelled())
        await tc

    async def test_cancel_release_race(self):
        
        
        
        loop = asyncio.get_running_loop()
        lock = asyncio.Lock()
        lock_count = 0
        call_count = 0

        async def lockit():
            nonlocal lock_count
            nonlocal call_count
            call_count += 1
            await lock.acquire()
            lock_count += 1

        def trigger():
            t1.cancel()
            lock.release()

        await lock.acquire()

        t1 = asyncio.create_task(lockit())
        t2 = asyncio.create_task(lockit())
        t3 = asyncio.create_task(lockit())

        
        await asyncio.sleep(0)

        loop.call_soon(trigger)
        with self.assertRaises(asyncio.CancelledError):
            
            await t1

        
        self.assertEqual(lock_count, 1)
        
        self.assertEqual(call_count, 3)
        self.assertTrue(t1.cancelled() and t2.done())

        
        t3.cancel()
        await asyncio.sleep(0)
        self.assertTrue(t3.cancelled())

    async def test_finished_waiter_cancelled(self):
        lock = asyncio.Lock()

        await lock.acquire()
        self.assertTrue(lock.locked())

        tb = asyncio.create_task(lock.acquire())
        await asyncio.sleep(0)
        self.assertEqual(len(lock._waiters), 1)

        
        
        tc = asyncio.create_task(lock.acquire())
        tb.cancel()
        lock.release()
        await asyncio.sleep(0)

        self.assertTrue(lock.locked())
        self.assertTrue(tb.cancelled())

        
        await tc

    async def test_release_not_acquired(self):
        lock = asyncio.Lock()

        self.assertRaises(RuntimeError, lock.release)

    async def test_release_no_waiters(self):
        lock = asyncio.Lock()
        await lock.acquire()
        self.assertTrue(lock.locked())

        lock.release()
        self.assertFalse(lock.locked())

    async def test_context_manager(self):
        lock = asyncio.Lock()
        self.assertFalse(lock.locked())

        async with lock:
            self.assertTrue(lock.locked())

        self.assertFalse(lock.locked())


class EventTests(unittest.IsolatedAsyncioTestCase):

    def test_repr(self):
        ev = asyncio.Event()
        self.assertTrue(repr(ev).endswith('[unset]>'))
        match = RGX_REPR.match(repr(ev))
        self.assertEqual(match.group('extras'), 'unset')

        ev.set()
        self.assertTrue(repr(ev).endswith('[set]>'))
        self.assertTrue(RGX_REPR.match(repr(ev)))

        ev._waiters.append(mock.Mock())
        self.assertTrue('waiters:1' in repr(ev))
        self.assertTrue(RGX_REPR.match(repr(ev)))

    async def test_wait(self):
        ev = asyncio.Event()
        self.assertFalse(ev.is_set())

        result = []

        async def c1(result):
            if await ev.wait():
                result.append(1)

        async def c2(result):
            if await ev.wait():
                result.append(2)

        async def c3(result):
            if await ev.wait():
                result.append(3)

        t1 = asyncio.create_task(c1(result))
        t2 = asyncio.create_task(c2(result))

        await asyncio.sleep(0)
        self.assertEqual([], result)

        t3 = asyncio.create_task(c3(result))

        ev.set()
        await asyncio.sleep(0)
        self.assertEqual([3, 1, 2], result)

        self.assertTrue(t1.done())
        self.assertIsNone(t1.result())
        self.assertTrue(t2.done())
        self.assertIsNone(t2.result())
        self.assertTrue(t3.done())
        self.assertIsNone(t3.result())

    async def test_wait_on_set(self):
        ev = asyncio.Event()
        ev.set()

        res = await ev.wait()
        self.assertTrue(res)

    async def test_wait_cancel(self):
        ev = asyncio.Event()

        wait = asyncio.create_task(ev.wait())
        asyncio.get_running_loop().call_soon(wait.cancel)
        with self.assertRaises(asyncio.CancelledError):
            await wait
        self.assertFalse(ev._waiters)

    async def test_clear(self):
        ev = asyncio.Event()
        self.assertFalse(ev.is_set())

        ev.set()
        self.assertTrue(ev.is_set())

        ev.clear()
        self.assertFalse(ev.is_set())

    async def test_clear_with_waiters(self):
        ev = asyncio.Event()
        result = []

        async def c1(result):
            if await ev.wait():
                result.append(1)
            return True

        t = asyncio.create_task(c1(result))
        await asyncio.sleep(0)
        self.assertEqual([], result)

        ev.set()
        ev.clear()
        self.assertFalse(ev.is_set())

        ev.set()
        ev.set()
        self.assertEqual(1, len(ev._waiters))

        await asyncio.sleep(0)
        self.assertEqual([1], result)
        self.assertEqual(0, len(ev._waiters))

        self.assertTrue(t.done())
        self.assertTrue(t.result())


class ConditionTests(unittest.IsolatedAsyncioTestCase):

    async def test_wait(self):
        cond = asyncio.Condition()
        result = []

        async def c1(result):
            await cond.acquire()
            if await cond.wait():
                result.append(1)
            return True

        async def c2(result):
            await cond.acquire()
            if await cond.wait():
                result.append(2)
            return True

        async def c3(result):
            await cond.acquire()
            if await cond.wait():
                result.append(3)
            return True

        t1 = asyncio.create_task(c1(result))
        t2 = asyncio.create_task(c2(result))
        t3 = asyncio.create_task(c3(result))

        await asyncio.sleep(0)
        self.assertEqual([], result)
        self.assertFalse(cond.locked())

        self.assertTrue(await cond.acquire())
        cond.notify()
        await asyncio.sleep(0)
        self.assertEqual([], result)
        self.assertTrue(cond.locked())

        cond.release()
        await asyncio.sleep(0)
        self.assertEqual([1], result)
        self.assertTrue(cond.locked())

        cond.notify(2)
        await asyncio.sleep(0)
        self.assertEqual([1], result)
        self.assertTrue(cond.locked())

        cond.release()
        await asyncio.sleep(0)
        self.assertEqual([1, 2], result)
        self.assertTrue(cond.locked())

        cond.release()
        await asyncio.sleep(0)
        self.assertEqual([1, 2, 3], result)
        self.assertTrue(cond.locked())

        self.assertTrue(t1.done())
        self.assertTrue(t1.result())
        self.assertTrue(t2.done())
        self.assertTrue(t2.result())
        self.assertTrue(t3.done())
        self.assertTrue(t3.result())

    async def test_wait_cancel(self):
        cond = asyncio.Condition()
        await cond.acquire()

        wait = asyncio.create_task(cond.wait())
        asyncio.get_running_loop().call_soon(wait.cancel)
        with self.assertRaises(asyncio.CancelledError):
            await wait
        self.assertFalse(cond._waiters)
        self.assertTrue(cond.locked())

    async def test_wait_cancel_contested(self):
        cond = asyncio.Condition()

        await cond.acquire()
        self.assertTrue(cond.locked())

        wait_task = asyncio.create_task(cond.wait())
        await asyncio.sleep(0)
        self.assertFalse(cond.locked())

        
        await cond.acquire()
        self.assertTrue(cond.locked())
        cond.notify()
        asyncio.get_running_loop().call_soon(wait_task.cancel)
        asyncio.get_running_loop().call_soon(cond.release)

        try:
            await wait_task
        except asyncio.CancelledError:
            
            pass

        self.assertTrue(cond.locked())

    async def test_wait_cancel_after_notify(self):
        
        waited = False

        cond = asyncio.Condition()

        async def wait_on_cond():
            nonlocal waited
            async with cond:
                waited = True  
                await cond.wait()

        waiter = asyncio.create_task(wait_on_cond())
        await asyncio.sleep(0)  

        await cond.acquire()
        cond.notify()
        await asyncio.sleep(0)  
        waiter.cancel()
        await asyncio.sleep(0)  
        cond.release()
        await asyncio.sleep(0)  

        self.assertTrue(waiter.cancelled())
        self.assertTrue(waited)

    async def test_wait_unacquired(self):
        cond = asyncio.Condition()
        with self.assertRaises(RuntimeError):
            await cond.wait()

    async def test_wait_for(self):
        cond = asyncio.Condition()
        presult = False

        def predicate():
            return presult

        result = []

        async def c1(result):
            await cond.acquire()
            if await cond.wait_for(predicate):
                result.append(1)
                cond.release()
            return True

        t = asyncio.create_task(c1(result))

        await asyncio.sleep(0)
        self.assertEqual([], result)

        await cond.acquire()
        cond.notify()
        cond.release()
        await asyncio.sleep(0)
        self.assertEqual([], result)

        presult = True
        await cond.acquire()
        cond.notify()
        cond.release()
        await asyncio.sleep(0)
        self.assertEqual([1], result)

        self.assertTrue(t.done())
        self.assertTrue(t.result())

    async def test_wait_for_unacquired(self):
        cond = asyncio.Condition()

        
        res = await cond.wait_for(lambda: [1, 2, 3])
        self.assertEqual([1, 2, 3], res)

        with self.assertRaises(RuntimeError):
            await cond.wait_for(lambda: False)

    async def test_notify(self):
        cond = asyncio.Condition()
        result = []

        async def c1(result):
            await cond.acquire()
            if await cond.wait():
                result.append(1)
                cond.release()
            return True

        async def c2(result):
            await cond.acquire()
            if await cond.wait():
                result.append(2)
                cond.release()
            return True

        async def c3(result):
            await cond.acquire()
            if await cond.wait():
                result.append(3)
                cond.release()
            return True

        t1 = asyncio.create_task(c1(result))
        t2 = asyncio.create_task(c2(result))
        t3 = asyncio.create_task(c3(result))

        await asyncio.sleep(0)
        self.assertEqual([], result)

        await cond.acquire()
        cond.notify(1)
        cond.release()
        await asyncio.sleep(0)
        self.assertEqual([1], result)

        await cond.acquire()
        cond.notify(1)
        cond.notify(2048)
        cond.release()
        await asyncio.sleep(0)
        self.assertEqual([1, 2, 3], result)

        self.assertTrue(t1.done())
        self.assertTrue(t1.result())
        self.assertTrue(t2.done())
        self.assertTrue(t2.result())
        self.assertTrue(t3.done())
        self.assertTrue(t3.result())

    async def test_notify_all(self):
        cond = asyncio.Condition()

        result = []

        async def c1(result):
            await cond.acquire()
            if await cond.wait():
                result.append(1)
                cond.release()
            return True

        async def c2(result):
            await cond.acquire()
            if await cond.wait():
                result.append(2)
                cond.release()
            return True

        t1 = asyncio.create_task(c1(result))
        t2 = asyncio.create_task(c2(result))

        await asyncio.sleep(0)
        self.assertEqual([], result)

        await cond.acquire()
        cond.notify_all()
        cond.release()
        await asyncio.sleep(0)
        self.assertEqual([1, 2], result)

        self.assertTrue(t1.done())
        self.assertTrue(t1.result())
        self.assertTrue(t2.done())
        self.assertTrue(t2.result())

    def test_notify_unacquired(self):
        cond = asyncio.Condition()
        self.assertRaises(RuntimeError, cond.notify)

    def test_notify_all_unacquired(self):
        cond = asyncio.Condition()
        self.assertRaises(RuntimeError, cond.notify_all)

    async def test_repr(self):
        cond = asyncio.Condition()
        self.assertTrue('unlocked' in repr(cond))
        self.assertTrue(RGX_REPR.match(repr(cond)))

        await cond.acquire()
        self.assertTrue('locked' in repr(cond))

        cond._waiters.append(mock.Mock())
        self.assertTrue('waiters:1' in repr(cond))
        self.assertTrue(RGX_REPR.match(repr(cond)))

        cond._waiters.append(mock.Mock())
        self.assertTrue('waiters:2' in repr(cond))
        self.assertTrue(RGX_REPR.match(repr(cond)))

    async def test_context_manager(self):
        cond = asyncio.Condition()
        self.assertFalse(cond.locked())
        async with cond:
            self.assertTrue(cond.locked())
        self.assertFalse(cond.locked())

    async def test_explicit_lock(self):
        async def f(lock=None, cond=None):
            if lock is None:
                lock = asyncio.Lock()
            if cond is None:
                cond = asyncio.Condition(lock)
            self.assertIs(cond._lock, lock)
            self.assertFalse(lock.locked())
            self.assertFalse(cond.locked())
            async with cond:
                self.assertTrue(lock.locked())
                self.assertTrue(cond.locked())
            self.assertFalse(lock.locked())
            self.assertFalse(cond.locked())
            async with lock:
                self.assertTrue(lock.locked())
                self.assertTrue(cond.locked())
            self.assertFalse(lock.locked())
            self.assertFalse(cond.locked())

        
        await f()
        await f(asyncio.Lock())
        lock = asyncio.Lock()
        await f(lock, asyncio.Condition(lock))

    async def test_ambiguous_loops(self):
        loop = asyncio.new_event_loop()
        self.addCleanup(loop.close)

        async def wrong_loop_in_lock():
            with self.assertRaises(TypeError):
                asyncio.Lock(loop=loop)  
            lock = asyncio.Lock()
            lock._loop = loop  
            async with lock:
                
                
                cond = asyncio.Condition(lock)
                
                
                with self.assertRaisesRegex(
                    RuntimeError,
                    "is bound to a different event loop",
                ):
                    await cond.acquire()

        async def wrong_loop_in_cond():
            
            lock = asyncio.Lock()
            async with lock:
                with self.assertRaises(TypeError):
                    asyncio.Condition(lock, loop=loop)
                cond = asyncio.Condition(lock)
                cond._loop = loop
                with self.assertRaisesRegex(
                    RuntimeError,
                    "is bound to a different event loop",
                ):
                    await cond.wait()

        await wrong_loop_in_lock()
        await wrong_loop_in_cond()

    async def test_timeout_in_block(self):
        condition = asyncio.Condition()
        async with condition:
            with self.assertRaises(asyncio.TimeoutError):
                await asyncio.wait_for(condition.wait(), timeout=0.5)


class SemaphoreTests(unittest.IsolatedAsyncioTestCase):

    def test_initial_value_zero(self):
        sem = asyncio.Semaphore(0)
        self.assertTrue(sem.locked())

    async def test_repr(self):
        sem = asyncio.Semaphore()
        self.assertTrue(repr(sem).endswith('[unlocked, value:1]>'))
        self.assertTrue(RGX_REPR.match(repr(sem)))

        await sem.acquire()
        self.assertTrue(repr(sem).endswith('[locked]>'))
        self.assertTrue('waiters' not in repr(sem))
        self.assertTrue(RGX_REPR.match(repr(sem)))

        if sem._waiters is None:
            sem._waiters = collections.deque()

        sem._waiters.append(mock.Mock())
        self.assertTrue('waiters:1' in repr(sem))
        self.assertTrue(RGX_REPR.match(repr(sem)))

        sem._waiters.append(mock.Mock())
        self.assertTrue('waiters:2' in repr(sem))
        self.assertTrue(RGX_REPR.match(repr(sem)))

    async def test_semaphore(self):
        sem = asyncio.Semaphore()
        self.assertEqual(1, sem._value)

        with self.assertRaisesRegex(
            TypeError,
            "object Semaphore can't be used in 'await' expression",
        ):
            await sem

        self.assertFalse(sem.locked())
        self.assertEqual(1, sem._value)

    def test_semaphore_value(self):
        self.assertRaises(ValueError, asyncio.Semaphore, -1)

    async def test_acquire(self):
        sem = asyncio.Semaphore(3)
        result = []

        self.assertTrue(await sem.acquire())
        self.assertTrue(await sem.acquire())
        self.assertFalse(sem.locked())

        async def c1(result):
            await sem.acquire()
            result.append(1)
            return True

        async def c2(result):
            await sem.acquire()
            result.append(2)
            return True

        async def c3(result):
            await sem.acquire()
            result.append(3)
            return True

        async def c4(result):
            await sem.acquire()
            result.append(4)
            return True

        t1 = asyncio.create_task(c1(result))
        t2 = asyncio.create_task(c2(result))
        t3 = asyncio.create_task(c3(result))

        await asyncio.sleep(0)
        self.assertEqual([1], result)
        self.assertTrue(sem.locked())
        self.assertEqual(2, len(sem._waiters))
        self.assertEqual(0, sem._value)

        t4 = asyncio.create_task(c4(result))

        sem.release()
        sem.release()
        self.assertEqual(0, sem._value)

        await asyncio.sleep(0)
        self.assertEqual(0, sem._value)
        self.assertEqual(3, len(result))
        self.assertTrue(sem.locked())
        self.assertEqual(1, len(sem._waiters))
        self.assertEqual(0, sem._value)

        self.assertTrue(t1.done())
        self.assertTrue(t1.result())
        race_tasks = [t2, t3, t4]
        done_tasks = [t for t in race_tasks if t.done() and t.result()]
        self.assertEqual(2, len(done_tasks))

        
        sem.release()
        await asyncio.gather(*race_tasks)

    async def test_acquire_cancel(self):
        sem = asyncio.Semaphore()
        await sem.acquire()

        acquire = asyncio.create_task(sem.acquire())
        asyncio.get_running_loop().call_soon(acquire.cancel)
        with self.assertRaises(asyncio.CancelledError):
            await acquire
        self.assertTrue((not sem._waiters) or
                        all(waiter.done() for waiter in sem._waiters))

    async def test_acquire_cancel_before_awoken(self):
        sem = asyncio.Semaphore(value=0)

        t1 = asyncio.create_task(sem.acquire())
        t2 = asyncio.create_task(sem.acquire())
        t3 = asyncio.create_task(sem.acquire())
        t4 = asyncio.create_task(sem.acquire())

        await asyncio.sleep(0)

        t1.cancel()
        t2.cancel()
        sem.release()

        await asyncio.sleep(0)
        await asyncio.sleep(0)
        num_done = sum(t.done() for t in [t3, t4])
        self.assertEqual(num_done, 1)
        self.assertTrue(t3.done())
        self.assertFalse(t4.done())

        t3.cancel()
        t4.cancel()
        await asyncio.sleep(0)

    async def test_acquire_hang(self):
        sem = asyncio.Semaphore(value=0)

        t1 = asyncio.create_task(sem.acquire())
        t2 = asyncio.create_task(sem.acquire())
        await asyncio.sleep(0)

        t1.cancel()
        sem.release()
        await asyncio.sleep(0)
        await asyncio.sleep(0)
        self.assertTrue(sem.locked())
        self.assertTrue(t2.done())

    async def test_acquire_no_hang(self):

        sem = asyncio.Semaphore(1)

        async def c1():
            async with sem:
                await asyncio.sleep(0)
            t2.cancel()

        async def c2():
            async with sem:
                self.assertFalse(True)

        t1 = asyncio.create_task(c1())
        t2 = asyncio.create_task(c2())

        r1, r2 = await asyncio.gather(t1, t2, return_exceptions=True)
        self.assertTrue(r1 is None)
        self.assertTrue(isinstance(r2, asyncio.CancelledError))

        await asyncio.wait_for(sem.acquire(), timeout=1.0)

    def test_release_not_acquired(self):
        sem = asyncio.BoundedSemaphore()

        self.assertRaises(ValueError, sem.release)

    async def test_release_no_waiters(self):
        sem = asyncio.Semaphore()
        await sem.acquire()
        self.assertTrue(sem.locked())

        sem.release()
        self.assertFalse(sem.locked())

    async def test_acquire_fifo_order(self):
        sem = asyncio.Semaphore(1)
        result = []

        async def coro(tag):
            await sem.acquire()
            result.append(f'{tag}_1')
            await asyncio.sleep(0.01)
            sem.release()

            await sem.acquire()
            result.append(f'{tag}_2')
            await asyncio.sleep(0.01)
            sem.release()

        async with asyncio.TaskGroup() as tg:
            tg.create_task(coro('c1'))
            tg.create_task(coro('c2'))
            tg.create_task(coro('c3'))

        self.assertEqual(
            ['c1_1', 'c2_1', 'c3_1', 'c1_2', 'c2_2', 'c3_2'],
            result
        )

    async def test_acquire_fifo_order_2(self):
        sem = asyncio.Semaphore(1)
        result = []

        async def c1(result):
            await sem.acquire()
            result.append(1)
            return True

        async def c2(result):
            await sem.acquire()
            result.append(2)
            sem.release()
            await sem.acquire()
            result.append(4)
            return True

        async def c3(result):
            await sem.acquire()
            result.append(3)
            return True

        t1 = asyncio.create_task(c1(result))
        t2 = asyncio.create_task(c2(result))
        t3 = asyncio.create_task(c3(result))

        await asyncio.sleep(0)

        sem.release()
        sem.release()

        tasks = [t1, t2, t3]
        await asyncio.gather(*tasks)
        self.assertEqual([1, 2, 3, 4], result)

    async def test_acquire_fifo_order_3(self):
        sem = asyncio.Semaphore(0)
        result = []

        async def c1(result):
            await sem.acquire()
            result.append(1)
            return True

        async def c2(result):
            await sem.acquire()
            result.append(2)
            return True

        async def c3(result):
            await sem.acquire()
            result.append(3)
            return True

        t1 = asyncio.create_task(c1(result))
        t2 = asyncio.create_task(c2(result))
        t3 = asyncio.create_task(c3(result))

        await asyncio.sleep(0)

        t1.cancel()

        await asyncio.sleep(0)

        sem.release()
        sem.release()

        tasks = [t1, t2, t3]
        await asyncio.gather(*tasks, return_exceptions=True)
        self.assertEqual([2, 3], result)


class BarrierTests(unittest.IsolatedAsyncioTestCase):

    async def asyncSetUp(self):
        await super().asyncSetUp()
        self.N = 5

    def make_tasks(self, n, coro):
        tasks = [asyncio.create_task(coro()) for _ in range(n)]
        return tasks

    async def gather_tasks(self, n, coro):
        tasks = self.make_tasks(n, coro)
        res = await asyncio.gather(*tasks)
        return res, tasks

    async def test_barrier(self):
        barrier = asyncio.Barrier(self.N)
        self.assertIn("filling", repr(barrier))
        with self.assertRaisesRegex(
            TypeError,
            "object Barrier can't be used in 'await' expression",
        ):
            await barrier

        self.assertIn("filling", repr(barrier))

    async def test_repr(self):
        barrier = asyncio.Barrier(self.N)

        self.assertTrue(RGX_REPR.match(repr(barrier)))
        self.assertIn("filling", repr(barrier))

        waiters = []
        async def wait(barrier):
            await barrier.wait()

        incr = 2
        for i in range(incr):
            waiters.append(asyncio.create_task(wait(barrier)))
        await asyncio.sleep(0)

        self.assertTrue(RGX_REPR.match(repr(barrier)))
        self.assertTrue(f"waiters:{incr}/{self.N}" in repr(barrier))
        self.assertIn("filling", repr(barrier))

        
        for i in range(barrier.parties - barrier.n_waiting):
            waiters.append(asyncio.create_task(wait(barrier)))
        await asyncio.sleep(0)

        self.assertTrue(RGX_REPR.match(repr(barrier)))
        self.assertIn("draining", repr(barrier))

        
        for i in range(incr):
            waiters.append(asyncio.create_task(wait(barrier)))
        await asyncio.sleep(0)
        
        await barrier.reset()

        self.assertTrue(RGX_REPR.match(repr(barrier)))
        self.assertIn("resetting", repr(barrier))

        
        for i in range(incr):
            waiters.append(asyncio.create_task(wait(barrier)))
        await asyncio.sleep(0)
        
        await barrier.abort()

        self.assertTrue(RGX_REPR.match(repr(barrier)))
        self.assertIn("broken", repr(barrier))
        self.assertTrue(barrier.broken)

        
        await asyncio.gather(*waiters, return_exceptions=True)

    async def test_barrier_parties(self):
        self.assertRaises(ValueError, lambda: asyncio.Barrier(0))
        self.assertRaises(ValueError, lambda: asyncio.Barrier(-4))

        self.assertIsInstance(asyncio.Barrier(self.N), asyncio.Barrier)

    async def test_context_manager(self):
        self.N = 3
        barrier = asyncio.Barrier(self.N)
        results = []

        async def coro():
            async with barrier as i:
                results.append(i)

        await self.gather_tasks(self.N, coro)

        self.assertListEqual(sorted(results), list(range(self.N)))
        self.assertEqual(barrier.n_waiting, 0)
        self.assertFalse(barrier.broken)

    async def test_filling_one_task(self):
        barrier = asyncio.Barrier(1)

        async def f():
            async with barrier as i:
                return True

        ret = await f()

        self.assertTrue(ret)
        self.assertEqual(barrier.n_waiting, 0)
        self.assertFalse(barrier.broken)

    async def test_filling_one_task_twice(self):
        barrier = asyncio.Barrier(1)

        t1 = asyncio.create_task(barrier.wait())
        await asyncio.sleep(0)
        self.assertEqual(barrier.n_waiting, 0)

        t2 = asyncio.create_task(barrier.wait())
        await asyncio.sleep(0)

        self.assertEqual(t1.result(), t2.result())
        self.assertEqual(t1.done(), t2.done())

        self.assertEqual(barrier.n_waiting, 0)
        self.assertFalse(barrier.broken)

    async def test_filling_task_by_task(self):
        self.N = 3
        barrier = asyncio.Barrier(self.N)

        t1 = asyncio.create_task(barrier.wait())
        await asyncio.sleep(0)
        self.assertEqual(barrier.n_waiting, 1)
        self.assertIn("filling", repr(barrier))

        t2 = asyncio.create_task(barrier.wait())
        await asyncio.sleep(0)
        self.assertEqual(barrier.n_waiting, 2)
        self.assertIn("filling", repr(barrier))

        t3 = asyncio.create_task(barrier.wait())
        await asyncio.sleep(0)

        await asyncio.wait([t1, t2, t3])

        self.assertEqual(barrier.n_waiting, 0)
        self.assertFalse(barrier.broken)

    async def test_filling_tasks_wait_twice(self):
        barrier = asyncio.Barrier(self.N)
        results = []

        async def coro():
            async with barrier:
                results.append(True)

                async with barrier:
                    results.append(False)

        await self.gather_tasks(self.N, coro)

        self.assertEqual(len(results), self.N*2)
        self.assertEqual(results.count(True), self.N)
        self.assertEqual(results.count(False), self.N)

        self.assertEqual(barrier.n_waiting, 0)
        self.assertFalse(barrier.broken)

    async def test_filling_tasks_check_return_value(self):
        barrier = asyncio.Barrier(self.N)
        results1 = []
        results2 = []

        async def coro():
            async with barrier:
                results1.append(True)

                async with barrier as i:
                    results2.append(True)
                    return i

        res, _ = await self.gather_tasks(self.N, coro)

        self.assertEqual(len(results1), self.N)
        self.assertTrue(all(results1))
        self.assertEqual(len(results2), self.N)
        self.assertTrue(all(results2))
        self.assertListEqual(sorted(res), list(range(self.N)))

        self.assertEqual(barrier.n_waiting, 0)
        self.assertFalse(barrier.broken)

    async def test_draining_state(self):
        barrier = asyncio.Barrier(self.N)
        results = []

        async def coro():
            async with barrier:
                
                results.append("draining" in repr(barrier))

        await self.gather_tasks(self.N, coro)

        self.assertEqual(len(results), self.N)
        self.assertEqual(results[-1], False)
        self.assertTrue(all(results[:self.N-1]))

        self.assertEqual(barrier.n_waiting, 0)
        self.assertFalse(barrier.broken)

    async def test_blocking_tasks_while_draining(self):
        rewait = 2
        barrier = asyncio.Barrier(self.N)
        barrier_nowaiting = asyncio.Barrier(self.N - rewait)
        results = []
        rewait_n = rewait
        counter = 0

        async def coro():
            nonlocal rewait_n

            
            await barrier.wait()

            
            if rewait_n > 0:
                rewait_n -= 1
                
                await barrier.wait()
            else:
                
                await barrier_nowaiting.wait()
                
                await barrier.wait()

        
        
        await self.gather_tasks(self.N, coro)

    async def test_filling_tasks_cancel_one(self):
        self.N = 3
        barrier = asyncio.Barrier(self.N)
        results = []

        async def coro():
            await barrier.wait()
            results.append(True)

        t1 = asyncio.create_task(coro())
        await asyncio.sleep(0)
        self.assertEqual(barrier.n_waiting, 1)

        t2 = asyncio.create_task(coro())
        await asyncio.sleep(0)
        self.assertEqual(barrier.n_waiting, 2)

        t1.cancel()
        await asyncio.sleep(0)
        self.assertEqual(barrier.n_waiting, 1)
        with self.assertRaises(asyncio.CancelledError):
            await t1
        self.assertTrue(t1.cancelled())

        t3 = asyncio.create_task(coro())
        await asyncio.sleep(0)
        self.assertEqual(barrier.n_waiting, 2)

        t4 = asyncio.create_task(coro())
        await asyncio.gather(t2, t3, t4)

        self.assertEqual(len(results), self.N)
        self.assertTrue(all(results))

        self.assertEqual(barrier.n_waiting, 0)
        self.assertFalse(barrier.broken)

    async def test_reset_barrier(self):
        barrier = asyncio.Barrier(1)

        asyncio.create_task(barrier.reset())
        await asyncio.sleep(0)

        self.assertEqual(barrier.n_waiting, 0)
        self.assertFalse(barrier.broken)

    async def test_reset_barrier_while_tasks_waiting(self):
        barrier = asyncio.Barrier(self.N)
        results = []

        async def coro():
            try:
                await barrier.wait()
            except asyncio.BrokenBarrierError:
                results.append(True)

        async def coro_reset():
            await barrier.reset()

        
        tasks  = self.make_tasks(self.N-1, coro)
        await asyncio.sleep(0)

        
        asyncio.create_task(coro_reset())
        await asyncio.gather(*tasks)

        self.assertEqual(len(results), self.N-1)
        self.assertTrue(all(results))
        self.assertEqual(barrier.n_waiting, 0)
        self.assertNotIn("resetting", repr(barrier))
        self.assertFalse(barrier.broken)

    async def test_reset_barrier_when_tasks_half_draining(self):
        barrier = asyncio.Barrier(self.N)
        results1 = []
        rest_of_tasks = self.N//2

        async def coro():
            try:
                await barrier.wait()
            except asyncio.BrokenBarrierError:
                
                results1.append(True)
            else:
                
                if rest_of_tasks == barrier._count:
                    
                    await barrier.reset()

        await self.gather_tasks(self.N, coro)

        self.assertEqual(results1, [True]*rest_of_tasks)
        self.assertEqual(barrier.n_waiting, 0)
        self.assertNotIn("resetting", repr(barrier))
        self.assertFalse(barrier.broken)

    async def test_reset_barrier_when_tasks_half_draining_half_blocking(self):
        barrier = asyncio.Barrier(self.N)
        results1 = []
        results2 = []
        blocking_tasks = self.N//2
        count = 0

        async def coro():
            nonlocal count
            try:
                await barrier.wait()
            except asyncio.BrokenBarrierError:
                
                results1.append(True)

                
                await barrier.wait()
            else:
                count += 1
                if count > blocking_tasks:
                    
                    await barrier.reset()

                    
                    await barrier.wait()
                else:
                    try:
                        await barrier.wait()
                    except asyncio.BrokenBarrierError:
                        
                        results2.append(True)

        await self.gather_tasks(self.N, coro)

        self.assertEqual(results1, [True]*blocking_tasks)
        self.assertEqual(results2, [])
        self.assertEqual(barrier.n_waiting, 0)
        self.assertNotIn("resetting", repr(barrier))
        self.assertFalse(barrier.broken)

    async def test_reset_barrier_while_tasks_waiting_and_waiting_again(self):
        barrier = asyncio.Barrier(self.N)
        results1 = []
        results2 = []

        async def coro1():
            try:
                await barrier.wait()
            except asyncio.BrokenBarrierError:
                results1.append(True)
            finally:
                await barrier.wait()
                results2.append(True)

        async def coro2():
            async with barrier:
                results2.append(True)

        tasks = self.make_tasks(self.N-1, coro1)

        
        asyncio.create_task(barrier.reset())
        await asyncio.sleep(0)

        
        asyncio.create_task(coro2())

        await asyncio.gather(*tasks)

        self.assertFalse(barrier.broken)
        self.assertEqual(len(results1), self.N-1)
        self.assertTrue(all(results1))
        self.assertEqual(len(results2), self.N)
        self.assertTrue(all(results2))

        self.assertEqual(barrier.n_waiting, 0)


    async def test_reset_barrier_while_tasks_draining(self):
        barrier = asyncio.Barrier(self.N)
        results1 = []
        results2 = []
        results3 = []
        count = 0

        async def coro():
            nonlocal count

            i = await barrier.wait()
            count += 1
            if count == self.N:
                
                await barrier.reset()

                
                await barrier.wait()
            else:
                try:
                    
                    await barrier.wait()

                    
                    results1.append(True)
                except Exception as e:
                    
                    results2.append(True)

            
            
            k = await barrier.wait()
            results3.append(True)

        await self.gather_tasks(self.N, coro)

        self.assertFalse(barrier.broken)
        self.assertTrue(all(results1))
        self.assertEqual(len(results1), self.N-1)
        self.assertEqual(len(results2), 0)
        self.assertEqual(len(results3), self.N)
        self.assertTrue(all(results3))

        self.assertEqual(barrier.n_waiting, 0)

    async def test_abort_barrier(self):
        barrier = asyncio.Barrier(1)

        asyncio.create_task(barrier.abort())
        await asyncio.sleep(0)

        self.assertEqual(barrier.n_waiting, 0)
        self.assertTrue(barrier.broken)

    async def test_abort_barrier_when_tasks_half_draining_half_blocking(self):
        barrier = asyncio.Barrier(self.N)
        results1 = []
        results2 = []
        blocking_tasks = self.N//2
        count = 0

        async def coro():
            nonlocal count
            try:
                await barrier.wait()
            except asyncio.BrokenBarrierError:
                
                results1.append(True)
            else:
                count += 1
                if count > blocking_tasks:
                    
                    await barrier.abort()
                else:
                    try:
                        await barrier.wait()
                    except asyncio.BrokenBarrierError:
                        
                        results2.append(True)

        await self.gather_tasks(self.N, coro)

        self.assertTrue(barrier.broken)
        self.assertEqual(results1, [True]*blocking_tasks)
        self.assertEqual(results2, [True]*(self.N-blocking_tasks-1))
        self.assertEqual(barrier.n_waiting, 0)
        self.assertNotIn("resetting", repr(barrier))

    async def test_abort_barrier_when_exception(self):
        
        barrier = asyncio.Barrier(self.N)
        results1 = []
        results2 = []

        async def coro():
            try:
                async with barrier as i :
                    if i == self.N//2:
                        raise RuntimeError
                async with barrier:
                    results1.append(True)
            except asyncio.BrokenBarrierError:
                results2.append(True)
            except RuntimeError:
                await barrier.abort()

        await self.gather_tasks(self.N, coro)

        self.assertTrue(barrier.broken)
        self.assertEqual(len(results1), 0)
        self.assertEqual(len(results2), self.N-1)
        self.assertTrue(all(results2))
        self.assertEqual(barrier.n_waiting, 0)

    async def test_abort_barrier_when_exception_then_resetting(self):
        
        barrier1 = asyncio.Barrier(self.N)
        barrier2 = asyncio.Barrier(self.N)
        results1 = []
        results2 = []
        results3 = []

        async def coro():
            try:
                i = await barrier1.wait()
                if i == self.N//2:
                    raise RuntimeError
                await barrier1.wait()
                results1.append(True)
            except asyncio.BrokenBarrierError:
                results2.append(True)
            except RuntimeError:
                await barrier1.abort()

            
            
            
            i = await barrier2.wait()
            if  i == self.N//2:
                await barrier1.reset()
            await barrier2.wait()
            await barrier1.wait()
            results3.append(True)

        await self.gather_tasks(self.N, coro)

        self.assertFalse(barrier1.broken)
        self.assertEqual(len(results1), 0)
        self.assertEqual(len(results2), self.N-1)
        self.assertTrue(all(results2))
        self.assertEqual(len(results3), self.N)
        self.assertTrue(all(results3))

        self.assertEqual(barrier1.n_waiting, 0)


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