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

How to use _enqueue_drone method in autotest

Best Python code snippet using autotest_python

drone_manager_unittest.py

Source:drone_manager_unittest.py

...76    def _test_choose_drone_for_execution_helper(self, processes_info_list,77                                                requested_processes):78        for index, process_info in enumerate(processes_info_list):79            active_processes, max_processes = process_info80            self.manager._enqueue_drone(MockDrone(index, active_processes,81                                                  max_processes))82        return self.manager._choose_drone_for_execution(requested_processes,83                                                        self._USERNAME, None)84    def test_choose_drone_for_execution(self):85        drone = self._test_choose_drone_for_execution_helper([(1, 2), (0, 2)],86                                                             1)87        self.assertEquals(drone.name, 1)88    def test_choose_drone_for_execution_some_full(self):89        drone = self._test_choose_drone_for_execution_helper([(0, 1), (1, 3)],90                                                             2)91        self.assertEquals(drone.name, 1)92    def test_choose_drone_for_execution_all_full(self):93        drone = self._test_choose_drone_for_execution_helper([(2, 1), (3, 2)],94                                                             1)95        self.assertEquals(drone.name, 1)96    def test_choose_drone_for_execution_all_full_same_percentage_capacity(self):97        drone = self._test_choose_drone_for_execution_helper([(5, 3), (10, 6)],98                                                             1)99        self.assertEquals(drone.name, 1)100    def test_user_restrictions(self):101        # this drone is restricted to a different user102        self.manager._enqueue_drone(MockDrone(1, max_processes=10,103                                              allowed_users=['fakeuser']))104        # this drone is allowed but has lower capacity105        self.manager._enqueue_drone(MockDrone(2, max_processes=2,106                                              allowed_users=[self._USERNAME]))107        self.assertEquals(2,108                          self.manager.max_runnable_processes(self._USERNAME,109                                                              None))110        drone = self.manager._choose_drone_for_execution(111            1, username=self._USERNAME, drone_hostnames_allowed=None)112        self.assertEquals(drone.name, 2)113    def test_user_restrictions_with_full_drone(self):114        # this drone is restricted to a different user115        self.manager._enqueue_drone(MockDrone(1, max_processes=10,116                                              allowed_users=['fakeuser']))117        # this drone is allowed but is full118        self.manager._enqueue_drone(MockDrone(2, active_processes=3,119                                              max_processes=2,120                                              allowed_users=[self._USERNAME]))121        self.assertEquals(0,122                          self.manager.max_runnable_processes(self._USERNAME,123                                                              None))124        drone = self.manager._choose_drone_for_execution(125            1, username=self._USERNAME, drone_hostnames_allowed=None)126        self.assertEquals(drone.name, 2)127    def _setup_test_drone_restrictions(self, active_processes=0):128        self.manager._enqueue_drone(MockDrone(129            1, active_processes=active_processes, max_processes=10))130        self.manager._enqueue_drone(MockDrone(131            2, active_processes=active_processes, max_processes=5))132        self.manager._enqueue_drone(MockDrone(133            3, active_processes=active_processes, max_processes=2))134    def test_drone_restrictions_allow_any(self):135        self._setup_test_drone_restrictions()136        self.assertEquals(10,137                          self.manager.max_runnable_processes(self._USERNAME,138                                                              None))139        drone = self.manager._choose_drone_for_execution(140            1, username=self._USERNAME, drone_hostnames_allowed=None)141        self.assertEqual(drone.name, 1)142    def test_drone_restrictions_under_capacity(self):143        self._setup_test_drone_restrictions()144        drone_hostnames_allowed = (2, 3)145        self.assertEquals(146            5, self.manager.max_runnable_processes(self._USERNAME,147                                                   drone_hostnames_allowed))148        drone = self.manager._choose_drone_for_execution(149            1, username=self._USERNAME,150            drone_hostnames_allowed=drone_hostnames_allowed)151        self.assertEqual(drone.name, 2)152    def test_drone_restrictions_over_capacity(self):153        self._setup_test_drone_restrictions(active_processes=6)154        drone_hostnames_allowed = (2, 3)155        self.assertEquals(156            0, self.manager.max_runnable_processes(self._USERNAME,157                                                   drone_hostnames_allowed))158        drone = self.manager._choose_drone_for_execution(159            7, username=self._USERNAME,160            drone_hostnames_allowed=drone_hostnames_allowed)161        self.assertEqual(drone.name, 2)162    def test_drone_restrictions_allow_none(self):163        self._setup_test_drone_restrictions()164        drone_hostnames_allowed = ()165        self.assertEquals(166            0, self.manager.max_runnable_processes(self._USERNAME,167                                                   drone_hostnames_allowed))168        drone = self.manager._choose_drone_for_execution(169            1, username=self._USERNAME,170            drone_hostnames_allowed=drone_hostnames_allowed)171        self.assertEqual(drone, None)172    def test_initialize(self):173        results_hostname = 'results_repo'174        results_install_dir = '/results/install'175        settings.override_value(scheduler_config.CONFIG_SECTION,176                                'results_host_installation_directory',177                                results_install_dir)178        (drones.get_drone.expect_call(self.mock_drone.name)179         .and_return(self.mock_drone))180        results_drone = MockDrone('results_drone')181        self.god.stub_function(results_drone, 'set_autotest_install_dir')182        drones.get_drone.expect_call(results_hostname).and_return(results_drone)183        results_drone.set_autotest_install_dir.expect_call(results_install_dir)184        self.manager.initialize(base_results_dir=self._RESULTS_DIR,185                                drone_hostnames=[self.mock_drone.name],186                                results_repository_hostname=results_hostname)187        self.assert_(self.mock_drone.was_call_queued(188            'initialize', self._DRONE_RESULTS_DIR + '/'))189        self.god.check_playback()190    def test_execute_command(self):191        self.manager._enqueue_drone(self.mock_drone)192        pidfile_name = 'my_pidfile'193        log_file = 'log_file'194        pidfile_id = self.manager.execute_command(195            command=['test', drone_manager.WORKING_DIRECTORY],196            working_directory=self._WORKING_DIRECTORY,197            pidfile_name=pidfile_name,198            num_processes=1,199            log_file=log_file)200        full_working_directory = os.path.join(self._DRONE_RESULTS_DIR,201                                              self._WORKING_DIRECTORY)202        self.assertEquals(pidfile_id.path,203                          os.path.join(full_working_directory, pidfile_name))204        self.assert_(self.mock_drone.was_call_queued(205            'execute_command', ['test', full_working_directory],206            full_working_directory,207            os.path.join(self._DRONE_RESULTS_DIR, log_file), pidfile_name))208    def test_attach_file_to_execution(self):209        self.manager._enqueue_drone(self.mock_drone)210        contents = 'my\ncontents'211        attached_path = self.manager.attach_file_to_execution(212            self._WORKING_DIRECTORY, contents)213        self.manager.execute_command(command=['test'],214                                     working_directory=self._WORKING_DIRECTORY,215                                     pidfile_name='mypidfile',216                                     num_processes=1,217                                     drone_hostnames_allowed=None)218        self.assert_(self.mock_drone.was_call_queued(219            'write_to_file',220            os.path.join(self._DRONE_RESULTS_DIR, attached_path),221            contents))222    def test_copy_results_on_drone(self):223        self.manager.copy_results_on_drone(self.mock_drone_process,...

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.