| #!/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. |
| try: |
| import boto |
| except ImportError: |
| boto = None |
| import ConfigParser |
| import logging |
| 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 |
| |
| |
| 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 testSetBotoKey(self): |
| """Ensure that the botokey path supplied is copied correctly.""" |
| self.setIsMoblab(True) |
| boto_key = '/tmp/boto' |
| moblab_rpc_interface.os.path = self.mox.CreateMockAnything() |
| moblab_rpc_interface.os.path.exists(boto_key).AndReturn( |
| True) |
| moblab_rpc_interface.shutil = self.mox.CreateMockAnything() |
| moblab_rpc_interface.shutil.copyfile( |
| boto_key, moblab_rpc_interface.MOBLAB_BOTO_LOCATION) |
| self.mox.ReplayAll() |
| moblab_rpc_interface.set_boto_key(boto_key) |
| |
| |
| 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, '_is_valid_boto_key') |
| self.mox.StubOutWithMock(moblab_rpc_interface, '_is_valid_bucket') |
| moblab_rpc_interface._is_valid_boto_key( |
| 'key', 'secret').AndReturn((True, None)) |
| moblab_rpc_interface._is_valid_bucket( |
| 'key', 'secret', 'bucket1').AndReturn((True, None)) |
| moblab_rpc_interface._is_valid_bucket( |
| 'key', 'secret', 'bucket2').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 testIsValidBotoKeyValid(self): |
| """Tests the boto key validation flow.""" |
| if boto is None: |
| logging.info('skip test since boto module not installed') |
| return |
| conn = self.mox.CreateMockAnything() |
| self.mox.StubOutWithMock(boto, 'connect_gs') |
| boto.connect_gs('key', 'secret').AndReturn(conn) |
| conn.get_all_buckets().AndReturn(['a', 'b']) |
| conn.close() |
| self.mox.ReplayAll() |
| valid, details = moblab_rpc_interface._is_valid_boto_key('key', 'secret') |
| self.assertTrue(valid) |
| self.mox.VerifyAll() |
| |
| |
| def testIsValidBotoKeyInvalid(self): |
| """Tests the boto key validation with invalid key.""" |
| if boto is None: |
| logging.info('skip test since boto module not installed') |
| return |
| conn = self.mox.CreateMockAnything() |
| self.mox.StubOutWithMock(boto, 'connect_gs') |
| boto.connect_gs('key', 'secret').AndReturn(conn) |
| conn.get_all_buckets().AndRaise( |
| boto.exception.GSResponseError('bad', 'reason')) |
| conn.close() |
| self.mox.ReplayAll() |
| valid, details = moblab_rpc_interface._is_valid_boto_key('key', 'secret') |
| self.assertFalse(valid) |
| self.assertEquals('The boto access key is not valid', details) |
| self.mox.VerifyAll() |
| |
| |
| def testIsValidBucketValid(self): |
| """Tests the bucket vaildation flow.""" |
| if boto is None: |
| logging.info('skip test since boto module not installed') |
| return |
| conn = self.mox.CreateMockAnything() |
| self.mox.StubOutWithMock(boto, 'connect_gs') |
| boto.connect_gs('key', 'secret').AndReturn(conn) |
| conn.lookup('bucket').AndReturn('bucket') |
| conn.close() |
| self.mox.ReplayAll() |
| valid, details = moblab_rpc_interface._is_valid_bucket( |
| 'key', 'secret', 'bucket') |
| self.assertTrue(valid) |
| self.mox.VerifyAll() |
| |
| |
| def testIsValidBucketInvalid(self): |
| """Tests the bucket validation flow with invalid key.""" |
| if boto is None: |
| logging.info('skip test since boto module not installed') |
| return |
| conn = self.mox.CreateMockAnything() |
| self.mox.StubOutWithMock(boto, 'connect_gs') |
| boto.connect_gs('key', 'secret').AndReturn(conn) |
| conn.lookup('bucket').AndReturn(None) |
| conn.close() |
| self.mox.ReplayAll() |
| valid, details = moblab_rpc_interface._is_valid_bucket( |
| 'key', 'secret', 'bucket') |
| self.assertFalse(valid) |
| self.assertEquals("Bucket bucket does not exist.", details) |
| self.mox.VerifyAll() |
| |
| |
| 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() |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |
