1#!/usr/bin/env vpython3 2# Copyright 2022 The Chromium Authors 3# Use of this source code is governed by a BSD-style license that can be 4# found in the LICENSE file. 5 6import io 7import json 8import os 9import sys 10import unittest 11from unittest import mock 12 13import update_product_bundles 14 15sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), 16 'test'))) 17 18import common 19 20 21class TestUpdateProductBundles(unittest.TestCase): 22 def setUp(self): 23 # By default, test in attended mode. 24 os.environ.pop('SWARMING_SERVER', None) 25 ffx_mock = mock.Mock() 26 ffx_mock.returncode = 0 27 self._ffx_patcher = mock.patch('common.run_ffx_command', 28 return_value=ffx_mock) 29 self._ffx_mock = self._ffx_patcher.start() 30 self.addCleanup(self._ffx_mock.stop) 31 32 def testConvertToProductBundleDefaultsUnknownImage(self): 33 self.assertEqual( 34 update_product_bundles.convert_to_products(['unknown-image']), 35 ['unknown-image']) 36 37 def testConvertToProductBundleRemovesReleaseSuffix(self): 38 self.assertEqual( 39 update_product_bundles.convert_to_products( 40 ['smart_display_eng.astro-release']), ['smart_display_eng.astro']) 41 42 def testConvertToProductBundleWarnsDeprecated(self): 43 with self.assertLogs(level='WARNING') as logs: 44 deprecated_images = [ 45 'qemu.arm64', 'qemu.x64', 'core.x64-dfv2-release', 46 'workstation_eng.chromebook-x64-release' 47 ] 48 self.assertEqual( 49 update_product_bundles.convert_to_products(deprecated_images), [ 50 'terminal.qemu-arm64', 'terminal.x64', 'core.x64-dfv2', 51 'workstation_eng.chromebook-x64' 52 ]) 53 for i, deprecated_image in enumerate(deprecated_images): 54 self.assertIn(f'Image name {deprecated_image} has been deprecated', 55 logs.output[i]) 56 57 58 @mock.patch('common.run_ffx_command') 59 def testRemoveRepositoriesRunsRemoveOnGivenRepos(self, ffx_mock): 60 update_product_bundles.remove_repositories(['foo', 'bar', 'fizz', 'buzz']) 61 62 ffx_mock.assert_has_calls([ 63 mock.call(cmd=('repository', 'remove', 'foo'), check=True), 64 mock.call(cmd=('repository', 'remove', 'bar'), check=True), 65 mock.call(cmd=('repository', 'remove', 'fizz'), check=True), 66 mock.call(cmd=('repository', 'remove', 'buzz'), check=True), 67 ]) 68 69 @mock.patch('os.path.exists') 70 @mock.patch('os.path.abspath') 71 def testGetRepositoriesPrunesReposThatDoNotExist(self, mock_abspath, 72 mock_exists): 73 with mock.patch('common.SDK_ROOT', 'some/path'): 74 self._ffx_mock.return_value.stdout = json.dumps([{ 75 "name": "terminal.x64", 76 "spec": { 77 "type": "pm", 78 "path": "some/path/that/exists" 79 } 80 }, { 81 "name": "workstation-eng.chromebook-x64", 82 "spec": { 83 "type": "pm", 84 "path": "some/path/that/does/not/exist" 85 } 86 }]) 87 mock_exists.side_effect = [True, False] 88 mock_abspath.side_effect = lambda x: x 89 90 self.assertEqual(update_product_bundles.get_repositories(), [{ 91 "name": "terminal.x64", 92 "spec": { 93 "type": "pm", 94 "path": "some/path/that/exists" 95 } 96 }]) 97 98 self._ffx_mock.assert_has_calls([ 99 mock.call(cmd=('--machine', 'json', 'repository', 'list'), 100 capture_output=True, 101 check=True), 102 mock.call(cmd=('repository', 'remove', 103 'workstation-eng.chromebook-x64'), 104 check=True) 105 ]) 106 107 @mock.patch('update_product_bundles.running_unattended', return_value=True) 108 # Disallow reading sdk_override. 109 @mock.patch('os.path.isfile', return_value=False) 110 @mock.patch('update_product_bundles.internal_hash', return_value='1.1.1') 111 def testLookupAndDownloadWithAuth(self, *_): 112 try: 113 common.get_host_os() 114 except: 115 # Ignore unsupported platforms. common.get_host_os used in 116 # update_product_bundles.main throws an unsupported exception. 117 return 118 auth_file = os.path.abspath( 119 os.path.join(os.path.dirname(__file__), 'get_auth_token.py')) 120 self._ffx_mock.return_value.stdout = json.dumps({ 121 "name": "core.x64", 122 "product_version": "17.20240106.2.1", 123 "transfer_manifest_url": "http://download-url" 124 }) 125 126 with mock.patch( 127 'sys.argv', 128 ['update_product_bundles.py', 'terminal.x64', '--internal']): 129 update_product_bundles.main() 130 self._ffx_mock.assert_has_calls([ 131 mock.call(cmd=[ 132 '--machine', 'json', 'product', 'lookup', 'terminal.x64', '1.1.1', 133 '--base-url', f'gs://fuchsia-sdk/development/1.1.1', '--auth', 134 auth_file 135 ], 136 check=True, 137 capture_output=True), 138 mock.call(cmd=[ 139 'product', 'download', 'http://download-url', 140 os.path.join(common.INTERNAL_IMAGES_ROOT, 'terminal', 'x64'), 141 '--auth', auth_file 142 ], 143 check=True) 144 ]) 145 146 @mock.patch('common.get_hash_from_sdk', return_value='2.2.2') 147 @mock.patch('update_product_bundles.get_current_signature', 148 return_value='2.2.2') 149 @mock.patch('update_sdk.GetSDKOverrideGCSPath', return_value=None) 150 def testIgnoreDownloadImagesWithSameHash(self, *_): 151 try: 152 common.get_host_os() 153 except: 154 # Ignore unsupported platforms. common.get_host_os used in 155 # update_product_bundles.main throws an unsupported exception. 156 return 157 with mock.patch('sys.argv', ['update_product_bundles.py', 'terminal.x64']): 158 update_product_bundles.main() 159 self.assertFalse(self._ffx_mock.called) 160 161 @mock.patch('common.get_hash_from_sdk', return_value='2.2.2') 162 @mock.patch('update_product_bundles.get_current_signature', 163 return_value='0.0') 164 @mock.patch('update_sdk.GetSDKOverrideGCSPath', return_value=None) 165 def testDownloadImagesWithDifferentHash(self, *_): 166 try: 167 common.get_host_os() 168 except: 169 # Ignore unsupported platforms. common.get_host_os used in 170 # update_product_bundles.main throws an unsupported exception. 171 return 172 self._ffx_mock.return_value.stdout = json.dumps({ 173 "name": 174 "core.x64", 175 "product_version": 176 "17.20240106.2.1", 177 "transfer_manifest_url": 178 "http://download-url" 179 }) 180 with mock.patch('sys.argv', ['update_product_bundles.py', 'terminal.x64']): 181 update_product_bundles.main() 182 self._ffx_mock.assert_has_calls([ 183 mock.call(cmd=[ 184 '--machine', 'json', 'product', 'lookup', 'terminal.x64', '2.2.2', 185 '--base-url', 'gs://fuchsia/development/2.2.2' 186 ], 187 check=True, 188 capture_output=True), 189 mock.call(cmd=[ 190 'product', 'download', 'http://download-url', 191 os.path.join(common.IMAGES_ROOT, 'terminal', 'x64') 192 ], 193 check=True) 194 ]) 195 196 @mock.patch('update_sdk.GetSDKOverrideGCSPath', 197 return_value='gs://my-bucket/sdk') 198 @mock.patch('common.get_hash_from_sdk', return_value='2.2.2') 199 def testSDKOverrideForSDKImages(self, *_): 200 try: 201 common.get_host_os() 202 except: 203 # Ignore unsupported platforms. common.get_host_os used in 204 # update_product_bundles.main throws an unsupported exception. 205 return 206 self._ffx_mock.return_value.stdout = json.dumps({ 207 "name": 208 "core.x64", 209 "product_version": 210 "17.20240106.2.1", 211 "transfer_manifest_url": 212 "http://download-url" 213 }) 214 with mock.patch('sys.argv', ['update_product_bundles.py', 'terminal.x64']): 215 update_product_bundles.main() 216 self._ffx_mock.assert_has_calls([ 217 mock.call(cmd=[ 218 '--machine', 'json', 'product', 'lookup', 'terminal.x64', '2.2.2', 219 '--base-url', 'gs://my-bucket' 220 ], 221 check=True, 222 capture_output=True), 223 mock.call(cmd=[ 224 'product', 'download', 'http://download-url', 225 os.path.join(common.IMAGES_ROOT, 'terminal', 'x64') 226 ], 227 check=True) 228 ]) 229 230 @mock.patch('update_product_bundles.internal_hash', return_value='1.1.1') 231 @mock.patch('update_product_bundles.get_current_signature', 232 return_value='1.1.1') 233 def testIgnoreDownloadInternalImagesWithSameHash(self, *_): 234 try: 235 common.get_host_os() 236 except: 237 # Ignore unsupported platforms. common.get_host_os used in 238 # update_product_bundles.main throws an unsupported exception. 239 return 240 with mock.patch( 241 'sys.argv', 242 ['update_product_bundles.py', 'terminal.x64', '--internal']): 243 update_product_bundles.main() 244 self.assertFalse(self._ffx_mock.called) 245 246 @mock.patch('update_product_bundles.get_current_signature', 247 return_value='0.0') 248 @mock.patch('update_product_bundles.internal_hash', return_value='1.1.1') 249 def testDownloadInternalImagesWithDifferentHash(self, *_): 250 try: 251 common.get_host_os() 252 except: 253 # Ignore unsupported platforms. common.get_host_os used in 254 # update_product_bundles.main throws an unsupported exception. 255 return 256 self._ffx_mock.return_value.stdout = json.dumps({ 257 "name": 258 "core.x64", 259 "product_version": 260 "17.20240106.2.1", 261 "transfer_manifest_url": 262 "http://download-url" 263 }) 264 with mock.patch( 265 'sys.argv', 266 ['update_product_bundles.py', 'terminal.x64', '--internal']): 267 update_product_bundles.main() 268 self._ffx_mock.assert_has_calls([ 269 mock.call(cmd=[ 270 '--machine', 'json', 'product', 'lookup', 'terminal.x64', '1.1.1', 271 '--base-url', 'gs://fuchsia-sdk/development/1.1.1' 272 ], 273 check=True, 274 capture_output=True), 275 mock.call(cmd=[ 276 'product', 'download', 'http://download-url', 277 os.path.join(common.INTERNAL_IMAGES_ROOT, 'terminal', 'x64') 278 ], 279 check=True) 280 ]) 281 282 @mock.patch('update_sdk.GetSDKOverrideGCSPath', 283 return_value='gs://my-bucket/sdk') 284 @mock.patch('update_product_bundles.internal_hash', return_value='1.1.1') 285 def testIgnoreSDKOverrideForInternalImages(self, *_): 286 try: 287 common.get_host_os() 288 except: 289 # Ignore unsupported platforms. common.get_host_os used in 290 # update_product_bundles.main throws an unsupported exception. 291 return 292 self._ffx_mock.return_value.stdout = json.dumps({ 293 "name": 294 "core.x64", 295 "product_version": 296 "17.20240106.2.1", 297 "transfer_manifest_url": 298 "http://download-url" 299 }) 300 with mock.patch( 301 'sys.argv', 302 ['update_product_bundles.py', 'terminal.x64', '--internal']): 303 update_product_bundles.main() 304 self._ffx_mock.assert_has_calls([ 305 mock.call(cmd=[ 306 '--machine', 'json', 'product', 'lookup', 'terminal.x64', '1.1.1', 307 '--base-url', 'gs://fuchsia-sdk/development/1.1.1' 308 ], 309 check=True, 310 capture_output=True), 311 mock.call(cmd=[ 312 'product', 'download', 'http://download-url', 313 os.path.join(common.INTERNAL_IMAGES_ROOT, 'terminal', 'x64') 314 ], 315 check=True) 316 ]) 317 318 319if __name__ == '__main__': 320 unittest.main() 321