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