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