| # Copyright 2015 The Chromium Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| # pylint: disable=unused-argument |
| |
| import os |
| import unittest |
| |
| from py_utils import cloud_storage |
| import mock |
| from pyfakefs import fake_filesystem_unittest |
| from pyfakefs import fake_filesystem |
| from pyfakefs import fake_filesystem_glob |
| |
| import dependency_manager |
| from dependency_manager import uploader |
| |
| |
| class BaseConfigCreationAndUpdateUnittests(fake_filesystem_unittest.TestCase): |
| def setUp(self): |
| self.addTypeEqualityFunc(uploader.CloudStorageUploader, |
| uploader.CloudStorageUploader.__eq__) |
| self.setUpPyfakefs() |
| self.dependencies = { |
| 'dep1': {'cloud_storage_bucket': 'bucket1', |
| 'cloud_storage_base_folder': 'dependencies_folder', |
| 'file_info': { |
| 'plat1': { |
| 'cloud_storage_hash': 'hash11', |
| 'download_path': '../../relative/dep1/path1'}, |
| 'plat2': { |
| 'cloud_storage_hash': 'hash12', |
| 'download_path': '../../relative/dep1/path2'}}}, |
| 'dep2': {'cloud_storage_bucket': 'bucket2', |
| 'file_info': { |
| 'plat1': { |
| 'cloud_storage_hash': 'hash21', |
| 'download_path': '../../relative/dep2/path1'}, |
| 'plat2': { |
| 'cloud_storage_hash': 'hash22', |
| 'download_path': '../../relative/dep2/path2'}}}} |
| |
| self.expected_file_lines = [ |
| # pylint: disable=bad-continuation |
| '{', '"config_type": "BaseConfig",', '"dependencies": {', |
| '"dep1": {', '"cloud_storage_base_folder": "dependencies_folder",', |
| '"cloud_storage_bucket": "bucket1",', '"file_info": {', |
| '"plat1": {', '"cloud_storage_hash": "hash11",', |
| '"download_path": "../../relative/dep1/path1"', '},', |
| '"plat2": {', '"cloud_storage_hash": "hash12",', |
| '"download_path": "../../relative/dep1/path2"', '}', '}', '},', |
| '"dep2": {', '"cloud_storage_bucket": "bucket2",', '"file_info": {', |
| '"plat1": {', '"cloud_storage_hash": "hash21",', |
| '"download_path": "../../relative/dep2/path1"', '},', |
| '"plat2": {', '"cloud_storage_hash": "hash22",', |
| '"download_path": "../../relative/dep2/path2"', '}', '}', '}', |
| '}', '}'] |
| |
| self.file_path = os.path.abspath(os.path.join( |
| 'path', 'to', 'config', 'file')) |
| |
| self.new_dep_path = 'path/to/new/dep' |
| self.fs.CreateFile(self.new_dep_path) |
| self.new_dep_hash = 'A23B56B7F23E798601F' |
| self.new_dependencies = { |
| 'dep1': {'cloud_storage_bucket': 'bucket1', |
| 'cloud_storage_base_folder': 'dependencies_folder', |
| 'file_info': { |
| 'plat1': { |
| 'cloud_storage_hash': 'hash11', |
| 'download_path': '../../relative/dep1/path1'}, |
| 'plat2': { |
| 'cloud_storage_hash': self.new_dep_hash, |
| 'download_path': '../../relative/dep1/path2'}}}, |
| 'dep2': {'cloud_storage_bucket': 'bucket2', |
| 'file_info': { |
| 'plat1': { |
| 'cloud_storage_hash': 'hash21', |
| 'download_path': '../../relative/dep2/path1'}, |
| 'plat2': { |
| 'cloud_storage_hash': 'hash22', |
| 'download_path': '../../relative/dep2/path2'}}}} |
| self.new_bucket = 'bucket1' |
| self.new_remote_path = 'dependencies_folder/dep1_%s' % self.new_dep_hash |
| self.new_pending_upload = uploader.CloudStorageUploader( |
| self.new_bucket, self.new_remote_path, self.new_dep_path) |
| self.expected_new_backup_path = '.'.join([self.new_remote_path, 'old']) |
| self.new_expected_file_lines = [ |
| # pylint: disable=bad-continuation |
| '{', '"config_type": "BaseConfig",', '"dependencies": {', |
| '"dep1": {', '"cloud_storage_base_folder": "dependencies_folder",', |
| '"cloud_storage_bucket": "bucket1",', '"file_info": {', |
| '"plat1": {', '"cloud_storage_hash": "hash11",', |
| '"download_path": "../../relative/dep1/path1"', '},', |
| '"plat2": {', '"cloud_storage_hash": "%s",' % self.new_dep_hash, |
| '"download_path": "../../relative/dep1/path2"', '}', '}', '},', |
| '"dep2": {', '"cloud_storage_bucket": "bucket2",', '"file_info": {', |
| '"plat1": {', '"cloud_storage_hash": "hash21",', |
| '"download_path": "../../relative/dep2/path1"', '},', |
| '"plat2": {', '"cloud_storage_hash": "hash22",', |
| '"download_path": "../../relative/dep2/path2"', '}', '}', '}', |
| '}', '}'] |
| |
| self.final_dep_path = 'path/to/final/dep' |
| self.fs.CreateFile(self.final_dep_path) |
| self.final_dep_hash = 'B34662F23B56B7F98601F' |
| self.final_bucket = 'bucket2' |
| self.final_remote_path = 'dep1_%s' % self.final_dep_hash |
| self.final_pending_upload = uploader.CloudStorageUploader( |
| self.final_bucket, self.final_remote_path, self.final_dep_path) |
| self.expected_final_backup_path = '.'.join([self.final_remote_path, |
| 'old']) |
| self.final_dependencies = { |
| 'dep1': {'cloud_storage_bucket': 'bucket1', |
| 'cloud_storage_base_folder': 'dependencies_folder', |
| 'file_info': { |
| 'plat1': { |
| 'cloud_storage_hash': 'hash11', |
| 'download_path': '../../relative/dep1/path1'}, |
| 'plat2': { |
| 'cloud_storage_hash': self.new_dep_hash, |
| 'download_path': '../../relative/dep1/path2'}}}, |
| 'dep2': {'cloud_storage_bucket': 'bucket2', |
| 'file_info': { |
| 'plat1': { |
| 'cloud_storage_hash': self.final_dep_hash, |
| 'download_path': '../../relative/dep2/path1'}, |
| 'plat2': { |
| 'cloud_storage_hash': 'hash22', |
| 'download_path': '../../relative/dep2/path2'}}}} |
| self.final_expected_file_lines = [ |
| # pylint: disable=bad-continuation |
| '{', '"config_type": "BaseConfig",', '"dependencies": {', |
| '"dep1": {', '"cloud_storage_base_folder": "dependencies_folder",', |
| '"cloud_storage_bucket": "bucket1",', '"file_info": {', |
| '"plat1": {', '"cloud_storage_hash": "hash11",', |
| '"download_path": "../../relative/dep1/path1"', '},', |
| '"plat2": {', '"cloud_storage_hash": "%s",' % self.new_dep_hash, |
| '"download_path": "../../relative/dep1/path2"', '}', '}', '},', |
| '"dep2": {', '"cloud_storage_bucket": "bucket2",', '"file_info": {', |
| '"plat1": {', '"cloud_storage_hash": "%s",' % self.final_dep_hash, |
| '"download_path": "../../relative/dep2/path1"', '},', |
| '"plat2": {', '"cloud_storage_hash": "hash22",', |
| '"download_path": "../../relative/dep2/path2"', '}', '}', '}', |
| '}', '}'] |
| |
| |
| def tearDown(self): |
| self.tearDownPyfakefs() |
| |
| # Init is not meant to be overridden, so we should be mocking the |
| # base_config's json module, even in subclasses. |
| def testCreateEmptyConfig(self): |
| expected_file_lines = ['{', |
| '"config_type": "BaseConfig",', |
| '"dependencies": {}', |
| '}'] |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| |
| file_module = fake_filesystem.FakeFileOpen(self.fs) |
| for line in file_module(self.file_path): |
| self.assertEqual(expected_file_lines.pop(0), line.strip()) |
| self.fs.CloseOpenFile(file_module(self.file_path)) |
| self.assertEqual({}, config._config_data) |
| self.assertEqual(self.file_path, config._config_path) |
| |
| def testCreateEmptyConfigError(self): |
| self.assertRaises(dependency_manager.EmptyConfigError, |
| dependency_manager.BaseConfig, self.file_path) |
| |
| def testCloudStorageRemotePath(self): |
| dependency = 'dep_name' |
| cs_hash = self.new_dep_hash |
| cs_base_folder = 'dependency_remote_folder' |
| expected_remote_path = '%s/%s_%s' % (cs_base_folder, dependency, cs_hash) |
| remote_path = dependency_manager.BaseConfig._CloudStorageRemotePath( |
| dependency, cs_hash, cs_base_folder) |
| self.assertEqual(expected_remote_path, remote_path) |
| |
| cs_base_folder = 'dependency_remote_folder' |
| expected_remote_path = '%s_%s' % (dependency, cs_hash) |
| remote_path = dependency_manager.BaseConfig._CloudStorageRemotePath( |
| dependency, cs_hash, cs_base_folder) |
| |
| def testGetEmptyJsonDict(self): |
| expected_json_dict = {'config_type': 'BaseConfig', |
| 'dependencies': {}} |
| json_dict = dependency_manager.BaseConfig._GetJsonDict() |
| self.assertEqual(expected_json_dict, json_dict) |
| |
| def testGetNonEmptyJsonDict(self): |
| expected_json_dict = {"config_type": "BaseConfig", |
| "dependencies": self.dependencies} |
| json_dict = dependency_manager.BaseConfig._GetJsonDict(self.dependencies) |
| self.assertEqual(expected_json_dict, json_dict) |
| |
| def testWriteEmptyConfigToFile(self): |
| expected_file_lines = ['{', '"config_type": "BaseConfig",', |
| '"dependencies": {}', '}'] |
| self.assertFalse(os.path.exists(self.file_path)) |
| dependency_manager.BaseConfig._WriteConfigToFile(self.file_path) |
| self.assertTrue(os.path.exists(self.file_path)) |
| file_module = fake_filesystem.FakeFileOpen(self.fs) |
| for line in file_module(self.file_path): |
| self.assertEqual(expected_file_lines.pop(0), line.strip()) |
| self.fs.CloseOpenFile(file_module(self.file_path)) |
| |
| def testWriteNonEmptyConfigToFile(self): |
| self.assertFalse(os.path.exists(self.file_path)) |
| dependency_manager.BaseConfig._WriteConfigToFile(self.file_path, |
| self.dependencies) |
| self.assertTrue(os.path.exists(self.file_path)) |
| expected_file_lines = list(self.expected_file_lines) |
| file_module = fake_filesystem.FakeFileOpen(self.fs) |
| for line in file_module(self.file_path): |
| self.assertEqual(expected_file_lines.pop(0), line.strip()) |
| self.fs.CloseOpenFile(file_module(self.file_path)) |
| |
| @mock.patch('dependency_manager.uploader.cloud_storage') |
| def testExecuteUpdateJobsNoOp(self, uploader_cs_mock): |
| self.fs.CreateFile(self.file_path, |
| contents='\n'.join(self.expected_file_lines)) |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| |
| self.assertFalse(config.ExecuteUpdateJobs()) |
| self.assertFalse(config._IsDirty()) |
| self.assertFalse(config._pending_uploads) |
| self.assertEqual(self.dependencies, config._config_data) |
| file_module = fake_filesystem.FakeFileOpen(self.fs) |
| expected_file_lines = list(self.expected_file_lines) |
| for line in file_module(self.file_path): |
| self.assertEqual(expected_file_lines.pop(0), line.strip()) |
| self.fs.CloseOpenFile(file_module(self.file_path)) |
| |
| @mock.patch('dependency_manager.uploader.cloud_storage') |
| def testExecuteUpdateJobsFailureOnInsertNoCSCollision( |
| self, uploader_cs_mock): |
| uploader_cs_mock.Exists.return_value = False |
| uploader_cs_mock.Insert.side_effect = cloud_storage.CloudStorageError |
| self.fs.CreateFile(self.file_path, |
| contents='\n'.join(self.expected_file_lines)) |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| config._config_data = self.new_dependencies.copy() |
| config._is_dirty = True |
| config._pending_uploads = [self.new_pending_upload] |
| self.assertEqual(self.new_dependencies, config._config_data) |
| self.assertTrue(config._is_dirty) |
| self.assertEqual(1, len(config._pending_uploads)) |
| self.assertEqual(self.new_pending_upload, config._pending_uploads[0]) |
| expected_exists_calls = [mock.call(self.new_bucket, self.new_remote_path)] |
| expected_insert_calls = [mock.call(self.new_bucket, self.new_remote_path, |
| self.new_dep_path)] |
| expected_copy_calls = [] |
| expected_delete_calls = [] |
| |
| self.assertRaises(cloud_storage.CloudStorageError, |
| config.ExecuteUpdateJobs) |
| self.assertTrue(config._is_dirty) |
| self.assertEqual(1, len(config._pending_uploads)) |
| self.assertEqual(self.new_pending_upload, config._pending_uploads[0]) |
| self.assertEqual(self.new_dependencies, config._config_data) |
| file_module = fake_filesystem.FakeFileOpen(self.fs) |
| expected_file_lines = list(self.expected_file_lines) |
| for line in file_module(self.file_path): |
| self.assertEqual(expected_file_lines.pop(0), line.strip()) |
| self.fs.CloseOpenFile(file_module(self.file_path)) |
| self.assertEqual(1, len(config._pending_uploads)) |
| self.assertEqual(self.new_pending_upload, config._pending_uploads[0]) |
| self.assertEqual(expected_insert_calls, |
| uploader_cs_mock.Insert.call_args_list) |
| self.assertEqual(expected_exists_calls, |
| uploader_cs_mock.Exists.call_args_list) |
| self.assertEqual(expected_copy_calls, |
| uploader_cs_mock.Copy.call_args_list) |
| self.assertEqual(expected_delete_calls, |
| uploader_cs_mock.Delete.call_args_list) |
| |
| @mock.patch('dependency_manager.uploader.cloud_storage') |
| def testExecuteUpdateJobsFailureOnInsertCSCollisionForce( |
| self, uploader_cs_mock): |
| uploader_cs_mock.Exists.return_value = True |
| uploader_cs_mock.Insert.side_effect = cloud_storage.CloudStorageError |
| self.fs.CreateFile(self.file_path, |
| contents='\n'.join(self.expected_file_lines)) |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| config._config_data = self.new_dependencies.copy() |
| config._is_dirty = True |
| config._pending_uploads = [self.new_pending_upload] |
| self.assertEqual(self.new_dependencies, config._config_data) |
| self.assertTrue(config._is_dirty) |
| self.assertEqual(1, len(config._pending_uploads)) |
| self.assertEqual(self.new_pending_upload, config._pending_uploads[0]) |
| expected_exists_calls = [mock.call(self.new_bucket, self.new_remote_path)] |
| expected_insert_calls = [mock.call(self.new_bucket, self.new_remote_path, |
| self.new_dep_path)] |
| expected_copy_calls = [mock.call(self.new_bucket, self.new_bucket, |
| self.new_remote_path, |
| self.expected_new_backup_path), |
| mock.call(self.new_bucket, self.new_bucket, |
| self.expected_new_backup_path, |
| self.new_remote_path)] |
| expected_delete_calls = [] |
| |
| self.assertRaises(cloud_storage.CloudStorageError, |
| config.ExecuteUpdateJobs, force=True) |
| self.assertTrue(config._is_dirty) |
| self.assertEqual(1, len(config._pending_uploads)) |
| self.assertEqual(self.new_pending_upload, config._pending_uploads[0]) |
| self.assertEqual(self.new_dependencies, config._config_data) |
| file_module = fake_filesystem.FakeFileOpen(self.fs) |
| expected_file_lines = list(self.expected_file_lines) |
| for line in file_module(self.file_path): |
| self.assertEqual(expected_file_lines.pop(0), line.strip()) |
| self.fs.CloseOpenFile(file_module(self.file_path)) |
| self.assertEqual(1, len(config._pending_uploads)) |
| self.assertEqual(self.new_pending_upload, config._pending_uploads[0]) |
| self.assertEqual(expected_insert_calls, |
| uploader_cs_mock.Insert.call_args_list) |
| self.assertEqual(expected_exists_calls, |
| uploader_cs_mock.Exists.call_args_list) |
| self.assertEqual(expected_copy_calls, |
| uploader_cs_mock.Copy.call_args_list) |
| self.assertEqual(expected_delete_calls, |
| uploader_cs_mock.Delete.call_args_list) |
| |
| @mock.patch('dependency_manager.uploader.cloud_storage') |
| def testExecuteUpdateJobsFailureOnInsertCSCollisionNoForce( |
| self, uploader_cs_mock): |
| uploader_cs_mock.Exists.return_value = True |
| uploader_cs_mock.Insert.side_effect = cloud_storage.CloudStorageError |
| self.fs.CreateFile(self.file_path, |
| contents='\n'.join(self.expected_file_lines)) |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| config._config_data = self.new_dependencies.copy() |
| config._is_dirty = True |
| config._pending_uploads = [self.new_pending_upload] |
| self.assertEqual(self.new_dependencies, config._config_data) |
| self.assertTrue(config._is_dirty) |
| self.assertEqual(1, len(config._pending_uploads)) |
| self.assertEqual(self.new_pending_upload, config._pending_uploads[0]) |
| expected_exists_calls = [mock.call(self.new_bucket, self.new_remote_path)] |
| expected_insert_calls = [] |
| expected_copy_calls = [] |
| expected_delete_calls = [] |
| |
| self.assertRaises(cloud_storage.CloudStorageError, |
| config.ExecuteUpdateJobs) |
| self.assertTrue(config._is_dirty) |
| self.assertEqual(1, len(config._pending_uploads)) |
| self.assertEqual(self.new_pending_upload, config._pending_uploads[0]) |
| self.assertEqual(self.new_dependencies, config._config_data) |
| file_module = fake_filesystem.FakeFileOpen(self.fs) |
| expected_file_lines = list(self.expected_file_lines) |
| for line in file_module(self.file_path): |
| self.assertEqual(expected_file_lines.pop(0), line.strip()) |
| self.fs.CloseOpenFile(file_module(self.file_path)) |
| self.assertEqual(1, len(config._pending_uploads)) |
| self.assertEqual(self.new_pending_upload, config._pending_uploads[0]) |
| self.assertEqual(expected_insert_calls, |
| uploader_cs_mock.Insert.call_args_list) |
| self.assertEqual(expected_exists_calls, |
| uploader_cs_mock.Exists.call_args_list) |
| self.assertEqual(expected_copy_calls, |
| uploader_cs_mock.Copy.call_args_list) |
| self.assertEqual(expected_delete_calls, |
| uploader_cs_mock.Delete.call_args_list) |
| |
| @mock.patch('dependency_manager.uploader.cloud_storage') |
| def testExecuteUpdateJobsFailureOnCopy( |
| self, uploader_cs_mock): |
| uploader_cs_mock.Exists.return_value = True |
| uploader_cs_mock.Copy.side_effect = cloud_storage.CloudStorageError |
| self.fs.CreateFile(self.file_path, |
| contents='\n'.join(self.expected_file_lines)) |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| config._config_data = self.new_dependencies.copy() |
| config._is_dirty = True |
| config._pending_uploads = [self.new_pending_upload] |
| self.assertEqual(self.new_dependencies, config._config_data) |
| self.assertTrue(config._is_dirty) |
| self.assertEqual(1, len(config._pending_uploads)) |
| self.assertEqual(self.new_pending_upload, config._pending_uploads[0]) |
| expected_exists_calls = [mock.call(self.new_bucket, self.new_remote_path)] |
| expected_insert_calls = [] |
| expected_copy_calls = [mock.call(self.new_bucket, self.new_bucket, |
| self.new_remote_path, |
| self.expected_new_backup_path)] |
| expected_delete_calls = [] |
| |
| self.assertRaises(cloud_storage.CloudStorageError, |
| config.ExecuteUpdateJobs, force=True) |
| self.assertTrue(config._is_dirty) |
| self.assertEqual(1, len(config._pending_uploads)) |
| self.assertEqual(self.new_pending_upload, config._pending_uploads[0]) |
| self.assertEqual(self.new_dependencies, config._config_data) |
| file_module = fake_filesystem.FakeFileOpen(self.fs) |
| expected_file_lines = list(self.expected_file_lines) |
| for line in file_module(self.file_path): |
| self.assertEqual(expected_file_lines.pop(0), line.strip()) |
| self.fs.CloseOpenFile(file_module(self.file_path)) |
| self.assertEqual(1, len(config._pending_uploads)) |
| self.assertEqual(self.new_pending_upload, config._pending_uploads[0]) |
| self.assertEqual(expected_insert_calls, |
| uploader_cs_mock.Insert.call_args_list) |
| self.assertEqual(expected_exists_calls, |
| uploader_cs_mock.Exists.call_args_list) |
| self.assertEqual(expected_copy_calls, |
| uploader_cs_mock.Copy.call_args_list) |
| self.assertEqual(expected_delete_calls, |
| uploader_cs_mock.Delete.call_args_list) |
| |
| @mock.patch('dependency_manager.uploader.cloud_storage') |
| def testExecuteUpdateJobsFailureOnSecondInsertNoCSCollision( |
| self, uploader_cs_mock): |
| uploader_cs_mock.Exists.return_value = False |
| uploader_cs_mock.Insert.side_effect = [ |
| True, cloud_storage.CloudStorageError] |
| self.fs.CreateFile(self.file_path, |
| contents='\n'.join(self.expected_file_lines)) |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| config._config_data = self.new_dependencies.copy() |
| config._is_dirty = True |
| config._pending_uploads = [self.new_pending_upload, |
| self.final_pending_upload] |
| self.assertEqual(self.new_dependencies, config._config_data) |
| self.assertTrue(config._is_dirty) |
| self.assertEqual(2, len(config._pending_uploads)) |
| self.assertEqual(self.new_pending_upload, config._pending_uploads[0]) |
| self.assertEqual(self.final_pending_upload, config._pending_uploads[1]) |
| expected_exists_calls = [mock.call(self.new_bucket, self.new_remote_path), |
| mock.call(self.final_bucket, |
| self.final_remote_path)] |
| expected_insert_calls = [mock.call(self.new_bucket, self.new_remote_path, |
| self.new_dep_path), |
| mock.call(self.final_bucket, |
| self.final_remote_path, |
| self.final_dep_path)] |
| expected_copy_calls = [] |
| expected_delete_calls = [mock.call(self.new_bucket, self.new_remote_path)] |
| |
| self.assertRaises(cloud_storage.CloudStorageError, |
| config.ExecuteUpdateJobs) |
| self.assertTrue(config._is_dirty) |
| self.assertEqual(2, len(config._pending_uploads)) |
| self.assertEqual(self.new_pending_upload, config._pending_uploads[0]) |
| self.assertEqual(self.final_pending_upload, config._pending_uploads[1]) |
| self.assertEqual(self.new_dependencies, config._config_data) |
| file_module = fake_filesystem.FakeFileOpen(self.fs) |
| expected_file_lines = list(self.expected_file_lines) |
| for line in file_module(self.file_path): |
| self.assertEqual(expected_file_lines.pop(0), line.strip()) |
| self.fs.CloseOpenFile(file_module(self.file_path)) |
| self.assertEqual(expected_insert_calls, |
| uploader_cs_mock.Insert.call_args_list) |
| self.assertEqual(expected_exists_calls, |
| uploader_cs_mock.Exists.call_args_list) |
| self.assertEqual(expected_copy_calls, |
| uploader_cs_mock.Copy.call_args_list) |
| self.assertEqual(expected_delete_calls, |
| uploader_cs_mock.Delete.call_args_list) |
| |
| @mock.patch('dependency_manager.uploader.cloud_storage') |
| def testExecuteUpdateJobsFailureOnSecondInsertCSCollisionForce( |
| self, uploader_cs_mock): |
| uploader_cs_mock.Exists.return_value = True |
| uploader_cs_mock.Insert.side_effect = [ |
| True, cloud_storage.CloudStorageError] |
| self.fs.CreateFile(self.file_path, |
| contents='\n'.join(self.expected_file_lines)) |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| config._config_data = self.new_dependencies.copy() |
| config._is_dirty = True |
| config._pending_uploads = [self.new_pending_upload, |
| self.final_pending_upload] |
| self.assertEqual(self.new_dependencies, config._config_data) |
| self.assertTrue(config._is_dirty) |
| self.assertEqual(2, len(config._pending_uploads)) |
| self.assertEqual(self.new_pending_upload, config._pending_uploads[0]) |
| self.assertEqual(self.final_pending_upload, config._pending_uploads[1]) |
| expected_exists_calls = [mock.call(self.new_bucket, self.new_remote_path), |
| mock.call(self.final_bucket, |
| self.final_remote_path)] |
| expected_insert_calls = [mock.call(self.new_bucket, self.new_remote_path, |
| self.new_dep_path), |
| mock.call(self.final_bucket, |
| self.final_remote_path, |
| self.final_dep_path)] |
| expected_copy_calls = [mock.call(self.new_bucket, self.new_bucket, |
| self.new_remote_path, |
| self.expected_new_backup_path), |
| mock.call(self.final_bucket, self.final_bucket, |
| self.final_remote_path, |
| self.expected_final_backup_path), |
| mock.call(self.final_bucket, self.final_bucket, |
| self.expected_final_backup_path, |
| self.final_remote_path), |
| mock.call(self.new_bucket, self.new_bucket, |
| self.expected_new_backup_path, |
| self.new_remote_path)] |
| expected_delete_calls = [] |
| |
| self.assertRaises(cloud_storage.CloudStorageError, |
| config.ExecuteUpdateJobs, force=True) |
| self.assertTrue(config._is_dirty) |
| self.assertEqual(2, len(config._pending_uploads)) |
| self.assertEqual(self.new_pending_upload, config._pending_uploads[0]) |
| self.assertEqual(self.final_pending_upload, config._pending_uploads[1]) |
| self.assertEqual(self.new_dependencies, config._config_data) |
| file_module = fake_filesystem.FakeFileOpen(self.fs) |
| expected_file_lines = list(self.expected_file_lines) |
| for line in file_module(self.file_path): |
| self.assertEqual(expected_file_lines.pop(0), line.strip()) |
| self.fs.CloseOpenFile(file_module(self.file_path)) |
| self.assertEqual(expected_insert_calls, |
| uploader_cs_mock.Insert.call_args_list) |
| self.assertEqual(expected_exists_calls, |
| uploader_cs_mock.Exists.call_args_list) |
| self.assertEqual(expected_copy_calls, |
| uploader_cs_mock.Copy.call_args_list) |
| self.assertEqual(expected_delete_calls, |
| uploader_cs_mock.Delete.call_args_list) |
| |
| @mock.patch('dependency_manager.uploader.cloud_storage') |
| def testExecuteUpdateJobsFailureOnSecondInsertFirstCSCollisionForce( |
| self, uploader_cs_mock): |
| uploader_cs_mock.Exists.side_effect = [True, False, True] |
| uploader_cs_mock.Insert.side_effect = [ |
| True, cloud_storage.CloudStorageError] |
| self.fs.CreateFile(self.file_path, |
| contents='\n'.join(self.expected_file_lines)) |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| config._config_data = self.new_dependencies.copy() |
| config._is_dirty = True |
| config._pending_uploads = [self.new_pending_upload, |
| self.final_pending_upload] |
| self.assertEqual(self.new_dependencies, config._config_data) |
| self.assertTrue(config._is_dirty) |
| self.assertEqual(2, len(config._pending_uploads)) |
| self.assertEqual(self.new_pending_upload, config._pending_uploads[0]) |
| self.assertEqual(self.final_pending_upload, config._pending_uploads[1]) |
| expected_exists_calls = [mock.call(self.new_bucket, self.new_remote_path), |
| mock.call(self.final_bucket, |
| self.final_remote_path)] |
| expected_insert_calls = [mock.call(self.new_bucket, self.new_remote_path, |
| self.new_dep_path), |
| mock.call(self.final_bucket, |
| self.final_remote_path, |
| self.final_dep_path)] |
| expected_copy_calls = [mock.call(self.new_bucket, self.new_bucket, |
| self.new_remote_path, |
| self.expected_new_backup_path), |
| mock.call(self.new_bucket, self.new_bucket, |
| self.expected_new_backup_path, |
| self.new_remote_path)] |
| expected_delete_calls = [] |
| |
| self.assertRaises(cloud_storage.CloudStorageError, |
| config.ExecuteUpdateJobs, force=True) |
| self.assertTrue(config._is_dirty) |
| self.assertEqual(2, len(config._pending_uploads)) |
| self.assertEqual(self.new_pending_upload, config._pending_uploads[0]) |
| self.assertEqual(self.final_pending_upload, config._pending_uploads[1]) |
| self.assertEqual(self.new_dependencies, config._config_data) |
| file_module = fake_filesystem.FakeFileOpen(self.fs) |
| expected_file_lines = list(self.expected_file_lines) |
| for line in file_module(self.file_path): |
| self.assertEqual(expected_file_lines.pop(0), line.strip()) |
| self.fs.CloseOpenFile(file_module(self.file_path)) |
| self.assertEqual(expected_insert_calls, |
| uploader_cs_mock.Insert.call_args_list) |
| self.assertEqual(expected_exists_calls, |
| uploader_cs_mock.Exists.call_args_list) |
| self.assertEqual(expected_copy_calls, |
| uploader_cs_mock.Copy.call_args_list) |
| self.assertEqual(expected_delete_calls, |
| uploader_cs_mock.Delete.call_args_list) |
| |
| @mock.patch('dependency_manager.uploader.cloud_storage') |
| def testExecuteUpdateJobsFailureOnFirstCSCollisionNoForce( |
| self, uploader_cs_mock): |
| uploader_cs_mock.Exists.side_effect = [True, False, True] |
| uploader_cs_mock.Insert.side_effect = [ |
| True, cloud_storage.CloudStorageError] |
| self.fs.CreateFile(self.file_path, |
| contents='\n'.join(self.expected_file_lines)) |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| config._config_data = self.new_dependencies.copy() |
| config._is_dirty = True |
| config._pending_uploads = [self.new_pending_upload, |
| self.final_pending_upload] |
| self.assertEqual(self.new_dependencies, config._config_data) |
| self.assertTrue(config._is_dirty) |
| self.assertEqual(2, len(config._pending_uploads)) |
| self.assertEqual(self.new_pending_upload, config._pending_uploads[0]) |
| self.assertEqual(self.final_pending_upload, config._pending_uploads[1]) |
| expected_exists_calls = [mock.call(self.new_bucket, self.new_remote_path)] |
| expected_insert_calls = [] |
| expected_copy_calls = [] |
| expected_delete_calls = [] |
| |
| self.assertRaises(cloud_storage.CloudStorageError, |
| config.ExecuteUpdateJobs) |
| self.assertTrue(config._is_dirty) |
| self.assertEqual(2, len(config._pending_uploads)) |
| self.assertEqual(self.new_pending_upload, config._pending_uploads[0]) |
| self.assertEqual(self.final_pending_upload, config._pending_uploads[1]) |
| self.assertEqual(self.new_dependencies, config._config_data) |
| file_module = fake_filesystem.FakeFileOpen(self.fs) |
| expected_file_lines = list(self.expected_file_lines) |
| for line in file_module(self.file_path): |
| self.assertEqual(expected_file_lines.pop(0), line.strip()) |
| self.fs.CloseOpenFile(file_module(self.file_path)) |
| self.assertEqual(expected_insert_calls, |
| uploader_cs_mock.Insert.call_args_list) |
| self.assertEqual(expected_exists_calls, |
| uploader_cs_mock.Exists.call_args_list) |
| self.assertEqual(expected_copy_calls, |
| uploader_cs_mock.Copy.call_args_list) |
| self.assertEqual(expected_delete_calls, |
| uploader_cs_mock.Delete.call_args_list) |
| |
| @mock.patch('dependency_manager.uploader.cloud_storage') |
| def testExecuteUpdateJobsFailureOnSecondCopyCSCollision( |
| self, uploader_cs_mock): |
| uploader_cs_mock.Exists.return_value = True |
| uploader_cs_mock.Insert.return_value = True |
| uploader_cs_mock.Copy.side_effect = [ |
| True, cloud_storage.CloudStorageError, True] |
| self.fs.CreateFile(self.file_path, |
| contents='\n'.join(self.expected_file_lines)) |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| config._config_data = self.new_dependencies.copy() |
| config._is_dirty = True |
| config._pending_uploads = [self.new_pending_upload, |
| self.final_pending_upload] |
| self.assertEqual(self.new_dependencies, config._config_data) |
| self.assertTrue(config._is_dirty) |
| self.assertEqual(2, len(config._pending_uploads)) |
| self.assertEqual(self.new_pending_upload, config._pending_uploads[0]) |
| self.assertEqual(self.final_pending_upload, config._pending_uploads[1]) |
| expected_exists_calls = [mock.call(self.new_bucket, self.new_remote_path), |
| mock.call(self.final_bucket, |
| self.final_remote_path)] |
| expected_insert_calls = [mock.call(self.new_bucket, self.new_remote_path, |
| self.new_dep_path)] |
| expected_copy_calls = [mock.call(self.new_bucket, self.new_bucket, |
| self.new_remote_path, |
| self.expected_new_backup_path), |
| mock.call(self.final_bucket, self.final_bucket, |
| self.final_remote_path, |
| self.expected_final_backup_path), |
| mock.call(self.new_bucket, self.new_bucket, |
| self.expected_new_backup_path, |
| self.new_remote_path)] |
| expected_delete_calls = [] |
| |
| self.assertRaises(cloud_storage.CloudStorageError, |
| config.ExecuteUpdateJobs, force=True) |
| self.assertTrue(config._is_dirty) |
| self.assertEqual(2, len(config._pending_uploads)) |
| self.assertEqual(self.new_pending_upload, config._pending_uploads[0]) |
| self.assertEqual(self.final_pending_upload, config._pending_uploads[1]) |
| self.assertEqual(self.new_dependencies, config._config_data) |
| file_module = fake_filesystem.FakeFileOpen(self.fs) |
| expected_file_lines = list(self.expected_file_lines) |
| for line in file_module(self.file_path): |
| self.assertEqual(expected_file_lines.pop(0), line.strip()) |
| self.fs.CloseOpenFile(file_module(self.file_path)) |
| self.assertEqual(expected_insert_calls, |
| uploader_cs_mock.Insert.call_args_list) |
| self.assertEqual(expected_exists_calls, |
| uploader_cs_mock.Exists.call_args_list) |
| self.assertEqual(expected_copy_calls, |
| uploader_cs_mock.Copy.call_args_list) |
| self.assertEqual(expected_delete_calls, |
| uploader_cs_mock.Delete.call_args_list) |
| |
| @mock.patch('dependency_manager.uploader.cloud_storage') |
| def testExecuteUpdateJobsFailureOnSecondCopyNoCSCollisionForce( |
| self, uploader_cs_mock): |
| uploader_cs_mock.Exists.side_effect = [False, True, False] |
| uploader_cs_mock.Copy.side_effect = cloud_storage.CloudStorageError |
| self.fs.CreateFile(self.file_path, |
| contents='\n'.join(self.expected_file_lines)) |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| config._config_data = self.new_dependencies.copy() |
| config._is_dirty = True |
| config._pending_uploads = [self.new_pending_upload, |
| self.final_pending_upload] |
| self.assertEqual(self.new_dependencies, config._config_data) |
| self.assertTrue(config._is_dirty) |
| self.assertEqual(2, len(config._pending_uploads)) |
| self.assertEqual(self.new_pending_upload, config._pending_uploads[0]) |
| self.assertEqual(self.final_pending_upload, config._pending_uploads[1]) |
| expected_exists_calls = [mock.call(self.new_bucket, self.new_remote_path), |
| mock.call(self.final_bucket, |
| self.final_remote_path)] |
| expected_insert_calls = [mock.call(self.new_bucket, self.new_remote_path, |
| self.new_dep_path)] |
| expected_copy_calls = [mock.call(self.final_bucket, self.final_bucket, |
| self.final_remote_path, |
| self.expected_final_backup_path)] |
| expected_delete_calls = [mock.call(self.new_bucket, self.new_remote_path)] |
| |
| self.assertRaises(cloud_storage.CloudStorageError, |
| config.ExecuteUpdateJobs, force=True) |
| self.assertTrue(config._is_dirty) |
| self.assertEqual(2, len(config._pending_uploads)) |
| self.assertEqual(self.new_pending_upload, config._pending_uploads[0]) |
| self.assertEqual(self.final_pending_upload, config._pending_uploads[1]) |
| self.assertEqual(self.new_dependencies, config._config_data) |
| file_module = fake_filesystem.FakeFileOpen(self.fs) |
| expected_file_lines = list(self.expected_file_lines) |
| for line in file_module(self.file_path): |
| self.assertEqual(expected_file_lines.pop(0), line.strip()) |
| self.fs.CloseOpenFile(file_module(self.file_path)) |
| self.assertEqual(expected_insert_calls, |
| uploader_cs_mock.Insert.call_args_list) |
| self.assertEqual(expected_exists_calls, |
| uploader_cs_mock.Exists.call_args_list) |
| self.assertEqual(expected_copy_calls, |
| uploader_cs_mock.Copy.call_args_list) |
| self.assertEqual(expected_delete_calls, |
| uploader_cs_mock.Delete.call_args_list) |
| |
| @mock.patch('dependency_manager.uploader.cloud_storage') |
| def testExecuteUpdateJobsFailureOnSecondCopyNoCSCollisionNoForce( |
| self, uploader_cs_mock): |
| uploader_cs_mock.Exists.side_effect = [False, True, False] |
| uploader_cs_mock.Copy.side_effect = cloud_storage.CloudStorageError |
| self.fs.CreateFile(self.file_path, |
| contents='\n'.join(self.expected_file_lines)) |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| config._config_data = self.new_dependencies.copy() |
| config._is_dirty = True |
| config._pending_uploads = [self.new_pending_upload, |
| self.final_pending_upload] |
| self.assertEqual(self.new_dependencies, config._config_data) |
| self.assertTrue(config._is_dirty) |
| self.assertEqual(2, len(config._pending_uploads)) |
| self.assertEqual(self.new_pending_upload, config._pending_uploads[0]) |
| self.assertEqual(self.final_pending_upload, config._pending_uploads[1]) |
| expected_exists_calls = [mock.call(self.new_bucket, self.new_remote_path), |
| mock.call(self.final_bucket, |
| self.final_remote_path)] |
| expected_insert_calls = [mock.call(self.new_bucket, self.new_remote_path, |
| self.new_dep_path)] |
| expected_copy_calls = [] |
| expected_delete_calls = [mock.call(self.new_bucket, self.new_remote_path)] |
| |
| self.assertRaises(cloud_storage.CloudStorageError, |
| config.ExecuteUpdateJobs) |
| self.assertTrue(config._is_dirty) |
| self.assertEqual(2, len(config._pending_uploads)) |
| self.assertEqual(self.new_pending_upload, config._pending_uploads[0]) |
| self.assertEqual(self.final_pending_upload, config._pending_uploads[1]) |
| self.assertEqual(self.new_dependencies, config._config_data) |
| file_module = fake_filesystem.FakeFileOpen(self.fs) |
| expected_file_lines = list(self.expected_file_lines) |
| for line in file_module(self.file_path): |
| self.assertEqual(expected_file_lines.pop(0), line.strip()) |
| self.fs.CloseOpenFile(file_module(self.file_path)) |
| self.assertEqual(expected_insert_calls, |
| uploader_cs_mock.Insert.call_args_list) |
| self.assertEqual(expected_exists_calls, |
| uploader_cs_mock.Exists.call_args_list) |
| self.assertEqual(expected_copy_calls, |
| uploader_cs_mock.Copy.call_args_list) |
| self.assertEqual(expected_delete_calls, |
| uploader_cs_mock.Delete.call_args_list) |
| |
| @mock.patch('dependency_manager.uploader.cloud_storage') |
| def testExecuteUpdateJobsSuccessOnePendingDepNoCloudStorageCollision( |
| self, uploader_cs_mock): |
| uploader_cs_mock.Exists.return_value = False |
| self.fs.CreateFile(self.file_path, |
| contents='\n'.join(self.expected_file_lines)) |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| config._config_data = self.new_dependencies.copy() |
| config._pending_uploads = [self.new_pending_upload] |
| self.assertEqual(self.new_dependencies, config._config_data) |
| self.assertTrue(config._IsDirty()) |
| self.assertEqual(1, len(config._pending_uploads)) |
| self.assertEqual(self.new_pending_upload, config._pending_uploads[0]) |
| expected_exists_calls = [mock.call(self.new_bucket, self.new_remote_path)] |
| expected_insert_calls = [mock.call(self.new_bucket, self.new_remote_path, |
| self.new_dep_path)] |
| expected_copy_calls = [] |
| expected_delete_calls = [] |
| |
| self.assertTrue(config.ExecuteUpdateJobs()) |
| self.assertFalse(config._IsDirty()) |
| self.assertFalse(config._pending_uploads) |
| self.assertEqual(self.new_dependencies, config._config_data) |
| file_module = fake_filesystem.FakeFileOpen(self.fs) |
| expected_file_lines = list(self.new_expected_file_lines) |
| for line in file_module(self.file_path): |
| self.assertEqual(expected_file_lines.pop(0), line.strip()) |
| self.fs.CloseOpenFile(file_module(self.file_path)) |
| self.assertFalse(config._pending_uploads) |
| self.assertEqual(expected_insert_calls, |
| uploader_cs_mock.Insert.call_args_list) |
| self.assertEqual(expected_exists_calls, |
| uploader_cs_mock.Exists.call_args_list) |
| self.assertEqual(expected_copy_calls, |
| uploader_cs_mock.Copy.call_args_list) |
| self.assertEqual(expected_delete_calls, |
| uploader_cs_mock.Delete.call_args_list) |
| |
| @mock.patch('dependency_manager.uploader.cloud_storage') |
| def testExecuteUpdateJobsSuccessOnePendingDepCloudStorageCollision( |
| self, uploader_cs_mock): |
| uploader_cs_mock.Exists.return_value = True |
| self.fs.CreateFile(self.file_path, |
| contents='\n'.join(self.expected_file_lines)) |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| config._config_data = self.new_dependencies.copy() |
| config._pending_uploads = [self.new_pending_upload] |
| self.assertEqual(self.new_dependencies, config._config_data) |
| self.assertTrue(config._IsDirty()) |
| self.assertEqual(1, len(config._pending_uploads)) |
| self.assertEqual(self.new_pending_upload, config._pending_uploads[0]) |
| expected_exists_calls = [mock.call(self.new_bucket, self.new_remote_path)] |
| expected_insert_calls = [mock.call(self.new_bucket, self.new_remote_path, |
| self.new_dep_path)] |
| expected_copy_calls = [mock.call(self.new_bucket, self.new_bucket, |
| self.new_remote_path, |
| self.expected_new_backup_path)] |
| |
| self.assertTrue(config.ExecuteUpdateJobs(force=True)) |
| self.assertFalse(config._IsDirty()) |
| self.assertFalse(config._pending_uploads) |
| self.assertEqual(self.new_dependencies, config._config_data) |
| file_module = fake_filesystem.FakeFileOpen(self.fs) |
| expected_file_lines = list(self.new_expected_file_lines) |
| for line in file_module(self.file_path): |
| self.assertEqual(expected_file_lines.pop(0), line.strip()) |
| self.fs.CloseOpenFile(file_module(self.file_path)) |
| self.assertFalse(config._pending_uploads) |
| self.assertEqual(expected_insert_calls, |
| uploader_cs_mock.Insert.call_args_list) |
| self.assertEqual(expected_exists_calls, |
| uploader_cs_mock.Exists.call_args_list) |
| self.assertEqual(expected_copy_calls, |
| uploader_cs_mock.Copy.call_args_list) |
| |
| @mock.patch('dependency_manager.uploader.cloud_storage') |
| def testExecuteUpdateJobsErrorOnePendingDepCloudStorageCollisionNoForce( |
| self, uploader_cs_mock): |
| uploader_cs_mock.Exists.return_value = True |
| self.fs.CreateFile(self.file_path, |
| contents='\n'.join(self.expected_file_lines)) |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| config._config_data = self.new_dependencies.copy() |
| config._is_dirty = True |
| config._pending_uploads = [self.new_pending_upload] |
| self.assertEqual(self.new_dependencies, config._config_data) |
| self.assertTrue(config._is_dirty) |
| self.assertEqual(1, len(config._pending_uploads)) |
| self.assertEqual(self.new_pending_upload, config._pending_uploads[0]) |
| expected_exists_calls = [mock.call(self.new_bucket, self.new_remote_path)] |
| expected_insert_calls = [] |
| expected_copy_calls = [] |
| |
| self.assertRaises(dependency_manager.CloudStorageUploadConflictError, |
| config.ExecuteUpdateJobs) |
| self.assertTrue(config._is_dirty) |
| self.assertTrue(config._pending_uploads) |
| self.assertEqual(self.new_dependencies, config._config_data) |
| self.assertEqual(1, len(config._pending_uploads)) |
| self.assertEqual(self.new_pending_upload, config._pending_uploads[0]) |
| file_module = fake_filesystem.FakeFileOpen(self.fs) |
| expected_file_lines = list(self.expected_file_lines) |
| for line in file_module(self.file_path): |
| self.assertEqual(expected_file_lines.pop(0), line.strip()) |
| self.fs.CloseOpenFile(file_module(self.file_path)) |
| self.assertEqual(expected_insert_calls, |
| uploader_cs_mock.Insert.call_args_list) |
| self.assertEqual(expected_exists_calls, |
| uploader_cs_mock.Exists.call_args_list) |
| self.assertEqual(expected_copy_calls, |
| uploader_cs_mock.Copy.call_args_list) |
| |
| @mock.patch('dependency_manager.uploader.cloud_storage') |
| def testExecuteUpdateJobsSuccessMultiplePendingDepsOneCloudStorageCollision( |
| self, uploader_cs_mock): |
| uploader_cs_mock.Exists.side_effect = [False, True] |
| self.fs.CreateFile(self.file_path, |
| contents='\n'.join(self.expected_file_lines)) |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| config._config_data = self.final_dependencies.copy() |
| config._pending_uploads = [self.new_pending_upload, |
| self.final_pending_upload] |
| self.assertEqual(self.final_dependencies, config._config_data) |
| self.assertTrue(config._IsDirty()) |
| self.assertEqual(2, len(config._pending_uploads)) |
| self.assertEqual(self.new_pending_upload, config._pending_uploads[0]) |
| self.assertEqual(self.final_pending_upload, config._pending_uploads[1]) |
| |
| expected_exists_calls = [mock.call(self.new_bucket, self.new_remote_path), |
| mock.call(self.final_bucket, |
| self.final_remote_path)] |
| expected_insert_calls = [mock.call(self.new_bucket, self.new_remote_path, |
| self.new_dep_path), |
| mock.call(self.final_bucket, |
| self.final_remote_path, |
| self.final_dep_path)] |
| expected_copy_calls = [mock.call(self.final_bucket, self.final_bucket, |
| self.final_remote_path, |
| self.expected_final_backup_path)] |
| |
| self.assertTrue(config.ExecuteUpdateJobs(force=True)) |
| self.assertFalse(config._IsDirty()) |
| self.assertFalse(config._pending_uploads) |
| self.assertEqual(self.final_dependencies, config._config_data) |
| file_module = fake_filesystem.FakeFileOpen(self.fs) |
| expected_file_lines = list(self.final_expected_file_lines) |
| for line in file_module(self.file_path): |
| self.assertEqual(expected_file_lines.pop(0), line.strip()) |
| self.fs.CloseOpenFile(file_module(self.file_path)) |
| self.assertFalse(config._pending_uploads) |
| self.assertEqual(expected_insert_calls, |
| uploader_cs_mock.Insert.call_args_list) |
| self.assertEqual(expected_exists_calls, |
| uploader_cs_mock.Exists.call_args_list) |
| self.assertEqual(expected_copy_calls, |
| uploader_cs_mock.Copy.call_args_list) |
| |
| @mock.patch('dependency_manager.uploader.cloud_storage') |
| def testUpdateCloudStorageDependenciesReadOnlyConfig( |
| self, uploader_cs_mock): |
| self.fs.CreateFile(self.file_path, |
| contents='\n'.join(self.expected_file_lines)) |
| config = dependency_manager.BaseConfig(self.file_path) |
| with self.assertRaises(dependency_manager.ReadWriteError): |
| config.AddCloudStorageDependencyUpdateJob( |
| 'dep', 'plat', 'path') |
| with self.assertRaises(dependency_manager.ReadWriteError): |
| config.AddCloudStorageDependencyUpdateJob( |
| 'dep', 'plat', 'path', version='1.2.3') |
| with self.assertRaises(dependency_manager.ReadWriteError): |
| config.AddCloudStorageDependencyUpdateJob( |
| 'dep', 'plat', 'path', execute_job=False) |
| with self.assertRaises(dependency_manager.ReadWriteError): |
| config.AddCloudStorageDependencyUpdateJob( |
| 'dep', 'plat', 'path', version='1.2.3', execute_job=False) |
| |
| @mock.patch('dependency_manager.uploader.cloud_storage') |
| def testUpdateCloudStorageDependenciesMissingDependency( |
| self, uploader_cs_mock): |
| self.fs.CreateFile(self.file_path, |
| contents='\n'.join(self.expected_file_lines)) |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| self.assertRaises(ValueError, config.AddCloudStorageDependencyUpdateJob, |
| 'dep', 'plat', 'path') |
| self.assertRaises(ValueError, config.AddCloudStorageDependencyUpdateJob, |
| 'dep', 'plat', 'path', version='1.2.3') |
| self.assertRaises(ValueError, config.AddCloudStorageDependencyUpdateJob, |
| 'dep', 'plat', 'path', execute_job=False) |
| self.assertRaises(ValueError, config.AddCloudStorageDependencyUpdateJob, |
| 'dep', 'plat', 'path', version='1.2.3', execute_job=False) |
| |
| @mock.patch('dependency_manager.uploader.cloud_storage') |
| @mock.patch('dependency_manager.base_config.cloud_storage') |
| def testUpdateCloudStorageDependenciesWrite( |
| self, base_config_cs_mock, uploader_cs_mock): |
| expected_dependencies = self.dependencies |
| self.fs.CreateFile(self.file_path, |
| contents='\n'.join(self.expected_file_lines)) |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| self.assertFalse(config._IsDirty()) |
| self.assertEqual(expected_dependencies, config._config_data) |
| |
| base_config_cs_mock.CalculateHash.return_value = self.new_dep_hash |
| uploader_cs_mock.Exists.return_value = False |
| expected_dependencies = self.new_dependencies |
| config.AddCloudStorageDependencyUpdateJob( |
| 'dep1', 'plat2', self.new_dep_path, execute_job=True) |
| self.assertFalse(config._IsDirty()) |
| self.assertFalse(config._pending_uploads) |
| self.assertEqual(expected_dependencies, config._config_data) |
| # check that file contents has been updated |
| file_module = fake_filesystem.FakeFileOpen(self.fs) |
| expected_file_lines = list(self.new_expected_file_lines) |
| for line in file_module(self.file_path): |
| self.assertEqual(expected_file_lines.pop(0), line.strip()) |
| self.fs.CloseOpenFile(file_module(self.file_path)) |
| |
| expected_dependencies = self.final_dependencies |
| base_config_cs_mock.CalculateHash.return_value = self.final_dep_hash |
| config.AddCloudStorageDependencyUpdateJob( |
| 'dep2', 'plat1', self.final_dep_path, execute_job=True) |
| self.assertFalse(config._IsDirty()) |
| self.assertFalse(config._pending_uploads) |
| self.assertEqual(expected_dependencies, config._config_data) |
| # check that file contents has been updated |
| expected_file_lines = list(self.final_expected_file_lines) |
| file_module = fake_filesystem.FakeFileOpen(self.fs) |
| for line in file_module(self.file_path): |
| self.assertEqual(expected_file_lines.pop(0), line.strip()) |
| self.fs.CloseOpenFile(file_module(self.file_path)) |
| |
| @mock.patch('dependency_manager.uploader.cloud_storage') |
| @mock.patch('dependency_manager.base_config.cloud_storage') |
| def testUpdateCloudStorageDependenciesNoWrite( |
| self, base_config_cs_mock, uploader_cs_mock): |
| self.fs.CreateFile(self.file_path, |
| contents='\n'.join(self.expected_file_lines)) |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| |
| self.assertRaises(ValueError, config.AddCloudStorageDependencyUpdateJob, |
| 'dep', 'plat', 'path') |
| self.assertRaises(ValueError, config.AddCloudStorageDependencyUpdateJob, |
| 'dep', 'plat', 'path', version='1.2.3') |
| |
| expected_dependencies = self.dependencies |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| self.assertFalse(config._IsDirty()) |
| self.assertFalse(config._pending_uploads) |
| self.assertEqual(expected_dependencies, config._config_data) |
| |
| base_config_cs_mock.CalculateHash.return_value = self.new_dep_hash |
| uploader_cs_mock.Exists.return_value = False |
| expected_dependencies = self.new_dependencies |
| config.AddCloudStorageDependencyUpdateJob( |
| 'dep1', 'plat2', self.new_dep_path, execute_job=False) |
| self.assertTrue(config._IsDirty()) |
| self.assertEqual(1, len(config._pending_uploads)) |
| self.assertEqual(self.new_pending_upload, config._pending_uploads[0]) |
| self.assertEqual(expected_dependencies, config._config_data) |
| # check that file contents have not been updated. |
| expected_file_lines = list(self.expected_file_lines) |
| file_module = fake_filesystem.FakeFileOpen(self.fs) |
| for line in file_module(self.file_path): |
| self.assertEqual(expected_file_lines.pop(0), line.strip()) |
| self.fs.CloseOpenFile(file_module(self.file_path)) |
| |
| expected_dependencies = self.final_dependencies |
| base_config_cs_mock.CalculateHash.return_value = self.final_dep_hash |
| config.AddCloudStorageDependencyUpdateJob( |
| 'dep2', 'plat1', self.final_dep_path, execute_job=False) |
| self.assertTrue(config._IsDirty()) |
| self.assertEqual(expected_dependencies, config._config_data) |
| # check that file contents have not been updated. |
| expected_file_lines = list(self.expected_file_lines) |
| file_module = fake_filesystem.FakeFileOpen(self.fs) |
| for line in file_module(self.file_path): |
| self.assertEqual(expected_file_lines.pop(0), line.strip()) |
| self.fs.CloseOpenFile(file_module(self.file_path)) |
| |
| |
| class BaseConfigDataManipulationUnittests(fake_filesystem_unittest.TestCase): |
| def setUp(self): |
| self.addTypeEqualityFunc(uploader.CloudStorageUploader, |
| uploader.CloudStorageUploader.__eq__) |
| self.setUpPyfakefs() |
| |
| self.cs_bucket = 'bucket1' |
| self.cs_base_folder = 'dependencies_folder' |
| self.cs_hash = 'hash12' |
| self.download_path = '../../relative/dep1/path2' |
| self.local_paths = ['../../../relative/local/path21', |
| '../../../relative/local/path22'] |
| self.platform_dict = {'cloud_storage_hash': self.cs_hash, |
| 'download_path': self.download_path, |
| 'local_paths': self.local_paths} |
| self.dependencies = { |
| 'dep1': { |
| 'cloud_storage_bucket': self.cs_bucket, |
| 'cloud_storage_base_folder': self.cs_base_folder, |
| 'file_info': { |
| 'plat1': { |
| 'cloud_storage_hash': 'hash11', |
| 'download_path': '../../relative/dep1/path1', |
| 'local_paths': ['../../../relative/local/path11', |
| '../../../relative/local/path12']}, |
| 'plat2': self.platform_dict |
| } |
| }, |
| 'dep2': { |
| 'cloud_storage_bucket': 'bucket2', |
| 'file_info': { |
| 'plat1': { |
| 'cloud_storage_hash': 'hash21', |
| 'download_path': '../../relative/dep2/path1', |
| 'local_paths': ['../../../relative/local/path31', |
| '../../../relative/local/path32']}, |
| 'plat2': { |
| 'cloud_storage_hash': 'hash22', |
| 'download_path': '../../relative/dep2/path2'}}}} |
| |
| self.file_path = os.path.abspath(os.path.join( |
| 'path', 'to', 'config', 'file')) |
| |
| |
| self.expected_file_lines = [ |
| # pylint: disable=bad-continuation |
| '{', '"config_type": "BaseConfig",', '"dependencies": {', |
| '"dep1": {', '"cloud_storage_base_folder": "dependencies_folder",', |
| '"cloud_storage_bucket": "bucket1",', '"file_info": {', |
| '"plat1": {', '"cloud_storage_hash": "hash11",', |
| '"download_path": "../../relative/dep1/path1",', |
| '"local_paths": [', '"../../../relative/local/path11",', |
| '"../../../relative/local/path12"', ']', '},', |
| '"plat2": {', '"cloud_storage_hash": "hash12",', |
| '"download_path": "../../relative/dep1/path2",', |
| '"local_paths": [', '"../../../relative/local/path21",', |
| '"../../../relative/local/path22"', ']', |
| '}', '}', '},', |
| '"dep2": {', '"cloud_storage_bucket": "bucket2",', '"file_info": {', |
| '"plat1": {', '"cloud_storage_hash": "hash21",', |
| '"download_path": "../../relative/dep2/path1",', |
| '"local_paths": [', '"../../../relative/local/path31",', |
| '"../../../relative/local/path32"', ']', '},', |
| '"plat2": {', '"cloud_storage_hash": "hash22",', |
| '"download_path": "../../relative/dep2/path2"', '}', '}', '}', |
| '}', '}'] |
| self.fs.CreateFile(self.file_path, |
| contents='\n'.join(self.expected_file_lines)) |
| |
| def testContaining(self): |
| config = dependency_manager.BaseConfig(self.file_path) |
| self.assertTrue('dep1' in config) |
| self.assertTrue('dep2' in config) |
| self.assertFalse('dep3' in config) |
| |
| def testAddNewDependencyNotWriteable(self): |
| config = dependency_manager.BaseConfig(self.file_path) |
| with self.assertRaises(dependency_manager.ReadWriteError): |
| config.AddNewDependency('dep4', 'foo', 'bar') |
| |
| def testAddNewDependencyWriteableButDependencyAlreadyExists(self): |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| with self.assertRaises(ValueError): |
| config.AddNewDependency('dep2', 'foo', 'bar') |
| |
| def testAddNewDependencySuccessfully(self): |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| config.AddNewDependency('dep3', 'foo', 'bar') |
| self.assertTrue('dep3' in config) |
| |
| def testSetDownloadPathNotWritable(self): |
| config = dependency_manager.BaseConfig(self.file_path) |
| with self.assertRaises(dependency_manager.ReadWriteError): |
| config.SetDownloadPath('dep2', 'plat1', '../../relative/dep1/path1') |
| |
| def testSetDownloadPathOnExistingPlatformSuccesfully(self): |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| download_path = '../../relative/dep1/foo.bar' |
| config.SetDownloadPath('dep2', 'plat1', download_path) |
| self.assertEqual( |
| download_path, |
| config._GetPlatformData('dep2', 'plat1', 'download_path')) |
| |
| def testSetDownloadPathOnNewPlatformSuccesfully(self): |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| download_path = '../../relative/dep1/foo.bar' |
| config.SetDownloadPath('dep2', 'newplat', download_path) |
| self.assertEqual( |
| download_path, |
| config._GetPlatformData('dep2', 'newplat', 'download_path')) |
| |
| |
| def testSetPlatformDataFailureNotWritable(self): |
| config = dependency_manager.BaseConfig(self.file_path) |
| self.assertRaises( |
| dependency_manager.ReadWriteError, config._SetPlatformData, |
| 'dep1', 'plat1', 'cloud_storage_bucket', 'new_bucket') |
| self.assertEqual(self.dependencies, config._config_data) |
| |
| def testSetPlatformDataFailure(self): |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| self.assertRaises(ValueError, config._SetPlatformData, 'missing_dep', |
| 'plat2', 'cloud_storage_bucket', 'new_bucket') |
| self.assertEqual(self.dependencies, config._config_data) |
| self.assertRaises(ValueError, config._SetPlatformData, 'dep1', |
| 'missing_plat', 'cloud_storage_bucket', 'new_bucket') |
| self.assertEqual(self.dependencies, config._config_data) |
| |
| |
| def testSetPlatformDataCloudStorageBucketSuccess(self): |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| updated_cs_dependencies = { |
| 'dep1': {'cloud_storage_bucket': 'new_bucket', |
| 'cloud_storage_base_folder': 'dependencies_folder', |
| 'file_info': { |
| 'plat1': { |
| 'cloud_storage_hash': 'hash11', |
| 'download_path': '../../relative/dep1/path1', |
| 'local_paths': ['../../../relative/local/path11', |
| '../../../relative/local/path12']}, |
| 'plat2': { |
| 'cloud_storage_hash': 'hash12', |
| 'download_path': '../../relative/dep1/path2', |
| 'local_paths': ['../../../relative/local/path21', |
| '../../../relative/local/path22']}}}, |
| 'dep2': {'cloud_storage_bucket': 'bucket2', |
| 'file_info': { |
| 'plat1': { |
| 'cloud_storage_hash': 'hash21', |
| 'download_path': '../../relative/dep2/path1', |
| 'local_paths': ['../../../relative/local/path31', |
| '../../../relative/local/path32']}, |
| 'plat2': { |
| 'cloud_storage_hash': 'hash22', |
| 'download_path': '../../relative/dep2/path2'}}}} |
| config._SetPlatformData('dep1', 'plat2', 'cloud_storage_bucket', |
| 'new_bucket') |
| self.assertEqual(updated_cs_dependencies, config._config_data) |
| |
| def testSetPlatformDataCloudStorageBaseFolderSuccess(self): |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| updated_cs_dependencies = { |
| 'dep1': {'cloud_storage_bucket': 'bucket1', |
| 'cloud_storage_base_folder': 'new_dependencies_folder', |
| 'file_info': { |
| 'plat1': { |
| 'cloud_storage_hash': 'hash11', |
| 'download_path': '../../relative/dep1/path1', |
| 'local_paths': ['../../../relative/local/path11', |
| '../../../relative/local/path12']}, |
| 'plat2': { |
| 'cloud_storage_hash': 'hash12', |
| 'download_path': '../../relative/dep1/path2', |
| 'local_paths': ['../../../relative/local/path21', |
| '../../../relative/local/path22']}}}, |
| 'dep2': {'cloud_storage_bucket': 'bucket2', |
| 'file_info': { |
| 'plat1': { |
| 'cloud_storage_hash': 'hash21', |
| 'download_path': '../../relative/dep2/path1', |
| 'local_paths': ['../../../relative/local/path31', |
| '../../../relative/local/path32']}, |
| 'plat2': { |
| 'cloud_storage_hash': 'hash22', |
| 'download_path': '../../relative/dep2/path2'}}}} |
| config._SetPlatformData('dep1', 'plat2', 'cloud_storage_base_folder', |
| 'new_dependencies_folder') |
| self.assertEqual(updated_cs_dependencies, config._config_data) |
| |
| def testSetPlatformDataHashSuccess(self): |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| updated_cs_dependencies = { |
| 'dep1': {'cloud_storage_bucket': 'bucket1', |
| 'cloud_storage_base_folder': 'dependencies_folder', |
| 'file_info': { |
| 'plat1': { |
| 'cloud_storage_hash': 'hash11', |
| 'download_path': '../../relative/dep1/path1', |
| 'local_paths': ['../../../relative/local/path11', |
| '../../../relative/local/path12']}, |
| 'plat2': { |
| 'cloud_storage_hash': 'new_hash', |
| 'download_path': '../../relative/dep1/path2', |
| 'local_paths': ['../../../relative/local/path21', |
| '../../../relative/local/path22']}}}, |
| 'dep2': {'cloud_storage_bucket': 'bucket2', |
| 'file_info': { |
| 'plat1': { |
| 'cloud_storage_hash': 'hash21', |
| 'download_path': '../../relative/dep2/path1', |
| 'local_paths': ['../../../relative/local/path31', |
| '../../../relative/local/path32']}, |
| 'plat2': { |
| 'cloud_storage_hash': 'hash22', |
| 'download_path': '../../relative/dep2/path2'}}}} |
| config._SetPlatformData('dep1', 'plat2', 'cloud_storage_hash', |
| 'new_hash') |
| self.assertEqual(updated_cs_dependencies, config._config_data) |
| |
| def testSetPlatformDataDownloadPathSuccess(self): |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| updated_cs_dependencies = { |
| 'dep1': {'cloud_storage_bucket': 'bucket1', |
| 'cloud_storage_base_folder': 'dependencies_folder', |
| 'file_info': { |
| 'plat1': { |
| 'cloud_storage_hash': 'hash11', |
| 'download_path': '../../relative/dep1/path1', |
| 'local_paths': ['../../../relative/local/path11', |
| '../../../relative/local/path12']}, |
| 'plat2': { |
| 'cloud_storage_hash': 'hash12', |
| 'download_path': '../../new/dep1/path2', |
| 'local_paths': ['../../../relative/local/path21', |
| '../../../relative/local/path22']}}}, |
| 'dep2': {'cloud_storage_bucket': 'bucket2', |
| 'file_info': { |
| 'plat1': { |
| 'cloud_storage_hash': 'hash21', |
| 'download_path': '../../relative/dep2/path1', |
| 'local_paths': ['../../../relative/local/path31', |
| '../../../relative/local/path32']}, |
| 'plat2': { |
| 'cloud_storage_hash': 'hash22', |
| 'download_path': '../../relative/dep2/path2'}}}} |
| config._SetPlatformData('dep1', 'plat2', 'download_path', |
| '../../new/dep1/path2') |
| self.assertEqual(updated_cs_dependencies, config._config_data) |
| |
| def testSetPlatformDataLocalPathSuccess(self): |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| updated_cs_dependencies = { |
| 'dep1': {'cloud_storage_bucket': 'bucket1', |
| 'cloud_storage_base_folder': 'dependencies_folder', |
| 'file_info': { |
| 'plat1': { |
| 'cloud_storage_hash': 'hash11', |
| 'download_path': '../../relative/dep1/path1', |
| 'local_paths': ['../../../relative/local/path11', |
| '../../../relative/local/path12']}, |
| 'plat2': { |
| 'cloud_storage_hash': 'hash12', |
| 'download_path': '../../relative/dep1/path2', |
| 'local_paths': ['../../new/relative/local/path21', |
| '../../new/relative/local/path22']}}}, |
| 'dep2': {'cloud_storage_bucket': 'bucket2', |
| 'file_info': { |
| 'plat1': { |
| 'cloud_storage_hash': 'hash21', |
| 'download_path': '../../relative/dep2/path1', |
| 'local_paths': ['../../../relative/local/path31', |
| '../../../relative/local/path32']}, |
| 'plat2': { |
| 'cloud_storage_hash': 'hash22', |
| 'download_path': '../../relative/dep2/path2'}}}} |
| config._SetPlatformData('dep1', 'plat2', 'local_paths', |
| ['../../new/relative/local/path21', |
| '../../new/relative/local/path22']) |
| self.assertEqual(updated_cs_dependencies, config._config_data) |
| |
| def testGetPlatformDataFailure(self): |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| self.assertRaises(ValueError, config._GetPlatformData, 'missing_dep', |
| 'plat2', 'cloud_storage_bucket') |
| self.assertEqual(self.dependencies, config._config_data) |
| self.assertRaises(ValueError, config._GetPlatformData, 'dep1', |
| 'missing_plat', 'cloud_storage_bucket') |
| self.assertEqual(self.dependencies, config._config_data) |
| |
| def testGetPlatformDataDictSuccess(self): |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| self.assertEqual(self.platform_dict, |
| config._GetPlatformData('dep1', 'plat2')) |
| self.assertEqual(self.dependencies, config._config_data) |
| |
| def testGetPlatformDataCloudStorageBucketSuccess(self): |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| self.assertEqual(self.cs_bucket, config._GetPlatformData( |
| 'dep1', 'plat2', 'cloud_storage_bucket')) |
| self.assertEqual(self.dependencies, config._config_data) |
| |
| def testGetPlatformDataCloudStorageBaseFolderSuccess(self): |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| self.assertEqual(self.cs_base_folder, config._GetPlatformData( |
| 'dep1', 'plat2', 'cloud_storage_base_folder')) |
| self.assertEqual(self.dependencies, config._config_data) |
| |
| def testGetPlatformDataHashSuccess(self): |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| self.assertEqual(self.cs_hash, config._GetPlatformData( |
| 'dep1', 'plat2', 'cloud_storage_hash')) |
| self.assertEqual(self.dependencies, config._config_data) |
| |
| def testGetPlatformDataDownloadPathSuccess(self): |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| self.assertEqual(self.download_path, config._GetPlatformData( |
| 'dep1', 'plat2', 'download_path')) |
| self.assertEqual(self.dependencies, config._config_data) |
| |
| def testGetPlatformDataLocalPathSuccess(self): |
| config = dependency_manager.BaseConfig(self.file_path, writable=True) |
| self.assertEqual(self.local_paths, config._GetPlatformData( |
| 'dep1', 'plat2', 'local_paths')) |
| self.assertEqual(self.dependencies, config._config_data) |
| |
| class BaseConfigTest(unittest.TestCase): |
| """ Subclassable unittests for BaseConfig. |
| For subclasses: override setUp, GetConfigDataFromDict, |
| and EndToEndExpectedConfigData as needed. |
| |
| setUp must set the following properties: |
| self.config_type: String returnedd from GetConfigType in config subclass. |
| self.config_class: the class for the config subclass. |
| self.config_module: importable module for the config subclass. |
| self.empty_dict: expected dictionary for an empty config, as it would be |
| stored in a json file. |
| self.one_dep_dict: example dictionary for a config with one dependency, |
| as it would be stored in a json file. |
| """ |
| def setUp(self): |
| self.config_type = 'BaseConfig' |
| self.config_class = dependency_manager.BaseConfig |
| self.config_module = 'dependency_manager.base_config' |
| |
| self.empty_dict = {'config_type': self.config_type, |
| 'dependencies': {}} |
| |
| dependency_dict = { |
| 'dep': { |
| 'cloud_storage_base_folder': 'cs_base_folder1', |
| 'cloud_storage_bucket': 'bucket1', |
| 'file_info': { |
| 'plat1_arch1': { |
| 'cloud_storage_hash': 'hash111', |
| 'download_path': 'download_path111', |
| 'cs_remote_path': 'cs_path111', |
| 'version_in_cs': 'version_111', |
| 'local_paths': ['local_path1110', 'local_path1111'] |
| }, |
| 'plat1_arch2': { |
| 'cloud_storage_hash': 'hash112', |
| 'download_path': 'download_path112', |
| 'cs_remote_path': 'cs_path112', |
| 'local_paths': ['local_path1120', 'local_path1121'] |
| }, |
| 'win_arch1': { |
| 'cloud_storage_hash': 'hash1w1', |
| 'download_path': 'download\\path\\1w1', |
| 'cs_remote_path': 'cs_path1w1', |
| 'local_paths': ['local\\path\\1w10', 'local\\path\\1w11'] |
| }, |
| 'all_the_variables': { |
| 'cloud_storage_hash': 'hash111', |
| 'download_path': 'download_path111', |
| 'cs_remote_path': 'cs_path111', |
| 'version_in_cs': 'version_111', |
| 'path_within_archive': 'path/within/archive', |
| 'local_paths': ['local_path1110', 'local_path1111'] |
| } |
| } |
| } |
| } |
| self.one_dep_dict = {'config_type': self.config_type, |
| 'dependencies': dependency_dict} |
| |
| def GetConfigDataFromDict(self, config_dict): |
| return config_dict.get('dependencies', {}) |
| |
| @mock.patch('os.path') |
| @mock.patch('__builtin__.open') |
| def testInitBaseProperties(self, open_mock, path_mock): |
| # Init is not meant to be overridden, so we should be mocking the |
| # base_config's json module, even in subclasses. |
| json_module = 'dependency_manager.base_config.json' |
| with mock.patch(json_module) as json_mock: |
| json_mock.load.return_value = self.empty_dict.copy() |
| config = self.config_class('file_path') |
| self.assertEqual('file_path', config._config_path) |
| self.assertEqual(self.config_type, config.GetConfigType()) |
| self.assertEqual(self.GetConfigDataFromDict(self.empty_dict), |
| config._config_data) |
| |
| @mock.patch('dependency_manager.dependency_info.DependencyInfo') |
| @mock.patch('os.path') |
| @mock.patch('__builtin__.open') |
| def testInitWithDependencies(self, open_mock, path_mock, dep_info_mock): |
| # Init is not meant to be overridden, so we should be mocking the |
| # base_config's json module, even in subclasses. |
| json_module = 'dependency_manager.base_config.json' |
| with mock.patch(json_module) as json_mock: |
| json_mock.load.return_value = self.one_dep_dict |
| config = self.config_class('file_path') |
| self.assertEqual('file_path', config._config_path) |
| self.assertEqual(self.config_type, config.GetConfigType()) |
| self.assertEqual(self.GetConfigDataFromDict(self.one_dep_dict), |
| config._config_data) |
| |
| def testFormatPath(self): |
| self.assertEqual(None, self.config_class._FormatPath(None)) |
| self.assertEqual('', self.config_class._FormatPath('')) |
| self.assertEqual('some_string', |
| self.config_class._FormatPath('some_string')) |
| |
| expected_path = os.path.join('some', 'file', 'path') |
| self.assertEqual(expected_path, |
| self.config_class._FormatPath('some/file/path')) |
| self.assertEqual(expected_path, |
| self.config_class._FormatPath('some\\file\\path')) |
| |
| @mock.patch('dependency_manager.base_config.json') |
| @mock.patch('dependency_manager.dependency_info.DependencyInfo') |
| @mock.patch('os.path.exists') |
| @mock.patch('__builtin__.open') |
| def testIterDependenciesError( |
| self, open_mock, exists_mock, dep_info_mock, json_mock): |
| # Init is not meant to be overridden, so we should be mocking the |
| # base_config's json module, even in subclasses. |
| json_mock.load.return_value = self.one_dep_dict |
| config = self.config_class('file_path', writable=True) |
| self.assertEqual(self.GetConfigDataFromDict(self.one_dep_dict), |
| config._config_data) |
| self.assertTrue(config._writable) |
| with self.assertRaises(dependency_manager.ReadWriteError): |
| for _ in config.IterDependencyInfo(): |
| pass |
| |
| @mock.patch('dependency_manager.base_config.json') |
| @mock.patch('dependency_manager.dependency_info.DependencyInfo') |
| @mock.patch('os.path.exists') |
| @mock.patch('__builtin__.open') |
| def testIterDependencies( |
| self, open_mock, exists_mock, dep_info_mock, json_mock): |
| json_mock.load.return_value = self.one_dep_dict |
| config = self.config_class('file_path') |
| self.assertEqual(self.GetConfigDataFromDict(self.one_dep_dict), |
| config._config_data) |
| expected_dep_info = ['dep_info0', 'dep_info1', 'dep_info2'] |
| dep_info_mock.side_effect = expected_dep_info |
| expected_calls = [ |
| mock.call('dep', 'plat1_arch1', 'file_path', cs_bucket='bucket1', |
| cs_hash='hash111', download_path='download_path111', |
| cs_remote_path='cs_path111', |
| local_paths=['local_path1110', 'local_path1111']), |
| mock.call('dep', 'plat1_arch1', 'file_path', cs_bucket='bucket1', |
| cs_hash='hash112', download_path='download_path112', |
| cs_remote_path='cs_path112', |
| local_paths=['local_path1120', 'local_path1121']), |
| mock.call('dep', 'win_arch1', 'file_path', cs_bucket='bucket1', |
| cs_hash='hash1w1', |
| download_path=os.path.join('download', 'path', '1w1'), |
| cs_remote_path='cs_path1w1', |
| local_paths=[os.path.join('download', 'path', '1w10'), |
| os.path.join('download', 'path', '1w11')])] |
| deps_seen = [] |
| for dep_info in config.IterDependencyInfo(): |
| deps_seen.append(dep_info) |
| dep_info_mock.assert_call_args(expected_calls) |
| self.assertItemsEqual(expected_dep_info, deps_seen) |
| |
| @mock.patch('dependency_manager.base_config.json') |
| @mock.patch('os.path.exists') |
| @mock.patch('__builtin__.open') |
| def testIterDependenciesStaleGlob(self, open_mock, exists_mock, json_mock): |
| json_mock.load.return_value = self.one_dep_dict |
| config = self.config_class('file_path') |
| |
| abspath = os.path.abspath |
| should_match = set(map(abspath, [ |
| 'dep_all_the_variables_0123456789abcdef0123456789abcdef01234567', |
| 'dep_all_the_variables_123456789abcdef0123456789abcdef012345678'])) |
| # Not testing case changes, because Windows is case-insensitive. |
| should_not_match = set(map(abspath, [ |
| # A configuration that doesn't unzip shouldn't clear any stale unzips. |
| 'dep_plat1_arch1_0123456789abcdef0123456789abcdef01234567', |
| # "Hash" component less than 40 characters (not a valid SHA1 hash). |
| 'dep_all_the_variables_0123456789abcdef0123456789abcdef0123456', |
| # "Hash" component greater than 40 characters (not a valid SHA1 hash). |
| 'dep_all_the_variables_0123456789abcdef0123456789abcdef012345678', |
| # "Hash" component not comprised of hex (not a valid SHA1 hash). |
| 'dep_all_the_variables_0123456789gggggg0123456789gggggg01234567'])) |
| |
| # Create a fake filesystem just for glob to use |
| fake_fs = fake_filesystem.FakeFilesystem() |
| fake_glob = fake_filesystem_glob.FakeGlobModule(fake_fs) |
| for stale_dir in set.union(should_match, should_not_match): |
| fake_fs.CreateDirectory(stale_dir) |
| fake_fs.CreateFile(os.path.join(stale_dir, 'some_file')) |
| |
| for dep_info in config.IterDependencyInfo(): |
| if dep_info.platform == 'all_the_variables': |
| cs_info = dep_info.cloud_storage_info |
| actual_glob = cs_info._archive_info._stale_unzip_path_glob |
| actual_matches = set(fake_glob.glob(actual_glob)) |
| self.assertItemsEqual(should_match, actual_matches) |
