• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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
5import collections
6import os
7import sys
8import unittest
9
10PERF_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
11sys.path.insert(0, os.path.join(os.path.dirname(PERF_ROOT), 'telemetry'))
12from telemetry.unittest import system_stub
13
14sys.path.insert(0, PERF_ROOT)
15from page_sets import PRESUBMIT
16
17
18class 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
29class 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
64class OutputAPIStub(object):
65  class PresubmitError(Exception):
66    pass
67
68  class PresubmitNotifyResult(Exception):
69    pass
70
71
72PRESUBMIT.LoadSupport(InputAPIStub([]))   # do this to support monkey patching
73
74class 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
155if __name__ == '__main__':
156  unittest.main()
157