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

How to use _drone_utility_path method in autotest

Best Python code snippet using autotest_python

drones.py

Source:drones.py

1import pickle, os, tempfile, logging2import common3from autotest_lib.scheduler import drone_utility, email_manager4from autotest_lib.client.common_lib import error, global_config5AUTOTEST_INSTALL_DIR = global_config.global_config.get_config_value('SCHEDULER',6                                                 'drone_installation_directory')7class _AbstractDrone(object):8    def __init__(self):9        self._calls = []10        self.hostname = None11        self.enabled = True12        self.max_processes = 013        self.active_processes = 014    def shutdown(self):15        pass16    def used_capacity(self):17        if self.max_processes == 0:18            return 1.019        return float(self.active_processes) / self.max_processes20    def _execute_calls_impl(self, calls):21        raise NotImplementedError22    def _execute_calls(self, calls):23        return_message = self._execute_calls_impl(calls)24        for warning in return_message['warnings']:25            subject = 'Warning from drone %s' % self.hostname26            logging.warn(subject + '\n' + warning)27            email_manager.manager.enqueue_notify_email(subject, warning)28        return return_message['results']29    def call(self, method, *args, **kwargs):30        return self._execute_calls(31            [drone_utility.call(method, *args, **kwargs)])32    def queue_call(self, method, *args, **kwargs):33        self._calls.append(drone_utility.call(method, *args, **kwargs))34    def clear_call_queue(self):35        self._calls = []36    def execute_queued_calls(self):37        if not self._calls:38            return39        self._execute_calls(self._calls)40        self.clear_call_queue()41class _LocalDrone(_AbstractDrone):42    def __init__(self):43        super(_LocalDrone, self).__init__()44        self.hostname = 'localhost'45        self._drone_utility = drone_utility.DroneUtility()46    def _execute_calls_impl(self, calls):47        return self._drone_utility.execute_calls(calls)48    def send_file_to(self, drone, source_path, destination_path,49                     can_fail=False):50        if drone.hostname == self.hostname:51            self.queue_call('copy_file_or_directory', source_path,52                            destination_path)53        else:54            self.queue_call('send_file_to', drone.hostname, source_path,55                            destination_path, can_fail)56class _RemoteDrone(_AbstractDrone):57    _temporary_directory = None58    def __init__(self, hostname):59        super(_RemoteDrone, self).__init__()60        self.hostname = hostname61        self._host = drone_utility.create_host(hostname)62        self._drone_utility_path = os.path.join(AUTOTEST_INSTALL_DIR,63                                                'scheduler',64                                                'drone_utility.py')65        try:66            self._host.run('mkdir -p ' + self._temporary_directory,67                           timeout=10)68        except error.AutoservError:69            pass70    @classmethod71    def set_temporary_directory(cls, temporary_directory):72        cls._temporary_directory = temporary_directory73    def shutdown(self):74        super(_RemoteDrone, self).shutdown()75        self._host.close()76    def _execute_calls_impl(self, calls):77        calls_fd, calls_filename = tempfile.mkstemp(suffix='.pickled_calls')78        calls_file = os.fdopen(calls_fd, 'w+')79        pickle.dump(calls, calls_file)80        calls_file.flush()81        calls_file.seek(0)82        try:83            logging.info("Running drone_utility on %s", self.hostname)84            result = self._host.run('python %s' % self._drone_utility_path,85                                    stdin=calls_file, connect_timeout=300)86        finally:87            calls_file.close()88            os.remove(calls_filename)89        try:90            return pickle.loads(result.stdout)91        except Exception: # pickle.loads can throw all kinds of exceptions92            logging.critical('Invalid response:\n---\n%s\n---', result.stdout)93            raise94    def send_file_to(self, drone, source_path, destination_path,95                     can_fail=False):96        if drone.hostname == self.hostname:97            self.queue_call('copy_file_or_directory', source_path,98                            destination_path)99        elif isinstance(drone, _LocalDrone):100            drone.queue_call('get_file_from', self.hostname, source_path,101                             destination_path)102        else:103            self.queue_call('send_file_to', drone.hostname, source_path,104                            destination_path, can_fail)105def set_temporary_directory(temporary_directory):106    _RemoteDrone.set_temporary_directory(temporary_directory)107def get_drone(hostname):108    """109    Use this factory method to get drone objects.110    """111    if hostname == 'localhost':112        return _LocalDrone()...

drones_unittest.py

Source:drones_unittest.py

1#!/usr/bin/python2#pylint: disable-msg=C01113"""Tests for autotest_lib.scheduler.drones."""4import cPickle5import unittest6import common7from autotest_lib.client.common_lib import utils8from autotest_lib.client.common_lib.test_utils import mock9from autotest_lib.scheduler import drones10from autotest_lib.server.hosts import ssh_host11class RemoteDroneTest(unittest.TestCase):12    def setUp(self):13        self.god = mock.mock_god()14        self._mock_host = self.god.create_mock_class(ssh_host.SSHHost,15                                                     'mock SSHHost')16        self.god.stub_function(drones.drone_utility, 'create_host')17        self.drone_utility_path = 'mock-drone-utility-path'18    def tearDown(self):19        self.god.unstub_all()20    def test_unreachable(self):21        drones.drone_utility.create_host.expect_call('fakehost').and_return(22                self._mock_host)23        self._mock_host.is_up.expect_call().and_return(False)24        self.assertRaises(drones.DroneUnreachable,25                          drones._RemoteDrone, 'fakehost')26    def test_execute_calls_impl(self):27        self.god.stub_with(drones._RemoteDrone, '_drone_utility_path',28                           self.drone_utility_path)29        drones.drone_utility.create_host.expect_call('fakehost').and_return(30                self._mock_host)31        self._mock_host.is_up.expect_call().and_return(True)32        mock_calls = ('foo',)33        mock_result = utils.CmdResult(stdout=cPickle.dumps('mock return'))34        self._mock_host.run.expect_call(35                'python %s' % self.drone_utility_path,36                stdin=cPickle.dumps(mock_calls), stdout_tee=None,37                connect_timeout=mock.is_instance_comparator(int)).and_return(38                        mock_result)39        drone = drones._RemoteDrone('fakehost', timestamp_remote_calls=False)40        self.assertEqual('mock return', drone._execute_calls_impl(mock_calls))41        self.god.check_playback()42    def test_execute_queued_calls(self):43        self.god.stub_with(drones._RemoteDrone, '_drone_utility_path',44                           self.drone_utility_path)45        drones.drone_utility.create_host.expect_call('fakehost').and_return(46                self._mock_host)47        self._mock_host.is_up.expect_call().and_return(True)48        drone = drones._RemoteDrone('fakehost', timestamp_remote_calls=False)49        mock_return={}50        mock_return['results'] = ['mock return']51        mock_return['warnings'] = []52        drone.queue_call('foo')53        mock_result = utils.CmdResult(stdout=cPickle.dumps(mock_return))54        self._mock_host.run.expect_call(55                'python %s' % self.drone_utility_path,56                stdin=cPickle.dumps(drone.get_calls()), stdout_tee=None,57                connect_timeout=mock.is_instance_comparator(int)).and_return(58                        mock_result)59        self.assertEqual(mock_return['results'], drone.execute_queued_calls())60        self.god.check_playback()61if __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.