KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _empty_pool method in toolium

Best Python code snippet using toolium_python

change_location.py

Source:change_location.py

...89            "Please select the node you would like to deploy your 3Bot on.", list(node_id_dict.keys()), required=True90        )91        self.selected_node = node_id_dict[node_id]92        self.available_farms = [farm for farm in self.available_farms if farm.id == self.selected_node.farm_id]93    def _empty_pool(self, pool):94        return pool.active_cu == 0 and pool.active_su == 095    def _max_pool(self, pool1, pool2):96        if pool1 is None:97            return pool298        if pool2 is None:99            return pool1100        if min(pool1.cus, pool1.sus) > min(pool2.cus, pool2.sus):101            return pool1102        else:103            return pool2104    def _contains_node(self, pool, node):105        if node is None:106            return True107        return node.node_id in pool.node_ids108    def _find_free_pool_in_farm(self, zos, farms):109        owner = self.threebot_name110        threebot = get_threebot_config_instance(owner, self.threebot_info["solution_uuid"])111        zos = get_threebot_zos(threebot)112        identity = generate_user_identity(threebot, self.password, zos)113        zos = j.sals.zos.get(identity.instance_name)114        pools = zos.pools.list()115        max_pool = None116        farm_name = None117        for pool in pools:118            try:119                pool_farm = deployer.get_pool_farm_name(pool=pool)120            except requests.exceptions.HTTPError:121                continue122            if (123                self._empty_pool(pool)124                and self._contains_node(pool, self.selected_node)125                and [1 for farm in farms if farm.name == pool_farm]126            ):127                max_pool = self._max_pool(max_pool, pool)128                farm_name = pool_farm129        return farm_name, max_pool130    @chatflow_step("Reserving a pool")131    def create_pool(self):132        owner = self.threebot_name133        threebot = get_threebot_config_instance(owner, self.threebot_info["solution_uuid"])134        zos = get_threebot_zos(threebot)135        identity = generate_user_identity(threebot, self.password, zos)136        zos = j.sals.zos.get(identity.instance_name)137        farm_name, existent_pool = self._find_free_pool_in_farm(zos, self.available_farms)...

server_backend.py

Source:server_backend.py

...69            return self.job_queue.qsize() > 070    def _full_pool(self):71        """Check if the thread pool has space."""72        return len(self.thread_pool) < 1073    def _empty_pool(self):74        """Check if the thread pool has space."""75        return len(self.thread_pool) == 076    def _add_job(self):77        """Add a job to the thread pool."""78        with self.loc:79            job_target_details = self.job_queue.get()80        target = job_target_details.get("hostname")81        setup_details = job_target_details.get("setup_details_link")82        job = ScriptRunner(target, setup_details, self.subthread_queue)83        self.thread_pool.append(job)84    def _clean_dead_threads(self):85        """Clean up the dead threads and do the logging."""86        # Remove the dead threads.87        self.thread_pool = [_t for _t in self.thread_pool if _t.alive]88        while not self.subthread_queue.empty():89            # Do logging.90            sub_info = self.subthread_queue.get()91            host = sub_info.get("hostname")92            status = sub_info.get("status")93            line = "{} {} \n".format(host, status)94            self._add_logline(line)95    def run(self):96        print 'server_backend PID:', os.getpid()97        # Allow the object to die with grace.98        signal.signal(signal.SIGINT, handler)99        self.loc.acquire()100        while self.alive.value:101            self.loc.release()102            try:103                if self._full_pool() and self._job_waiting():104                    # If the pool has space and we have a job.105                    self._add_job()106                self._clean_dead_threads()107                if self._full_pool() or (not self._job_waiting()):108                    # Don't spin if we can't do work.109                    time.sleep(0.5)110            finally:111                # Take the lock before we loop around.112                self.loc.acquire()113        # Release the lock from the loop.114        self.loc.release()115        while not self._empty_pool():116            self._clean_dead_threads()117            time.sleep(0.5)118class ScriptRunner(threading.Thread):119    """Subthread update script runner."""120    def __init__(self, target, setup_details, shared_queue):121        super(ScriptRunner, self).__init__()122        self.target = target123        self.setup_details = setup_details124        self.alive = True125        self.start()126        if isinstance(shared_queue, Queue.Queue):127            self.shared_queue = shared_queue128        else:129            self.shared_queue = Queue.Queue()...

test_page_nested_groups.py

Source:test_page_nested_groups.py

...32        self.outer = OuterGroup(By.XPATH, '//div')33@pytest.fixture34def driver_wrapper():35    # Reset wrappers pool values36    DriverWrappersPool._empty_pool()37    DriverWrapper.config_properties_filenames = None38    # Create a new wrapper39    driver_wrapper = DriverWrappersPool.get_default_wrapper()40    driver_wrapper.driver = mock.MagicMock()41    return driver_wrapper42def test_reset_object_nested_groups(driver_wrapper):43    # Mock Driver.save_web_element = True44    driver_wrapper.config = mock.MagicMock()45    driver_wrapper.config.getboolean_optional.return_value = True46    # Create mock element47    mock_element = mock.MagicMock(spec=WebElement)48    driver_wrapper.driver.find_element.return_value = mock_element49    nested_page = NestedPageObject()50    # Check that web elements are empty...

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.