我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用asyncio.Handle()。
def __init__(self, client: Session, ignoreHTTPSErrors: Any, options: dict = None, **kwargs: Any) -> None: """Make new navigator watcher.""" if options is None: options = {} options.update(kwargs) self._client = client self._ignoreHTTPSErrors = ignoreHTTPSErrors self._timeout = options.get('timeout', 3000) self._idleTime = options.get('networkIdleTimeout', 1000) self._idleTimer: Optional[Union[asyncio.Future, asyncio.Handle]] = None self._idleInflight = options.get('networkIdleInflight', 2) self._waitUntil = options.get('waitUntil', 'load') if self._waitUntil not in ('load', 'networkidle'): raise ValueError( f'Unknown value for options.waitUntil: {self._waitUntil}')
def __init__(self, id: int, args: Dict[str, str], monitor_def: ActiveMonitorDef, state: str, state_ts: float, msg: str, alert_id: Union[int, None], checks_enabled: bool, alerts_enabled: bool, manager: ActiveMonitorManager) -> None: self.id = id self.args = args self.monitor_def = monitor_def self.state = state self.manager = manager self.last_check_state = None # type: Optional[str] self.consecutive_checks = 0 self.last_check = time.time() self.msg = msg self.alert_id = alert_id self.state_ts = state_ts if not self.state_ts: self.state_ts = time.time() self.monitoring = False self.deleted = False self.checks_enabled = checks_enabled self.alerts_enabled = alerts_enabled self._pending_reset = False self.scheduled_job = None # type: Optional[asyncio.Handle] self.scheduled_job_ts = 0.0 event.running('CREATE_ACTIVE_MONITOR', monitor=self) stats.inc('num_monitors', 'ACT_MON')
def test_callback_with_exception(self): def callback(): raise ValueError() self.loop = mock.Mock() self.loop.call_exception_handler = mock.Mock() h = asyncio.Handle(callback, (), self.loop) h._run() self.loop.call_exception_handler.assert_called_with({ 'message': test_utils.MockPattern('Exception in callback.*'), 'exception': mock.ANY, 'handle': h, 'source_traceback': h._source_traceback, })
def test_run_once_in_executor_plain(self): def cb(): pass h = asyncio.Handle(cb, (), self.loop) f = asyncio.Future(loop=self.loop) executor = mock.Mock() executor.submit.return_value = f self.loop.set_default_executor(executor) res = self.loop.run_in_executor(None, h) self.assertIs(f, res) executor = mock.Mock() executor.submit.return_value = f res = self.loop.run_in_executor(executor, h) self.assertIs(f, res) self.assertTrue(executor.submit.called) f.cancel() # Don't complain about abandoned Future.
def __init__(self, name, tag, *args, prefix=None, untagged_resp_name=None, loop=asyncio.get_event_loop(), timeout=None): self.name = name self.tag = tag self.args = args self.prefix = prefix + ' ' if prefix else None self.untagged_resp_name = untagged_resp_name or name self.response = None self._exception = None self._event = asyncio.Event(loop=loop) self._loop = loop self._timeout = timeout self._timer = asyncio.Handle(lambda: None, None, loop) # fake timer self._set_timer() self._literal_data = None self._expected_size = 0
def clearTimeout(fut: Optional[Union[asyncio.Future, asyncio.Handle]]) -> None: """Cancel timer task.""" if fut: fut.cancel()
def test_handle(self): def callback(*args): return args args = () h = asyncio.Handle(callback, args, self.loop) self.assertIs(h._callback, callback) self.assertIs(h._args, args) self.assertFalse(h._cancelled) h.cancel() self.assertTrue(h._cancelled)
def test_handle_from_handle(self): def callback(*args): return args h1 = asyncio.Handle(callback, (), loop=self.loop) self.assertRaises( AssertionError, asyncio.Handle, h1, (), self.loop)
def test_handle_weakref(self): wd = weakref.WeakValueDictionary() h = asyncio.Handle(lambda: None, (), self.loop) wd['h'] = h # Would fail without __weakref__ slot.
def test_handle_repr(self): self.loop.get_debug.return_value = False # simple function h = asyncio.Handle(noop, (1, 2), self.loop) filename, lineno = test_utils.get_function_source(noop) self.assertEqual(repr(h), '<Handle noop(1, 2) at %s:%s>' % (filename, lineno)) # cancelled handle h.cancel() self.assertEqual(repr(h), '<Handle cancelled>') # decorated function cb = asyncio.coroutine(noop) h = asyncio.Handle(cb, (), self.loop) self.assertEqual(repr(h), '<Handle noop() at %s:%s>' % (filename, lineno)) # partial function cb = functools.partial(noop, 1, 2) h = asyncio.Handle(cb, (3,), self.loop) regex = (r'^<Handle noop\(1, 2\)\(3\) at %s:%s>$' % (re.escape(filename), lineno)) self.assertRegex(repr(h), regex) # partial method if sys.version_info >= (3, 4): method = HandleTests.test_handle_repr cb = functools.partialmethod(method) filename, lineno = test_utils.get_function_source(method) h = asyncio.Handle(cb, (), self.loop) cb_regex = r'<function HandleTests.test_handle_repr .*>' cb_regex = (r'functools.partialmethod\(%s, , \)\(\)' % cb_regex) regex = (r'^<Handle %s at %s:%s>$' % (cb_regex, re.escape(filename), lineno)) self.assertRegex(repr(h), regex)
def test_timer_comparison(self): def callback(*args): return args when = time.monotonic() h1 = asyncio.TimerHandle(when, callback, (), self.loop) h2 = asyncio.TimerHandle(when, callback, (), self.loop) # TODO: Use assertLess etc. self.assertFalse(h1 < h2) self.assertFalse(h2 < h1) self.assertTrue(h1 <= h2) self.assertTrue(h2 <= h1) self.assertFalse(h1 > h2) self.assertFalse(h2 > h1) self.assertTrue(h1 >= h2) self.assertTrue(h2 >= h1) self.assertTrue(h1 == h2) self.assertFalse(h1 != h2) h2.cancel() self.assertFalse(h1 == h2) h1 = asyncio.TimerHandle(when, callback, (), self.loop) h2 = asyncio.TimerHandle(when + 10.0, callback, (), self.loop) self.assertTrue(h1 < h2) self.assertFalse(h2 < h1) self.assertTrue(h1 <= h2) self.assertFalse(h2 <= h1) self.assertFalse(h1 > h2) self.assertTrue(h2 > h1) self.assertFalse(h1 >= h2) self.assertTrue(h2 >= h1) self.assertFalse(h1 == h2) self.assertTrue(h1 != h2) h3 = asyncio.Handle(callback, (), self.loop) self.assertIs(NotImplemented, h1.__eq__(h3)) self.assertIs(NotImplemented, h1.__ne__(h3))
def test__add_callback_handle(self): h = asyncio.Handle(lambda: False, (), self.loop) self.loop._add_callback(h) self.assertFalse(self.loop._scheduled) self.assertIn(h, self.loop._ready)
def test__add_callback_cancelled_handle(self): h = asyncio.Handle(lambda: False, (), self.loop) h.cancel() self.loop._add_callback(h) self.assertFalse(self.loop._scheduled) self.assertFalse(self.loop._ready)
def test_call_soon(self): def cb(): pass h = self.loop.call_soon(cb) self.assertEqual(h._callback, cb) self.assertIsInstance(h, asyncio.Handle) self.assertIn(h, self.loop._ready)
def test_run_once_in_executor_handle(self): def cb(): pass self.assertRaises( AssertionError, self.loop.run_in_executor, None, asyncio.Handle(cb, (), self.loop), ('',)) self.assertRaises( AssertionError, self.loop.run_in_executor, None, asyncio.TimerHandle(10, cb, (), self.loop))
def test_default_exc_handler_callback(self): self.loop._process_events = mock.Mock() def zero_error(fut): fut.set_result(True) 1/0 # Test call_soon (events.Handle) with mock.patch('asyncio.base_events.logger') as log: fut = asyncio.Future(loop=self.loop) self.loop.call_soon(zero_error, fut) fut.add_done_callback(lambda fut: self.loop.stop()) self.loop.run_forever() log.error.assert_called_with( test_utils.MockPattern('Exception in callback.*zero'), exc_info=(ZeroDivisionError, MOCK_ANY, MOCK_ANY)) # Test call_later (events.TimerHandle) with mock.patch('asyncio.base_events.logger') as log: fut = asyncio.Future(loop=self.loop) self.loop.call_later(0.01, zero_error, fut) fut.add_done_callback(lambda fut: self.loop.stop()) self.loop.run_forever() log.error.assert_called_with( test_utils.MockPattern('Exception in callback.*zero'), exc_info=(ZeroDivisionError, MOCK_ANY, MOCK_ANY))
def test_log_slow_callbacks(self, m_logger): def stop_loop_cb(loop): loop.stop() @asyncio.coroutine def stop_loop_coro(loop): yield from () loop.stop() asyncio.set_event_loop(self.loop) self.loop.set_debug(True) self.loop.slow_callback_duration = 0.0 # slow callback self.loop.call_soon(stop_loop_cb, self.loop) self.loop.run_forever() fmt, *args = m_logger.warning.call_args[0] self.assertRegex(fmt % tuple(args), "^Executing <Handle.*stop_loop_cb.*> " "took .* seconds$") # slow task asyncio.async(stop_loop_coro(self.loop), loop=self.loop) self.loop.run_forever() fmt, *args = m_logger.warning.call_args[0] self.assertRegex(fmt % tuple(args), "^Executing <Task.*stop_loop_coro.*> " "took .* seconds$")
def test_handle_signal_cancelled_handler(self): h = asyncio.Handle(mock.Mock(), (), loop=mock.Mock()) h.cancel() self.loop._signal_handlers[signal.NSIG + 1] = h self.loop.remove_signal_handler = mock.Mock() self.loop._handle_signal(signal.NSIG + 1) self.loop.remove_signal_handler.assert_called_with(signal.NSIG + 1)
def test_add_signal_handler(self, m_signal): m_signal.NSIG = signal.NSIG cb = lambda: True self.loop.add_signal_handler(signal.SIGHUP, cb) h = self.loop._signal_handlers.get(signal.SIGHUP) self.assertIsInstance(h, asyncio.Handle) self.assertEqual(h._callback, cb)
def remove_timeout(handle: asyncio.Handle): return handle.cancel()
def test_log_slow_callbacks(self, m_logger): def stop_loop_cb(loop): loop.stop() @asyncio.coroutine def stop_loop_coro(loop): yield from () loop.stop() asyncio.set_event_loop(self.loop) self.loop.set_debug(True) self.loop.slow_callback_duration = 0.0 # slow callback self.loop.call_soon(stop_loop_cb, self.loop) self.loop.run_forever() fmt, *args = m_logger.warning.call_args[0] self.assertRegex(fmt % tuple(args), "^Executing <Handle.*stop_loop_cb.*> " "took .* seconds$") # slow task asyncio.ensure_future(stop_loop_coro(self.loop), loop=self.loop) self.loop.run_forever() fmt, *args = m_logger.warning.call_args[0] self.assertRegex(fmt % tuple(args), "^Executing <Task.*stop_loop_coro.*> " "took .* seconds$")
def test_set_ttl_handle(self, memory): await memory._set(pytest.KEY, "value", ttl=100) assert pytest.KEY in memory._handlers assert isinstance(memory._handlers[pytest.KEY], asyncio.Handle)
def test_expire_no_handle_ttl(self, memory): SimpleMemoryBackend._cache.__contains__.return_value = True await memory._expire(pytest.KEY, 1) assert isinstance(memory._handlers.get(pytest.KEY), asyncio.Handle)
def test_expire_handle_ttl(self, memory): fake = MagicMock() SimpleMemoryBackend._handlers[pytest.KEY] = fake SimpleMemoryBackend._cache.__contains__.return_value = True await memory._expire(pytest.KEY, 1) assert fake.cancel.call_count == 1 assert isinstance(memory._handlers.get(pytest.KEY), asyncio.Handle)
