src/third_party/chromite/lib/paygen/urilib_unittest.py

   1 #!/usr/bin/python
   2 # Copyright (c) 2012 The Chromium OS 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 """Test the urilib module."""
   7
   8 from __future__ import print_function
   9
  10 import os
  11
  12 import fixup_path
  13 fixup_path.FixupPath()
  14
  15 from chromite.lib import cros_test_lib
  16 from chromite.lib import osutils
  17
  18 from chromite.lib.paygen import filelib
  19 from chromite.lib.paygen import gslib
  20 from chromite.lib.paygen import unittest_lib
  21 from chromite.lib.paygen import urilib
  22
  23
  24 # We access private members to test them.
  25 # pylint: disable-msg=E1101,W0201
  26
  27
  28 class FakeHttpResponse(object):
  29   """For simulating http response objects."""
  30
  31   class FakeHeaders(object):
  32     """Helper class for faking HTTP headers in a response."""
  33     def __init__(self, headers_dict):
  34       self.headers_dict = headers_dict
  35
  36     def getheader(self, name):
  37       return self.headers_dict.get(name)
  38
  39   def __init__(self, code, headers_dict=None):
  40     self.code = code
  41     self.headers = FakeHttpResponse.FakeHeaders(headers_dict)
  42
  43   def getcode(self):
  44     return self.code
  45
  46
  47 class TestFileManipulation(unittest_lib.TestCase):
  48   """Test general urilib file methods together."""
  49   # pylint: disable-msg=E1101
  50
  51   FILE1 = 'file1'
  52   FILE2 = 'file2'
  53   SUBDIR = 'subdir'
  54   SUBFILE = '%s/file3' % SUBDIR
  55
  56   FILE1_CONTENTS = 'Howdy doody there dandy'
  57   FILE2_CONTENTS = 'Once upon a time in a galaxy far far away.'
  58   SUBFILE_CONTENTS = 'Five little monkeys jumped on the bed.'
  59
  60   GS_DIR = 'gs://chromeos-releases-public/unittest'
  61
  62   def _SetUpDirs(self):
  63     self.file1_local = os.path.join(self.tempdir, self.FILE1)
  64     self.file2_local = os.path.join(self.tempdir, self.FILE2)
  65     self.subdir_local = os.path.join(self.tempdir, self.SUBDIR)
  66     self.subfile_local = os.path.join(self.tempdir, self.SUBFILE)
  67
  68     self.file1_gs = os.path.join(self.GS_DIR, self.FILE1)
  69     self.file2_gs = os.path.join(self.GS_DIR, self.FILE2)
  70     self.subdir_gs = os.path.join(self.GS_DIR, self.SUBDIR)
  71     self.subfile_gs = os.path.join(self.GS_DIR, self.SUBFILE)
  72
  73     # Pre-populate local dir with contents.
  74     with open(self.file1_local, 'w') as out1:
  75       out1.write(self.FILE1_CONTENTS)
  76
  77     with open(self.file2_local, 'w') as out2:
  78       out2.write(self.FILE2_CONTENTS)
  79
  80     os.makedirs(self.subdir_local)
  81
  82     with open(self.subfile_local, 'w') as out3:
  83       out3.write(self.SUBFILE_CONTENTS)
  84
  85     # Make sure gs:// directory is ready (empty).
  86     gslib.Remove(os.path.join(self.GS_DIR, '*'), recurse=True,
  87                  ignore_no_match=True)
  88
  89   @cros_test_lib.NetworkTest()
  90   @osutils.TempDirDecorator
  91   def testIntegration(self):
  92     self._SetUpDirs()
  93
  94     self.assertTrue(urilib.Exists(self.tempdir, as_dir=True))
  95     self.assertTrue(urilib.Exists(self.file1_local))
  96     self.assertTrue(urilib.Exists(self.file2_local))
  97     self.assertTrue(urilib.Exists(self.subfile_local))
  98     self.assertTrue(urilib.Exists(self.subdir_local, as_dir=True))
  99
 100     self.assertFalse(urilib.Exists(self.file1_gs))
 101     self.assertFalse(urilib.Exists(self.file2_gs))
 102     self.assertFalse(urilib.Exists(self.subfile_gs))
 103
 104     shallow_local_files = [self.file1_local, self.file2_local]
 105     deep_local_files = shallow_local_files + [self.subfile_local]
 106     shallow_gs_files = [self.file1_gs, self.file2_gs]
 107     deep_gs_files = shallow_gs_files + [self.subfile_gs]
 108
 109     # Test ListFiles, local version.
 110     self.assertEquals(set(shallow_local_files),
 111                       set(urilib.ListFiles(self.tempdir)))
 112     self.assertEquals(set(deep_local_files),
 113                       set(urilib.ListFiles(self.tempdir, recurse=True)))
 114
 115     # Test CopyFiles, from local to GS.
 116     self.assertEquals(set(deep_gs_files),
 117                       set(urilib.CopyFiles(self.tempdir, self.GS_DIR)))
 118
 119     # Test ListFiles, GS version.
 120     self.assertEquals(set(shallow_gs_files),
 121                       set(urilib.ListFiles(self.GS_DIR)))
 122     self.assertEquals(set(deep_gs_files),
 123                       set(urilib.ListFiles(self.GS_DIR, recurse=True)))
 124
 125     # Test Cmp between some files.
 126     self.assertTrue(urilib.Cmp(self.file1_local, self.file1_gs))
 127     self.assertFalse(urilib.Cmp(self.file2_local, self.file1_gs))
 128
 129     # Test RemoveDirContents, local version.
 130     urilib.RemoveDirContents(self.tempdir)
 131     self.assertFalse(urilib.ListFiles(self.tempdir))
 132
 133     # Test CopyFiles, from GS to local.
 134     self.assertEquals(set(deep_local_files),
 135                       set(urilib.CopyFiles(self.GS_DIR, self.tempdir)))
 136
 137     # Test RemoveDirContents, GS version.
 138     urilib.RemoveDirContents(self.GS_DIR)
 139     self.assertFalse(urilib.ListFiles(self.GS_DIR))
 140
 141
 142 class TestUrilib(unittest_lib.MoxTestCase):
 143   """Test urilib module."""
 144
 145   def testExtractProtocol(self):
 146     tests = {'gs': ['gs://',
 147                     'gs://foo',
 148                     'gs://foo/bar'
 149                     ],
 150              'abc': ['abc://',
 151                      'abc://foo',
 152                      'abc://foo/bar',
 153                      ],
 154              None: ['foo/bar',
 155                     '/foo/bar',
 156                     '://garbage/path',
 157                     ],
 158              }
 159
 160     for protocol in tests:
 161       for uri in tests[protocol]:
 162         self.assertEquals(protocol, urilib.ExtractProtocol(uri))
 163
 164   def testGetUriType(self):
 165     tests = {'gs': ['gs://',
 166                     'gs://foo',
 167                     'gs://foo/bar'
 168                     ],
 169              'abc': ['abc://',
 170                      'abc://foo',
 171                      'abc://foo/bar',
 172                      ],
 173              'file': ['foo/bar',
 174                       '/foo/bar',
 175                       '://garbage/path',
 176                       '/cnsfoo/bar',
 177                       ],
 178              }
 179
 180     for uri_type in tests:
 181       for uri in tests[uri_type]:
 182         self.assertEquals(uri_type, urilib.GetUriType(uri))
 183
 184   def testSplitURI(self):
 185     tests = [
 186       ['gs', 'foo', 'gs://foo'],
 187       ['gs', 'foo/bar', 'gs://foo/bar'],
 188       ['file', '/foo/bar', 'file:///foo/bar'],
 189       [None, '/foo/bar', '/foo/bar'],
 190       ]
 191
 192     for test in tests:
 193       uri = test[2]
 194       protocol, path = urilib.SplitURI(uri)
 195       self.assertEquals(test[0], protocol)
 196       self.assertEquals(test[1], path)
 197
 198   def testIsGsURI(self):
 199     tests_true = ('gs://',
 200                   'gs://foo',
 201                   'gs://foo/bar',
 202                   )
 203     for test in tests_true:
 204       self.assertTrue(urilib.IsGsURI(test))
 205
 206     tests_false = ('gsfoo/bar',
 207                    'gs/foo/bar',
 208                    'gs',
 209                    '/foo/bar',
 210                    '/gs',
 211                    '/gs/foo/bar'
 212                    'file://foo/bar',
 213                    'http://foo/bar',
 214                    )
 215     for test in tests_false:
 216       self.assertFalse(urilib.IsGsURI(test))
 217
 218   def testIsFileURI(self):
 219     tests_true = ('file://',
 220                   'file://foo/bar',
 221                   'file:///foo/bar',
 222                   '/foo/bar',
 223                   'foo/bar',
 224                   'foo',
 225                   '',
 226                   )
 227     for test in tests_true:
 228       self.assertTrue(urilib.IsFileURI(test))
 229
 230     tests_false = ('gs://',
 231                    'foo://',
 232                    'gs://foo/bar',
 233                    )
 234     for test in tests_false:
 235       self.assertFalse(urilib.IsFileURI(test))
 236
 237   def testIsHttpURI(self):
 238     tests_true = ('http://',
 239                   'http://foo',
 240                   'http://foo/bar',
 241                   )
 242     for test in tests_true:
 243       self.assertTrue(urilib.IsHttpURI(test))
 244
 245     tests_https_true = ('https://',
 246                         'https://foo',
 247                         'https://foo/bar',
 248                         )
 249     for test in tests_https_true:
 250       self.assertTrue(urilib.IsHttpURI(test, https_ok=True))
 251     for test in tests_https_true:
 252       self.assertFalse(urilib.IsHttpURI(test))
 253
 254     tests_false = ('httpfoo/bar',
 255                    'http/foo/bar',
 256                    'http',
 257                    '/foo/bar',
 258                    '/http',
 259                    '/http/foo/bar'
 260                    'file:///foo/bar',
 261                    'gs://foo/bar',
 262                    )
 263     for test in tests_false:
 264       self.assertFalse(urilib.IsHttpURI(test))
 265
 266   def testIsHttpsURI(self):
 267     tests_true = ('https://',
 268                   'https://foo',
 269                   'https://foo/bar',
 270                   )
 271     for test in tests_true:
 272       self.assertTrue(urilib.IsHttpsURI(test))
 273
 274     tests_false = ('http://',
 275                    'http://foo',
 276                    'http://foo/bar',
 277                    'httpfoo/bar',
 278                    'http/foo/bar',
 279                    'http',
 280                    '/foo/bar',
 281                    '/http',
 282                    '/http/foo/bar'
 283                    'file:///foo/bar',
 284                    'gs://foo/bar',
 285                    )
 286     for test in tests_false:
 287       self.assertFalse(urilib.IsHttpsURI(test))
 288
 289   def testMD5Sum(self):
 290     gs_path = 'gs://bucket/some/path'
 291     local_path = '/some/local/path'
 292     http_path = 'http://host.domain/some/path'
 293
 294     self.mox.StubOutWithMock(gslib, 'MD5Sum')
 295     self.mox.StubOutWithMock(filelib, 'MD5Sum')
 296
 297     # Set up the test replay script.
 298     # Run 1, GS.
 299     gslib.MD5Sum(gs_path).AndReturn('TheResult')
 300     # Run 3, local file.
 301     filelib.MD5Sum(local_path).AndReturn('TheResult')
 302     self.mox.ReplayAll()
 303
 304     # Run the test verification.
 305     self.assertEquals('TheResult', urilib.MD5Sum(gs_path))
 306     self.assertEquals('TheResult', urilib.MD5Sum(local_path))
 307     self.assertRaises(urilib.NotSupportedForType, urilib.MD5Sum, http_path)
 308     self.mox.VerifyAll()
 309
 310   def testCmp(self):
 311     gs_path = 'gs://bucket/some/path'
 312     local_path = '/some/local/path'
 313     http_path = 'http://host.domain/some/path'
 314
 315     result = 'TheResult'
 316
 317     self.mox.StubOutWithMock(gslib, 'Cmp')
 318     self.mox.StubOutWithMock(filelib, 'Cmp')
 319
 320     # Set up the test replay script.
 321     # Run 1, two local files.
 322     filelib.Cmp(local_path, local_path + '.1').AndReturn(result)
 323     # Run 2, local and GS.
 324     gslib.Cmp(local_path, gs_path).AndReturn(result)
 325     # Run 4, GS and GS
 326     gslib.Cmp(gs_path, gs_path + '.1').AndReturn(result)
 327     # Run 7, local and HTTP
 328     self.mox.ReplayAll()
 329
 330     # Run the test verification.
 331     self.assertEquals(result, urilib.Cmp(local_path, local_path + '.1'))
 332     self.assertEquals(result, urilib.Cmp(local_path, gs_path))
 333     self.assertEquals(result, urilib.Cmp(gs_path, gs_path + '.1'))
 334     self.assertRaises(urilib.NotSupportedBetweenTypes, urilib.Cmp,
 335                       local_path, http_path)
 336     self.mox.VerifyAll()
 337
 338   @cros_test_lib.NetworkTest()
 339   @osutils.TempDirDecorator
 340   def testURLRetrieve(self):
 341     good_url = 'https://codereview.chromium.org/download/issue11731004_1_2.diff'
 342     bad_domain_url = 'http://notarealdomainireallyhope.com/some/path'
 343     bad_path_url = 'https://dl.google.com/dl/edgedl/x/y/z/a/b/c/foobar'
 344     local_path = os.path.join(self.tempdir, 'downloaded_file')
 345     bad_local_path = '/tmp/a/b/c/d/x/y/z/foobar'
 346
 347     git_index1 = 'e6c0d72a5122171deb4c458991d1c7547f31a2f0'
 348     git_index2 = '3d0f7d3edfd8146031e66dc3f45926920d3ded78'
 349     expected_contents = """Index: LICENSE
 350 diff --git a/LICENSE b/LICENSE
 351 index %s..%s 100644
 352 --- a/LICENSE
 353 +++ b/LICENSE
 354 @@ -1,4 +1,4 @@
 355 -// Copyright (c) 2012 The Chromium Authors. All rights reserved.
 356 +// Copyright (c) 2013 The Chromium Authors. All rights reserved.
 357  //
 358  // Redistribution and use in source and binary forms, with or without
 359  // modification, are permitted provided that the following conditions are
 360 """ % (git_index1, git_index2)
 361
 362     self.assertRaises(urilib.MissingURLError, urilib.URLRetrieve,
 363                       bad_path_url, local_path)
 364     self.assertRaises(urilib.MissingURLError, urilib.URLRetrieve,
 365                       bad_domain_url, local_path)
 366
 367     urilib.URLRetrieve(good_url, local_path)
 368     with open(local_path, 'r') as f:
 369       actual_contents = f.read()
 370     self.assertEqual(expected_contents, actual_contents)
 371
 372     self.assertRaises(IOError, urilib.URLRetrieve, good_url, bad_local_path)
 373
 374
 375   def testCopy(self):
 376     gs_path = 'gs://bucket/some/path'
 377     local_path = '/some/local/path'
 378     http_path = 'http://host.domain/some/path'
 379
 380     result = 'TheResult'
 381
 382     self.mox.StubOutWithMock(gslib, 'Copy')
 383     self.mox.StubOutWithMock(filelib, 'Copy')
 384     self.mox.StubOutWithMock(urilib, 'URLRetrieve')
 385
 386     # Set up the test replay script.
 387     # Run 1, two local files.
 388     filelib.Copy(local_path, local_path + '.1').AndReturn(result)
 389     # Run 2, local and GS.
 390     gslib.Copy(local_path, gs_path).AndReturn(result)
 391     # Run 4, GS and GS
 392     gslib.Copy(gs_path, gs_path + '.1').AndReturn(result)
 393     # Run 7, HTTP and local
 394     urilib.URLRetrieve(http_path, local_path).AndReturn(result)
 395     # Run 8, local and HTTP
 396     self.mox.ReplayAll()
 397
 398     # Run the test verification.
 399     self.assertEquals(result, urilib.Copy(local_path, local_path + '.1'))
 400     self.assertEquals(result, urilib.Copy(local_path, gs_path))
 401     self.assertEquals(result, urilib.Copy(gs_path, gs_path + '.1'))
 402     self.assertEquals(result, urilib.Copy(http_path, local_path))
 403     self.assertRaises(urilib.NotSupportedBetweenTypes, urilib.Copy,
 404                       local_path, http_path)
 405     self.mox.VerifyAll()
 406
 407   def testRemove(self):
 408     gs_path = 'gs://bucket/some/path'
 409     local_path = '/some/local/path'
 410     http_path = 'http://host.domain/some/path'
 411
 412     self.mox.StubOutWithMock(gslib, 'Remove')
 413     self.mox.StubOutWithMock(filelib, 'Remove')
 414
 415     # Set up the test replay script.
 416     # Run 1, two local files.
 417     filelib.Remove(local_path, local_path + '.1')
 418     # Run 2, local and GS.
 419     gslib.Remove(local_path, gs_path, ignore_no_match=True)
 420     # Run 4, GS and GS
 421     gslib.Remove(gs_path, gs_path + '.1',
 422                    ignore_no_match=True, recurse=True)
 423     # Run 7, local and HTTP
 424     self.mox.ReplayAll()
 425
 426     # Run the test verification.
 427     urilib.Remove(local_path, local_path + '.1')
 428     urilib.Remove(local_path, gs_path, ignore_no_match=True)
 429     urilib.Remove(gs_path, gs_path + '.1', ignore_no_match=True, recurse=True)
 430     self.assertRaises(urilib.NotSupportedForTypes, urilib.Remove,
 431                       local_path, http_path)
 432     self.mox.VerifyAll()
 433
 434   def testSize(self):
 435     gs_path = 'gs://bucket/some/path'
 436     local_path = '/some/local/path'
 437     http_path = 'http://host.domain/some/path'
 438     ftp_path = 'ftp://host.domain/some/path'
 439
 440     result = 100
 441     http_response = FakeHttpResponse(200, {'Content-Length': str(result)})
 442
 443     self.mox.StubOutWithMock(gslib, 'FileSize')
 444     self.mox.StubOutWithMock(filelib, 'Size')
 445     self.mox.StubOutWithMock(urilib.urllib2, 'urlopen')
 446
 447     # Set up the test replay script.
 448     # Run 1, local.
 449     filelib.Size(local_path).AndReturn(result)
 450     # Run 2, GS.
 451     gslib.FileSize(gs_path).AndReturn(result)
 452     # Run 4, HTTP.
 453     urilib.urllib2.urlopen(http_path).AndReturn(http_response)
 454     # Run 5, FTP.
 455     self.mox.ReplayAll()
 456
 457     # Run the test verification.
 458     self.assertEquals(result, urilib.Size(local_path))
 459     self.assertEquals(result, urilib.Size(gs_path))
 460     self.assertEquals(result, urilib.Size(http_path))
 461     self.assertRaises(urilib.NotSupportedForType, urilib.Size, ftp_path)
 462     self.mox.VerifyAll()
 463
 464   def testExists(self):
 465     gs_path = 'gs://bucket/some/path'
 466     local_path = '/some/local/path'
 467     http_path = 'http://host.domain/some/path'
 468     ftp_path = 'ftp://host.domain/some/path'
 469
 470     result = 'TheResult'
 471
 472     self.mox.StubOutWithMock(gslib, 'Exists')
 473     self.mox.StubOutWithMock(filelib, 'Exists')
 474     self.mox.StubOutWithMock(urilib.urllib2, 'urlopen')
 475
 476     # Set up the test replay script.
 477     # Run 1, local, as_dir=False
 478     filelib.Exists(local_path, as_dir=False).AndReturn(result)
 479     # Run 2, GS, as_dir=False.
 480     gslib.Exists(gs_path).AndReturn(result)
 481     # Run 3, GS, as_dir=True.
 482     # Run 6, HTTP, as_dir=False, code=200.
 483     urilib.urllib2.urlopen(http_path).AndReturn(FakeHttpResponse(200))
 484     # Run 7, HTTP, as_dir=False, code=404.
 485     urilib.urllib2.urlopen(http_path).AndReturn(FakeHttpResponse(404))
 486     # Run 8, HTTP, as_dir=False, HTTPError.
 487     urilib.urllib2.urlopen(http_path).AndRaise(
 488         urilib.urllib2.HTTPError('url', 404, 'msg', None, None))
 489     # Run 9, HTTP, as_dir=True.
 490     # Run 10, FTP, as_dir=False.
 491     self.mox.ReplayAll()
 492
 493     # Run the test verification.
 494     self.assertEquals(result, urilib.Exists(local_path))
 495     self.assertEquals(result, urilib.Exists(gs_path))
 496     self.assertEquals(False, urilib.Exists(gs_path, as_dir=True))
 497     self.assertTrue(urilib.Exists(http_path))
 498     self.assertFalse(urilib.Exists(http_path))
 499     self.assertFalse(urilib.Exists(http_path))
 500     self.assertRaises(urilib.NotSupportedForType,
 501                       urilib.Exists, http_path, as_dir=True)
 502     self.assertRaises(urilib.NotSupportedForType, urilib.Exists, ftp_path)
 503     self.mox.VerifyAll()
 504
 505   def testListFiles(self):
 506     gs_path = 'gs://bucket/some/path'
 507     local_path = '/some/local/path'
 508     http_path = 'http://host.domain/some/path'
 509
 510     result = 'TheResult'
 511     patt = 'TheFilePattern'
 512
 513     self.mox.StubOutWithMock(gslib, 'ListFiles')
 514     self.mox.StubOutWithMock(filelib, 'ListFiles')
 515
 516     # Set up the test replay script.
 517     # Run 1, local.
 518     filelib.ListFiles(local_path, recurse=True, filepattern=None, sort=False
 519                       ).AndReturn(result)
 520     # Run 2, GS.
 521     gslib.ListFiles(gs_path, recurse=False, filepattern=patt, sort=True
 522                       ).AndReturn(result)
 523     # Run 4, HTTP.
 524     self.mox.ReplayAll()
 525
 526     # Run the test verification.
 527     self.assertEquals(result, urilib.ListFiles(local_path, recurse=True))
 528     self.assertEquals(result, urilib.ListFiles(gs_path, filepattern=patt,
 529                                                sort=True))
 530     self.assertRaises(urilib.NotSupportedForType, urilib.ListFiles, http_path)
 531     self.mox.VerifyAll()
 532
 533
 534 if __name__ == '__main__':
 535   cros_test_lib.main()