How to use _wait_for_stop method in yandex-tank

Best Python code snippet using yandex-tank

tango.py

Source:tango.py

...45            state = 'alarm'46        else:47            raise ValueError("Unknown Tango state '{}'".format(tango_state))48        return state49    async def _wait_for_stop(self):50        await asyncio.sleep(TANGO_SLEEP_TIME)51        while True:52            if await self.get_state() == "moving":53                await asyncio.sleep(TANGO_SLEEP_TIME)54            else:55                break56    async def _stop(self):57        await self._device.StopMove()58        await self._wait_for_stop()59    async def _home(self):60        try:61            await self._device.MoveHome()62            await self._wait_for_stop()63        except asyncio.CancelledError:64            await self.stop()65            raise66    async def _get_position(self):67        return (await self._device['position']).value * self['position'].unit68    async def _set_position(self, position):69        try:70            await self._device.write_attribute('position',71                                               position.to(self['position'].unit).magnitude)72            await self._wait_for_stop()73            if await self._in_hard_limit():74                raise HardLimitError()75        except asyncio.CancelledError:76            await self.stop()77            raise78    async def __aenter__(self):79        self._device.lock()80        await self._lock.acquire()81        return self82    async def __aexit__(self, exc_type, exc, tb):83        self._device.unlock()84        self._lock.release()85class LinearMotor(_TangoMixin, base.LinearMotor):86    """A linear motor based on DESY Tango motor interface."""...

2_wait_stop.py

Source:2_wait_stop.py

...21  while not lp.wait_for_stop(1):22    logging.info('Sleeping again...')23  logging.info('Clean termination of _sleep node')24  time.sleep(2)25def _wait_for_stop():26  lp.wait_for_stop()27  logging.info('Clean termination of _wait_for_stop node')28  time.sleep(2)29def _stop_event():30  lp.stop_event().wait()31  logging.info('Clean termination of _stop_event node')32  time.sleep(2)33def _register_stop_handler():34  """Showcases the use of lp.register_stop_handler."""35  stop = threading.Event()36  def _stop_handler():37    logging.info('_stop_handler called')38    time.sleep(5)39    logging.info('setting stop event!!')...

launch.py

Source:launch.py

...21  while not lp.wait_for_stop(1):22    logging.info('Sleeping again...')23  logging.info('Clean termination of _sleep node')24  time.sleep(2)25def _wait_for_stop():26  lp.wait_for_stop()27  logging.info('Clean termination of _wait_for_stop node')28  time.sleep(2)29def _stop_event():30  lp.stop_event().wait()31  logging.info('Clean termination of _stop_event node')32  time.sleep(2)33def _register_stop_handler():34  """Showcases the use of lp.register_stop_handler."""35  stop = threading.Event()36  def _stop_handler():37    logging.info('_stop_handler called')38    stop.set()39  lp.register_stop_handler(_stop_handler)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.