def run(self, params):

        additional_timeout = params.get("timeout", 300)

        stats = {
            "requeued": 0,
            "started": 0
        }

        # There shouldn't be that much "started" jobs so we can quite safely
        # iterate over them.

        fields = {"_id": 1, "datestarted": 1, "queue": 1, "path": 1, "retry_count": 1}
        for job_data in connections.mongodb_jobs.mrq_jobs.find(
                {"status": "started"}, projection=fields):
            job = Job(job_data["_id"])
            job.set_data(job_data)

            stats["started"] += 1

            expire_date = datetime.datetime.utcnow(
            ) - datetime.timedelta(seconds=job.timeout + additional_timeout)

            if job_data["datestarted"] < expire_date:
                log.debug("Requeueing job %s" % job.id)
                job.requeue()
                stats["requeued"] += 1

        return stats

def test_cancel_by_path(worker):

  # Start the worker with only one greenlet so that tasks execute sequentially
  worker.start(flags="--gevent 1")

  job_id1 = worker.send_task("tests.tasks.general.MongoInsert", {"a": 41, "sleep": 2}, block=False)

  worker.send_task("mrq.basetasks.utils.JobAction", {
    "path": "tests.tasks.general.MongoInsert",
    "status": "queued",
    "action": "cancel"
  }, block=False)

  job_id2 = worker.send_task("tests.tasks.general.MongoInsert", {"a": 43}, block=False)

  Job(job_id2).wait(poll_interval=0.01)

  # Leave some time to unqueue job_id2 without executing.
  time.sleep(1)
  worker.stop(deps=False)

  job1 = Job(job_id1).fetch().data
  job2 = Job(job_id2).fetch().data

  assert job1["status"] == "success"
  assert job1["result"] == {"a": 41, "sleep": 2}

  assert job2["status"] == "cancel"
  assert job2["dateexpires"] > job2["dateupdated"]

  assert job2.get("result") is None

  assert worker.mongodb_logs.tests_inserts.count() == 1

  assert Queue("default").size() == 0

def test_interrupt_worker_double_sigint(worker, p_flags):
    """ Test what happens when we interrupt a running worker with 2 SIGINTs. """

    start_time = time.time()

    worker.start(flags=p_flags)

    job_id = worker.send_task("tests.tasks.general.Add", {"a": 41, "b": 1, "sleep": 20}, block=False)

    while Job(job_id).fetch().data["status"] == "queued":
        time.sleep(0.1)

    job = Job(job_id).fetch().data
    assert job["status"] == "started"

    # Stop the worker gracefully. first job should still finish!
    worker.stop(block=False, deps=False)

    time.sleep(1)

    # Should not be accepting new jobs!
    job_id2 = worker.send_task("tests.tasks.general.Add", {"a": 42, "b": 1, "sleep": 20}, block=False)

    time.sleep(1)

    job2 = Job(job_id2).fetch().data
    assert job2.get("status") == "queued"

    job = Job(job_id).fetch().data
    assert job["status"] == "started"

    # Sending a second kill -2 should make it stop
    worker.stop(block=True, deps=False, force=True)

    while Job(job_id).fetch().data["status"] == "started":
        time.sleep(0.1)

    job = Job(job_id).fetch().data
    assert job["status"] == "interrupt"

    assert time.time() - start_time < 15

    # Then try the cleaning task that requeues interrupted jobs

    assert Queue("default").size() == 1

    worker.start(queues="cleaning", deps=False, flush=False)

    res = worker.send_task("mrq.basetasks.cleaning.RequeueInterruptedJobs", {}, block=True, queue="cleaning")

    assert res["requeued"] == 1

    assert Queue("default").size() == 2

    Queue("default").list_job_ids() == [str(job_id2), str(job_id)]

    job = Job(job_id).fetch().data
    assert job["status"] == "queued"
    assert job["queue"] == "default"

    def run(self, params):

        # If there are more than this much items on the queue, we don't try to check if our mongodb
        # jobs are still queued.
        max_queue_items = params.get("max_queue_items", 1000)

        stats = {"fetched": 0, "requeued": 0}

        # This was only checking in Redis and wasn't resistant to a redis-wide flush.
        # Doing Queue.all() is slower but covers more edge cases.
        # all_queues = Queue.all_known()

        all_queues = Queue.all()

        log.info("Checking %s queues" % len(all_queues))

        for queue_name in all_queues:

            queue = Queue(queue_name)
            queue_size = queue.size()

            # If the queue is raw, the jobs were only stored in redis so they are lost for good.
            if queue.is_raw:
                continue

            log.info("Checking queue %s" % queue_name)

            if queue_size > max_queue_items:
                log.info("Stopping because queue %s has %s items" % (queue_name, queue_size))
                continue

            queue_jobs_ids = set(queue.list_job_ids(limit=max_queue_items + 1))
            if len(queue_jobs_ids) >= max_queue_items:
                log.info(
                    "Stopping because queue %s actually had more than %s items" % (queue_name, len(queue_jobs_ids))
                )
                continue

            for job_data in connections.mongodb_jobs.mrq_jobs.find(
                {"queue": queue_name, "status": "queued"}, projection={"_id": 1}
            ).sort([["_id", 1]]):

                stats["fetched"] += 1

                if str(job_data["_id"]) in queue_jobs_ids:
                    log.info("Found job %s on queue %s. Stopping" % (job_data["_id"], queue.id))
                    break

                # At this point, this job is not on the queue and we're sure
                # the queue is less than max_queue_items
                # We can safely requeue the job.
                log.info("Requeueing %s on %s" % (job_data["_id"], queue.id))

                stats["requeued"] += 1
                job = Job(job_data["_id"])
                job.requeue(queue=queue_name)

        return stats

def test_retry(worker):

  job_id = worker.send_task("mrq.basetasks.tests.general.Retry", {"queue": "noexec", "countdown": 60}, block=False)

  job_data = Job(job_id).wait(poll_interval=0.01, full_data=True)

  assert job_data["queue"] == "noexec"
  assert job_data["status"] == "retry"
  assert job_data["dateretry"] > datetime.datetime.utcnow()
  assert job_data.get("result") is None

def get_job():

    job_id = request.args['job_id']

    job = Job(job_id)
    job.fetch()

    if job.data["params"].get("user"):
        if not g.user.is_authenticated() or (job.data["params"].get("user") != str(g.user.id)):
            return "Unauthorized."

    return json.dumps({k: v for k, v in job.data.iteritems() if k in ("status", "result")})

def test_retry_cancel_on_retry(worker):

  job_id = worker.send_task("mrq.basetasks.tests.general.Retry", {
    "queue": "noexec",
    "countdown": 60,
    "cancel_on_retry": True
  }, block=False)

  job_data = Job(job_id).wait(poll_interval=0.01, full_data=True)

  assert job_data["status"] == "cancel"
  assert job_data["queue"] == "default"
  assert job_data.get("result") is None

  def wait_for_tasks_results(self, job_ids, block=True, accept_statuses=["success"]):

    if not block:
      return job_ids

    results = []

    for job_id in job_ids:
      job = Job(job_id).wait(poll_interval=0.01)
      assert job.get("status") in accept_statuses, "Job had status %s, not in %s. Dump: %s" % (job.get("status"), accept_statuses, job)

      results.append(job.get("result"))

    return results

def test_cli_run_nonblocking(worker):

  worker.start_deps()

  job_id1 = worker.send_task_cli("tests.tasks.general.Add", {"a": 41, "b": 1}, block=False)

  job1 = Job(job_id1).fetch()

  job1.wait(poll_interval=0.01)

  job1.fetch()

  assert job1.data["status"] == "success"
  assert job1.data["result"] == 42

def test_interrupt_maxconcurrency(worker):

    # The worker will raise a maxconcurrency on the second job
    worker.start(flags="--greenlets=2")

    job_ids = worker.send_tasks(
        "tests.tasks.concurrency.LockedAdd", [{"a": i, "b": 1, "sleep": 2} for i in range(2)], block=False
    )

    worker.wait_for_tasks_results(job_ids, accept_statuses=["success", "failed", "maxconcurrency"])
    job_statuses = [Job(job_id).fetch().data["status"] for job_id in job_ids]
    assert set(job_statuses) == set(["success", "maxconcurrency"])

    # the job concurrency key must be equal to 0
    last_job_id = worker.send_task("tests.tasks.concurrency.LockedAdd", {"a": 1, "b": 1, "sleep": 2}, block=False)

    last_job = Job(last_job_id).wait(poll_interval=0.01)
    assert last_job.get("status") == "success"

  def run(self, params):

    self.collection = connections.mongodb_jobs.mrq_jobs

    # If there are more than this much items on the queue, we don't try to check if our mongodb
    # jobs are still queued.
    max_queue_items = params.get("max_queue_items", 1000)

    stats = {
      "fetched": 0,
      "requeued": 0
    }

    for job_data in self.collection.find({
      "status": "queued"
    }, fields={"_id": 1, "queue": 1}).sort([("_id", 1)]):

      stats["fetched"] += 1

      queue = Queue(job_data["queue"])
      queue_size = queue.size()
      if queue_size > max_queue_items:
        log.info("Stopping because queue %s has %s items" % (queue, queue_size))
        break

      queue_jobs_ids = set(queue.list_job_ids(limit=max_queue_items + 1))
      if len(queue_jobs_ids) >= max_queue_items:
        log.info("Stopping because queue %s actually had more than %s items" % (queue, len(queue_jobs_ids)))
        break

      if str(job_data["_id"]) in queue_jobs_ids:
        log.info("Stopping because we found job %s in redis" % job_data["_id"])
        break

      # At this point, this job is not on the queue and we're sure the queue is less than max_queue_items
      # We can safely requeue the job.
      log.info("Requeueing %s on %s" % (job_data["_id"], queue.id))

      stats["requeued"] += 1
      job = Job(job_data["_id"])
      job.requeue(queue=job_data["queue"])

    return stats

def test_interrupt_worker_gracefully(worker, p_flags):
    """ Test what happens when we interrupt a running worker gracefully. """

    worker.start(flags=p_flags)

    job_id = worker.send_task(
        "tests.tasks.general.Add", {"a": 41, "b": 1, "sleep": 5}, block=False)

    time.sleep(2)

    job = Job(job_id).fetch().data
    assert job["status"] == "started"

    # Stop the worker gracefully. first job should still finish!
    worker.stop(block=False, deps=False)

    time.sleep(1)

    # Should not be accepting new jobs!
    job_id2 = worker.send_task(
        "tests.tasks.general.Add", {"a": 42, "b": 1, "sleep": 4}, block=False)

    time.sleep(1)

    job = Job(job_id2).fetch().data
    assert job.get("status") == "queued"

    time.sleep(4)

    job = Job(job_id).fetch().data
    assert job["status"] == "success"
    assert job["result"] == 42

    job = Job(job_id2).fetch().data
    assert job.get("status") == "queued"

def test_raw_no_storage(worker):
    """ Test tasks that don't store unless they go to error status like 'failed' """

    worker.start(
        flags="--config tests/fixtures/config-raw1.py",
        queues="default testnostorage_raw"
    )

    jobs_collection = worker.mongodb_jobs.mrq_jobs
    test_collection = worker.mongodb_logs.tests_inserts

    worker.send_raw_tasks("testnostorage_raw", [
        "tests.tasks.general.MongoInsert 3"
    ], block=False)

    time.sleep(2)

    # No started inserted.
    assert jobs_collection.count() == 0

    time.sleep(2)

    # No success either, but we did insert
    assert test_collection.count() == 1
    assert jobs_collection.count() == 0
    test_collection.remove({})

    # However failed tasks get stored.

    worker.send_raw_tasks("testnostorage_raw", [
        "tests.tasks.general.RaiseException 0"
    ], block=False)

    time.sleep(2)

    # Failed was inserted.
    assert jobs_collection.count({"status": "failed", "path": "tests.tasks.general.RaiseException"}) == 1

    # If we requeue and don't raise, should be OK and inserted this time, even in success
    # no_storage depends on a raw queue, not a task path.
    _id = jobs_collection.find_one()["_id"]
    jobs_collection.update({"_id": _id}, {"$set": {"path": "tests.tasks.general.MongoInsert"}})
    job = Job(_id).fetch(full_data=True)
    job.requeue(queue="default")

    time.sleep(1)
    assert test_collection.count() == 1
    assert jobs_collection.count() == 1
    assert jobs_collection.count({"status": "success"}) == 1

    jobs_collection.remove({})

    # Test with retry: should be inserted
    worker.send_raw_tasks("testnostorage_raw", [
        "tests.tasks.general.Retry 0"
    ], block=False)

    assert jobs_collection.count({"status": "started"}) == 0

    time.sleep(2)

    assert jobs_collection.count({"status": "retry"}) == 1