buildbucket: support unique change urls

scripts/master/buildbucket/integration.py

 # Copyright 2015 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.
 
 """BuildBucketIntegrator integrates Buildbot and Buildbucket."""
 
 import datetime
-import itertools
 import json
 import logging
-import traceback
 
 from buildbot.util import deferredLocked
-from master.buildbucket import common, changestore
-from master.buildbucket.common import log
+from master.buildbucket import changestore
 from twisted.internet import defer, reactor
 from twisted.internet.defer import inlineCallbacks, returnValue
 
 from . import common
 
 # buildbucket API-related constants.
 # time enough to schedule and start a build.
 LEASE_DURATION = datetime.timedelta(hours=1)
 DEFAULT_HEARTBEAT_INTERVAL = datetime.timedelta(minutes=5)
 # Maximum integer that max_builds buildbucket parameter can take.
@@ skipping 21 matching lines @@
   # It is a dict build_id -> lease, where lease is a dict with keys:
   # - key: lease key
   # - build_request: BuildRequestGateway
   # - build: if build_started with same build_id was called,
   #          the associated build.
   # _lease is None on creation, but it gets loaded from the database in
   # _ensure_leases_loaded().
 
   def __init__(
       self, buckets, build_params_hook=None, max_lease_count=None,
-      heartbeat_interval=None):
+      heartbeat_interval=None, change_store_factory=None):
     """Creates a BuildBucketIntegrator.
 
     Args:
       buckets (list of str): poll only builds in any of |buckets|.
       build_params_hook (func): If not None, a callable with arguments
         (params, build) that can modify the supplied parameters during
         validation.
 
         If a ValueError is raised, the build will be marked as an
         INVALID_BUILD_DEFINITION failure and the error message will be
         propagated to the BuildBucket status.
       max_lease_count (int): maximum number of builds that can be leased at a
         time. Defaults to the number of connected slaves.
       heartbeat_interval (datetime.timedelta): frequency of build heartbeats.
         Defaults to 1 minute.
+      change_store_factory: function (BuildbotGateway) => ChangeStore.
     """
     assert buckets, 'Buckets not specified'
     assert max_lease_count is None or isinstance(max_lease_count, int)
     if max_lease_count is not None:
       assert max_lease_count >= 1
     self.buckets = buckets[:]
     self.build_params_hook = build_params_hook
     self.buildbot = None
     self.buildbucket_service = None
     self._find_change_cache = None
     self.started = False
     self.changes = None
     self.poll_lock = defer.DeferredLock()
     self._max_lease_count = max_lease_count
     self._leases = None
     self.heartbeat_interval = heartbeat_interval or DEFAULT_HEARTBEAT_INTERVAL
+    self.change_store_factory = change_store_factory or changestore.ChangeStore
 
   def get_max_lease_count(self):
     if self._max_lease_count:
       return self._max_lease_count
     if not self.buildbot:
       return 0
     return len(self.buildbot.get_connected_slaves())
 
   def log(self, message, level=None):
     common.log(message, level)
 
-  def start(self, buildbot, buildbucket_service, change_store_factory=None):
+  def start(self, buildbot, buildbucket_service):
     assert not self.started, 'BuildBucketIntegrator is already started'
     assert buildbot
     assert buildbucket_service
-    change_store_factory = change_store_factory or changestore.ChangeStore
     self.buildbot = buildbot
     self.buildbucket_service = buildbucket_service
     self.buildbucket_service.start()
-    self.changes = change_store_factory(buildbot)
+    self.changes = self.change_store_factory(buildbot)
     self.started = True
     self.log('integrator started')
     self._do_until_stopped(self.heartbeat_interval, self.send_heartbeats)
     self._do_until_stopped(
         LEASE_CLEANUP_INTERVAL, self.clean_completed_build_requests)
 
   @inlineCallbacks
   def _get_leases_from_db(self):
     """Returns currently held leases from the database.
 
@@ skipping 524 matching lines @@
             'properties': properties_to_send,
         }, sort_keys=True),
     }
     if status == 'SUCCESS':
       yield self._leased_build_call('succeed', build, body)
     else:
       body['failure_reason'] = (
           'BUILD_FAILURE' if status == 'FAILURE'
           else 'INFRA_FAILURE')
       yield self._leased_build_call('fail', build, body)