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

How to use _get_user_current_count method in locust

Best Python code snippet using locust

test_dispatch.py

Source:test_dispatch.py

...2883                        list(users_dispatcher)2884                        for user_class in user_classes:2885                            if user_class.fixed_count:2886                                self.assertEqual(2887                                    users_dispatcher._get_user_current_count(user_class.__name__),2888                                    user_class.fixed_count,2889                                )2890                        # Ramp-down to go to `down_to_count`2891                        # and ensure the fixed users was decreased too2892                        users_dispatcher.new_dispatch(target_user_count=down_to_count, spawn_rate=1)2893                        users_dispatcher._wait_between_dispatch = 02894                        list(users_dispatcher)2895                        for user_class in user_classes:2896                            if user_class.fixed_count:2897                                self.assertNotEqual(2898                                    users_dispatcher._get_user_current_count(user_class.__name__),2899                                    user_class.fixed_count,2900                                )2901                        # Ramp-up go back to `target_user_count` and ensure2902                        # the fixed users return to their counts2903                        users_dispatcher.new_dispatch(target_user_count=target_user_count, spawn_rate=1)2904                        users_dispatcher._wait_between_dispatch = 02905                        list(users_dispatcher)2906                        for user_class in user_classes:2907                            if user_class.fixed_count:2908                                self.assertEqual(2909                                    users_dispatcher._get_user_current_count(user_class.__name__),2910                                    user_class.fixed_count,2911                                )2912def _aggregate_dispatched_users(d: Dict[str, Dict[str, int]]) -> Dict[str, int]:2913    user_classes = list(next(iter(d.values())).keys())2914    return {u: sum(d[u] for d in d.values()) for u in user_classes}2915def _user_count(d: Dict[str, Dict[str, int]]) -> int:2916    return sum(map(sum, map(dict.values, d.values())))  # type: ignore2917def _user_count_on_worker(d: Dict[str, Dict[str, int]], worker_node_id: str) -> int:...

dispatch.py

Source:dispatch.py

...228            self._current_user_count -= 1229            self._try_dispatch_fixed = True230            if self._current_user_count == 0 or self._current_user_count <= current_user_count_target:231                return self._users_on_workers232    def _get_user_current_count(self, user: str) -> int:233        count = 0234        for users_on_node in self._users_on_workers.values():235            count += users_on_node.get(user, 0)236        return count237    def _distribute_users(238        self, target_user_count: int239    ) -> Tuple[240        Dict[str, Dict[str, int]], Generator[Optional[str], None, None], itertools.cycle, List[Tuple["WorkerNode", str]]241    ]:242        """243        This function might take some time to complete if the `target_user_count` is a big number. A big number244        is typically > 50 000. However, this function is only called if a worker is added or removed while a test245        is running. Such a situation should be quite rare.246        """247        user_gen = self._user_gen()248        worker_gen = itertools.cycle(self._worker_nodes)249        users_on_workers = {250            worker_node.id: {user_class.__name__: 0 for user_class in self._user_classes}251            for worker_node in self._worker_nodes252        }253        active_users = []254        user_count = 0255        while user_count < target_user_count:256            user = next(user_gen)257            if not user:258                break259            worker_node = next(worker_gen)260            users_on_workers[worker_node.id][user] += 1261            user_count += 1262            active_users.append((worker_node, user))263        return users_on_workers, user_gen, worker_gen, active_users264    def _user_gen(self) -> Generator[Optional[str], None, None]:265        """266        This method generates users according to their weights using267        a smooth weighted round-robin algorithm implemented by https://github.com/linnik/roundrobin.268        For example, given users A, B with weights 5 and 1 respectively, this algorithm269        will yield AAABAAAAABAA. The smooth aspect of this algorithm is what makes it possible270        to keep the distribution during ramp-up and ramp-down. If we were to use a normal271        weighted round-robin algorithm, we'd get AAAAABAAAAAB which would make the distribution272        less accurate during ramp-up/down.273        """274        def infinite_cycle_gen(users: List[Tuple[Type[User], int]]) -> itertools.cycle:275            if not users:276                return itertools.cycle([None])277            # Normalize the weights so that the smallest weight will be equal to "target_min_weight".278            # The value "2" was experimentally determined because it gave a better distribution especially279            # when dealing with weights which are close to each others, e.g. 1.5, 2, 2.4, etc.280            target_min_weight = 2281            # 'Value' here means weight or fixed count282            normalized_values = [283                (284                    user.__name__,285                    round(target_min_weight * value / min(u[1] for u in users)),286                )287                for user, value in users288            ]289            generation_length_to_get_proper_distribution = sum(290                normalized_val[1] for normalized_val in normalized_values291            )292            gen = smooth(normalized_values)293            # Instead of calling `gen()` for each user, we cycle through a generator of fixed-length294            # `generation_length_to_get_proper_distribution`. Doing so greatly improves performance because295            # we only ever need to call `gen()` a relatively small number of times. The length of this generator296            # is chosen as the sum of the normalized weights. So, for users A, B, C of weights 2, 5, 6, the length is297            # 2 + 5 + 6 = 13 which would yield the distribution `CBACBCBCBCABC` that gets repeated over and over298            # until the target user count is reached.299            return itertools.cycle(gen() for _ in range(generation_length_to_get_proper_distribution))300        fixed_users = {u.__name__: u for u in self._user_classes if u.fixed_count}301        cycle_fixed_gen = infinite_cycle_gen([(u, u.fixed_count) for u in fixed_users.values()])302        cycle_weighted_gen = infinite_cycle_gen([(u, u.weight) for u in self._user_classes if not u.fixed_count])303        # Spawn users304        while True:305            if self._try_dispatch_fixed:306                self._try_dispatch_fixed = False307                current_fixed_users_count = {u: self._get_user_current_count(u) for u in fixed_users}308                spawned_classes: Set[str] = set()309                while len(spawned_classes) != len(fixed_users):310                    user_name: Optional[str] = next(cycle_fixed_gen)311                    if not user_name:312                        break313                    if current_fixed_users_count[user_name] < fixed_users[user_name].fixed_count:314                        current_fixed_users_count[user_name] += 1315                        if current_fixed_users_count[user_name] == fixed_users[user_name].fixed_count:316                            spawned_classes.add(user_name)317                        yield user_name318                        # 'self._try_dispatch_fixed' was changed outhere,  we have to recalculate current count319                        if self._try_dispatch_fixed:320                            current_fixed_users_count = {u: self._get_user_current_count(u) for u in fixed_users}321                            spawned_classes.clear()322                            self._try_dispatch_fixed = False323            yield next(cycle_weighted_gen)324    @staticmethod325    def _fast_users_on_workers_copy(users_on_workers: Dict[str, Dict[str, int]]) -> Dict[str, Dict[str, int]]:326        """deepcopy is too slow, so we use this custom copy function.327        The implementation was profiled and compared to other implementations such as dict-comprehensions328        and the one below is the most efficient.329        """330        # type is ignored due to: https://github.com/python/mypy/issues/1507...

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.