src/tools/perf/page_sets/presubmit_unittest.py

   1 # Copyright 2013 The Chromium Authors. All rights reserved.
   2 # Use of this source code is governed by a BSD-style license that can be
   3 # found in the LICENSE file.
   4
   5 import collections
   6 import os
   7 import sys
   8 import unittest
   9
  10 PERF_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
  11 sys.path.insert(0, os.path.join(os.path.dirname(PERF_ROOT), 'telemetry'))
  12 from telemetry.unittest import system_stub
  13
  14 sys.path.insert(0, PERF_ROOT)
  15 from page_sets import PRESUBMIT
  16
  17
  18 class AffectedFileStub(object):
  19   def __init__(self, absolute_local_path):
  20     self._absolute_local_path = absolute_local_path
  21
  22   def AbsoluteLocalPath(self):
  23     return self._absolute_local_path
  24
  25
  26 class InputAPIStub(object):
  27   def __init__(self, paths, deleted_paths=None):
  28     self._paths = paths
  29     if deleted_paths:
  30       self._deleted_paths = deleted_paths
  31     else:
  32       self._deleted_paths = []
  33
  34   def AffectedFiles(self, include_deletes=True):
  35     affected_files = [AffectedFileStub(path) for path in self._paths]
  36     if include_deletes:
  37       affected_files += [AffectedFileStub(path) for path in self._deleted_paths]
  38     return affected_files
  39
  40   def AbsoluteLocalPaths(self):
  41     return [af.AbsoluteLocalPath() for af in self.AffectedFiles()]
  42
  43   def PresubmitLocalPath(self):
  44     return PRESUBMIT.__file__
  45
  46
  47 class OutputAPIStub(object):
  48   class PresubmitError(Exception):
  49     pass
  50
  51   class PresubmitNotifyResult(Exception):
  52     pass
  53
  54
  55 PRESUBMIT.LoadSupport(InputAPIStub([]))   # do this to support monkey patching
  56
  57 class PresubmitTest(unittest.TestCase):
  58   def setUp(self):
  59     success_file_hash = 'da39a3ee5e6b4b0d3255bfef95601890afd80709'
  60
  61     self._stubs = system_stub.Override(
  62         PRESUBMIT, ['cloud_storage', 'os', 'raw_input'])
  63     self._stubs.raw_input.input = 'public'
  64     # Files in Cloud Storage.
  65     self._stubs.cloud_storage.remote_paths = [
  66         'skip'.zfill(40),
  67     ]
  68     # Local data files and their hashes.
  69     self._stubs.cloud_storage.local_file_hashes = {
  70         '/path/to/skip.wpr': 'skip'.zfill(40),
  71         '/path/to/success.wpr': success_file_hash,
  72         '/path/to/wrong_hash.wpr': success_file_hash,
  73     }
  74     # Local data files.
  75     self._stubs.os.path.files = (
  76         self._stubs.cloud_storage.local_file_hashes.keys())
  77     # Local hash files and their contents.
  78     self._stubs.cloud_storage.local_hash_files = {
  79         '/path/to/invalid_hash.wpr.sha1': 'invalid_hash',
  80         '/path/to/missing.wpr.sha1': 'missing'.zfill(40),
  81         '/path/to/success.wpr.sha1': success_file_hash,
  82         '/path/to/skip.wpr.sha1': 'skip'.zfill(40),
  83         '/path/to/wrong_hash.wpr.sha1': 'wronghash'.zfill(40),
  84     }
  85
  86   def tearDown(self):
  87     self._stubs.Restore()
  88
  89   def assertResultCount(self, results, expected_errors, expected_notifications):
  90     counts = collections.defaultdict(int)
  91     for result in results:
  92       counts[type(result)] += 1
  93     actual_errors = counts[OutputAPIStub.PresubmitError]
  94     actual_notifications = counts[OutputAPIStub.PresubmitNotifyResult]
  95     self.assertEqual(expected_errors, actual_errors,
  96         msg='Expected %d errors, but got %d. Results: %s' %
  97         (expected_errors, actual_errors, results))
  98     self.assertEqual(expected_notifications, actual_notifications,
  99         msg='Expected %d notifications, but got %d. Results: %s' %
 100         (expected_notifications, actual_notifications, results))
 101
 102   def _CheckUpload(self, paths, deleted_paths=None):
 103     input_api = InputAPIStub(paths, deleted_paths)
 104     return PRESUBMIT.CheckChangeOnUpload(input_api, OutputAPIStub())
 105
 106   def testIgnoreDeleted(self):
 107     results = self._CheckUpload([], ['/path/to/deleted.wpr.sha1'])
 108     self.assertResultCount(results, 0, 0)
 109
 110   def testIgnoreNonHashes(self):
 111     results = self._CheckUpload(['/path/to/irrelevant.py'])
 112     self.assertResultCount(results, 0, 0)
 113
 114   def testInvalidHash(self):
 115     results = self._CheckUpload(['/path/to/invalid_hash.wpr.sha1'])
 116     self.assertResultCount(results, 1, 0)
 117     self.assertTrue('valid SHA-1 hash' in str(results[0]), msg=results[0])
 118
 119   def testMissingFile(self):
 120     results = self._CheckUpload(['/path/to/missing.wpr.sha1'])
 121     self.assertResultCount(results, 1, 0)
 122     self.assertTrue('not found' in str(results[0]), msg=results[0])
 123
 124   def testSkip(self):
 125     results = self._CheckUpload(['/path/to/skip.wpr.sha1'])
 126     self.assertResultCount(results, 0, 0)
 127
 128   def testSuccess(self):
 129     results = self._CheckUpload(['/path/to/success.wpr.sha1'])
 130     self.assertResultCount(results, 0, 1)
 131     self.assertTrue('Uploaded' in str(results[0]), msg=results[0])
 132
 133   def testWrongHash(self):
 134     results = self._CheckUpload(['/path/to/wrong_hash.wpr.sha1'])
 135     self.assertTrue('does not match' in str(results[0]), msg=results[0])
 136
 137
 138 if __name__ == '__main__':
 139   unittest.main()