Add relative path support.

tests/pending_manager_test.py

 #!/usr/bin/python
 # Copyright (c) 2010 The Chromium Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
 """Unit tests for pending_manager.py."""
 
 import logging
 import os
+import re
 import sys
 import unittest
 import urllib2
 
 ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
 sys.path.insert(0, os.path.join(ROOT_DIR, '..'))
 
+import find_depot_tools  # pylint: disable=W0611
+import patch
+
 import pending_manager
 from verification import base
 from verification import fake
+from verification import project_base
 
 
 # Access to a protected member xxx of a client class
 # pylint: disable=W0212
 
-def _SendStackFail(last_tb, stack, url=None, maxlen=50):
+SVN_PATCH = (
+    'Index: chrome/file.cc\n'
+    '===================================================================\n'
+    '--- chrome/file.cc\t(revision 74690)\n'
+    '+++ chrome/file.cc\t(working copy)\n'
+    '@@ -80,10 +80,13 @@\n'
+    ' // Foo\n'
+    ' // Bar\n'
+    ' void foo() {\n'
+    '-   return bar;\n'
+    '+   return foo;\n'
+    ' }\n'
+    ' \n'
+    ' \n')
+
+
+def _SendStackFail(last_tb, stack, *args, **kwargs):
   """Fails a test that calls SendStack.
 
   In practice it doesn't happen when a test pass but will when a test fails so
   hook it here so breakpad doesn't send too many stack traces to maintainers.
   """
+  print '_SendStackFail() was called'
   print last_tb
   print stack
   assert False
 
 
 def read(filename):
   f = open(filename, 'rb')
   content = f.read()
   f.close()
   return content
 
 
 def write(filename, content):
   f = open(filename, 'wb')
   f.write(content)
   f.close()
 
 
 def trim(x):
   return x.replace(' ', '').replace('\n', '')
 
 
 class FakeRietveld(object):
   url = 'http://nowhere'
 
-  def __init__(self, test):
-    self.test = test
-
-  def get_issue_properties(self, issue_id, _):
-    self.test.assertEquals(issue_id, 31337)
-    return {
+  def __init__(self):
+    self.issues = {
+      31337: {
         "description": "foo",
         "created": "2010-12-27 03:23:31.149045",
         "cc": ["cc@example.com",],
         "reviewers": ["rev@example.com"],
         "owner_email": "author@example.com",
         "patchsets": [1],
         "modified": "2011-01-10 20:52:39.127231",
         "private": False,
         "base_url": "svn://fake/repo",
         "closed": False,
         "owner": "Author",
-        "issue": issue_id,
+        "issue": 31337,
         "subject": 'foo',
         "messages": [
           {
             "date": "2010-12-27 03:23:32.489999",
             "text": "hi!",
             "sender": "author@example.com",
             "recipients": ["rev@example.com", "cc@example.com"],
           },
         ],
         "commit": True,
+      },
     }
 
+  def get_pending_issues(self):
+    return self.issues.keys()
+
+  def get_issue_properties(self, issue_id, _):
+    return self.issues[issue_id]
+
   def close_issue(self, issue):
     pass
 
   def update_description(self, issue, text):
     pass
 
   def set_flag(self, issue, patchset, flag, value):
     pass
 
   def add_comment(self, issue, text):
     pass
 
+  def get_patch(self, issue, patchset):
+    return patch.PatchSet([
+        patch.FilePatchDiff('chrome/file.cc', SVN_PATCH, []),
+        patch.FilePatchDelete('other/place/foo', True),
+        patch.FilePatchBinary('foo', 'data', []),
+    ])
+
 
 class FakeCheckout(object):
   project_path = os.getcwd()
 
   def __init__(self):
-    self._prepare = False
-    self._commit = False
+    self._prepare = 0
+    self._commit = []
+    self._apply_patch = []
 
   def prepare(self):
-    self._prepare = True
+    self._prepare += 1
+
+  def apply_patch(self, patchset):
+    self._apply_patch.append(patchset)
 
   def commit(self, message, author):
-    self._commit = True
+    self._commit.append((message, author))
 
   def get_settings(self, key):
     return None
 
 
 class PendingManager(unittest.TestCase):
   def setUp(self):
     super(PendingManager, self).setUp()
     self.root_dir = ROOT_DIR
     self._old_send_stack = pending_manager.breakpad.SendStack
@@ skipping 19 matching lines @@
       self.assertEquals(pc.queue.pending_commits, [])
       pc.save(filename)
       self.assertEquals(trim(empty), trim(read(filename)))
     finally:
       os.remove(filename)
       if os.path.exists(filename + '.old'):
         os.remove(filename + '.old')
 
   def _get_pc(self, verifiers_no_patch, verifiers):
     pc = pending_manager.PendingManager(
-        FakeRietveld(self), FakeCheckout(), verifiers_no_patch, verifiers)
+        FakeRietveld(), FakeCheckout(), verifiers_no_patch, verifiers)
 
-    def fake_issues():
-      return [31337]
     def apply_patch(pending, revision):
       pc._patch_applied = True
 
-    pc._fetch_pending_issues = fake_issues
     pc._apply_patch = apply_patch
     pc._patch_applied = False
     return pc
 
   def testNoVerification(self):
     try:
       pending_manager.PendingManager(
-          FakeRietveld(self), FakeCheckout(), [], [])
+          FakeRietveld(), FakeCheckout(), [], [])
       self.fail()
     except AssertionError:
       pass
     try:
       pending_manager.PendingManager(
-          FakeRietveld(self),
+          FakeRietveld(),
           FakeCheckout(),
           [fake.FakeVerifier(base.SUCCEEDED)],
           [fake.FakeVerifier(base.SUCCEEDED)])
       self.fail()
     except AssertionError:
       pass
 
   def _check_1(self, pc, result):
     self.assertEquals(len(pc.queue.pending_commits), 0)
     pc.look_for_new_pending_commit()
     self.assertEquals(len(pc.queue.pending_commits), 1)
     commit = pc.queue.pending_commits[0]
     self.assertEquals(len(commit.verifications), 0)
     pc.process_new_pending_commit()
     commit = pc.queue.pending_commits[0]
     self.assertEquals(commit.verifications['fake'].get_state(), result)
     self.assertEquals(len(commit.verifications), 1)
     pc.update_status()
     commit = pc.queue.pending_commits[0]
     self.assertEquals(commit.verifications['fake'].get_state(), result)
+    self.assertEquals('', commit.relpath)
     self.assertEquals(len(commit.verifications), 1)
     pc.scan_results()
     self.assertEquals(len(pc.queue.pending_commits), 0)
-    self.assertEquals(pc.checkout._commit, result == base.SUCCEEDED)
+    self.assertEquals(len(pc.checkout._commit), result == base.SUCCEEDED)
 
   def testNoPatchVerification(self):
     pc = self._get_pc([fake.FakeVerifier(base.SUCCEEDED)], [])
     self._check_1(pc, base.SUCCEEDED)
 
   def testPatchVerification(self):
     pc = self._get_pc([], [fake.FakeVerifier(base.SUCCEEDED)])
     self._check_1(pc, base.SUCCEEDED)
 
   def testNoPatchVerificationFail(self):
@@ skipping 22 matching lines @@
     self.assertEquals([True], result)
 
   def _check_defer_1(self, pc, result):
     self.assertEquals(len(pc.queue.pending_commits), 0)
     pc.look_for_new_pending_commit()
     self.assertEquals(len(pc.queue.pending_commits), 1)
     commit = pc.queue.pending_commits[0]
     self.assertEquals(len(commit.verifications), 0)
     pc.process_new_pending_commit()
     commit = pc.queue.pending_commits[0]
+    self.assertEquals('', commit.relpath)
     self.assertEquals(commit.verifications['fake'].get_state(), base.PROCESSING)
     self.assertEquals(len(commit.verifications), 1)
     pc.update_status()
     commit = pc.queue.pending_commits[0]
+    self.assertEquals('', commit.relpath)
     self.assertEquals(commit.verifications['fake'].get_state(), result)
     self.assertEquals(len(commit.verifications), 1)
     pc.scan_results()
     self.assertEquals(len(pc.queue.pending_commits), 0)
-    self.assertEquals(pc.checkout._commit, result == base.SUCCEEDED)
+    self.assertEquals(len(pc.checkout._commit), result == base.SUCCEEDED)
 
   def testDeferNoPatchVerification(self):
     pc = self._get_pc([fake.DeferredFakeVerifier(base.SUCCEEDED)], [])
     self._check_defer_1(pc, base.SUCCEEDED)
 
   def testDeferPatchVerification(self):
     pc = self._get_pc([], [fake.DeferredFakeVerifier(base.SUCCEEDED)])
     self._check_defer_1(pc, base.SUCCEEDED)
 
   def testDeferNoPatchVerificationFail(self):
@@ skipping 41 matching lines @@
     if nb >= 2:
       self.assertEquals(commit.verifications['fake2'].get_state(), f2)
     if nb >= 3:
       self.assertEquals(commit.verifications['fake3'].get_state(), f3)
     if nb >= 4:
       self.assertEquals(commit.verifications['fake4'].get_state(), f4)
     self.assertEquals(len(commit.verifications), nb)
     pc.scan_results()
     self.assertEquals(len(pc.queue.pending_commits), 0)
     self.assertEquals(
-        pc.checkout._commit,
+        len(pc.checkout._commit),
         all(x == base.SUCCEEDED for x in (f1, f2, f3, f4)))
 
   def test4thVerificationFail(self):
     self._check_4(base.SUCCEEDED, base.SUCCEEDED, base.SUCCEEDED, base.FAILED)
 
   def test4Verification(self):
     self._check_4(
         base.SUCCEEDED, base.SUCCEEDED, base.SUCCEEDED, base.SUCCEEDED)
 
   def test4Verification3rdFail(self):
@@ skipping 23 matching lines @@
     self.assertEquals(
         commit.verifications['fake3'].get_state(), base.PROCESSING)
     self.assertEquals(
         commit.verifications['fake4'].get_state(), base.PROCESSING)
     self.assertEquals(len(commit.verifications), 4)
     pc.update_status()
     self.assertEquals(commit.verifications['fake1'].get_state(), f1)
     self.assertEquals(commit.verifications['fake2'].get_state(), f2)
     self.assertEquals(commit.verifications['fake3'].get_state(), f3)
     self.assertEquals(commit.verifications['fake4'].get_state(), f4)
+    self.assertEquals('', commit.relpath)
     pc.scan_results()
     self.assertEquals(len(pc.queue.pending_commits), 0)
     self.assertEquals(
-        pc.checkout._commit,
+        len(pc.checkout._commit),
         all(x == base.SUCCEEDED for x in (f1, f2, f3, f4)))
 
   def testDefer4thVerificationFail(self):
     self._check_defer_4(
         base.SUCCEEDED, base.SUCCEEDED, base.SUCCEEDED, base.FAILED)
 
   def testDefer4Verification(self):
     self._check_defer_4(
         base.SUCCEEDED, base.SUCCEEDED, base.SUCCEEDED, base.SUCCEEDED)
 
   def testDefer4Verification3rdFail(self):
     self._check_defer_4(
         base.SUCCEEDED, base.SUCCEEDED, base.FAILED, base.SUCCEEDED)
 
+  def testRelPath(self):
+    verifiers = [
+        project_base.ProjectBaseUrlVerifier(
+          [r'^%s(.*)$' % re.escape(r'http://example.com/')]),
+    ]
+    pc = pending_manager.PendingManager(
+        FakeRietveld(), FakeCheckout(), verifiers, [])
+    pc.rietveld.issues[31337]['base_url'] = 'http://example.com/sub/dir'
+    pc.look_for_new_pending_commit()
+    self.assertEquals(1, len(pc.queue.pending_commits))
+    pc.process_new_pending_commit()
+    self.assertEquals('sub/dir', pc.queue.pending_commits[0].relpath)
+    self.assertEquals(1, pc.checkout._prepare)
+    self.assertEquals(1, len(pc.checkout._apply_patch))
+    self.assertEquals([], pc.checkout._commit)
+    pc.update_status()
+    self.assertEquals(1, pc.checkout._prepare)
+    self.assertEquals(1, len(pc.checkout._apply_patch))
+    self.assertEquals([], pc.checkout._commit)
+    patches = pc.checkout._apply_patch[0]
+    self.assertEquals(
+        ['sub/dir/chrome/file.cc', 'sub/dir/other/place/foo', 'sub/dir/foo'],
+        patches.filenames)
+    pc.scan_results()
+    self.assertEquals(2, pc.checkout._prepare)
+    self.assertEquals(2, len(pc.checkout._apply_patch))
+    self.assertEquals(
+        [('foo\n\nReview URL: http://nowhere/31337', 'author@example.com')],
+        pc.checkout._commit)
+
 
 if __name__ == '__main__':
   logging.basicConfig(level=logging.ERROR)
   unittest.main()