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 9 import mock 10 import unittest 11 12 import common 13 from autotest_lib.client.common_lib import control_data 14 from autotest_lib.frontend import setup_django_environment 15 from autotest_lib.frontend.afe import frontend_test_utils 16 from autotest_lib.frontend.afe import models 17 from autotest_lib.frontend.afe import rpc_utils 18 19 20 class 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 30 class 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 53 class 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 101 class AllowedHostsForMasterJobTest(unittest.TestCase): 102 """Unit tests for _allowed_hosts_for_master_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_master_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_master_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_master_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_master_job([1, 2]) 135 self.assertTrue(got) 136 137 138 if __name__ == '__main__': 139 unittest.main() 140
