Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _request_start method in Molotov

Best Python code snippet using molotov_python

service.py

Source:service.py

...23    ServiceState,24)25class SystemdService(ServicePlugin):26    BIN_PATH = '/usr/bin/systemctl'27    def _request_start(self):28        super()._request_start()29        return self._vm.ssh([self.BIN_PATH, 'start', self._name])30    def _request_stop(self):31        super()._request_stop()32        return self._vm.ssh([self.BIN_PATH, 'stop', self._name])33    def state(self):34        ret = self._vm.ssh([self.BIN_PATH, 'status --lines=0', self._name])35        if not ret:36            return ServiceState.ACTIVE37        out = ret.out.decode('utf-8')38        lines = [l.strip() for l in out.split('\n')]39        loaded = [l for l in lines if l.startswith('Loaded:')].pop()40        if loaded.split()[1] == 'loaded':41            return ServiceState.INACTIVE42        return ServiceState.MISSING43class SysVInitService(ServicePlugin):44    BIN_PATH = '/sbin/service'45    def _request_start(self):46        super()._request_start()47        return self._vm.ssh([self.BIN_PATH, self._name, 'start'])48    def _request_stop(self):49        super()._request_stop()50        return self._vm.ssh([self.BIN_PATH, self._name, 'stop'])51    def state(self):52        ret = self._vm.ssh([self.BIN_PATH, self._name, 'status'])53        if ret.code == 0:54            return ServiceState.ACTIVE55        out = ret.out.decode('utf-8')56        if out.strip().endswith('is stopped'):57            return ServiceState.INACTIVE58        return ServiceState.MISSING59class SystemdContainerService(ServicePlugin):60    BIN_PATH = '/usr/bin/docker'61    HOST_BIN_PATH = '/usr/bin/systemctl'62    def _request_start(self):63        super()._request_start()64        ret = self._vm.ssh(65            [self.BIN_PATH, 'exec vdsmc systemctl start', self._name]66        )67        if ret.code == 0:68            return ret69        return self._vm.ssh([self.HOST_BIN_PATH, 'start', self._name])70    def _request_stop(self):71        super()._request_stop()72        ret = self._vm.ssh(73            [self.BIN_PATH, 'exec vdsmc systemctl stop', self._name]74        )75        if ret.code == 0:76            return ret77        return self._vm.ssh([self.HOST_BIN_PATH, 'stop', self._name])...

provider1pool.py

Source:provider1pool.py

1# -*- coding: UTF-8 -*-2import time3import threading4from datanal.config.api_settings import api_config5class Provider1Pool(object):6    _auth_token = None7    _request_start = None8    _request_permission = True9    def __init__(self):10        self._lock = threading.Lock()11    @property12    def auth_token(self) -> str:13        return self._auth_token14    @auth_token.setter15    def auth_token(self, value: str) -> None:16        self._auth_token = value17    @property18    def request_permission(self) -> str:19        self._lock.acquire()20        try:21            self._request_start = time.perf_counter()22            if self._request_permission is not True:23                self.request_permission = True24            self.request_permission = False25            return True26        finally:27            self._request_start = None28            self._lock.release()29    @request_permission.setter30    def request_permission(self, value: bool) -> None:31        if value is True:32            while ((time.perf_counter() - self._request_start)33                    < (1 / api_config['provider1']['requests_per_second'])):34                time.sleep(0.1)...

provider2pool.py

Source:provider2pool.py

1# -*- coding: UTF-8 -*-2import time3import threading4from datanal.config.api_settings import api_config5class Provider2Pool(object):6    _request_start = None7    _request_permission = True8    def __init__(self):9        self._lock = threading.Lock()10    @property11    def request_permission(self) -> str:12        self._lock.acquire()13        try:14            self._request_start = time.perf_counter()15            if self._request_permission is not True:16                self.request_permission = True17            self.request_permission = False18            return True19        finally:20            self._request_start = None21            self._lock.release()22    @request_permission.setter23    def request_permission(self, value: bool) -> None:24        if value is True:25            while ((time.perf_counter() - self._request_start)26                    < (1 / api_config['provider2']['requests_per_second'])):27                time.sleep(0.1)...

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.