def create(self, request, body):
        if (body is None or body.get('job') is None or
                body['job'].get('schedule_id') is None):
            raise webob.exc.HTTPBadRequest()
        job = body['job']

        try:
            schedule = self.db_api.schedule_get_by_id(job['schedule_id'])
        except exception.NotFound:
            raise webob.exc.HTTPNotFound()

        # Check integrity of schedule and update next run
        expected_next_run = job.get('next_run')
        if expected_next_run:
            expected_next_run = timeutils.parse_isotime(job.get('next_run'))

        next_run = api_utils.schedule_to_next_run(schedule, timeutils.utcnow())
        try:
            self.db_api.schedule_test_and_set_next_run(schedule['id'],
                        expected_next_run, next_run)

        except exception.NotFound:
            msg = _("Specified next run does not match the current next run"
                    " value. This could mean schedule has either changed"
                    "or has already been scheduled since you last expected.")
            raise webob.exc.HTTPConflict(explanation=msg)

        # Update schedule last_scheduled
        values = {}
        values['last_scheduled'] = timeutils.utcnow()
        self.db_api.schedule_update(schedule['id'], values)

        # Create job
        values = {}
        values.update(job)
        values['tenant'] = schedule['tenant']
        values['action'] = schedule['action']
        values['status'] = 'QUEUED'

        job_metadata = []
        for metadata in schedule['schedule_metadata']:
            job_metadata.append({
                    'key': metadata['key'],
                    'value': metadata['value']
                    })

        values['job_metadata'] = job_metadata

        job_action = values['action']
        if not 'timeout' in values:
            values['timeout'] = api_utils.get_new_timeout_by_action(job_action)
            values['hard_timeout'] = \
                api_utils.get_new_timeout_by_action(job_action)

        job = self.db_api.job_create(values)
        utils.serialize_datetimes(job)
        api_utils.serialize_job_metadata(job)
        job = {'job': job}
        utils.generate_notification(None, 'qonos.job.create', job, 'INFO')
        return job

def _gen_base_attributes(item_id=None):
    values = {}
    if item_id is None:
        values['id'] = str(uuid.uuid4())
    values['created_at'] = timeutils.utcnow()
    values['updated_at'] = timeutils.utcnow()
    return copy.deepcopy(values)

 def test_schedule_test_and_set_next_run_invalid(self):
     fixture = {
         'id': str(uuid.uuid4()),
         'tenant': str(uuid.uuid4()),
         'action': 'snapshot',
         'minute': 30,
         'hour': 2,
     }
     bad_expected_next_run = timeutils.utcnow()
     timeutils.advance_time_seconds(10)
     schedule = self.db_api.schedule_create(fixture)
     self.assertRaises(exception.NotFound,
                       self.db_api.schedule_test_and_set_next_run,
                       schedule['id'], bad_expected_next_run,
                       timeutils.utcnow())

    def _create_jobs(self):
        now = timeutils.utcnow()
        timeout = now + datetime.timedelta(hours=1)
        hard_timeout = now + datetime.timedelta(hours=4)
        fixture = {
            'id': unit_utils.JOB_UUID1,
            'action': 'snapshot',
            'tenant': unit_utils.TENANT1,
            'schedule_id': unit_utils.SCHEDULE_UUID1,
            'worker_id': unit_utils.WORKER_UUID1,
            'status': 'queued',
            'timeout': timeout,
            'hard_timeout': hard_timeout,
            'retry_count': 0,
        }
        self.job_1 = self.db_api.job_create(fixture)

        fixture = {
            'id': unit_utils.JOB_UUID2,
            'action': 'snapshot',
            'tenant': unit_utils.TENANT1,
            'schedule_id': unit_utils.SCHEDULE_UUID2,
            'worker_id': unit_utils.WORKER_UUID2,
            'status': 'error',
            'timeout': timeout,
            'hard_timeout': hard_timeout,
            'retry_count': 0,
        }
        self.job_2 = self.db_api.job_create(fixture)

    def test_get_next_job_assigned_once_due_to_timeout(self):
        now = timeutils.utcnow()
        timeout = now - datetime.timedelta(hours=1)
        new_timeout = now + datetime.timedelta(hours=3)
        hard_timeout = now + datetime.timedelta(hours=4)
        job_fixture = {
            'action': 'snapshot',
            'tenant': unit_utils.TENANT1,
            'schedule_id': unit_utils.SCHEDULE_UUID2,
            'worker_id': unit_utils.WORKER_UUID2,
            'status': 'queued',
            'timeout': timeout,
            'hard_timeout': hard_timeout,
            'retry_count': 0,
        }

        retries = 2
        self._create_jobs(10, job_fixture)
        job = db_api.job_get_and_assign_next_by_action('snapshot',
                                                       unit_utils.WORKER_UUID1,
                                                       retries,
                                                       new_timeout)

        job2 = db_api.job_get_and_assign_next_by_action('snapshot',
                                                       unit_utils.WORKER_UUID1,
                                                       retries,
                                                       new_timeout)
        self.assertEqual(job2, None)

    def test_cron_string_to_datetime(self):
        minute = timeutils.utcnow().minute
        if minute == 0:
            minute = 59
        else:
            minute -= 1

        hour = timeutils.utcnow().hour
        if hour == 0:
            hour = 23
        else:
            hour -= 1
        next_run = utils.cron_string_to_next_datetime(minute=minute,
                                                      hour=hour)

        self.assertTrue(next_run > timeutils.utcnow())

def job_get_and_assign_next_by_action(action, worker_id, max_retry, new_timeout):
    """Get the next available job for the given action and assign it
    to the worker for worker_id."""
    now = timeutils.utcnow()
    session = get_session()

    job_ref = _job_get_next_by_action(session, now, action, max_retry)

    if not job_ref:
        return None

    # Make sure the job has not changed unexpectedly since
    # retrieving it
    try:
        query = (
            session.query(models.Job)
            .filter_by(id=job_ref["id"])
            .update({"worker_id": worker_id, "timeout": new_timeout, "retry_count": job_ref["retry_count"] + 1})
        )
    except sa_orm.exc.NoResultFound:
        # In case the job was deleted during assignment return nothing
        return None
    except sa_orm.exc.StaleDataError:
        # In case the job was picked up by another transaction return nothing
        return None

    if not query:
        return None

    return _job_get_by_id(job_ref["id"])

def job_get_and_assign_next_by_action(action, worker_id, max_retry,
                                      new_timeout):
    """Get the next available job for the given action and assign it
    to the worker for worker_id.
    This must be an atomic action!"""
    job_ref = None
    now = timeutils.utcnow()
    jobs = _jobs_get_sorted()
    for job in jobs:
        if job['action'] == action and \
                job['retry_count'] < max_retry and \
                job['hard_timeout'] > now and \
                job['status'] not in ['DONE', 'CANCELLED'] and \
                (job['worker_id'] is None or job['timeout'] <= now):
            job_ref = job
            break

    if job_ref is None:
        return None

    job_id = job_ref['id']
    DATA['jobs'][job_id]['worker_id'] = worker_id
    DATA['jobs'][job_id]['timeout'] = new_timeout
    DATA['jobs'][job_id]['retry_count'] = job_ref['retry_count'] + 1
    job = copy.deepcopy(DATA['jobs'][job_id])
    job['job_metadata'] = job_meta_get_all_by_job_id(job_id)

    return job

    def test_create_zero_hour(self):
        hour = 0
        fixture = {'schedule': {
            'id': unit_utils.SCHEDULE_UUID5,
            'tenant': unit_utils.TENANT1,
            'action': 'snapshot',
            'minute': 30,
            'hour': hour,
        }}
        expected = fixture['schedule']
        request = unit_utils.get_fake_request(method='POST')

        actual = self.controller.create(request, fixture)['schedule']

        self.assertNotEqual(actual.get('id'), None)
        self.assertNotEqual(actual.get('created_at'), None)
        self.assertNotEqual(actual.get('updated_at'), None)
        now = timeutils.utcnow()
        if not (now.hour == hour and now.minute < 30):
            now = now + datetime.timedelta(days=1)
        expected_next_run = timeutils.isotime(
            now.replace(hour=hour, minute=30, second=0,
                        microsecond=0))
        self.assertEqual(expected_next_run, actual['next_run'])
        self.assertEqual(expected['tenant'], actual['tenant'])
        self.assertEqual(expected['action'], actual['action'])
        self.assertEqual(expected['minute'], actual['minute'])
        self.assertEqual(expected['hour'], actual['hour'])

def job_update(job_id, job_values):
    global DATA
    values = job_values.copy()
    if job_id not in DATA['jobs']:
        raise exception.NotFound()

    metadata = None
    if 'job_metadata' in values:
        metadata = values['job_metadata']
        del values['job_metadata']

    if len(values) > 0:
        job = DATA['jobs'][job_id]
        #NOTE(ameade): This must come before update specified values since
        # we may be trying to manually set updated_at
        job['updated_at'] = timeutils.utcnow()
        job['version_id'] = str(uuid.uuid4())
        job.update(values)

    if metadata is not None:
        DATA['job_metadata'][job_id] = {}
        for metadatum in metadata:
            job_meta_create(job_id, metadatum)

    return job_get_by_id(job_id)

def job_get_and_assign_next_by_action(action, worker_id, new_timeout):
    """Get the next available job for the given action and assign it
    to the worker for worker_id.
    This must be an atomic action!"""
    job_ref = None
    now = timeutils.utcnow().replace(second=0, microsecond=0)
    jobs = _jobs_get_sorted()
    statuses = ['DONE', 'CANCELLED', 'HARD_TIMED_OUT', 'MAX_RETRIED']
    for job in jobs:
        if job['action'] == action and \
                job['status'] not in statuses and \
                (job['worker_id'] is None or job['timeout'] <= now):
            job_ref = job
            break

    if job_ref is None:
        return None

    job_id = job_ref['id']
    DATA['jobs'][job_id]['worker_id'] = worker_id
    DATA['jobs'][job_id]['timeout'] = new_timeout
    DATA['jobs'][job_id]['retry_count'] = job_ref['retry_count'] + 1
    DATA['jobs'][job_id]['version_id'] = str(uuid.uuid4())
    job = copy.deepcopy(DATA['jobs'][job_id])
    job['job_metadata'] = job_meta_get_all_by_job_id(job_id)

    return job

    def test_create_zero_hour(self):
        hour = 0
        fixture = {
            "schedule": {
                "id": unit_utils.SCHEDULE_UUID5,
                "tenant": unit_utils.TENANT1,
                "action": "snapshot",
                "minute": 30,
                "hour": hour,
            }
        }
        expected = fixture["schedule"]
        request = unit_utils.get_fake_request(method="POST")

        actual = self.controller.create(request, fixture)["schedule"]

        self.assertNotEqual(actual.get("id"), None)
        self.assertNotEqual(actual.get("created_at"), None)
        self.assertNotEqual(actual.get("updated_at"), None)
        now = timeutils.utcnow()
        if not (now.hour == hour and now.minute < 30):
            now = now + datetime.timedelta(days=1)
        expected_next_run = timeutils.isotime(now.replace(hour=hour, minute=30, second=0, microsecond=0))
        self.assertEqual(expected_next_run, actual["next_run"])
        self.assertEqual(expected["tenant"], actual["tenant"])
        self.assertEqual(expected["action"], actual["action"])
        self.assertEqual(expected["minute"], actual["minute"])
        self.assertEqual(expected["hour"], actual["hour"])

 def fake_next_datetime(min, h, dom, m, dow, start_time):
     self.called = True
     self.assertEqual(min, '*')
     self.assertEqual(h, '*')
     self.assertEqual(dom, '*')
     self.assertEqual(m, '*')
     self.assertEqual(dow, '*')
     self.assertEqual(timeutils.utcnow(), start_time)

def _jobs_cleanup_hard_timed_out():
    """Find all jobs with hard_timeout values which have passed
    and delete them, logging the timeout / failure as appropriate"""
    now = timeutils.utcnow()
    session = get_session()
    num_del = session.query(models.Job).filter(models.Job.hard_timeout <= now).delete()
    session.flush()
    return num_del

def get_new_timeout_by_action(action):
    now = timeutils.utcnow()

    group = 'action_' + action
    if group not in CONF:
        group = 'action_default'
    job_timeout_seconds = CONF.get(group).timeout_seconds
    return now + datetime.timedelta(seconds=job_timeout_seconds)

    def _create_images_list(self, instance_id, image_count):
        images = []
        base_time = timeutils.utcnow()
        one_day = datetime.timedelta(days=1)
        for i in range(image_count):
            images.append(self._create_image(instance_id, base_time))
            base_time = base_time - one_day

        return images

 def test_get_next_job_too_many_retries(self):
     now = timeutils.utcnow()
     new_timeout = now + datetime.timedelta(hours=3)
     now = timeutils.utcnow()
     retries = 2
     self.job_fixture_2['retry_count'] = 3
     self.job_fixture_2['timeout'] = now + datetime.timedelta(seconds=5)
     self._create_jobs(10, self.job_fixture_2, self.job_fixture_1)
     job = db_api.job_get_and_assign_next_by_action('snapshot',
                                                    unit_utils.WORKER_UUID1,
                                                    retries,
                                                    new_timeout)
     expected = self.jobs[1]
     self.assertEqual(job['id'], expected['id'])
     self.assertEqual(job['worker_id'], unit_utils.WORKER_UUID1)
     self.assertEqual(job['timeout'], new_timeout)
     self.assertEqual(job['hard_timeout'], expected['hard_timeout'])
     self.assertEqual(job['retry_count'], expected['retry_count'] + 1)

    def _assign_jobs_for_concurrent_workers(self, workers):
        new_timeout = timeutils.utcnow() + datetime.timedelta(hours=3)

        worker1_job = base.db_api.job_get_and_assign_next_by_action(
            'snapshot', workers[0], new_timeout)
        worker2_job = base.db_api.job_get_and_assign_next_by_action(
            'snapshot', workers[1], new_timeout)

        return worker1_job, worker2_job

    def _create_jobs(self):
        now = timeutils.utcnow()
        timeout = now + datetime.timedelta(hours=1)
        hard_timeout = now + datetime.timedelta(hours=4)

        fixture = {
            "id": unit_utils.JOB_UUID1,
            "schedule_id": self.schedule_1["id"],
            "tenant": unit_utils.TENANT1,
            "worker_id": None,
            "action": "snapshot",
            "status": None,
            "timeout": timeout,
            "hard_timeout": hard_timeout,
            "retry_count": 0,
        }
        self.job_1 = db_api.job_create(fixture)
        fixture = {
            "id": unit_utils.JOB_UUID2,
            "schedule_id": self.schedule_2["id"],
            "tenant": unit_utils.TENANT2,
            "worker_id": unit_utils.WORKER_UUID2,
            "action": "snapshot",
            "status": None,
            "timeout": timeout,
            "hard_timeout": hard_timeout,
            "retry_count": 1,
            "job_metadata": [{"key": "instance_id", "value": "my_instance"}],
        }
        self.job_2 = db_api.job_create(fixture)
        fixture = {
            "id": unit_utils.JOB_UUID3,
            "schedule_id": self.schedule_3["id"],
            "tenant": unit_utils.TENANT3,
            "worker_id": unit_utils.WORKER_UUID2,
            "action": "snapshot",
            "status": None,
            "timeout": timeout,
            "hard_timeout": hard_timeout,
            "retry_count": 1,
            "job_metadata": [{"key": "instance_id", "value": "my_instance"}],
        }
        self.job_3 = db_api.job_create(fixture)
        fixture = {
            "id": unit_utils.JOB_UUID4,
            "schedule_id": self.schedule_4["id"],
            "tenant": unit_utils.TENANT4,
            "worker_id": unit_utils.WORKER_UUID2,
            "action": "snapshot",
            "status": None,
            "timeout": timeout,
            "hard_timeout": hard_timeout,
            "retry_count": 1,
            "job_metadata": [{"key": "instance_id", "value": "my_instance"}],
        }
        self.job_4 = db_api.job_create(fixture)

def schedule_to_next_run(schedule, start_time=None):
    start_time = start_time or timeutils.utcnow()
    minute = schedule.get('minute', '*')
    hour = schedule.get('hour', '*')
    day_of_month = schedule.get('day_of_month', '*')
    month = schedule.get('month', '*')
    day_of_week = schedule.get('day_of_week', '*')
    return utils.cron_string_to_next_datetime(minute, hour, day_of_month,
                                              month, day_of_week,
                                              start_time)