How to use _enqueue_drone method in autotest

Best Python code snippet using autotest_python

drone_manager_unittest.py

Source:drone_manager_unittest.py Github

copy

Full Screen

...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,...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run autotest automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful