KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference
How to use was_call_queued method in autotest

Best Python code snippet using autotest_python
drone_manager_unittest.py
Source:drone_manager_unittest.py
...37        if not was_called:38            print 'Recorded args:', recorded_arg_list39            print 'Expected:', arguments40        return was_called41    def was_call_queued(self, method, *args, **kwargs):42        return self._check_for_recorded_call('queue_call',43                                             (method, args, kwargs))44    def was_file_sent(self, drone, source_path, destination_path):45        return self._check_for_recorded_call('send_file_to',46                                             (drone, source_path,47                                              destination_path))48class DroneManager(unittest.TestCase):49    _DRONE_INSTALL_DIR = '/drone/install/dir'50    _DRONE_RESULTS_DIR = os.path.join(_DRONE_INSTALL_DIR, 'results')51    _RESULTS_DIR = '/results/dir'52    _SOURCE_PATH = 'source/path'53    _DESTINATION_PATH = 'destination/path'54    _WORKING_DIRECTORY = 'working/directory'55    _USERNAME = 'my_user'56    def setUp(self):57        self.god = mock.mock_god()58        self.god.stub_with(drones, 'AUTOTEST_INSTALL_DIR',59                           self._DRONE_INSTALL_DIR)60        self.manager = drone_manager.DroneManager()61        self.god.stub_with(self.manager, '_results_dir', self._RESULTS_DIR)62        # we don't want this to ever actually get called63        self.god.stub_function(drones, 'get_drone')64        # we don't want the DroneManager to go messing with global config65        def do_nothing():66            pass67        self.god.stub_with(self.manager, 'refresh_drone_configs', do_nothing)68        # set up some dummy drones69        self.mock_drone = MockDrone('mock_drone')70        self.manager._drones[self.mock_drone.name] = self.mock_drone71        self.results_drone = MockDrone('results_drone', 0, 10)72        self.manager._results_drone = self.results_drone73        self.mock_drone_process = drone_manager.Process(self.mock_drone.name, 0)74    def tearDown(self):75        self.god.unstub_all()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,224                                           self._SOURCE_PATH,225                                           self._DESTINATION_PATH)226        self.assert_(self.mock_drone.was_call_queued(227            'copy_file_or_directory',228            os.path.join(self._DRONE_RESULTS_DIR, self._SOURCE_PATH),229            os.path.join(self._DRONE_RESULTS_DIR, self._DESTINATION_PATH)))230    def test_copy_to_results_repository(self):231        self.manager.copy_to_results_repository(self.mock_drone_process,232                                                self._SOURCE_PATH)233        self.assert_(self.mock_drone.was_file_sent(234            self.results_drone,235            os.path.join(self._DRONE_RESULTS_DIR, self._SOURCE_PATH),236            os.path.join(self._RESULTS_DIR, self._SOURCE_PATH)))237    def test_write_lines_to_file(self):238        file_path = 'file/path'239        lines = ['line1', 'line2']240        written_data = 'line1\nline2\n'241        # write to results repository242        self.manager.write_lines_to_file(file_path, lines)243        self.assert_(self.results_drone.was_call_queued(244            'write_to_file', os.path.join(self._RESULTS_DIR, file_path),245            written_data))246        # write to a drone247        self.manager.write_lines_to_file(248            file_path, lines, paired_with_process=self.mock_drone_process)249        self.assert_(self.mock_drone.was_call_queued(250            'write_to_file',251            os.path.join(self._DRONE_RESULTS_DIR, file_path), written_data))252    def test_pidfile_expiration(self):253        self.god.stub_with(self.manager, '_get_max_pidfile_refreshes',254                           lambda: 0)255        pidfile_id = self.manager.get_pidfile_id_from('tag', 'name')256        self.manager.register_pidfile(pidfile_id)257        self.manager._drop_old_pidfiles()258        self.manager._drop_old_pidfiles()259        self.assertFalse(self.manager._registered_pidfile_info)260if __name__ == '__main__':...
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.