1 #!/usr/bin/env python 2 # Copyright 2015 The Chromium Authors. All rights reserved. 3 # Use of this source code is governed by a BSD-style license that can be 4 # found in the LICENSE file. 5 6 '''Unittests for update.py. 7 8 They set up a temporary directory that is used to mock a bucket, the directory 9 containing the configuration files and the android sdk directory. 10 11 Tests run the script with various inputs and check the status of the filesystem 12 ''' 13 14 import shutil 15 import tempfile 16 import unittest 17 import os 18 import sys 19 import zipfile 20 import contextlib 21 22 sys.path.append(os.path.join(os.path.dirname(__file__), os.pardir)) 23 from play_services import update 24 25 26 class TestFunctions(unittest.TestCase): 27 DEFAULT_CONFIG_VERSION = 42 28 DEFAULT_LICENSE = 'Default License' 29 DEFAULT_ZIP_SHA1 = 'zip0and0filling0to0forty0chars0000000000' 30 31 def __init__(self, *args, **kwargs): 32 super(TestFunctions, self).__init__(*args, **kwargs) 33 self.paths = None # Initialized in SetUpWorkdir 34 self.workdir = None # Initialized in setUp 35 36 #override 37 def setUp(self): 38 self.workdir = tempfile.mkdtemp() 39 40 #override 41 def tearDown(self): 42 shutil.rmtree(self.workdir) 43 self.workdir = None 44 45 def testUpload(self): 46 version = 1337 47 self.SetUpWorkdir( 48 xml_version=version, 49 gms_lib=True, 50 source_prop=True) 51 52 status = update.main([ 53 'upload', 54 '--dry-run', 55 '--skip-git', 56 '--bucket', self.paths.bucket, 57 '--config', self.paths.config_file, 58 '--sdk-root', self.paths.sdk_root 59 ]) 60 self.assertEqual(status, 0, 'the command should have succeeded.') 61 62 # bucket should contain license, name = license.sha1 63 self.assertTrue(os.path.isfile(self.paths.config_license_sha1)) 64 license_sha1 = _GetFileContent(self.paths.config_license_sha1) 65 bucket_license = os.path.join(self.paths.bucket, str(version), 66 license_sha1) 67 self.assertTrue(os.path.isfile(bucket_license)) 68 self.assertEqual(_GetFileContent(bucket_license), self.DEFAULT_LICENSE) 69 70 # bucket should contain zip, name = zip.sha1 71 self.assertTrue(os.path.isfile(self.paths.config_zip_sha1)) 72 bucket_zip = os.path.join(self.paths.bucket, str(version), 73 _GetFileContent(self.paths.config_zip_sha1)) 74 self.assertTrue(os.path.isfile(bucket_zip)) 75 76 # unzip, should contain expected files 77 with zipfile.ZipFile(bucket_zip, "r") as bucket_zip_file: 78 self.assertEqual(bucket_zip_file.namelist(), 79 ['dummy_file', 'res/values/version.xml']) 80 81 def testUploadAlreadyLatestVersion(self): 82 self.SetUpWorkdir( 83 xml_version=self.DEFAULT_CONFIG_VERSION, 84 gms_lib=True, 85 source_prop=True) 86 87 status = update.main([ 88 'upload', 89 '--dry-run', 90 '--skip-git', 91 '--bucket', self.paths.bucket, 92 '--config', self.paths.config_file, 93 '--sdk-root', self.paths.sdk_root, 94 ]) 95 self.assertEqual(status, 0, 'the command should have succeeded.') 96 97 # bucket should be empty 98 self.assertFalse(os.listdir(self.paths.bucket)) 99 self.assertFalse(os.path.isfile(self.paths.config_license_sha1)) 100 self.assertFalse(os.path.isfile(self.paths.config_zip_sha1)) 101 102 def testDownload(self): 103 self.SetUpWorkdir(populate_bucket=True) 104 105 with _MockedInput('y'): 106 status = update.main([ 107 'download', 108 '--dry-run', 109 '--bucket', self.paths.bucket, 110 '--config', self.paths.config_file, 111 '--sdk-root', self.paths.sdk_root, 112 ]) 113 114 self.assertEqual(status, 0, 'the command should have succeeded.') 115 116 # sdk_root should contain zip contents, zip sha1, license 117 self.assertTrue(os.path.isfile(os.path.join(self.paths.gms_lib, 118 'dummy_file'))) 119 self.assertTrue(os.path.isfile(self.paths.gms_root_sha1)) 120 self.assertTrue(os.path.isfile(self.paths.gms_root_license)) 121 self.assertEquals(_GetFileContent(self.paths.gms_root_license), 122 self.DEFAULT_LICENSE) 123 124 def testDownloadBot(self): 125 self.SetUpWorkdir(populate_bucket=True, bot_env=True) 126 127 # No need to type 'y' on bots 128 status = update.main([ 129 'download', 130 '--dry-run', 131 '--bucket', self.paths.bucket, 132 '--config', self.paths.config_file, 133 '--sdk-root', self.paths.sdk_root, 134 ]) 135 136 self.assertEqual(status, 0, 'the command should have succeeded.') 137 138 # sdk_root should contain zip contents, zip sha1, license 139 self.assertTrue(os.path.isfile(os.path.join(self.paths.gms_lib, 140 'dummy_file'))) 141 self.assertTrue(os.path.isfile(self.paths.gms_root_sha1)) 142 self.assertTrue(os.path.isfile(self.paths.gms_root_license)) 143 self.assertEquals(_GetFileContent(self.paths.gms_root_license), 144 self.DEFAULT_LICENSE) 145 146 def testDownloadAlreadyUpToDate(self): 147 self.SetUpWorkdir( 148 populate_bucket=True, 149 existing_zip_sha1=self.DEFAULT_ZIP_SHA1) 150 151 status = update.main([ 152 'download', 153 '--dry-run', 154 '--bucket', self.paths.bucket, 155 '--config', self.paths.config_file, 156 '--sdk-root', self.paths.sdk_root, 157 ]) 158 159 self.assertEqual(status, 0, 'the command should have succeeded.') 160 161 # there should not be new files downloaded to sdk_root 162 self.assertFalse(os.path.isfile(os.path.join(self.paths.gms_lib, 163 'dummy_file'))) 164 self.assertFalse(os.path.isfile(self.paths.gms_root_license)) 165 166 def testDownloadAcceptedLicense(self): 167 self.SetUpWorkdir( 168 populate_bucket=True, 169 existing_license=self.DEFAULT_LICENSE) 170 171 # License already accepted, no need to type 172 status = update.main([ 173 'download', 174 '--dry-run', 175 '--bucket', self.paths.bucket, 176 '--config', self.paths.config_file, 177 '--sdk-root', self.paths.sdk_root, 178 ]) 179 180 self.assertEqual(status, 0, 'the command should have succeeded.') 181 182 # sdk_root should contain zip contents, zip sha1, license 183 self.assertTrue(os.path.isfile(os.path.join(self.paths.gms_lib, 184 'dummy_file'))) 185 self.assertTrue(os.path.isfile(self.paths.gms_root_sha1)) 186 self.assertTrue(os.path.isfile(self.paths.gms_root_license)) 187 self.assertEquals(_GetFileContent(self.paths.gms_root_license), 188 self.DEFAULT_LICENSE) 189 190 def testDownloadNewLicense(self): 191 self.SetUpWorkdir( 192 populate_bucket=True, 193 existing_license='Old license') 194 195 with _MockedInput('y'): 196 status = update.main([ 197 'download', 198 '--dry-run', 199 '--bucket', self.paths.bucket, 200 '--config', self.paths.config_file, 201 '--sdk-root', self.paths.sdk_root, 202 ]) 203 204 self.assertEqual(status, 0, 'the command should have succeeded.') 205 206 # sdk_root should contain zip contents, zip sha1, NEW license 207 self.assertTrue(os.path.isfile(os.path.join(self.paths.gms_lib, 208 'dummy_file'))) 209 self.assertTrue(os.path.isfile(self.paths.gms_root_sha1)) 210 self.assertTrue(os.path.isfile(self.paths.gms_root_license)) 211 self.assertEquals(_GetFileContent(self.paths.gms_root_license), 212 self.DEFAULT_LICENSE) 213 214 def testDownloadRefusedLicense(self): 215 self.SetUpWorkdir( 216 populate_bucket=True, 217 existing_license='Old license') 218 219 with _MockedInput('n'): 220 status = update.main([ 221 'download', 222 '--dry-run', 223 '--bucket', self.paths.bucket, 224 '--config', self.paths.config_file, 225 '--sdk-root', self.paths.sdk_root, 226 ]) 227 228 self.assertEqual(status, 0, 'the command should have succeeded.') 229 230 # there should not be new files downloaded to sdk_root 231 self.assertFalse(os.path.isfile(os.path.join(self.paths.gms_lib, 232 'dummy_file'))) 233 self.assertEquals(_GetFileContent(self.paths.gms_root_license), 234 'Old license') 235 236 def testDownloadNoAndroidSDK(self): 237 self.SetUpWorkdir( 238 populate_bucket=True, 239 existing_license='Old license') 240 241 non_existing_sdk_root = os.path.join(self.workdir, 'non_existing_sdk_root') 242 # Should not run, no typing needed 243 status = update.main([ 244 'download', 245 '--dry-run', 246 '--bucket', self.paths.bucket, 247 '--config', self.paths.config_file, 248 '--sdk-root', non_existing_sdk_root, 249 ]) 250 251 self.assertEqual(status, 0, 'the command should have succeeded.') 252 self.assertFalse(os.path.isdir(non_existing_sdk_root)) 253 254 def SetUpWorkdir(self, 255 bot_env=False, 256 config_version=DEFAULT_CONFIG_VERSION, 257 existing_license=None, 258 existing_zip_sha1=None, 259 gms_lib=False, 260 populate_bucket=False, 261 source_prop=None, 262 xml_version=None): 263 '''Prepares workdir by putting it in the specified state 264 265 Args: 266 - general 267 bot_env: sets or unsets CHROME_HEADLESS 268 269 - bucket 270 populate_bucket: boolean. Populate the bucket with a zip and license 271 file. The sha1s will be copied to the config directory 272 273 - config 274 config_version: number. Version of the current SDK. Defaults to 275 `self.DEFAULT_CONFIG_VERSION` 276 277 - sdk_root 278 existing_license: string. Create a LICENSE file setting the specified 279 text as content of the currently accepted license. 280 existing_zip_sha1: string. Create a sha1 file setting the specified 281 hash as hash of the SDK supposed to be installed 282 gms_lib: boolean. Create a dummy file in the location of the play 283 services SDK. 284 source_prop: boolean. Create a source.properties file that contains 285 the license to upload. 286 xml_version: number. Create a version.xml file with the specified 287 version that is used when uploading 288 ''' 289 self.paths = Paths(self.workdir) 290 291 # Create the main directories 292 _MakeDirs(self.paths.sdk_root) 293 _MakeDirs(self.paths.config_dir) 294 _MakeDirs(self.paths.bucket) 295 296 # is not configured via argument. 297 update.SHA1_DIRECTORY = self.paths.config_dir 298 299 os.environ['CHROME_HEADLESS'] = '1' if bot_env else '' 300 301 if config_version: 302 _MakeDirs(os.path.dirname(self.paths.config_file)) 303 with open(self.paths.config_file, 'w') as stream: 304 stream.write(('{"version_number":%d,' 305 '"version_xml_path": "res/values/version.xml"}' 306 '\n') % config_version) 307 308 if existing_license: 309 _MakeDirs(self.paths.gms_root) 310 with open(self.paths.gms_root_license, 'w') as stream: 311 stream.write(existing_license) 312 313 if existing_zip_sha1: 314 _MakeDirs(self.paths.gms_root) 315 with open(self.paths.gms_root_sha1, 'w') as stream: 316 stream.write(existing_zip_sha1) 317 318 if gms_lib: 319 _MakeDirs(self.paths.gms_lib) 320 with open(os.path.join(self.paths.gms_lib, 'dummy_file'), 'w') as stream: 321 stream.write('foo\n') 322 323 if source_prop: 324 _MakeDirs(os.path.dirname(self.paths.source_prop)) 325 with open(self.paths.source_prop, 'w') as stream: 326 stream.write('Foo=Bar\n' 327 'Pkg.License=%s\n' 328 'Baz=Fizz\n' % self.DEFAULT_LICENSE) 329 330 if populate_bucket: 331 _MakeDirs(self.paths.config_dir) 332 bucket_dir = os.path.join(self.paths.bucket, str(config_version)) 333 _MakeDirs(bucket_dir) 334 335 # TODO(dgn) should we use real sha1s? comparison with the real sha1 is 336 # done but does not do anything other than displaying a message. 337 config_license_sha1 = 'license0and0filling0to0forty0chars000000' 338 with open(self.paths.config_license_sha1, 'w') as stream: 339 stream.write(config_license_sha1) 340 341 with open(os.path.join(bucket_dir, config_license_sha1), 'w') as stream: 342 stream.write(self.DEFAULT_LICENSE) 343 344 config_zip_sha1 = self.DEFAULT_ZIP_SHA1 345 with open(self.paths.config_zip_sha1, 'w') as stream: 346 stream.write(config_zip_sha1) 347 348 pre_zip_lib = os.path.join(self.workdir, 'pre_zip_lib') 349 post_zip_lib = os.path.join(bucket_dir, config_zip_sha1) 350 _MakeDirs(pre_zip_lib) 351 with open(os.path.join(pre_zip_lib, 'dummy_file'), 'w') as stream: 352 stream.write('foo\n') 353 shutil.make_archive(post_zip_lib, 'zip', pre_zip_lib) 354 # make_archive appends .zip 355 shutil.move(post_zip_lib + '.zip', post_zip_lib) 356 357 if xml_version: 358 _MakeDirs(os.path.dirname(self.paths.xml_version)) 359 with open(self.paths.xml_version, 'w') as stream: 360 stream.write( 361 '<?xml version="1.0" encoding="utf-8"?>\n' 362 '<resources>\n' 363 ' <integer name="google_play_services_version">%d</integer>\n' 364 '</resources>\n' % xml_version) 365 366 367 class Paths(object): 368 '''Declaration of the paths commonly manipulated in the tests.''' 369 370 def __init__(self, workdir): 371 self.bucket = os.path.join(workdir, 'bucket') 372 373 self.config_dir = os.path.join(workdir, 'config') 374 self.config_file = os.path.join(self.config_dir, 'config.json') 375 self.config_license_sha1 = os.path.join(self.config_dir, 'LICENSE.sha1') 376 self.config_zip_sha1 = os.path.join( 377 self.config_dir, 378 'google_play_services_library.zip.sha1') 379 380 self.sdk_root = os.path.join(workdir, 'sdk_root') 381 self.gms_root = os.path.join(self.sdk_root, 'extras', 'google', 382 'google_play_services') 383 self.gms_root_sha1 = os.path.join(self.gms_root, 384 'google_play_services_library.zip.sha1') 385 self.gms_root_license = os.path.join(self.gms_root, 'LICENSE') 386 self.source_prop = os.path.join(self.gms_root, 'source.properties') 387 self.gms_lib = os.path.join(self.gms_root, 'libproject', 388 'google-play-services_lib') 389 self.xml_version = os.path.join(self.gms_lib, 'res', 'values', 390 'version.xml') 391 392 393 def _GetFileContent(file_path): 394 with open(file_path, 'r') as stream: 395 return stream.read() 396 397 398 def _MakeDirs(path): 399 '''Avoids having to do the error handling everywhere.''' 400 if not os.path.exists(path): 401 os.makedirs(path) 402 403 404 @contextlib.contextmanager 405 def _MockedInput(typed_string): 406 '''Makes raw_input return |typed_string| while inside the context.''' 407 try: 408 original_raw_input = __builtins__.raw_input 409 __builtins__.raw_input = lambda _: typed_string 410 yield 411 finally: 412 __builtins__.raw_input = original_raw_input 413 414 415 if __name__ == '__main__': 416 unittest.main() 417
