| #!/usr/bin/python |
| # |
| # Copyright (c) 2016 The Chromium OS Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """Unit tests for frontend/afe/moblab_rpc_interface.py.""" |
| |
| import __builtin__ |
| # The boto module is only available/used in Moblab for validation of cloud |
| # storage access. The module is not available in the test lab environment, |
| # and the import error is handled. |
| import ConfigParser |
| import mox |
| import StringIO |
| import unittest |
| |
| import common |
| |
| from autotest_lib.client.common_lib import error |
| from autotest_lib.client.common_lib import global_config |
| from autotest_lib.client.common_lib import lsbrelease_utils |
| from autotest_lib.frontend import setup_django_environment |
| from autotest_lib.frontend.afe import frontend_test_utils |
| from autotest_lib.frontend.afe import moblab_rpc_interface |
| from autotest_lib.frontend.afe import rpc_utils |
| from autotest_lib.server import utils |
| from autotest_lib.server.hosts import moblab_host |
| from autotest_lib.client.common_lib import utils as common_lib_utils |
| |
| |
| class MoblabRpcInterfaceTest(mox.MoxTestBase, |
| frontend_test_utils.FrontendTestMixin): |
| """Unit tests for functions in moblab_rpc_interface.py.""" |
| |
| def setUp(self): |
| super(MoblabRpcInterfaceTest, self).setUp() |
| self._frontend_common_setup(fill_data=False) |
| |
| |
| def tearDown(self): |
| self._frontend_common_teardown() |
| |
| |
| def setIsMoblab(self, is_moblab): |
| """Set utils.is_moblab result. |
| |
| @param is_moblab: Value to have utils.is_moblab to return. |
| """ |
| self.mox.StubOutWithMock(utils, 'is_moblab') |
| utils.is_moblab().AndReturn(is_moblab) |
| |
| |
| def _mockReadFile(self, path, lines=[]): |
| """Mock out reading a file line by line. |
| |
| @param path: Path of the file we are mock reading. |
| @param lines: lines of the mock file that will be returned when |
| readLine() is called. |
| """ |
| mockFile = self.mox.CreateMockAnything() |
| for line in lines: |
| mockFile.readline().AndReturn(line) |
| mockFile.readline() |
| mockFile.close() |
| open(path).AndReturn(mockFile) |
| |
| |
| def testMoblabOnlyDecorator(self): |
| """Ensure the moblab only decorator gates functions properly.""" |
| self.setIsMoblab(False) |
| self.mox.ReplayAll() |
| self.assertRaises(error.RPCException, |
| moblab_rpc_interface.get_config_values) |
| |
| |
| def testGetConfigValues(self): |
| """Ensure that the config object is properly converted to a dict.""" |
| self.setIsMoblab(True) |
| config_mock = self.mox.CreateMockAnything() |
| moblab_rpc_interface._CONFIG = config_mock |
| config_mock.get_sections().AndReturn(['section1', 'section2']) |
| config_mock.config = self.mox.CreateMockAnything() |
| config_mock.config.items('section1').AndReturn([('item1', 'value1'), |
| ('item2', 'value2')]) |
| config_mock.config.items('section2').AndReturn([('item3', 'value3'), |
| ('item4', 'value4')]) |
| |
| rpc_utils.prepare_for_serialization( |
| {'section1' : [('item1', 'value1'), |
| ('item2', 'value2')], |
| 'section2' : [('item3', 'value3'), |
| ('item4', 'value4')]}) |
| self.mox.ReplayAll() |
| moblab_rpc_interface.get_config_values() |
| |
| |
| def testUpdateConfig(self): |
| """Ensure that updating the config works as expected.""" |
| self.setIsMoblab(True) |
| moblab_rpc_interface.os = self.mox.CreateMockAnything() |
| |
| self.mox.StubOutWithMock(__builtin__, 'open') |
| self._mockReadFile(global_config.DEFAULT_CONFIG_FILE) |
| |
| self.mox.StubOutWithMock(lsbrelease_utils, 'is_moblab') |
| lsbrelease_utils.is_moblab().AndReturn(True) |
| |
| self._mockReadFile(global_config.DEFAULT_MOBLAB_FILE, |
| ['[section1]', 'item1: value1']) |
| |
| moblab_rpc_interface.os = self.mox.CreateMockAnything() |
| moblab_rpc_interface.os.path = self.mox.CreateMockAnything() |
| moblab_rpc_interface.os.path.exists( |
| moblab_rpc_interface._CONFIG.shadow_file).AndReturn( |
| True) |
| mockShadowFile = self.mox.CreateMockAnything() |
| mockShadowFileContents = StringIO.StringIO() |
| mockShadowFile.__enter__().AndReturn(mockShadowFileContents) |
| mockShadowFile.__exit__(mox.IgnoreArg(), mox.IgnoreArg(), |
| mox.IgnoreArg()) |
| open(moblab_rpc_interface._CONFIG.shadow_file, |
| 'w').AndReturn(mockShadowFile) |
| moblab_rpc_interface.os.system('sudo reboot') |
| |
| self.mox.ReplayAll() |
| moblab_rpc_interface.update_config_handler( |
| {'section1' : [('item1', 'value1'), |
| ('item2', 'value2')], |
| 'section2' : [('item3', 'value3'), |
| ('item4', 'value4')]}) |
| |
| # item1 should not be in the new shadow config as its updated value |
| # matches the original config's value. |
| self.assertEquals( |
| mockShadowFileContents.getvalue(), |
| '[section2]\nitem3 = value3\nitem4 = value4\n\n' |
| '[section1]\nitem2 = value2\n\n') |
| |
| |
| def testResetConfig(self): |
| """Ensure that reset opens the shadow_config file for writing.""" |
| self.setIsMoblab(True) |
| config_mock = self.mox.CreateMockAnything() |
| moblab_rpc_interface._CONFIG = config_mock |
| config_mock.shadow_file = 'shadow_config.ini' |
| self.mox.StubOutWithMock(__builtin__, 'open') |
| mockFile = self.mox.CreateMockAnything() |
| file_contents = self.mox.CreateMockAnything() |
| mockFile.__enter__().AndReturn(file_contents) |
| mockFile.__exit__(mox.IgnoreArg(), mox.IgnoreArg(), mox.IgnoreArg()) |
| open(config_mock.shadow_file, 'w').AndReturn(mockFile) |
| moblab_rpc_interface.os = self.mox.CreateMockAnything() |
| moblab_rpc_interface.os.system('sudo reboot') |
| self.mox.ReplayAll() |
| moblab_rpc_interface.reset_config_settings() |
| |
| |
| def testSetLaunchControlKey(self): |
| """Ensure that the Launch Control key path supplied is copied correctly. |
| """ |
| self.setIsMoblab(True) |
| launch_control_key = '/tmp/launch_control' |
| moblab_rpc_interface.os = self.mox.CreateMockAnything() |
| moblab_rpc_interface.os.path = self.mox.CreateMockAnything() |
| moblab_rpc_interface.os.path.exists(launch_control_key).AndReturn( |
| True) |
| moblab_rpc_interface.shutil = self.mox.CreateMockAnything() |
| moblab_rpc_interface.shutil.copyfile( |
| launch_control_key, |
| moblab_host.MOBLAB_LAUNCH_CONTROL_KEY_LOCATION) |
| moblab_rpc_interface.os.system('sudo restart moblab-devserver-init') |
| self.mox.ReplayAll() |
| moblab_rpc_interface.set_launch_control_key(launch_control_key) |
| |
| |
| def testGetNetworkInfo(self): |
| """Ensure the network info is properly converted to a dict.""" |
| self.setIsMoblab(True) |
| |
| self.mox.StubOutWithMock(moblab_rpc_interface, '_get_network_info') |
| moblab_rpc_interface._get_network_info().AndReturn(('10.0.0.1', True)) |
| self.mox.StubOutWithMock(rpc_utils, 'prepare_for_serialization') |
| |
| rpc_utils.prepare_for_serialization( |
| {'is_connected': True, 'server_ips': ['10.0.0.1']}) |
| self.mox.ReplayAll() |
| moblab_rpc_interface.get_network_info() |
| self.mox.VerifyAll() |
| |
| |
| def testGetNetworkInfoWithNoIp(self): |
| """Queries network info with no public IP address.""" |
| self.setIsMoblab(True) |
| |
| self.mox.StubOutWithMock(moblab_rpc_interface, '_get_network_info') |
| moblab_rpc_interface._get_network_info().AndReturn((None, False)) |
| self.mox.StubOutWithMock(rpc_utils, 'prepare_for_serialization') |
| |
| rpc_utils.prepare_for_serialization( |
| {'is_connected': False}) |
| self.mox.ReplayAll() |
| moblab_rpc_interface.get_network_info() |
| self.mox.VerifyAll() |
| |
| |
| def testGetNetworkInfoWithNoConnectivity(self): |
| """Queries network info with public IP address but no connectivity.""" |
| self.setIsMoblab(True) |
| |
| self.mox.StubOutWithMock(moblab_rpc_interface, '_get_network_info') |
| moblab_rpc_interface._get_network_info().AndReturn(('10.0.0.1', False)) |
| self.mox.StubOutWithMock(rpc_utils, 'prepare_for_serialization') |
| |
| rpc_utils.prepare_for_serialization( |
| {'is_connected': False, 'server_ips': ['10.0.0.1']}) |
| self.mox.ReplayAll() |
| moblab_rpc_interface.get_network_info() |
| self.mox.VerifyAll() |
| |
| |
| def testGetCloudStorageInfo(self): |
| """Ensure the cloud storage info is properly converted to a dict.""" |
| self.setIsMoblab(True) |
| config_mock = self.mox.CreateMockAnything() |
| moblab_rpc_interface._CONFIG = config_mock |
| config_mock.get_config_value( |
| 'CROS', 'image_storage_server').AndReturn('gs://bucket1') |
| config_mock.get_config_value( |
| 'CROS', 'results_storage_server', default=None).AndReturn( |
| 'gs://bucket2') |
| self.mox.StubOutWithMock(moblab_rpc_interface, '_get_boto_config') |
| moblab_rpc_interface._get_boto_config().AndReturn(config_mock) |
| config_mock.sections().AndReturn(['Credentials', 'b']) |
| config_mock.options('Credentials').AndReturn( |
| ['gs_access_key_id', 'gs_secret_access_key']) |
| config_mock.get( |
| 'Credentials', 'gs_access_key_id').AndReturn('key') |
| config_mock.get( |
| 'Credentials', 'gs_secret_access_key').AndReturn('secret') |
| rpc_utils.prepare_for_serialization( |
| { |
| 'gs_access_key_id': 'key', |
| 'gs_secret_access_key' : 'secret', |
| 'use_existing_boto_file': True, |
| 'image_storage_server' : 'gs://bucket1', |
| 'results_storage_server' : 'gs://bucket2' |
| }) |
| self.mox.ReplayAll() |
| moblab_rpc_interface.get_cloud_storage_info() |
| self.mox.VerifyAll() |
| |
| |
| def testValidateCloudStorageInfo(self): |
| """ Ensure the cloud storage info validation flow.""" |
| self.setIsMoblab(True) |
| cloud_storage_info = { |
| 'use_existing_boto_file': False, |
| 'gs_access_key_id': 'key', |
| 'gs_secret_access_key': 'secret', |
| 'image_storage_server': 'gs://bucket1', |
| 'results_storage_server': 'gs://bucket2'} |
| self.mox.StubOutWithMock(moblab_rpc_interface, |
| '_run_bucket_performance_test') |
| moblab_rpc_interface._run_bucket_performance_test( |
| 'key', 'secret', 'gs://bucket1').AndReturn((True, None)) |
| rpc_utils.prepare_for_serialization({'status_ok': True }) |
| self.mox.ReplayAll() |
| moblab_rpc_interface.validate_cloud_storage_info(cloud_storage_info) |
| self.mox.VerifyAll() |
| |
| |
| def testGetBucketNameFromUrl(self): |
| """Gets bucket name from bucket URL.""" |
| self.assertEquals( |
| 'bucket_name-123', |
| moblab_rpc_interface._get_bucket_name_from_url( |
| 'gs://bucket_name-123')) |
| self.assertEquals( |
| 'bucket_name-123', |
| moblab_rpc_interface._get_bucket_name_from_url( |
| 'gs://bucket_name-123/')) |
| self.assertEquals( |
| 'bucket_name-123', |
| moblab_rpc_interface._get_bucket_name_from_url( |
| 'gs://bucket_name-123/a/b/c')) |
| self.assertIsNone(moblab_rpc_interface._get_bucket_name_from_url( |
| 'bucket_name-123/a/b/c')) |
| |
| |
| def testGetShadowConfigFromPartialUpdate(self): |
| """Tests getting shadow configuration based on partial upate.""" |
| partial_config = { |
| 'section1': [ |
| ('opt1', 'value1'), |
| ('opt2', 'value2'), |
| ('opt3', 'value3'), |
| ('opt4', 'value4'), |
| ] |
| } |
| shadow_config_str = "[section1]\nopt2 = value2_1\nopt4 = value4_1" |
| shadow_config = ConfigParser.ConfigParser() |
| shadow_config.readfp(StringIO.StringIO(shadow_config_str)) |
| original_config = self.mox.CreateMockAnything() |
| self.mox.StubOutWithMock(moblab_rpc_interface, '_read_original_config') |
| self.mox.StubOutWithMock(moblab_rpc_interface, '_read_raw_config') |
| moblab_rpc_interface._read_original_config().AndReturn(original_config) |
| moblab_rpc_interface._read_raw_config( |
| moblab_rpc_interface._CONFIG.shadow_file).AndReturn(shadow_config) |
| original_config.get_config_value( |
| 'section1', 'opt1', |
| allow_blank=True, default='').AndReturn('value1') |
| original_config.get_config_value( |
| 'section1', 'opt2', |
| allow_blank=True, default='').AndReturn('value2') |
| original_config.get_config_value( |
| 'section1', 'opt3', |
| allow_blank=True, default='').AndReturn('blah') |
| original_config.get_config_value( |
| 'section1', 'opt4', |
| allow_blank=True, default='').AndReturn('blah') |
| self.mox.ReplayAll() |
| shadow_config = moblab_rpc_interface._get_shadow_config_from_partial_update( |
| partial_config) |
| # opt1 same as the original. |
| self.assertFalse(shadow_config.has_option('section1', 'opt1')) |
| # opt2 reverts back to original |
| self.assertFalse(shadow_config.has_option('section1', 'opt2')) |
| # opt3 is updated from original. |
| self.assertEquals('value3', shadow_config.get('section1', 'opt3')) |
| # opt3 in shadow but updated again. |
| self.assertEquals('value4', shadow_config.get('section1', 'opt4')) |
| self.mox.VerifyAll() |
| |
| |
| def testGetShadowConfigFromPartialUpdateWithNewSection(self): |
| """ |
| Test getting shadown configuration based on partial update with new section. |
| """ |
| partial_config = { |
| 'section2': [ |
| ('opt5', 'value5'), |
| ('opt6', 'value6'), |
| ], |
| } |
| shadow_config_str = "[section1]\nopt2 = value2_1\n" |
| shadow_config = ConfigParser.ConfigParser() |
| shadow_config.readfp(StringIO.StringIO(shadow_config_str)) |
| original_config = self.mox.CreateMockAnything() |
| self.mox.StubOutWithMock(moblab_rpc_interface, '_read_original_config') |
| self.mox.StubOutWithMock(moblab_rpc_interface, '_read_raw_config') |
| moblab_rpc_interface._read_original_config().AndReturn(original_config) |
| moblab_rpc_interface._read_raw_config( |
| moblab_rpc_interface._CONFIG.shadow_file).AndReturn(shadow_config) |
| original_config.get_config_value( |
| 'section2', 'opt5', |
| allow_blank=True, default='').AndReturn('value5') |
| original_config.get_config_value( |
| 'section2', 'opt6', |
| allow_blank=True, default='').AndReturn('blah') |
| self.mox.ReplayAll() |
| shadow_config = moblab_rpc_interface._get_shadow_config_from_partial_update( |
| partial_config) |
| # opt2 is still in shadow |
| self.assertEquals('value2_1', shadow_config.get('section1', 'opt2')) |
| # opt5 is not changed. |
| self.assertFalse(shadow_config.has_option('section2', 'opt5')) |
| # opt6 is updated. |
| self.assertEquals('value6', shadow_config.get('section2', 'opt6')) |
| self.mox.VerifyAll() |
| |
| def testGetBuildsForInDirectory(self): |
| config_mock = self.mox.CreateMockAnything() |
| moblab_rpc_interface._CONFIG = config_mock |
| config_mock.get_config_value( |
| 'CROS', 'image_storage_server').AndReturn('gs://bucket1/') |
| self.mox.StubOutWithMock(common_lib_utils, 'run') |
| output = self.mox.CreateMockAnything() |
| self.mox.StubOutWithMock(StringIO, 'StringIO', use_mock_anything=True) |
| StringIO.StringIO().AndReturn(output) |
| output.getvalue().AndReturn( |
| """gs://bucket1/dummy/R53-8480.0.0/\ngs://bucket1/dummy/R53-8530.72.0/\n |
| gs://bucket1/dummy/R54-8712.0.0/\ngs://bucket1/dummy/R54-8717.0.0/\n |
| gs://bucket1/dummy/R55-8759.0.0/\n |
| gs://bucket1/dummy/R55-8760.0.0-b5849/\n |
| gs://bucket1/dummy/R56-8995.0.0/\ngs://bucket1/dummy/R56-9001.0.0/\n |
| gs://bucket1/dummy/R57-9202.66.0/\ngs://bucket1/dummy/R58-9331.0.0/\n |
| gs://bucket1/dummy/R58-9334.15.0/\ngs://bucket1/dummy/R58-9334.17.0/\n |
| gs://bucket1/dummy/R58-9334.18.0/\ngs://bucket1/dummy/R58-9334.19.0/\n |
| gs://bucket1/dummy/R58-9334.22.0/\ngs://bucket1/dummy/R58-9334.28.0/\n |
| gs://bucket1/dummy/R58-9334.3.0/\ngs://bucket1/dummy/R58-9334.30.0/\n |
| gs://bucket1/dummy/R58-9334.36.0/\ngs://bucket1/dummy/R58-9334.55.0/\n |
| gs://bucket1/dummy/R58-9334.6.0/\ngs://bucket1/dummy/R58-9334.7.0/\n |
| gs://bucket1/dummy/R58-9334.9.0/\ngs://bucket1/dummy/R59-9346.0.0/\n |
| gs://bucket1/dummy/R59-9372.0.0/\ngs://bucket1/dummy/R59-9387.0.0/\n |
| gs://bucket1/dummy/R59-9436.0.0/\ngs://bucket1/dummy/R59-9452.0.0/\n |
| gs://bucket1/dummy/R59-9453.0.0/\ngs://bucket1/dummy/R59-9455.0.0/\n |
| gs://bucket1/dummy/R59-9460.0.0/\ngs://bucket1/dummy/R59-9460.11.0/\n |
| gs://bucket1/dummy/R59-9460.16.0/\ngs://bucket1/dummy/R59-9460.25.0/\n |
| gs://bucket1/dummy/R59-9460.8.0/\ngs://bucket1/dummy/R59-9460.9.0/\n |
| gs://bucket1/dummy/R60-9472.0.0/\ngs://bucket1/dummy/R60-9491.0.0/\n |
| gs://bucket1/dummy/R60-9492.0.0/\ngs://bucket1/dummy/R60-9497.0.0/\n |
| gs://bucket1/dummy/R60-9500.0.0/""") |
| |
| output.close() |
| |
| self.mox.StubOutWithMock(moblab_rpc_interface.GsUtil, 'get_gsutil_cmd') |
| moblab_rpc_interface.GsUtil.get_gsutil_cmd().AndReturn( |
| '/path/to/gsutil') |
| |
| common_lib_utils.run('/path/to/gsutil', |
| args=('ls', 'gs://bucket1/dummy'), |
| stdout_tee=mox.IgnoreArg()).AndReturn(output) |
| self.mox.ReplayAll() |
| expected_results = ['dummy/R60-9500.0.0', 'dummy/R60-9497.0.0', |
| 'dummy/R60-9492.0.0', 'dummy/R60-9491.0.0', 'dummy/R60-9472.0.0', |
| 'dummy/R59-9460.25.0', 'dummy/R59-9460.16.0', 'dummy/R59-9460.11.0', |
| 'dummy/R59-9460.9.0', 'dummy/R59-9460.8.0', 'dummy/R58-9334.55.0', |
| 'dummy/R58-9334.36.0', 'dummy/R58-9334.30.0', 'dummy/R58-9334.28.0', |
| 'dummy/R58-9334.22.0'] |
| actual_results = moblab_rpc_interface._get_builds_for_in_directory( |
| "dummy",3, 5) |
| self.assertEquals(expected_results, actual_results) |
| self.mox.VerifyAll() |
| |
| def testRunBucketPerformanceTestFail(self): |
| self.mox.StubOutWithMock(moblab_rpc_interface.GsUtil, 'get_gsutil_cmd') |
| moblab_rpc_interface.GsUtil.get_gsutil_cmd().AndReturn( |
| '/path/to/gsutil') |
| self.mox.StubOutWithMock(common_lib_utils, 'run') |
| common_lib_utils.run('/path/to/gsutil', |
| args=( |
| '-o', 'Credentials:gs_access_key_id=key', |
| '-o', 'Credentials:gs_secret_access_key=secret', |
| 'perfdiag', '-s', '1K', |
| '-o', 'testoutput', |
| '-n', '10', |
| 'gs://bucket1')).AndRaise( |
| error.CmdError("fakecommand", common_lib_utils.CmdResult(), |
| "xxxxxx<Error>yyyyyyyyyy</Error>")) |
| |
| self.mox.ReplayAll() |
| self.assertRaisesRegexp( |
| moblab_rpc_interface.BucketPerformanceTestException, |
| '<Error>yyyyyyyyyy', |
| moblab_rpc_interface._run_bucket_performance_test, |
| 'key', 'secret', 'gs://bucket1', '1K', '10', 'testoutput') |
| self.mox.VerifyAll() |
| |
| def testEnableNotificationUsingCredentialsInBucketFail(self): |
| config_mock = self.mox.CreateMockAnything() |
| moblab_rpc_interface._CONFIG = config_mock |
| config_mock.get_config_value( |
| 'CROS', 'image_storage_server').AndReturn('gs://bucket1/') |
| |
| self.mox.StubOutWithMock(moblab_rpc_interface.GsUtil, 'get_gsutil_cmd') |
| moblab_rpc_interface.GsUtil.get_gsutil_cmd().AndReturn( |
| '/path/to/gsutil') |
| |
| self.mox.StubOutWithMock(common_lib_utils, 'run') |
| common_lib_utils.run('/path/to/gsutil', |
| args=('cp', 'gs://bucket1/pubsub-key-do-not-delete.json', |
| '/tmp')).AndRaise( |
| error.CmdError("fakecommand", common_lib_utils.CmdResult(), "")) |
| self.mox.ReplayAll() |
| moblab_rpc_interface._enable_notification_using_credentials_in_bucket() |
| |
| def testEnableNotificationUsingCredentialsInBucketSuccess(self): |
| config_mock = self.mox.CreateMockAnything() |
| moblab_rpc_interface._CONFIG = config_mock |
| config_mock.get_config_value( |
| 'CROS', 'image_storage_server').AndReturn('gs://bucket1/') |
| |
| self.mox.StubOutWithMock(moblab_rpc_interface.GsUtil, 'get_gsutil_cmd') |
| moblab_rpc_interface.GsUtil.get_gsutil_cmd().AndReturn( |
| '/path/to/gsutil') |
| |
| self.mox.StubOutWithMock(common_lib_utils, 'run') |
| common_lib_utils.run('/path/to/gsutil', |
| args=('cp', 'gs://bucket1/pubsub-key-do-not-delete.json', |
| '/tmp')) |
| moblab_rpc_interface.shutil = self.mox.CreateMockAnything() |
| moblab_rpc_interface.shutil.copyfile( |
| '/tmp/pubsub-key-do-not-delete.json', |
| moblab_host.MOBLAB_SERVICE_ACCOUNT_LOCATION) |
| self.mox.StubOutWithMock(moblab_rpc_interface, '_update_partial_config') |
| moblab_rpc_interface._update_partial_config( |
| {'CROS': [(moblab_rpc_interface._CLOUD_NOTIFICATION_ENABLED, True)]} |
| ) |
| self.mox.ReplayAll() |
| moblab_rpc_interface._enable_notification_using_credentials_in_bucket() |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |
