Method test_distribute_users

locust/test/test_dispatch.py:2127–2146

(self)

    mixed_users = weighted_user_classes[:25] + fixed_user_classes_10k[25:]

    def test_distribute_users(self):
        for user_classes in [self.weighted_user_classes, self.fixed_user_classes_1M, self.mixed_users]:
            workers = [WorkerNode(str(i)) for i in range(10_000)]

            target_user_count = 1_000_000

            users_dispatcher = UsersDispatcher(worker_nodes=workers, user_classes=user_classes)

            ts = time.perf_counter()
            users_on_workers, user_gen, worker_gen, active_users = users_dispatcher._distribute_users(
                target_user_count=target_user_count
            )
            delta = time.perf_counter() - ts

            # Because tests are run with coverage, the code will be slower.
            # We set the pass criterion to 7000ms, but in real life, the
            # `_distribute_users` method runs faster than this.
            self.assertLessEqual(1000 * delta, 7000)

            self.assertEqual(_user_count(users_on_workers), target_user_count)

    def test_ramp_up_from_0_to_100_000_users_with_50_user_classes_and_1000_workers_and_5000_spawn_rate(self):
        for user_classes in [