1#!/usr/bin/python 2#pylint: disable-msg=C0111 3 4"""Tests for autotest_lib.scheduler.drones.""" 5 6import cPickle 7import unittest 8 9import common 10from autotest_lib.client.common_lib import utils 11from autotest_lib.client.common_lib.test_utils import mock 12from autotest_lib.scheduler import drones 13from autotest_lib.server.hosts import ssh_host 14 15 16class RemoteDroneTest(unittest.TestCase): 17 18 def setUp(self): 19 self.god = mock.mock_god() 20 self._mock_host = self.god.create_mock_class(ssh_host.SSHHost, 21 'mock SSHHost') 22 self.god.stub_function(drones.drone_utility, 'create_host') 23 self.drone_utility_path = 'mock-drone-utility-path' 24 25 26 def tearDown(self): 27 self.god.unstub_all() 28 29 30 def test_unreachable(self): 31 drones.drone_utility.create_host.expect_call('fakehost').and_return( 32 self._mock_host) 33 self._mock_host.is_up.expect_call().and_return(False) 34 self.assertRaises(drones.DroneUnreachable, 35 drones._RemoteDrone, 'fakehost') 36 37 38 def test_execute_calls_impl(self): 39 self.god.stub_with(drones._RemoteDrone, '_drone_utility_path', 40 self.drone_utility_path) 41 drones.drone_utility.create_host.expect_call('fakehost').and_return( 42 self._mock_host) 43 self._mock_host.is_up.expect_call().and_return(True) 44 mock_calls = ('foo',) 45 mock_result = utils.CmdResult(stdout=cPickle.dumps('mock return')) 46 self._mock_host.run.expect_call( 47 'python %s' % self.drone_utility_path, 48 stdin=cPickle.dumps(mock_calls), stdout_tee=None, 49 connect_timeout=mock.is_instance_comparator(int)).and_return( 50 mock_result) 51 drone = drones._RemoteDrone('fakehost', timestamp_remote_calls=False) 52 self.assertEqual('mock return', drone._execute_calls_impl(mock_calls)) 53 self.god.check_playback() 54 55 56 def test_execute_queued_calls(self): 57 self.god.stub_with(drones._RemoteDrone, '_drone_utility_path', 58 self.drone_utility_path) 59 drones.drone_utility.create_host.expect_call('fakehost').and_return( 60 self._mock_host) 61 self._mock_host.is_up.expect_call().and_return(True) 62 drone = drones._RemoteDrone('fakehost', timestamp_remote_calls=False) 63 mock_return={} 64 mock_return['results'] = ['mock return'] 65 mock_return['warnings'] = [] 66 drone.queue_call('foo') 67 mock_result = utils.CmdResult(stdout=cPickle.dumps(mock_return)) 68 self._mock_host.run.expect_call( 69 'python %s' % self.drone_utility_path, 70 stdin=cPickle.dumps(drone.get_calls()), stdout_tee=None, 71 connect_timeout=mock.is_instance_comparator(int)).and_return( 72 mock_result) 73 self.assertEqual(mock_return['results'], drone.execute_queued_calls()) 74 self.god.check_playback() 75 76 77 78if __name__ == '__main__': 79 unittest.main() 80