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, action): 20 self._absolute_local_path = absolute_local_path 21 self.action = action 22 23 def AbsoluteLocalPath(self): 24 return self._absolute_local_path 25 26 def Action(self): 27 return self.action 28 29 class InputAPIStub(object): 30 def __init__(self, paths, deleted_paths=None, added_paths=None): 31 self._paths = paths 32 self._deleted_paths = deleted_paths if deleted_paths else [] 33 self._added_paths = added_paths if added_paths else [] 34 35 def AffectedFiles(self, include_deletes=True, file_filter=None): 36 if not file_filter: 37 file_filter = lambda x: True 38 39 affected_files = [] 40 for path in self._paths: 41 affected_file_stub = AffectedFileStub(path, 'M') 42 if file_filter(affected_file_stub): 43 affected_files.append(affected_file_stub) 44 45 for path in self._added_paths: 46 affected_file_stub = AffectedFileStub(path, 'A') 47 if file_filter(affected_file_stub): 48 affected_files.append(affected_file_stub) 49 50 if include_deletes: 51 for path in self._deleted_paths: 52 affected_file_stub = AffectedFileStub(path, 'D') 53 if file_filter(affected_file_stub): 54 affected_files.append(affected_file_stub) 55 return affected_files 56 57 def AbsoluteLocalPaths(self): 58 return [af.AbsoluteLocalPath() for af in self.AffectedFiles()] 59 60 def PresubmitLocalPath(self): 61 return PRESUBMIT.__file__ 62 63 64 class OutputAPIStub(object): 65 class PresubmitError(Exception): 66 pass 67 68 class PresubmitNotifyResult(Exception): 69 pass 70 71 72 PRESUBMIT.LoadSupport(InputAPIStub([])) # do this to support monkey patching 73 74 class PresubmitTest(unittest.TestCase): 75 def setUp(self): 76 success_file_hash = 'da39a3ee5e6b4b0d3255bfef95601890afd80709' 77 78 self._stubs = system_stub.Override( 79 PRESUBMIT, ['cloud_storage', 'os', 'raw_input']) 80 self._stubs.raw_input.input = 'public' 81 # Files in Cloud Storage. 82 self._stubs.cloud_storage.remote_paths = [ 83 'skip'.zfill(40), 84 ] 85 # Local data files and their hashes. 86 self._stubs.cloud_storage.local_file_hashes = { 87 '/path/to/skip.wpr': 'skip'.zfill(40), 88 '/path/to/success.wpr': success_file_hash, 89 '/path/to/wrong_hash.wpr': success_file_hash, 90 } 91 # Local data files. 92 self._stubs.os.path.files = ( 93 self._stubs.cloud_storage.local_file_hashes.keys()) 94 # Local hash files and their contents. 95 self._stubs.cloud_storage.local_hash_files = { 96 '/path/to/invalid_hash.wpr.sha1': 'invalid_hash', 97 '/path/to/missing.wpr.sha1': 'missing'.zfill(40), 98 '/path/to/success.wpr.sha1': success_file_hash, 99 '/path/to/skip.wpr.sha1': 'skip'.zfill(40), 100 '/path/to/wrong_hash.wpr.sha1': 'wronghash'.zfill(40), 101 } 102 103 def tearDown(self): 104 self._stubs.Restore() 105 106 def assertResultCount(self, results, expected_errors, expected_notifications): 107 counts = collections.defaultdict(int) 108 for result in results: 109 counts[type(result)] += 1 110 actual_errors = counts[OutputAPIStub.PresubmitError] 111 actual_notifications = counts[OutputAPIStub.PresubmitNotifyResult] 112 self.assertEqual(expected_errors, actual_errors, 113 msg='Expected %d errors, but got %d. Results: %s' % 114 (expected_errors, actual_errors, results)) 115 self.assertEqual(expected_notifications, actual_notifications, 116 msg='Expected %d notifications, but got %d. Results: %s' % 117 (expected_notifications, actual_notifications, results)) 118 119 def _CheckUpload(self, paths, deleted_paths=None, added_paths=None): 120 input_api = InputAPIStub(paths, deleted_paths, added_paths) 121 return PRESUBMIT.CheckChangeOnUpload(input_api, OutputAPIStub()) 122 123 def testIgnoreDeleted(self): 124 results = self._CheckUpload([], ['/path/to/deleted.wpr.sha1']) 125 self.assertResultCount(results, 0, 0) 126 127 def testIgnoreNonHashes(self): 128 results = self._CheckUpload(['/path/to/irrelevant.py']) 129 self.assertResultCount(results, 0, 0) 130 131 def testInvalidHash(self): 132 results = self._CheckUpload(['/path/to/invalid_hash.wpr.sha1']) 133 self.assertResultCount(results, 1, 0) 134 self.assertTrue('valid SHA-1 hash' in str(results[0]), msg=results[0]) 135 136 def testMissingFile(self): 137 results = self._CheckUpload(['/path/to/missing.wpr.sha1']) 138 self.assertResultCount(results, 1, 0) 139 self.assertTrue('not found' in str(results[0]), msg=results[0]) 140 141 def testSkip(self): 142 results = self._CheckUpload(['/path/to/skip.wpr.sha1']) 143 self.assertResultCount(results, 0, 0) 144 145 def testSuccess(self): 146 results = self._CheckUpload(['/path/to/success.wpr.sha1']) 147 self.assertResultCount(results, 0, 1) 148 self.assertTrue('Uploaded' in str(results[0]), msg=results[0]) 149 150 def testWrongHash(self): 151 results = self._CheckUpload(['/path/to/wrong_hash.wpr.sha1']) 152 self.assertTrue('does not match' in str(results[0]), msg=results[0]) 153 154 155 if __name__ == '__main__': 156 unittest.main() 157
