def test_stopService_flushes(self):
        wfd = defer.waitForDeferred(
            self.makeBuilder())
        yield wfd
        wfd.getResult()

        # just check that stopService calls this and waits
        # for the deferred to fire
        events = []

        long_d = defer.Deferred()
        long_d.addCallback(lambda _ : events.append('long_d'))
        self.bldr.maybeStartBuild = lambda : long_d

        stop_d = self.bldr.stopService()
        stop_d.addCallback(lambda _ : events.append('stop_d'))

        # nothing should have happened yet
        self.assertEqual(events, [])

        # finish the maybeStartBuild invocation..
        long_d.callback(None)

        wfd = defer.waitForDeferred(stop_d)
        yield wfd
        wfd.getResult()

        # and then check that things happened in the right order
        self.assertEqual(events, [ 'long_d', 'stop_d' ])

    def stableTimerFired(self, timer_name):
        # if the service has already been stoppd then just bail out
        if not self._stable_timers[timer_name]:
            return

        # delete this now-fired timer
        del self._stable_timers[timer_name]

        wfd = defer.waitForDeferred(
                self.getChangeClassificationsForTimer(self.schedulerid,
                                                      timer_name))
        yield wfd
        classifications = wfd.getResult()

        # just in case: databases do weird things sometimes!
        if not classifications: # pragma: no cover
            return

        changeids = sorted(classifications.keys())
        wfd = defer.waitForDeferred(
                self.addBuildsetForChanges(reason='scheduler',
                                           changeids=changeids))
        yield wfd
        wfd.getResult()

        max_changeid = changeids[-1] # (changeids are sorted)
        wfd = defer.waitForDeferred(
                self.master.db.schedulers.flushChangeClassifications(
                            self.schedulerid, less_than=max_changeid+1))
        yield wfd
        wfd.getResult()

    def produce(self):
        """Produce entries."""
        while not self._paused:
            wfd = defer.waitForDeferred(self._write_pending())
            yield wfd
            wfd.getResult()

            if self.finished:
                self.finishProducing()
                return
            first = self._last_fetching
            while (self._currently_fetching <= self._max_currently_fetching and
                   self._last_fetching <= self._end and
                   self._currently_stored <= FLAGS.entries_buffer):
                last = min(self._last_fetching + self._batch_size - 1, self._end,
                   self._last_fetching + self._max_currently_fetching
                           - self._currently_fetching + 1)
                self._batches.put(self._create_fetch_deferred(first, last))
                self._currently_fetching += last - first + 1
                first = last + 1
                self._last_fetching = first

            wfd = defer.waitForDeferred(self._batches.get())
            # Pause here until the body of the response is available.
            yield wfd
            # The producer may have been paused while waiting for the response,
            # or errored out upon receiving it: do not write the entries out
            # until after the next self._paused check.
            self._pending = wfd.getResult()

    def test_reconfigService_default_changed(self):
        config = mock.Mock()
        config.slavePortnum = '9876'
        self.attachChangeSource(pb.PBChangeSource())

        self.startChangeSource()
        wfd = defer.waitForDeferred(
                self.changesource.reconfigService(config))
        yield wfd
        wfd.getResult()

        self.assertRegistered('9876', 'change', 'changepw')

        config.slavePortnum = '1234'

        wfd = defer.waitForDeferred(
                self.changesource.reconfigService(config))
        yield wfd
        wfd.getResult()

        self.assertUnregistered('9876', 'change', 'changepw')
        self.assertRegistered('1234', 'change', 'changepw')

        wfd = defer.waitForDeferred(
                self.stopChangeSource())
        yield wfd
        wfd.getResult()

        self.assertUnregistered('1234', 'change', 'changepw')

def deleteResource(request, resource, resource_uri, depth="0"):
    """
    Handle a resource delete with proper quota etc updates
    """
    if not resource.exists():
        log.error("File not found: %s" % (resource,))
        raise HTTPError(responsecode.NOT_FOUND)

    # Do quota checks before we start deleting things
    myquota = waitForDeferred(resource.quota(request))
    yield myquota
    myquota = myquota.getResult()
    if myquota is not None:
        old_size = waitForDeferred(resource.quotaSize(request))
        yield old_size
        old_size = old_size.getResult()
    else:
        old_size = 0

    # Do delete
    x = waitForDeferred(delete(resource_uri, resource.fp, depth))
    yield x
    result = x.getResult()

    # Adjust quota
    if myquota is not None:
        d = waitForDeferred(resource.quotaSizeAdjust(request, -old_size))
        yield d
        d.getResult()
    
    yield result

    def start(self):
        args = self.args
        # args['dir'] is relative to Builder directory, and is required.
        assert args["dir"] is not None
        dirnames = args["dir"]

        self.timeout = args.get("timeout", 120)
        self.maxTime = args.get("maxTime", None)
        self.rc = 0
        if type(dirnames) is list:
            assert len(dirnames) != 0
            for dirname in dirnames:
                wfd = defer.waitForDeferred(self.removeSingleDir(dirname))
                yield wfd
                res = wfd.getResult()
                # Even if single removal of single file/dir consider it as
                # failure of whole command, but continue removing other files
                # Send 'rc' to master to handle failure cases
                if res != 0:
                    self.rc = res
        else:
            wfd = defer.waitForDeferred(self.removeSingleDir(dirnames))
            yield wfd
            self.rc = wfd.getResult()

        self.sendStatus({"rc": self.rc})

    def full(self, _):
        if self.method == 'clobber':
            wfd = defer.waitForDeferred(self.clobber())
            yield wfd
            wfd.getResult()
            return
        elif self.method in ['copy', 'export']:
            wfd = defer.waitForDeferred(self.copy())
            yield wfd
            wfd.getResult()
            return

        wfd = defer.waitForDeferred(self._sourcedirIsUpdatable())
        yield wfd
        updatable = wfd.getResult()
        if not updatable:
            d = self._dovccmd(['checkout', self.svnurl, '.'])
        elif self.method == 'clean':
            d = self.clean()
        elif self.method == 'fresh':
            d = self.fresh()

        wfd = defer.waitForDeferred(d)
        yield wfd
        wfd.getResult()

def test():
    print "STARTING TEST"
    foo = paisley.CouchDB('localhost')
    print "\nCreate database 'testdb':"
    d = foo.createDB('testdb1')
    d.addCallback(lambda f: log.err(f))
    wfd = defer.waitForDeferred(d)
    yield wfd
    try:
        print wfd.getResult()
    except Exception as e:
        # FIXME: not sure why Error.status is a str compared to http constants
        if hasattr(e, 'status') and e.status == str(http.UNAUTHORIZED):
            print "\nError: not allowed to create databases"
            reactor.stop()
            return
        else:
            raise

    print "\nList databases on server:"
    d = foo.listDB()
    wfd = defer.waitForDeferred(d)
    yield wfd
    print wfd.getResult()
    reactor.stop()

    def incremental(self, _):
        wfd = defer.waitForDeferred(
            self._sourcedirIsUpdatable())
        yield wfd
        updatable = wfd.getResult()

        if not updatable:
            # blow away the old (un-updatable) directory
            wfd = defer.waitForDeferred(
                self._rmdir(self.workdir))
            yield wfd
            wfd.getResult()

            # and plan to do a checkout
            command = ['checkout', self.repourl, '.']
        else:
            # otherwise, do an update
            command = ['update']
        if self.revision:
            command.extend(['--revision', str(self.revision)])

        wfd = defer.waitForDeferred(
            self._dovccmd(command))
        yield wfd
        wfd.getResult()

    def test_reconfigService_class_name_change(self):
        sch1 = self.makeSched(self.ReconfigSched, 'sch1')
        self.new_config.schedulers = dict(sch1=sch1)

        wfd = defer.waitForDeferred(
            self.sm.reconfigService(self.new_config))
        yield wfd
        wfd.getResult()

        self.assertIdentical(sch1.parent, self.sm)
        self.assertIdentical(sch1.master, self.master)
        self.assertEqual(sch1.reconfig_count, 1)

        sch1_new = self.makeSched(self.ReconfigSched2, 'sch1')
        self.new_config.schedulers = dict(sch1=sch1_new)

        wfd = defer.waitForDeferred(
            self.sm.reconfigService(self.new_config))
        yield wfd
        wfd.getResult()

        # sch1 had its class name change, so sch1_new is now the active
        # instance
        self.assertIdentical(sch1_new.parent, self.sm)
        self.assertIdentical(sch1_new.master, self.master)

    def test_reconfigService_add_and_change_and_remove_no_reconfig(self):
        sch1 = self.makeSched(self.Sched, 'sch1', attr='alpha')
        self.new_config.schedulers = dict(sch1=sch1)

        wfd = defer.waitForDeferred(
            self.sm.reconfigService(self.new_config))
        yield wfd
        wfd.getResult()

        self.assertIdentical(sch1.parent, self.sm)
        self.assertIdentical(sch1.master, self.master)

        sch1_new = self.makeSched(self.Sched, 'sch1', attr='beta')
        sch2 = self.makeSched(self.Sched, 'sch2', attr='alpha')
        self.new_config.schedulers = dict(sch1=sch1_new, sch2=sch2)

        wfd = defer.waitForDeferred(
            self.sm.reconfigService(self.new_config))
        yield wfd
        wfd.getResult()

        # sch1 is not longer active, and sch1_new is
        self.assertIdentical(sch1.parent, None)
        self.assertIdentical(sch1.master, None)
        self.assertIdentical(sch1_new.parent, self.sm)
        self.assertIdentical(sch1_new.master, self.master)
        self.assertIdentical(sch2.parent, self.sm)
        self.assertIdentical(sch2.master, self.master)

    def test_queue_collapsing(self):
        # just to check that we're practicing with the right queue size (so
        # QUEUE_SIZE_FACTOR is 10)
        self.assertEqual(self.lru.max_queue, 30)

        for c in 'a' + 'x' * 27 + 'ab':
            wfd = defer.waitForDeferred(
                    self.lru.get(c))
            yield wfd
            res = wfd.getResult()
        self.check_result(res, short('b'), 27, 3)

        # at this point, we should have 'x', 'a', and 'b' in the cache, and
        # 'axx..xxab' in the queue.
        self.assertEqual(len(self.lru.queue), 30)

        # This 'get' operation for an existing key should cause compaction
        wfd = defer.waitForDeferred(
                self.lru.get('b'))
        yield wfd
        res = wfd.getResult()
        self.check_result(res, short('b'), 28, 3)

        self.assertEqual(len(self.lru.queue), 3)

        # expect a cached short('a')
        self.lru.miss_fn = self.long_miss_fn
        wfd = defer.waitForDeferred(
                self.lru.get('a'))
        yield wfd
        res = wfd.getResult()
        self.check_result(res, short('a'), 29, 3)

        def _deferred(deferredResult, caller):
            """only callable by ``newfunc``"""
            from twisted.internet import reactor

            result = None
            attempt = 0
            while attempt <= maximum:
                try:
                    wfd = defer.waitForDeferred(deferredResult)
                    yield wfd
                    result = wfd.getResult()
                    break  # didn't throw exception then we have the result
                except SystemExit:
                    result = Failure()
                    break
                except KeyboardInterrupt:
                    result = Failure()
                    break
                except:
                    attempt += 1
                    if attempt > maximum:
                        result = Failure()
                        break  # failure captured
                    d = defer.Deferred()
                    reactor.callLater(delay, d.callback, None)
                    wfd = defer.waitForDeferred(d)
                    yield wfd
                    wfd.getResult()
                    # fortunatly the caller has been set up in advance
                    caller.write(">>> Retry attempt %d" % attempt)
                    deferredResult = caller(*caller.args, **caller.kwargs)
            yield result

    def _checkCompletedBuildsets(self, bsid, result):
        wfd = defer.waitForDeferred(
            self.master.db.buildsets.getSubscribedBuildsets(self.schedulerid))
        yield wfd
        subs = wfd.getResult()

        for (sub_bsid, sub_sssetid, sub_complete, sub_results) in subs:
            # skip incomplete builds, handling the case where the 'complete'
            # column has not been updated yet
            if not sub_complete and sub_bsid != bsid:
                continue

            # build a dependent build if the status is appropriate
            if sub_results in (SUCCESS, WARNINGS):
                wfd = defer.waitForDeferred(
                    self.addBuildsetForSourceStamp(setid=sub_sssetid,
                                               reason='downstream'))
                yield wfd
                wfd.getResult()

            # and regardless of status, remove the subscription
            wfd = defer.waitForDeferred(
                self.master.db.buildsets.unsubscribeFromBuildset(
                                          self.schedulerid, sub_bsid))
            yield wfd
            wfd.getResult()

    def asDict_async(self):
        result = {}

        wfd = defer.waitForDeferred(
                self.getSourceStamp())
        yield wfd
        ss = wfd.getResult()
        result['source'] = ss.asDict()

        result['builderName'] = self.getBuilderName()

        wfd = defer.waitForDeferred(
                self.getSubmitTime())
        yield wfd
        submittedAt = wfd.getResult()
        result['submittedAt'] = submittedAt

        wfd = defer.waitForDeferred(
            self.getBuilds())
        yield wfd
        builds = wfd.getResult()

        result['builds'] = [ build.asDict() for build in builds ]

        yield result

    def _sourcedirIsUpdatable(self):
        # first, perform a stat to ensure that this is really an svn directory
        cmd = buildstep.RemoteCommand('stat', {'file': self.workdir + '/.svn',
                                               'logEnviron': self.logEnviron,})
        cmd.useLog(self.stdio_log, False)
        wfd = defer.waitForDeferred(
            self.runCommand(cmd))
        yield wfd
        wfd.getResult()

        if cmd.rc != 0:
            yield False
            return

        # then run 'svn info' to check that the URL matches our repourl
        wfd = defer.waitForDeferred(
            self._dovccmd(['info'], collectStdout=True))
        yield wfd
        stdout = wfd.getResult()

        # extract the URL, handling whitespace carefully so that \r\n works
        # is a line terminator
        mo = re.search('^URL:\s*(.*?)\s*$', stdout, re.M)
        yield mo and mo.group(1) == self.repourl
        return

    def _mergeRequests(self, breq, unclaimed_requests, mergeRequests_fn):
        """Use C{mergeRequests_fn} to merge C{breq} against
        C{unclaimed_requests}, where both are build request dictionaries"""
        # short circuit if there is no merging to do
        if not mergeRequests_fn or len(unclaimed_requests) == 1:
            yield [ breq ]
            return

        # we'll need BuildRequest objects, so get those first
        wfd = defer.waitForDeferred(
            defer.gatherResults(
                [ self._brdictToBuildRequest(brdict)
                  for brdict in unclaimed_requests ]))
        yield wfd
        unclaimed_request_objects = wfd.getResult()
        breq_object = unclaimed_request_objects.pop(
                unclaimed_requests.index(breq))

        # gather the mergeable requests
        merged_request_objects = [breq_object]
        for other_breq_object in unclaimed_request_objects:
            wfd = defer.waitForDeferred(
                defer.maybeDeferred(lambda :
                    mergeRequests_fn(self, breq_object, other_breq_object)))
            yield wfd
            if wfd.getResult():
                merged_request_objects.append(other_breq_object)

        # convert them back to brdicts and return
        merged_requests = [ br.brdict for br in merged_request_objects ]
        yield merged_requests

def SubmitTryJobChanges(self, changes):
  """ Override of SVNPoller.submit_changes:
  http://src.chromium.org/viewvc/chrome/trunk/tools/build/scripts/master/try_job_svn.py?view=markup

  We modify it so that the patch file url is added to the build properties.
  This allows the slave to download the patch directly rather than receiving
  it from the master.
  """
  for chdict in changes:
    # pylint: disable=E1101
    parsed = self.parent.parse_options(text_to_dict(chdict['comments']))

    # 'fix' revision.
    # LKGR must be known before creating the change object.
    wfd = defer.waitForDeferred(self.parent.get_lkgr(parsed))
    yield wfd
    wfd.getResult()

    wfd = defer.waitForDeferred(self.master.addChange(
      author=','.join(parsed['email']),
      revision=parsed['revision'],
      comments='',
      properties={'patch_file_url': chdict['repository'] + '/' + \
                      chdict['files'][0]}))
    yield wfd
    change = wfd.getResult()

    self.parent.addChangeInner(chdict['files'], parsed, change.number)

    def test_brdictToBuildRequest(self):
        self.makeBuilder()
        # set up all of the data required for a BuildRequest object
        wfd = defer.waitForDeferred(
            self.db.insertTestData([
                fakedb.SourceStamp(id=234),
                fakedb.Buildset(id=30, sourcestampid=234, reason='foo',
                    submitted_at=1300305712, results=-1),
                fakedb.BuildRequest(id=19, buildsetid=30, buildername='bldr',
                    priority=13, submitted_at=1300305712, results=-1),
            ]))
        yield wfd
        wfd.getResult()

        wfd = defer.waitForDeferred(
            self.db.buildrequests.getBuildRequest(19))
        yield wfd
        brdict = wfd.getResult()

        wfd = defer.waitForDeferred(
            self.bldr._brdictToBuildRequest(brdict))
        yield wfd
        br = wfd.getResult()

        # just check that the BuildRequest looks reasonable -
        # test_process_buildrequest checks the whole thing
        self.assertEqual(br.reason, 'foo')

        # and check that the cross-pointers are correct
        self.assertIdentical(br.brdict, brdict)
        self.assertIdentical(brdict['brobj'], br)

        self.bldr._breakBrdictRefloops([brdict])

    def test_reconfig_bad_config(self):
        reactor = self.make_reactor()
        self.master.reconfigService = mock.Mock(
                side_effect=lambda n : defer.succeed(None))

        wfd = defer.waitForDeferred(
            self.master.startService(_reactor=reactor))
        yield wfd
        wfd.getResult()


        # reset, since startService called reconfigService
        self.master.reconfigService.reset_mock()

        # reconfig, with a failure
        self.patch_loadConfig_fail()
        wfd = defer.waitForDeferred(
            self.master.reconfig())
        yield wfd
        wfd.getResult()

        self.master.stopService()

        self.assertLogged("reconfig aborted without")
        self.failIf(self.master.reconfigService.called)