Botswana football predictions today

How accurate are your predictions?

Our predictions are based on comprehensive data analysis and expert insights. While no prediction can guarantee 100% accuracy, we strive to provide the most reliable forecasts available. <|repo_name|>chrisjw/python-asyncio<|file_sep|>/test/test_loop.py # Copyright (c) 2012-2015 Denis Bilenko # # Permission is hereby granted, free of charge, to any person obtaining a copy # of this software and associated documentation files (the "Software"), to deal # in the Software without restriction, including without limitation the rights # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell # copies of the Software, and to permit persons to whom the Software is # furnished to do so, subject to the following conditions: # # The above copyright notice and this permission notice shall be included in # all copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, # EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF # MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. # IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, # DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, # TORT OR OTHERWISE, ARISING FROM, # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN # THE SOFTWARE. import asyncio import gc import os import signal import sys from unittest import TestCase from .base import TestBase if sys.version_info >= (3,) or 'PyPy' in sys.version: # Python 2's signal module doesn't have SIGWINCH. _SIGWINCH = signal.SIGWINCH else: _SIGWINCH = None class TestLoop(TestBase): def test_create(self): l = asyncio.get_event_loop() self.assertIsInstance(l._ready._ready_queue._queue[0][0], asyncio.Event) # Check that there's only one loop. l1 = asyncio.get_event_loop() l2 = asyncio.get_event_loop() self.assertIs(l1._default_executor(), l2._default_executor()) self.assertIs(l1._default_watcher(), l2._default_watcher()) self.assertIs(l1._default_socketpair(), l2._default_socketpair()) self.assertIs(l1._default_proactor(), l2._default_proactor()) self.assertIs(l1._default_selector(), l2._default_selector()) def test_close(self): # Verify that close() clears internal references. l = asyncio.get_event_loop() f = asyncio.Future() def cb(): f.set_result(None) t1 = asyncio.TimerHandle(0.001) t1.add_done_callback(cb) t1.close() self.assertFalse(t1.is_closed()) t2 = asyncio.TimerHandle(0.001) t2.close() self.assertTrue(t2.is_closed()) ioloop = asyncio.IOHandle(os.pipe()[0]) ioloop.read().add_done_callback(cb) ioloop.close() self.assertTrue(ioloop.is_closed()) # Shouldn't fail even if we close twice. ioloop.close() iowatcher = l.add_reader(ioloop.fileno(), lambda: None) iowatcher.close() # Shouldn't fail even if we close twice. iowatcher.close() f.result() def test_run_until_complete(self): loop = asyncio.get_event_loop() async def coro(): await asyncio.sleep(0) return 42 self.assertEqual(loop.run_until_complete(coro()), 42) def test_run_forever(self): loop = asyncio.get_event_loop() result = [] async def coro(): await asyncio.sleep(0) result.append(42) raise RuntimeError('foo') return 43 try: loop.run_until_complete(coro()) self.fail('should not reach here') except RuntimeError: pass try: loop.run_forever() self.fail('should not reach here') except RuntimeError: pass self.assertEqual(result[0], 42) def test_run_forever_exception(self): loop = asyncio.get_event_loop() async def coro(): await asyncio.sleep(0) raise RuntimeError('foo') try: loop.run_until_complete(coro()) self.fail('should not reach here') except RuntimeError: pass try: loop.run_forever() self.fail('should not reach here') except RuntimeError: pass def test_default_executor(self): loop = asyncio.get_event_loop() class C(object): def __init__(self): self.closed = False def __enter__(self): return self def __exit__(self,a,b,c): self.closed = True class MyExecutor(asyncio.AbstractExecutor): def __init__(self): super(MyExecutor,self).__init__() self.closed = False def submit(self,f,*args,**kwargs): f(*args,**kwargs) return None def close(self): super(MyExecutor,self).close() self.closed = True old_executor_factory = loop.default_executor_factory old_context_manager_factory = loop.default_context_manager_factory try: loop.default_executor_factory(lambda: C()) executor = loop.default_executor() executor.submit(lambda: None) executor.close() self.assertTrue(executor.closed) loop.default_context_manager_factory(lambda: C()) cmf = loop.default_context_manager_factory() cmf.__enter__() cmf.__exit__(None,None,None) self.assertTrue(cmf.closed) loop.default_executor_factory(lambda: MyExecutor()) executor2 = loop.default_executor() executor2.submit(lambda: None) executor2.close() self.assertTrue(executor2.closed) finally: # Reset back original factory values. loop.default_context_manager_factory(old_context_manager_factory) loop.default_executor_factory(old_executor_factory) def test_default_watcher(self): loop = asyncio.get_event_loop() class C(object): def __init__(self): self.closed = False def __enter__(self): return self def __exit__(self,a,b,c): self.closed = True class MyWatcher(asyncio.AbstractWatcher): def __init__(self,*args,**kwargs): super(MyWatcher,self).__init__(*args,**kwargs) self.closed = False def close(self): super(MyWatcher,self).close() self.closed=True old_watcher_factory = loop.default_watcher_factory try: loop.default_watcher_factory(lambda: C()) watcher=loop.default_watcher() watcher.close() assert watcher.closed loop.default_watcher_factory(lambda: MyWatcher()) watcher=loop.default_watcher() watcher.close() assert watcher.closed finally: # Reset back original factory values. loop.default_watcher_factory(old_watcher_factory) def test_default_socketpair(self): class C(object): def __init__(self,a,b): super(C,self).__init__() assert isinstance(a,int) or isinstance(a,type(None)) assert isinstance(b,int) or isinstance(b,type(None)) if not isinstance(a,int) or not isinstance(b,int): return assert os.read(a,'') == '' assert os.read(b,'') == '' os.write(a,'ok') os.write(b,'ok') assert os.read(a,'ok') == 'ok' assert os.read(b,'ok') == 'ok' os.write(a,'') os.write(b,'') assert os.read(a,'') == '' assert os.read(b,'') == '' os.close(a) os.close(b) # Must support context manager interface. def __enter__(self): return self def __exit__(self,a,b,c): pass old_socketpair=loop.default_socketpair try: # Test that default socketpair factory works. # Try default socketpair implementation. if hasattr(os,'socketpair'): if _SIGWINCH is not None: socketpair=os.socketpair else: socketpair=lambda*args,**kwargs: (os.socketpair(*args,**kwargs)[0],os.socketpair(*args,**kwargs)[1]) with C(*socketpair()) as c: pass socketpair=os.socketpair if _SIGWINCH is not None: socketpair=lambda*args,**kwargs: (os.socketpair(*args,**kwargs)[0],os.socketpair(*args,**kwargs)[1]) with C(*socketpair()) as c: pass # Now test custom implementation. socket_pair=os.pipe socket_pair=lambda*args,**kwargs: (os.pipe()[0],os.pipe()[1]) with C(*socket_pair()) as c: pass socket_pair=os.pipe socket_pair=lambda*args,**kwargs: (os.pipe()[0],os.pipe()[1]) with C(*socket_pair()) as c: pass finally: # Reset back original factory values. loop.default_socketpair(old_socketpair) def test_default_proactor(self): class C(object): def __init__(self): super(C,self).__init__() if sys.platform.startswith('win32'): s=os.pipe() w=asyncio.ProactorEventLoop(s[0]) r=asyncio.ProactorEventLoop(s[1]) w.add_reader(r,s[0],lambda r,w,s,wakeup=None:self.loop(wakeup=wakeup)) r.add_writer(w,s[1],lambda r,w,s,wakeup=None:self.loop(wakeup=wakeup)) w.call_soon(lambda x: r.remove_writer(w)) r.call_soon(lambda x: w.remove_reader(r)) w.run_forever() w.close() r.run_forever() r.close() @staticmethod def loop(wakeup=None): if wakeup is not None: wakeup.send_bytes(b'x') wakeup.send_bytes(b'y