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

How to use verify_tasks method in locust

Best Python code snippet using locust

test_tasks.py

Source:test_tasks.py

...17    :param data: Dictionary containing tasks18    :return: tasks in the dictionary19    """20    return [Task(**task) for task in data]21def verify_tasks(data: list[dict[str, Any]]):22    """23    Check tasks properties24    :param data: Dictionary containing tasks25    :return: True if the tasks properties are correct26    """27    tasks = get_tasks(data)28    for task in tasks:29        # primitive attributes30        assert task.id != ""31        assert task.user_id != ""32        assert task.title != ""33    return True34class TestTasksRouter(unittest.TestCase):35    """36    Class where task router test are performed37    """38    class Endpoints(str, enum.Enum):39        """40        Task router endpoints41        """42        tasks = "/tasks/"43        task_by_id = "/tasks/{}"44    def test_tasks(self):45        """46        Test the method that allows to get all tasks47        """48        response = client.get(self.Endpoints.tasks)49        self.assertEqual(status.HTTP_200_OK, response.status_code)50        result = GetAllResult(**response.json())51        self.assertEqual(TASKS_COUNT, result.total_items)52        self.assertEqual(result.total_items, len(result.data))53        self.assertTrue(verify_tasks(result.data))54    def test_tasks_empty_parameters(self):55        """56        Test the method that allows to get all tasks, without passing parameters values57        """58        response = client.get(self.Endpoints.tasks, params={"title": "", "completed": None})59        self.assertEqual(status.HTTP_200_OK, response.status_code)60        result = GetAllResult(**response.json())61        self.assertEqual(TASKS_COUNT, result.total_items)62        self.assertEqual(result.total_items, len(result.data))63        self.assertTrue(verify_tasks(result.data))64    def test_tasks_by_title_ok(self):65        """66        Test the method that allows to get all tasks, by title67        """68        words = ["delectus", "aut", "autem", "veritatis", "pariatur"]69        for word in words:70            response = client.get(self.Endpoints.tasks, params={"title": word})71            self.assertEqual(status.HTTP_200_OK, response.status_code)72            result = GetAllResult(**response.json())73            self.assertEqual(result.total_items, len(result.data))74            self.assertTrue(verify_tasks(result.data))75    def test_tasks_by_title_empty(self):76        """77        Test the method that allows to get all tasks, by title78        """79        words = ["messi", "maradona"]80        for word in words:81            response = client.get(self.Endpoints.tasks, params={"title": word})82            self.assertEqual(status.HTTP_200_OK, response.status_code)83            result = GetAllResult(**response.json())84            self.assertEqual(result.total_items, len(result.data))85            self.assertEqual(0, result.total_items)86    def test_tasks_by_status(self):87        """88        Test the method that allows to get all tasks, by status89        """90        response_completed_tasks = client.get(91            self.Endpoints.tasks, params={"completed": True}92        )93        self.assertEqual(status.HTTP_200_OK, response_completed_tasks.status_code)94        result_completed_tasks = GetAllResult(**response_completed_tasks.json())95        self.assertEqual(96            result_completed_tasks.total_items, len(result_completed_tasks.data)97        )98        self.assertTrue(verify_tasks(result_completed_tasks.data))99        response_not_completed_tasks = client.get(100            self.Endpoints.tasks, params={"completed": False}101        )102        self.assertEqual(status.HTTP_200_OK, response_not_completed_tasks.status_code)103        result_not_completed_tasks = GetAllResult(**response_not_completed_tasks.json())104        self.assertEqual(105            result_not_completed_tasks.total_items, len(result_not_completed_tasks.data)106        )107        self.assertTrue(verify_tasks(result_not_completed_tasks.data))108        tasks_count_by_status = (109                result_completed_tasks.total_items + result_not_completed_tasks.total_items110        )111        response_all_tasks = client.get(self.Endpoints.tasks)112        self.assertEqual(status.HTTP_200_OK, response_all_tasks.status_code)113        result_all_tasks = GetAllResult(**response_all_tasks.json())114        self.assertEqual(result_all_tasks.total_items, len(result_all_tasks.data))115        self.assertTrue(verify_tasks(result_all_tasks.data))116        self.assertEqual(117            TASKS_COUNT, result_all_tasks.total_items, tasks_count_by_status118        )119    def test_task_by_id_ok(self):120        """121        Test the method that allows you to get all the tasks by their identifier,122        when the identifier passed as a parameter corresponds to an existing task123        """124        for task_id in range(1, TASKS_COUNT + 1):125            response = client.get(self.Endpoints.task_by_id.format(task_id))126            self.assertEqual(status.HTTP_200_OK, response.status_code)127            task = response.json()128            self.assertTrue(verify_tasks([task]))129    def test_task_by_id_fail(self):130        """131        Try the method that allows you to get all the tasks by their identifier,132        when the identifier passed as a parameter does not correspond to an existing task133        """134        response = client.get(self.Endpoints.task_by_id.format(TASKS_COUNT + 1))135        self.assertEqual(status.HTTP_404_NOT_FOUND, response.status_code)136if __name__ == "__main__":...

test_env.py

Source:test_env.py

...31            e.exception.args[0],32            "The following user classes have the same class name: locust.test.fake_module1_for_env_test.MyUserWithSameName, locust.test.fake_module2_for_env_test.MyUserWithSameName",33        )34    def test_assign_equal_weights(self):35        def verify_tasks(u, target_tasks):36            self.assertEqual(len(u.tasks), len(target_tasks))37            tasks = [t.__name__ for t in u.tasks]38            self.assertEqual(len(tasks), len(set(tasks)))39            self.assertEqual(set(tasks), set(target_tasks))40        # Base case41        class MyUser1(User):42            wait_time = constant(0)43            @task(4)44            def my_task(self):45                pass46            @task(1)47            def my_task_2(self):48                pass49        environment = Environment(user_classes=[MyUser1])50        environment.assign_equal_weights()51        u = environment.user_classes[0]52        verify_tasks(u, ["my_task", "my_task_2"])53        # Testing nested task sets54        class MyUser2(User):55            @task56            class TopLevelTaskSet(TaskSet):57                @task58                class IndexTaskSet(TaskSet):59                    @task(10)60                    def index(self):61                        self.client.get("/")62                    @task63                    def stop(self):64                        self.client.get("/hi")65                @task(2)66                def stats(self):67                    self.client.get("/stats/requests")68        environment = Environment(user_classes=[MyUser2])69        environment.assign_equal_weights()70        u = environment.user_classes[0]71        verify_tasks(u, ["index", "stop", "stats"])72        # Testing task assignment via instance variable73        def outside_task():74            pass75        def outside_task_2():76            pass77        class SingleTaskSet(TaskSet):78            tasks = [outside_task, outside_task, outside_task_2]79        class MyUser3(User):80            tasks = [SingleTaskSet, outside_task]81        environment = Environment(user_classes=[MyUser3])82        environment.assign_equal_weights()83        u = environment.user_classes[0]84        verify_tasks(u, ["outside_task", "outside_task_2"])85        # Testing task assignment via dict86        class DictTaskSet(TaskSet):87            def dict_task_1():88                pass89            def dict_task_2():90                pass91            def dict_task_3():92                pass93            tasks = {94                dict_task_1: 5,95                dict_task_2: 3,96                dict_task_3: 1,97            }98        class MyUser4(User):99            tasks = [DictTaskSet, SingleTaskSet, SingleTaskSet]100        # Assign user tasks in dict101        environment = Environment(user_classes=[MyUser4])102        environment.assign_equal_weights()103        u = environment.user_classes[0]104        verify_tasks(u, ["outside_task", "outside_task_2", "dict_task_1", "dict_task_2", "dict_task_3"])105        class MyUser5(User):106            tasks = {107                DictTaskSet: 5,108                SingleTaskSet: 3,109                outside_task: 6,110            }111        environment = Environment(user_classes=[MyUser5])112        environment.assign_equal_weights()113        u = environment.user_classes[0]...

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.