• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python2
2#
3# Copyright (c) 2016 The Chromium OS Authors. All rights reserved.
4# Use of this source code is governed by a BSD-style license that can be
5# found in the LICENSE file.
6
7"""Unit tests for frontend/afe/rpc_utils.py."""
8
9import mock
10import unittest
11
12import common
13from autotest_lib.client.common_lib import control_data
14from autotest_lib.frontend import setup_django_environment
15from autotest_lib.frontend.afe import frontend_test_utils
16from autotest_lib.frontend.afe import models
17from autotest_lib.frontend.afe import rpc_utils
18
19
20class DjangoModelTest(unittest.TestCase):
21    """Unit tests for verifying Django behavior."""
22
23    def test_model_equal_by_id(self):
24        """Test that model instances compare equal by id."""
25        x = models.Host(id=10, hostname='foo')
26        y = models.Host(id=10, hostname='bar')
27        self.assertEqual(x, y)
28
29
30class RpcUtilsTest(unittest.TestCase,
31                   frontend_test_utils.FrontendTestMixin):
32    """Unit tests for functions in rpc_utils.py."""
33    def setUp(self):
34        self._frontend_common_setup()
35
36
37    def tearDown(self):
38        self._frontend_common_teardown()
39
40
41    def testCheckIsServer(self):
42        """Ensure that test type check is correct."""
43        self.assertFalse(rpc_utils._check_is_server_test(None))
44        self.assertFalse(rpc_utils._check_is_server_test(
45            control_data.CONTROL_TYPE.CLIENT))
46        self.assertFalse(rpc_utils._check_is_server_test('Client'))
47        self.assertTrue(rpc_utils._check_is_server_test(
48            control_data.CONTROL_TYPE.SERVER))
49        self.assertTrue(rpc_utils._check_is_server_test('Server'))
50        self.assertFalse(rpc_utils._check_is_server_test('InvalidType'))
51
52
53class ConvertToKwargsOnlyTest(unittest.TestCase):
54    """Unit tests for _convert_to_kwargs_only()."""
55
56    # pylint: disable=unused-argument,missing-docstring
57
58    def test_no_kwargs_in_spec(self):
59        """Test with function without kwargs."""
60        def func(a, b):
61            pass
62        got = rpc_utils._convert_to_kwargs_only(func, (1, 2), {})
63        self.assertEquals(got, {'a': 1, 'b': 2})
64
65    def test_pass_by_keyword(self):
66        """Test passing required args by keyword."""
67        def func(a, b):
68            pass
69        got = rpc_utils._convert_to_kwargs_only(func, (), {'a': 1, 'b': 2})
70        self.assertEquals(got, {'a': 1, 'b': 2})
71
72    def test_with_kwargs(self):
73        """Test with custom keyword arg."""
74        def func(a, b, **kwargs):
75            pass
76        got = rpc_utils._convert_to_kwargs_only(func, (1, 2), {'c': 3})
77        self.assertEquals(got, {'a': 1, 'b': 2, 'c': 3})
78
79    def test_with_kwargs_pass_by_keyword(self):
80        """Test passing required parameter by keyword."""
81        def func(a, b, **kwargs):
82            pass
83        got = rpc_utils._convert_to_kwargs_only(func, (1,), {'b': 2, 'c': 3})
84        self.assertEquals(got, {'a': 1, 'b': 2, 'c': 3})
85
86    def test_empty_kwargs(self):
87        """Test without passing kwargs."""
88        def func(a, b, **kwargs):
89            pass
90        got = rpc_utils._convert_to_kwargs_only(func, (1, 2), {})
91        self.assertEquals(got, {'a': 1, 'b': 2})
92
93    def test_with_varargs(self):
94        """Test against vararg function."""
95        def func(a, b, *args):
96            pass
97        got = rpc_utils._convert_to_kwargs_only(func, (1, 2, 3), {})
98        self.assertEquals(got, {'a': 1, 'b': 2, 'args': (3,)})
99
100
101class AllowedHostsForMainJobTest(unittest.TestCase):
102    """Unit tests for _allowed_hosts_for_main_job()."""
103
104    # pylint: disable=missing-docstring
105
106    @mock.patch.object(rpc_utils, 'bucket_hosts_by_shard', autospec=True)
107    def test_multiple_shards(self, bucket_mock):
108        bucket_mock.return_value = {
109            'shard1': [],
110            'shard2': [],
111        }
112        got = rpc_utils._allowed_hosts_for_main_job([])
113        self.assertFalse(got)
114
115    @mock.patch.object(rpc_utils, 'bucket_hosts_by_shard', autospec=True)
116    def test_one_shard_with_less_hosts(self, bucket_mock):
117        bucket_mock.return_value = {
118            'shard1': [1],
119        }
120        got = rpc_utils._allowed_hosts_for_main_job([1, 2])
121        self.assertFalse(got)
122
123    @mock.patch.object(rpc_utils, 'bucket_hosts_by_shard', autospec=True)
124    def test_one_shard_with_equal_hosts(self, bucket_mock):
125        bucket_mock.return_value = {
126            'shard1': [1, 2],
127        }
128        got = rpc_utils._allowed_hosts_for_main_job([1, 2])
129        self.assertTrue(got)
130
131    @mock.patch.object(rpc_utils, 'bucket_hosts_by_shard', autospec=True)
132    def test_no_shards(self, bucket_mock):
133        bucket_mock.return_value = {}
134        got = rpc_utils._allowed_hosts_for_main_job([1, 2])
135        self.assertTrue(got)
136
137
138if __name__ == '__main__':
139    unittest.main()
140