How to use history_remove method in stestr

Best Python code snippet using stestr_python

itemqueue.py

Source:itemqueue.py

...78            eva.core.log_traceback()79            return False80        finally:81            self.actions_lock.release()82    def history_remove(self, action):83        if not self.actions_lock.acquire(timeout=eva.core.config.timeout):84            logging.critical('ActiveItemQueue::history_remove locking broken')85            eva.core.critical()86            return False87        try:88            if not self.enterprise_layout:89                self.actions_by_item_id[action.item.item_id].remove(action)90            self.actions_by_item_full_id[action.item.full_id].remove(action)91            self.actions.remove(action)92            del self.actions_by_id[action.uuid]93            return True94        except:95            eva.core.log_traceback()96            return False97        finally:98            self.actions_lock.release()99    def start(self):100        if self.keep_history is None:101            self.keep_history = eva.core.config.keep_action_history102        self.action_cleaner_interval = eva.core.config.action_cleaner_interval103        self.action_cleaner = BackgroundIntervalWorker(104            fn=action_cleaner,105            name='primary_action_cleaner',106            delay=self.action_cleaner_interval,107            o=self,108            on_error=eva.core.log_traceback,109            loop='cleaners')110        self.action_cleaner.start()111        self.action_processor = BackgroundQueueWorker(112            fn=action_processor,113            name='primary_action_processor',114            on_error=eva.core.log_traceback,115            queue=asyncio.queues.PriorityQueue,116            o=self)117        self.action_processor.start()118    def stop(self):119        self.action_cleaner.stop()120        self.action_processor.stop()121    def process_action(self, action):122        return action.item.q_put_task(action)123async def action_processor(action, **kwargs):124    if not action.item:125        return126    o = kwargs.get('o')127    logging.debug('new action to toss, uuid: %s, priority: %u' % \128            (action.uuid, action.priority))129    try:130        if o.process_action(action):131            logging.debug(132                    'action %s requeued into local queue of %s' % \133                    (action.uuid, action.item.full_id))134        else:135            logging.debug(136             'action %s failed to requeue into local queue of %s' %\137             (action.uuid, action.item.full_id))138    except:139        eva.core.log_traceback()140async def action_cleaner(**kwargs):141    o = kwargs.get('o')142    if not o.actions_lock.acquire(timeout=eva.core.config.timeout):143        logging.critical('ActiveItemQueue::_t_action_cleanup locking broken')144        eva.core.critical()145        return146    logging.debug('cleaning old actions')147    try:148        _actions = o.actions.copy()149    except:150        _actions = []151        eva.core.log_traceback()152    finally:153        o.actions_lock.release()154    for a in _actions:155        try:156            tk = list(a.time.keys()).copy()157        except:158            eva.core.log_traceback()159        maxtime = 0160        for t in tk:161            try:162                maxtime = max(maxtime, a.time[t])163            except:164                pass165        if maxtime and maxtime < time.time() - o.keep_history:166            if a.is_finished():167                logging.debug(168                        '%s action %s too old, removing' % \169                        (o.q_id, a.uuid))170                o.history_remove(a)171            else:172                logging.warning(173                    '%s action %s too old, status is %s ' % \174                    (o.q_id, a.uuid,...

test_frontend_messaging.py

Source:test_frontend_messaging.py

...29        headers = {"Authorization": "apikey apikey"}30        resp = requests.post(self.url("api"), headers=headers, json=data)31        resp.raise_for_status()32        return resp.json()33    def history_remove(self, channel: str) -> None:34        data = {"method": "history_remove", "params": {"channel": channel}}35        self._api_send(data)36    def history(self, channel: str) -> Any:37        data = {"method": "history", "params": {"channel": channel, "limit": 10}}38        return self._api_send(data)39@pytest.fixture40def centrifugo_server() -> CentrifugoServer:41    server = CentrifugoServer()42    start_time = datetime.now()43    while not server.ping():44        elapsed = datetime.now() - start_time45        assert (46            elapsed.total_seconds() < 3047        ), "Timeout waiting for Centrifugo server to be ready"48        time.sleep(1.0)49    server.history_remove("heartbeat")50    server.history_remove("proxied:opc_data")51    server.history_remove("proxied:opc_status")52    return server53class CentrifugoClient:54    def __init__(self, url: URL) -> None:55        self._command_id = 056        self.client = websocket.WebSocket()57        self.client.connect(str(url))58        self._send_command(Method.CONNECT, {})59    def _send_command(self, method: Method, params: dict[str, Any]) -> None:60        self._command_id += 161        command_data = {62            "id": self._command_id,63            "method": method.value,64            "params": params,65        }...

websocket_test1.py

Source:websocket_test1.py

...16messages = client.history("public:chat")17clients = client.presence("public:chat")18channels = client.channels()19stats = client.info()...

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.