Teach CQ status app to check login status of users.

appengine/chromium_cq_status/handlers/patch_summary.py

 # Copyright 2014 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.
 
 from datetime import datetime
 from itertools import chain
 import logging
 
 import webapp2
 
 from model.record import Record
 from shared.config import (
   TAG_START,
   TAG_STOP,
   TAG_ISSUE,
   TAG_PATCHSET,
   TAG_CODEREVIEW_HOSTNAME,
   TRYJOBVERIFIER,
   JOB_STATE,
 )
 from shared.parsing import parse_rietveld_timestamp
-from shared.utils import (
-  cross_origin_json,
-  to_unix_timestamp,
-  guess_legacy_codereview_hostname,
-)
+from shared import utils
 
 
 class PatchSummary(webapp2.RequestHandler):
-  @cross_origin_json
+  @utils.cross_origin_json
+  @utils.read_access
   def get(self, issue, patch): # pylint: disable=W0221
-    now = to_unix_timestamp(datetime.utcnow())
-    codereview_hostname = guess_legacy_codereview_hostname(issue)
+    now = utils.to_unix_timestamp(datetime.utcnow())
+    codereview_hostname = utils.guess_legacy_codereview_hostname(issue)
     return summarize_patch(codereview_hostname, issue, patch, now)
 
 
 class PatchSummaryV2(webapp2.RequestHandler):
-  @cross_origin_json
+  @utils.cross_origin_json
+  @utils.read_access
   def get(self, codereview_hostname, issue, patch): # pylint: disable=W0221
-    now = to_unix_timestamp(datetime.utcnow())
+    now = utils.to_unix_timestamp(datetime.utcnow())
     return summarize_patch(codereview_hostname, issue, patch, now)
 
 
 def summarize_patch(codereview_hostname, issue, patch, now):
   attempts = [
     summarize_attempt(raw_attempt, now)
     for raw_attempt in get_raw_attempts(codereview_hostname, issue, patch)
   ][::-1]
   return {
     'success': any(attempt['success'] for attempt in attempts),
@@ skipping 35 matching lines @@
       if TAG_STOP in record.tags:
         raw_attempts.append(raw_attempt)
         raw_attempt = None
   if raw_attempt != None and len(raw_attempt) > 0:  # pragma: no cover
     raw_attempts.append(raw_attempt)
   return raw_attempts
 
 
 def summarize_attempt(raw_attempt, now):
   assert len(raw_attempt) > 0
-  start_timestamp = to_unix_timestamp(raw_attempt[0].timestamp)
+  start_timestamp = utils.to_unix_timestamp(raw_attempt[0].timestamp)
   summary = blank_attempt_summary()
   job_tracker = AttemptJobTracker(start_timestamp)
   durations = summary['durations']
   last_patch_action = None
   last_patch_timestamp = None
   verifier_start_timestamp = None
   for record in raw_attempt:
     action = record.fields.get('action')
     # patch_ready_to_commit signals are noisy and not useful.
     if action == 'patch_ready_to_commit':
       continue
     verifier = record.fields.get('verifier')
-    timestamp = to_unix_timestamp(record.timestamp)
+    timestamp = utils.to_unix_timestamp(record.timestamp)
     if last_patch_action:
       patch_state_duration = timestamp - last_patch_timestamp
 
     # Verifier job updates.
     if verifier == TRYJOBVERIFIER:
       if action == 'verifier_start':
         verifier_start_timestamp = timestamp
       elif action == 'verifier_jobs_update':
         job_tracker.update_jobs(record)
       elif (action in ('verifier_pass', 'verifier_fail') and
@@ skipping 171 matching lines @@
     return min(iterable)
   except ValueError:  # pragma: no cover
     return None
 
 
 def maybe_max(iterable):
   try:
     return max(iterable)
   except ValueError:  # pragma: no cover
     return None