• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3
2# Copyright 2016 The Chromium Authors. All rights reserved.
3# Use of this source code is governed by a BSD-style license that can be
4# found in the LICENSE file.
5
6import unittest
7from unittest import mock
8import six
9
10import common
11from autotest_lib.site_utils import diagnosis_utils
12from autotest_lib.server import frontend
13
14
15class DiagnosisUtilsTest(unittest.TestCase):
16    """Tests for diagnosis_utils."""
17
18    def setUp(self):
19        """Set up test."""
20        self.afe_mock = mock.MagicMock()
21
22    def _constructRPCHelper(self):
23        """Method to construct RPCHelper"""
24        return diagnosis_utils.RPCHelper(self.afe_mock)
25
26    def _mockZeroHost(self):
27        """Mock method to return zero host"""
28        return ()
29
30    def _mockTwoAvailableHosts(self):
31        """Mock method to return two available hosts"""
32        host_a = _StubHost(status='Ready', locked=False)
33        host_b = _StubHost(status='Ready', locked=False)
34        return (host_a, host_b)
35
36    def _mockTwoFailedHosts(self):
37        """Mock method to return two unavailable hosts"""
38        host_a = _StubHost(status='Repair Failed', locked=False)
39        host_b = _StubHost(status='Repairing', locked=False)
40        return (host_a, host_b)
41
42    def testCheckDutAvailable(self):
43        """Test check_dut_availability with different scenarios"""
44        rpc_helper = self._constructRPCHelper()
45        labels = ('board:test_board', 'pool:test_pool', 'model:test_model')
46
47        # Mock aef.get_hosts to return 0 host
48        self.afe_mock.get_hosts.return_value = self._mockZeroHost()
49        skip_duts_check = False
50
51        # When minimum_duts is 0, do not force available DUTs
52        minimum_duts = 0
53        rpc_helper.check_dut_availability(labels,
54                                          minimum_duts=minimum_duts,
55                                          skip_duts_check=skip_duts_check)
56
57        # When skip_duts_check = False and minimum_duts > 0 and there's no host,
58        # should raise DUTsNotAvailableError
59        minimum_duts = 1
60        with self.assertRaises(diagnosis_utils.DUTsNotAvailableError):
61            rpc_helper.check_dut_availability(labels,
62                                              minimum_duts=minimum_duts,
63                                              skip_duts_check=skip_duts_check)
64
65        # Mock aef.get_hosts to return 2 available hosts
66        self.afe_mock.get_hosts.return_value = self._mockTwoAvailableHosts()
67
68        # Set skip_duts_check to True, should not force checking avialble DUTs
69        # although available DUTs are less then minimum_duts
70        minimum_duts = 4
71        skip_duts_check = True
72        rpc_helper.check_dut_availability(labels,
73                                          minimum_duts=minimum_duts,
74                                          skip_duts_check=skip_duts_check)
75
76        # Set skip_duts_check to False and current available DUTs are less
77        # then minimum_duts, should raise NotEnoughDutsError
78        skip_duts_check = False
79        with self.assertRaises(diagnosis_utils.NotEnoughDutsError):
80            rpc_helper.check_dut_availability(labels,
81                                              minimum_duts=minimum_duts,
82                                              skip_duts_check=skip_duts_check)
83
84        # When skip_duts_check is False and current available DUTs
85        # satisfy minimum_duts, no errors
86        minimum_duts = 2
87        rpc_helper.check_dut_availability(labels,
88                                          minimum_duts=minimum_duts,
89                                          skip_duts_check=skip_duts_check)
90
91        # Mock aef.get_hosts to return 2 failed hosts
92        self.afe_mock.get_hosts.return_value = self._mockTwoFailedHosts()
93
94        # When skip_duts_check is False and the two hosts are not available,
95        # should raise NotEnoughDutsError
96        with self.assertRaises(diagnosis_utils.NotEnoughDutsError):
97            rpc_helper.check_dut_availability(labels,
98                                              minimum_duts=minimum_duts,
99                                              skip_duts_check=skip_duts_check)
100
101
102class _StubHost(object):
103
104    def __init__(self, status, locked):
105        self.status = status
106        self.locked = locked
107
108    if six.PY2:
109        is_available = frontend.Host.is_available.__func__
110    elif six.PY3:
111        is_available = frontend.Host.is_available
112
113
114if __name__ == '__main__':
115    unittest.main()
116