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, 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()