Switch CheckOwners to use input_api.rietveld.

tests/presubmit_unittest.py

 #!/usr/bin/env python
 # Copyright (c) 2011 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 presubmit_support.py and presubmit_canned_checks.py."""
 
 # pylint is too confused.
 # pylint: disable=E1101,E1103,R0201,W0212,W0403
 
 import StringIO
 import sys
 
 # Fixes include path.
 from super_mox import mox, SuperMoxTestBase
 
 import owners
 import presubmit_support as presubmit
+import rietveld
 # Shortcut.
 presubmit_canned_checks = presubmit.presubmit_canned_checks
 
 
 class PresubmitTestsBase(SuperMoxTestBase):
   """Setups and tear downs the mocks but doesn't test anything as-is."""
   presubmit_text = """
 def CheckChangeOnUpload(input_api, output_api):
   if not input_api.change.NOSUCHKEY:
     return [output_api.PresubmitError("!!")]
@@ skipping 1302 matching lines @@
 
 class CannedChecksUnittest(PresubmitTestsBase):
   """Tests presubmit_canned_checks.py."""
 
   def setUp(self):
     PresubmitTestsBase.setUp(self)
 
   def MockInputApi(self, change, committing):
     input_api = self.mox.CreateMock(presubmit.InputApi)
     input_api.cStringIO = presubmit.cStringIO
+    input_api.json = presubmit.json
     input_api.os_listdir = self.mox.CreateMockAnything()
     input_api.os_walk = self.mox.CreateMockAnything()
     input_api.os_path = presubmit.os.path
     input_api.re = presubmit.re
+    input_api.rietveld = self.mox.CreateMock(rietveld.Rietveld)
     input_api.traceback = presubmit.traceback
     input_api.urllib2 = self.mox.CreateMock(presubmit.urllib2)
     input_api.unittest = unittest
     input_api.subprocess = self.mox.CreateMock(presubmit.subprocess)
     class fake_CalledProcessError(Exception):
       def __str__(self):
         return 'foo'
     input_api.subprocess.CalledProcessError = fake_CalledProcessError
 
     input_api.change = change
@@ skipping 453 matching lines @@
     self.mox.ReplayAll()
     results = presubmit_canned_checks.CheckTreeIsOpen(
         input_api, presubmit.OutputApi,
         url='url_to_closed', closed='.*closed.*')
     self.assertEquals(len(results), 1)
     self.assertEquals(results[0].__class__,
                       presubmit.OutputApi.PresubmitError)
 
   def testCannedCheckJsonTreeIsOpenOpen(self):
     input_api = self.MockInputApi(None, True)
-    input_api.json = presubmit.json
     connection = self.mox.CreateMockAnything()
     input_api.urllib2.urlopen('url_to_open').AndReturn(connection)
     status = {
         'can_commit_freely': True,
         'general_state': 'open',
         'message': 'The tree is open'
     }
     connection.read().AndReturn(input_api.json.dumps(status))
     connection.close()
     self.mox.ReplayAll()
     results = presubmit_canned_checks.CheckTreeIsOpen(
         input_api, presubmit.OutputApi, json_url='url_to_open')
     self.assertEquals(results, [])
 
   def testCannedCheckJsonTreeIsOpenClosed(self):
     input_api = self.MockInputApi(None, True)
-    input_api.json = presubmit.json
     connection = self.mox.CreateMockAnything()
     input_api.urllib2.urlopen('url_to_closed').AndReturn(connection)
     status = {
         'can_commit_freely': False,
         'general_state': 'closed',
         'message': 'The tree is close',
     }
     connection.read().AndReturn(input_api.json.dumps(status))
     connection.close()
     self.mox.ReplayAll()
@@ skipping 115 matching lines @@
 
     results = presubmit_canned_checks.CheckRietveldTryJobExecution(
         input_api, presubmit.OutputApi, 'uurl', ('mac', 'linux', 'amiga'),
         'georges')
     self.assertEquals(len(results), 1)
     self.assertEquals(results[0].__class__,
         presubmit.OutputApi.PresubmitPromptWarning)
 
   def testCheckBuildbotPendingBuildsBad(self):
     input_api = self.MockInputApi(None, True)
-    input_api.json = presubmit.json
     connection = self.mox.CreateMockAnything()
     input_api.urllib2.urlopen('uurl').AndReturn(connection)
     connection.read().AndReturn('foo')
     connection.close()
     self.mox.ReplayAll()
 
     results = presubmit_canned_checks.CheckBuildbotPendingBuilds(
         input_api, presubmit.OutputApi, 'uurl', 2, ('foo'))
     self.assertEquals(len(results), 1)
     self.assertEquals(results[0].__class__,
         presubmit.OutputApi.PresubmitNotifyResult)
 
   def testCheckBuildbotPendingBuildsGood(self):
     input_api = self.MockInputApi(None, True)
-    input_api.json = presubmit.json
     connection = self.mox.CreateMockAnything()
     input_api.urllib2.urlopen('uurl').AndReturn(connection)
     connection.read().AndReturn("""
     {
       'b1': { 'pending_builds': [0, 1, 2, 3, 4, 5, 6, 7] },
       'foo': { 'pending_builds': [0, 1, 2, 3, 4, 5, 6, 7] },
       'b2': { 'pending_builds': [0] }
     }""")
     connection.close()
     self.mox.ReplayAll()
 
     results = presubmit_canned_checks.CheckBuildbotPendingBuilds(
         input_api, presubmit.OutputApi, 'uurl', 2, ('foo'))
     self.assertEquals(len(results), 1)
     self.assertEquals(results[0].__class__,
         presubmit.OutputApi.PresubmitNotifyResult)
 
   def AssertOwnersWorks(self, tbr=False, issue='1', approvers=None,
-      rietveld_response=None, host_url=None,
-      uncovered_files=None, expected_output=''):
+      rietveld_response=None, uncovered_files=None, expected_output=''):
     if approvers is None:
       approvers = set()
     if uncovered_files is None:
       uncovered_files = set()
 
     change = self.mox.CreateMock(presubmit.Change)
     change.issue = issue
     affected_file = self.mox.CreateMock(presubmit.SvnAffectedFile)
     input_api = self.MockInputApi(change, False)
     fake_db = self.mox.CreateMock(owners.Database)
     fake_db.email_regexp = input_api.re.compile(owners.BASIC_EMAIL_REGEXP)
     input_api.owners_db = fake_db
     input_api.is_committing = True
     input_api.tbr = tbr
 
     if not tbr and issue:
       affected_file.LocalPath().AndReturn('foo.cc')
       change.AffectedFiles(file_filter=None).AndReturn([affected_file])
-
-      expected_host = 'http://localhost'
-      if host_url:
-        input_api.host_url = host_url
-        if host_url.startswith('https'):
-          expected_host = host_url
-
       owner_email = 'john@example.com'
       if not rietveld_response:
-        messages = (
-          '{"sender": "%s", "text": "I approve", "approval": true}' % a
-          for a in approvers)
-        rietveld_response = '{"owner_email": "%s", "messages": [%s]}' % (
-            owner_email, ','.join(messages))
-      input_api.urllib2.urlopen(
-          expected_host + '/api/1?messages=true').AndReturn(
-          StringIO.StringIO(rietveld_response))
-      input_api.json = presubmit.json
+        rietveld_response = {
+          "owner_email": owner_email,
+          "messages": [
+            {"sender": a, "text": "I approve", "approval": True}
+            for a in approvers
+          ],
+        }
+      input_api.rietveld.get_issue_properties(
+          int(input_api.change.issue), True).AndReturn(rietveld_response)
       fake_db.files_not_covered_by(set(['foo.cc']),
-         approvers.union(set([owner_email]))).AndReturn(uncovered_files)
+          approvers.union(set([owner_email]))).AndReturn(uncovered_files)
 
     self.mox.ReplayAll()
     output = presubmit.PresubmitOutput()
     results = presubmit_canned_checks.CheckOwners(input_api,
         presubmit.OutputApi)
     if results:
       results[0].handle(output)
     self.assertEquals(output.getvalue(), expected_output)
 
-  def testCannedCheckOwners_LGTMPhrases(self):
-    def phrase_test(approval, approvers=None, expected_output=''):
-      if approvers is None:
-        approvers = set(['ben@example.com'])
-      self.AssertOwnersWorks(approvers=approvers,
-          rietveld_response=(
-            '{"owner_email": "john@example.com",'
-            '"messages": [{"sender": "ben@example.com",'
-            '  "text": "foo", "approval": %s}]}') % approval,
-          expected_output=expected_output)
+  def testCannedCheckOwners_Approved(self):
+    response = {
+      "owner_email": "john@example.com",
+      "messages": [
+        {
+          "sender": "ben@example.com", "text": "foo", "approval": True,
+        },
+      ],
+    }
+    self.AssertOwnersWorks(approvers=set(['ben@example.com']),
+        rietveld_response=response,
+        expected_output='')
 
-    phrase_test('true')
-    phrase_test('false', approvers=set(),
-                expected_output='Missing LGTM from someone other than '
-                                'john@example.com\n')
-
-  def testCannedCheckOwners_HTTPS_HostURL(self):
-    self.AssertOwnersWorks(approvers=set(['ben@example.com']),
-                           host_url='https://localhost')
-
-  def testCannedCheckOwners_MissingSchemeInHostURL(self):
-    self.AssertOwnersWorks(approvers=set(['ben@example.com']),
-                           host_url='localhost')
+  def testCannedCheckOwners_NotApproved(self):
+    response = {
+      "owner_email": "john@example.com",
+      "messages": [
+        {
+          "sender": "ben@example.com", "text": "foo", "approval": False,
+        },
+      ],
+    }
+    self.AssertOwnersWorks(
+        approvers=set(),
+        rietveld_response=response,
+        expected_output=
+            'Missing LGTM from someone other than john@example.com\n')
 
   def testCannedCheckOwners_NoIssue(self):
     self.AssertOwnersWorks(issue=None,
         expected_output="OWNERS check failed: this change has no Rietveld "
                         "issue number, so we can't check it for approvals.\n")
 
   def testCannedCheckOwners_NoLGTM(self):
     self.AssertOwnersWorks(expected_output='Missing LGTM from someone '
                                            'other than john@example.com\n')
 
@@ skipping 71 matching lines @@
         whitelist=['^a$', '^b$'],
         blacklist=['a'])
     self.assertEqual(results, [])
     self.checkstdout(
         'Running %s\n' % presubmit.os.path.join('random_directory', 'b'))
 
 
 if __name__ == '__main__':
   import unittest
   unittest.main()