def check_pending_upload(session, pu, _test_shim=lambda: None):
    # we can check the upload any time between the expiration of the URL
    # (after which the user can't make any more changes, but the upload
    # may yet be incomplete) and 1 day afterward (ample time for the upload
    # to complete)
    sha512 = pu.file.sha512
    size = pu.file.size

    log = logger.bind(tooltool_sha512=sha512, mozdef=True)

    if time.now() < pu.expires:
        # URL is not expired yet
        return
    elif time.now() > pu.expires + timedelta(days=1):
        # Upload will probably never complete
        log.info(
            "Deleting abandoned pending upload for {}".format(sha512))
        session.delete(pu)
        return

    # connect and see if the file exists..
    s3 = current_app.aws.connect_to('s3', pu.region)
    cfg = current_app.config.get('TOOLTOOL_REGIONS')
    if not cfg or pu.region not in cfg:
        log.warning("Pending upload for {} was to an un-configured "
                    "region".format(sha512))
        session.delete(pu)
        return

    bucket = s3.get_bucket(cfg[pu.region], validate=False)
    key = bucket.get_key(util.keyname(sha512))
    if not key:
        # not uploaded yet
        return

    # commit the session before verifying the file instance, since the
    # DB connection may otherwise go away while we're distracted.
    session.commit()
    _test_shim()

    if not verify_file_instance(sha512, size, key):
        log.warning(
            "Upload of {} was invalid; deleting key".format(sha512))
        key.delete()
        session.delete(pu)
        session.commit()
        return

    log.info("Upload of {} considered valid".format(sha512))
    # add a file instance, but it's OK if it already exists
    try:
        tables.FileInstance(file=pu.file, region=pu.region)
        session.commit()
    except sa.exc.IntegrityError:
        session.rollback()

    # and delete the pending upload
    session.delete(pu)
    session.commit()

def upload_complete(digest):
    """Signal that a file has been uploaded and the server should begin
    validating it.  This is merely an optimization: the server also polls
    occasionally for uploads and validates them when they appear.

    Uploads cannot be safely validated until the upload URL has expired, which
    occurs a short time after the URL is generated (currently 60 seconds but
    subject to change).

    If the upload URL has expired, then the response is an HTTP 202 indicating
    that the signal has been accepted.  If the URL has not expired, then the
    response is an HTTP 409, and the ``X-Retry-After`` header gives a time,
    in seconds, that the client should wait before trying again."""
    if not is_valid_sha512(digest):
        raise BadRequest("Invalid sha512 digest")

    # if the pending upload is still valid, then we can't check this file
    # yet, so return 409 Conflict.  If there is no PU, or it's expired,
    # then we can proceed.
    file = tables.File.query.filter(tables.File.sha512 == digest).first()
    if file:
        for pu in file.pending_uploads:
            until = pu.expires - time.now()
            if until > datetime.timedelta(0):
                # add 1 second to avoid rounding / skew errors
                hdr = {'X-Retry-After': str(1 + int(until.total_seconds()))}
                return Response(status=409, headers=hdr)

    # start a celery task in the background and return immediately
    grooming.check_file_pending_uploads.delay(digest)
    return '{}', 202

def renew_tracker_pending_expiry(tracker):
    pending_expires_at = now() + datetime.timedelta(seconds=PENDING_EXPIRES_IN)
    session = current_app.db.session('relengapi')
    logger.info("renewing tracker {} with pending expiry: {}".format(
                tracker.id, pending_expires_at), archiver_task=tracker.task_id)
    tracker.pending_expires_at = pending_expires_at
    session.commit()

def get_archive(src_url, key, preferred_region):
    """
    A generic getter for retrieving an s3 location of an archive where the archive is based off a
    src_url.

    sub-dir: hg.mozilla.org supports archives of sub directories within a repository. This
    flexibility allows for creating archives of only a portion of what would normally be an entire
    repo archive.

    logic flow:
     If their is already a key within s3, a re-direct link is given for the
    s3 location. If the key does not exist, download the archive from src url, upload it to s3
    for each region supported and return all uploaded s3 url locations.

     When the key does not exist, the remaining work will be assigned to a celery background task
    with a url location returned immediately for obtaining task state updates.
    """
    buckets = current_app.config['ARCHIVER_S3_BUCKETS']
    random_region = buckets.keys()[randint(0, len(buckets.keys()) - 1)]
    # use preferred region if available otherwise choose a valid one at random
    region = preferred_region if preferred_region and preferred_region in buckets else random_region
    bucket = buckets[region]
    s3 = current_app.aws.connect_to('s3', region)
    session = current_app.db.session('relengapi')

    # first, see if the key exists
    if not s3.get_bucket(bucket).get_key(key):
        task_id = key.replace('/', '_')  # keep things simple and avoid slashes in task url
        # can't use unique support:
        # api.pub.build.mozilla.org/docs/development/databases/#unique-row-support-get-or-create
        # because we want to know when the row doesn't exist before creating it
        tracker = tables.ArchiverTask.query.filter(tables.ArchiverTask.task_id == task_id).first()
        if tracker and tracker.state in FINISHED_STATES:
            log = logger.bind(archiver_task=task_id, archiver_task_state=tracker.state)
            log.info('Task tracker: {} exists but finished with state: '
                     '{}'.format(task_id, tracker.state))
            # remove tracker and try celery task again
            delete_tracker(tracker)
            tracker = None
        if not tracker:
            log = logger.bind(archiver_task=task_id)
            log.info("Creating new celery task and task tracker for: {}".format(task_id))
            task = create_and_upload_archive.apply_async(args=[src_url, key], task_id=task_id)
            if task and task.id:
                pending_expires_at = now() + datetime.timedelta(seconds=PENDING_EXPIRES_IN)
                session.add(tables.ArchiverTask(task_id=task.id, s3_key=key, created_at=now(),
                                                pending_expires_at=pending_expires_at,
                                                src_url=src_url, state="PENDING"))
                session.commit()
            else:
                return {}, 500
        return {}, 202, {'Location': url_for('archiver.task_status', task_id=task_id)}

    logger.info("generating GET URL to {}, expires in {}s".format(key, GET_EXPIRES_IN))
    # return 302 pointing to s3 url with archive
    signed_url = s3.generate_url(
        method='GET', expires_in=GET_EXPIRES_IN,
        bucket=bucket, key=key
    )
    return redirect(signed_url)

def cleanup_old_jobs(job_status):
    session = current_app.db.session('relengapi')
    Task = tables.BadpennyTask
    Job = tables.BadpennyJob

    old_job_days = current_app.config.get('BADPENNY_OLD_JOB_DAYS', 7)
    old = time.now() - datetime.timedelta(days=old_job_days)
    deleted = 0

    for task in Task.query.all():
        # Iterate until we find a job that's not too old.  Only
        # delete on the next iteration to avoid deleting the most
        # recent job.
        to_delete = None
        for job in Job.query.filter(Job.task_id == task.id).order_by(Job.created_at):
            if to_delete:
                for log in to_delete.logs:
                    session.delete(log)
                session.delete(to_delete)
                to_delete = None
                deleted += 1

            if job.created_at < old:
                to_delete = job
            else:
                break

    if deleted:
        logger.info("removed %d old jobs", deleted)
        session.commit()

def update_tree_status(session, tree, status=None, reason=None,
                       tags=[], message_of_the_day=None):
    """Update the given tree's status; note that this does not commit
    the session.  Supply a tree object or name."""
    if status is not None:
        tree.status = status
    if reason is not None:
        tree.reason = reason
    if message_of_the_day is not None:
        tree.message_of_the_day = message_of_the_day

    # log it if the reason or status have changed
    if status or reason:
        if status is None:
            status = 'no change'
        if reason is None:
            reason = 'no change'
        l = model.DbLog(
            tree=tree.tree,
            when=relengapi_time.now(),
            who=str(current_user),
            status=status,
            reason=reason,
            tags=tags)
        session.add(l)

    tree_cache_invalidate(tree.tree)

    def run_task(self, task):
        """Actually run a task, inserting a DB row and generating the celery task."""
        job = tables.BadpennyJob(
            task_id=task.task_id,
            created_at=time.now())
        current_app.db.session('relengapi').add(job)
        current_app.db.session('relengapi').commit()

        execution.submit_job(task_name=task.name, job_id=job.id)

    def run(self, parser, args):
        logger.info("Synchronizing tasks into the DB")
        self.sync_tasks()

        logger.info("Creating jobs for overdue tasks")
        now = time.now()
        for task in self.runnable_tasks(now):
            logger.info("Running %r", task.name)
            self.run_task(task)

def add_batch_to_db(app, author, message, files):
    with app.app_context():
        session = app.db.session("relengapi")
        batch = tables.Batch(author=author, message=message, uploaded=relengapi_time.now())
        session.add(batch)
        for filename, file in files.iteritems():
            session.add(tables.BatchFile(filename=filename, batch=batch, file=file))
        session.commit()
        return batch

def test_check_pending_upload_not_expired(app):
    """check_pending_upload doesn't check anything if the URL isn't expired yet"""
    with app.app_context(), set_time():
        expires = time.now() + timedelta(seconds=10)  # 10s shy
        pu_row, file_row = add_pending_upload_and_file_row(
            len(DATA), DATA_DIGEST, expires, 'us-west-2')
        session = app.db.session('relengapi')
        grooming.check_pending_upload(session, pu_row)
        session.commit()
        eq_(len(tables.PendingUpload.query.all()), 1)  # PU still exists

def test_check_pending_upload_bad_region(app):
    """check_pending_upload deletes a pending upload with a bad region"""
    with app.app_context(), set_time():
        expires = time.now() - timedelta(seconds=90)
        pu_row, file_row = add_pending_upload_and_file_row(
            len(DATA), DATA_DIGEST, expires, 'us-west-1')
        session = app.db.session('relengapi')
        grooming.check_pending_upload(session, pu_row)
        session.commit()
        eq_(tables.PendingUpload.query.all(), [])  # PU is deleted

    def _finish(self, successful):
        session = current_app.db.session('relengapi')

        self.job.completed_at = time.now()
        self.job.successful = successful
        if self._log_output:
            content = u'\n'.join(self._log_output)
            l = tables.BadpennyJobLog(id=self.job.id, content=content)
            session.add(l)
        session.commit()

def cleanup_old_tasks(job_status):
    """delete any tracker task if it is older than the time a task can live for."""
    session = current_app.db.session('relengapi')
    expiry_cutoff = now() - datetime.timedelta(seconds=TASK_TIME_OUT)
    table = tables.ArchiverTask
    for tracker in session.query(table).order_by(table.created_at):
        if tracker.created_at < expiry_cutoff:
            delete_tracker(tracker)
        else:
            break

 def _finish(self, successful):
     self._update_job({
         tables.BadpennyJob.completed_at: time.now(),
         tables.BadpennyJob.successful: successful,
     })
     if self._log_output:
         session = current_app.db.session('relengapi')
         content = u'\n'.join(self._log_output)
         l = tables.BadpennyJobLog(id=self.job_id, content=content)
         session.add(l)
         session.commit()

def test_check_file_pending_uploads(app):
    """check_file_pending_uploads calls check_pending_upload for each PU for the file"""
    with app.app_context(), set_time():
        expires = time.now() - timedelta(seconds=90)
        pu_row, file_row = add_pending_upload_and_file_row(
            len(DATA), DATA_DIGEST, expires, 'us-west-2')
        with mock.patch('relengapi.blueprints.tooltool.grooming.check_pending_upload') as cpu:
            pending_uploads = []
            cpu.side_effect = lambda sess, pu: pending_uploads.append(pu)
            grooming.check_file_pending_uploads(DATA_DIGEST)
            assert len(pending_uploads) == 1

def test_check_pending_upload_no_upload(app):
    """check_pending_upload leaves the PU in place if the upload is
    not complete"""
    with app.app_context(), set_time():
        expires = time.now() - timedelta(seconds=90)
        pu_row, file_row = add_pending_upload_and_file_row(
            len(DATA), DATA_DIGEST, expires, 'us-west-2')
        session = app.db.session('relengapi')
        grooming.check_pending_upload(session, pu_row)
        session.commit()
        # PU has not been deleted
        assert tables.PendingUpload.query.first().file.sha512 == DATA_DIGEST

def update_trees(body):
    """
    Update trees' status.

    If the update indicates that the previous state should be saved, then a new
    change will be added to the stack containing the previous status and
    reason.  In this case, both reason and status must be supplied.

    The `tags` property must not be empty if `status` is `closed`.
    """
    session = current_app.db.session('relengapi')
    trees = [session.query(model.DbTree).get(t) for t in body.trees]
    if not all(trees):
        raise NotFound("one or more trees not found")

    if body.status == 'closed' and not body.tags:
        raise BadRequest("tags are required when closing a tree")

    if body.remember:
        if body.status is Unset or body.reason is Unset:
            raise BadRequest("must specify status and reason to remember the change")
        # add a new stack entry with the new and existing states
        ch = model.DbStatusChange(
            who=str(current_user),
            reason=body.reason,
            when=relengapi_time.now(),
            status=body.status)
        for tree in trees:
            stt = model.DbStatusChangeTree(
                tree=tree.tree,
                last_state=json.dumps(
                    {'status': tree.status, 'reason': tree.reason}))
            ch.trees.append(stt)
        session.add(ch)

    # update the trees as requested
    def unset_to_none(x):
        return x if x is not Unset else None
    new_status = unset_to_none(body.status)
    new_reason = unset_to_none(body.reason)
    new_motd = unset_to_none(body.message_of_the_day)
    new_tags = unset_to_none(body.tags) or []

    for tree in trees:
        update_tree_status(session, tree,
                           status=new_status,
                           reason=new_reason,
                           message_of_the_day=new_motd,
                           tags=new_tags)

    session.commit()
    return None, 204

def test_check_pending_upload_not_valid(app):
    """check_pending_upload deletes the PU and the key if the upload is
    invalid."""
    with app.app_context(), set_time():
        expires = time.now() - timedelta(seconds=90)
        pu_row, file_row = add_pending_upload_and_file_row(
            len(DATA), DATA_DIGEST, expires, 'us-west-2')
        make_key(app, 'us-west-2', 'tt-usw2', DATA_KEY, 'xxx')
        session = app.db.session('relengapi')
        grooming.check_pending_upload(session, pu_row)
        session.commit()
        eq_(tables.PendingUpload.query.all(), [])  # PU is deleted
        assert not key_exists(app, 'us-west-2', 'tt-usw2', DATA_KEY)

def test_check_pending_upload_success(app):
    """check_pending_upload deletes the PU and adds a FileInstance if valid"""
    with app.app_context(), set_time():
        expires = time.now() - timedelta(seconds=90)
        pu_row, file_row = add_pending_upload_and_file_row(
            len(DATA), DATA_DIGEST, expires, 'us-west-2')
        make_key(app, 'us-west-2', 'tt-usw2', DATA_KEY, DATA)
        session = app.db.session('relengapi')
        grooming.check_pending_upload(session, pu_row)
        session.commit()
        eq_(tables.PendingUpload.query.all(), [])  # PU is deleted
        eq_(len(tables.File.query.first().instances), 1)  # FileInstance exists
        assert key_exists(app, 'us-west-2', 'tt-usw2', DATA_KEY)

def run_task_now(task_name):
    """Force the given badpenny task to run now."""
    t = tables.BadpennyTask.query.filter(
        tables.BadpennyTask.name == task_name).first()
    if not t:
        raise NotFound

    session = current_app.db.session('relengapi')
    job = tables.BadpennyJob(task=t, created_at=time.now())
    session.add(job)
    session.commit()

    execution.submit_job(task_name=t.name, job_id=job.id)
    return job.to_jsonjob()