Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@tb.silence_long_exec_warning()
def test_signals_sigint_pycode_stop(self):
async def runner():
PROG = R"""\
import asyncio
import uvloop
import time
from uvloop import _testbase as tb
async def worker():
print('READY', flush=True)
time.sleep(200)
@tb.silence_long_exec_warning()
def run():
loop = """ + self.NEW_LOOP + """
@tb.silence_long_exec_warning()
def test_signals_sigint_uvcode(self):
async def runner():
PROG = R"""\
import asyncio
import uvloop
srv = None
async def worker():
global srv
cb = lambda *args: None
srv = await asyncio.start_server(cb, '127.0.0.1', 0)
print('READY', flush=True)
loop = """ + self.NEW_LOOP + """
asyncio.set_event_loop(loop)
@tb.silence_long_exec_warning()
def test_signals_sigint_pycode_continue(self):
async def runner():
PROG = R"""\
import asyncio
import uvloop
import time
from uvloop import _testbase as tb
async def worker():
print('READY', flush=True)
try:
time.sleep(200)
except KeyboardInterrupt:
print("oups")
await asyncio.sleep(0.5)
@tb.silence_long_exec_warning()
def test_signals_sigint_and_custom_handler(self):
async def runner():
PROG = R"""\
import asyncio
import signal
import uvloop
srv = None
async def worker():
global srv
cb = lambda *args: None
srv = await asyncio.start_server(cb, '127.0.0.1', 0)
print('READY', flush=True)
def handler_sig(say):
@tb.silence_long_exec_warning()
def test_signals_and_custom_handler_1(self):
async def runner():
PROG = R"""\
import asyncio
import signal
import uvloop
srv = None
async def worker():
global srv
cb = lambda *args: None
srv = await asyncio.start_server(cb, '127.0.0.1', 0)
print('READY', flush=True)
def handler1():