Python utils.generate_notification函数代码示例

OGeek|极客世界-中国程序员成长平台 › 门户 › 编程› Python›Python编程经验

原作者: [db:作者] 来自: [db:来源] 收藏邀请

本文整理汇总了Python中qonos.common.utils.generate_notification函数的典型用法代码示例。如果您正苦于以下问题：Python generate_notification函数的具体用法？Python generate_notification怎么用？Python generate_notification使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。

在下文中一共展示了generate_notification函数的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: test_generate_notification

    def test_generate_notification(self):

        notification = {}

        def fake_publisher_id(service, host=None):
            if not host:
                host = 'localhost'
            return "%s.%s" % (service, host)

        def fake_notify(context, publisher_id, event_type, priority, payload):
            notification['context'] = context
            notification['publisher_id'] = publisher_id
            notification['event_type'] = event_type
            notification['priority'] = priority
            notification['payload'] = payload

        self.stubs.Set(notifier_api, 'publisher_id', fake_publisher_id)
        self.stubs.Set(notifier_api, 'notify', fake_notify)
        payload = {'id': 'fake-id'}
        self.assertEqual(notification, {})
        self.config(host='localhost')
        utils.generate_notification(None, 'qonos.fake.event', payload)
        expected_notification = {
                                    'context': None,
                                    'publisher_id': 'qonos.localhost',
                                    'event_type': 'qonos.fake.event',
                                    'priority': 'INFO',
                                    'payload': {'id': 'fake-id'}
                                }

        self.assertEqual(notification, expected_notification)

开发者ID:amalaba，项目名称:qonos，代码行数:31，代码来源:test_utils.py

示例2: test_doesnt_delete_images_on_retention_error

    def test_doesnt_delete_images_on_retention_error(self):
        timeutils.set_time_override()
        self.nova_client.servers.get(mox.IsA(str)).AndReturn(MockServer())
        self.nova_client.servers.create_image(mox.IsA(str),
            mox.IsA(str), self.snapshot_meta).AndReturn(IMAGE_ID)
        self.nova_client.images.get(IMAGE_ID).AndReturn(
            MockImageStatus('ACTIVE'))
        self.nova_client.rax_scheduled_images_python_novaclient_ext.\
            get(mox.IsA(str)).AndRaise(Exception())
        mock_server = MockServer(retention=None)
        self._init_worker_mock()
        self.mox.StubOutWithMock(utils, 'generate_notification')
        utils.generate_notification(None, 'qonos.job.run.start', mox.IsA(dict),
                                    mox.IsA(str))
        utils.generate_notification(None, 'qonos.job.run.end', mox.IsA(dict),
                                    mox.IsA(str))
        self.worker.get_qonos_client().AndReturn(self.qonos_client)
        self.qonos_client.delete_schedule(mox.IsA(str))
        self.worker.update_job(fakes.JOB_ID, 'DONE', timeout=None,
                               error_message=None)
        self.mox.ReplayAll()

        processor = TestableSnapshotProcessor(self.nova_client)
        processor.init_processor(self.worker)

        processor.process_job(self.job)

        self.mox.VerifyAll()

开发者ID:amalaba，项目名称:qonos，代码行数:28，代码来源:test_snapshot.py

示例3: create

    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

开发者ID:pperezrubio，项目名称:qonos，代码行数:60，代码来源:jobs.py

示例4: test_process_job_should_not_create_on_job_error_image_ok

    def test_process_job_should_not_create_on_job_error_image_ok(self):
        timeutils.set_time_override()
        self.job['metadata']['image_id'] = IMAGE_ID
        self.job['status'] = 'ERROR'

        self.nova_client.images.get(IMAGE_ID).AndReturn(
            MockImageStatus('SAVING'))
        self.nova_client.images.get(IMAGE_ID).AndReturn(
            MockImageStatus('ACTIVE'))

        mock_retention = MockRetention()
        self.nova_client.rax_scheduled_images_python_novaclient_ext.\
            get(mox.IsA(str)).AndReturn(mock_retention)

        self._simple_prepare_worker_mock(skip_metadata_update=True)

        self.worker.update_job(fakes.JOB_ID, 'DONE', timeout=None,
                               error_message=None)
        self.mox.StubOutWithMock(utils, 'generate_notification')
        utils.generate_notification(None, mox.IsA(str), mox.IsA(dict),
                                    mox.IsA(str)).MultipleTimes()
        self.mox.ReplayAll()

        processor = TestableSnapshotProcessor(self.nova_client)
        processor.init_processor(self.worker)

        processor.process_job(self.job)

        self.mox.VerifyAll()

开发者ID:clefelhocz，项目名称:qonos，代码行数:29，代码来源:test_snapshot.py

示例5: test_deletes_images_more_than_retention

    def test_deletes_images_more_than_retention(self):
        timeutils.set_time_override()
        instance_id = self.job["metadata"]["instance_id"]
        self.nova_client.servers.get(mox.IsA(str)).AndReturn(MockServer())
        self.nova_client.servers.create_image(mox.IsA(str), mox.IsA(str), self.snapshot_meta).AndReturn(IMAGE_ID)
        self.nova_client.images.get(IMAGE_ID).AndReturn(MockImageStatus("ACTIVE"))
        mock_retention = MockRetention(3)
        self.nova_client.rax_scheduled_images_python_novaclient_ext.get(mox.IsA(str)).AndReturn(mock_retention)
        mock_server = MockServer(instance_id=instance_id)
        image_list = self._create_images_list(mock_server.id, 5)
        self.nova_client.images.list(detailed=True).AndReturn(image_list)
        # The image list happens to be in descending created order
        self.nova_client.images.delete(image_list[-2].id)
        self.nova_client.images.delete(image_list[-1].id)
        self._init_worker_mock()
        self.worker.update_job(fakes.JOB_ID, "DONE", timeout=None, error_message=None)
        self.mox.StubOutWithMock(utils, "generate_notification")
        utils.generate_notification(None, "qonos.job.run.start", mox.IsA(dict), mox.IsA(str))
        utils.generate_notification(None, "qonos.job.run.end", mox.IsA(dict), mox.IsA(str))
        self.mox.ReplayAll()

        processor = TestableSnapshotProcessor(self.nova_client)
        processor.init_processor(self.worker)

        processor.process_job(self.job)

        self.mox.VerifyAll()

开发者ID:hmakkapati，项目名称:qonos，代码行数:27，代码来源:test_snapshot.py

示例6: test_process_job_should_cancel_if_schedule_deleted

    def test_process_job_should_cancel_if_schedule_deleted(self):
        self._init_qonos_client()
        self.mox.StubOutWithMock(utils, 'generate_notification')
        utils.generate_notification(None, 'qonos.job.run.start', mox.IsA(dict),
                                    mox.IsA(str))
        response = {'status': 'CANCELLED', 'timeout': self.job['timeout']}
        self.worker.update_job(fakes.JOB_ID, 'CANCELLED', timeout=None,
                               error_message=mox.IsA(str)).AndReturn(response)
        expected_payload = {'job': {'status': 'CANCELLED',
                   'hard_timeout': self.job['hard_timeout'],
                   'created_at': self.job['created_at'],
                   'modified_at': self.job['modified_at'],
                   'retry_count': 1,
                   'schedule_id': '33333333-3333-3333-3333-33333333',
                   'worker_id': '11111111-1111-1111-1111-11111111',
                   'timeout': self.job['timeout'],
                   'action': 'snapshot',
                   'id': '22222222-2222-2222-2222-22222222',
                   'tenant': '44444444-4444-4444-4444-44444444',
                   'metadata': {'instance_id':
                   '55555555-5555-5555-5555-55555555'}}}
        utils.generate_notification(None, 'qonos.job.failed', expected_payload,
                                    mox.IsA(str))
        self.mox.ReplayAll()

        processor = TestableSnapshotProcessor(self.nova_client)
        processor.init_processor(self.worker)

        processor.process_job(self.job)

        self.mox.VerifyAll()

开发者ID:clefelhocz，项目名称:qonos，代码行数:31，代码来源:test_snapshot.py

示例7: test_doesnt_delete_images_from_another_instance

    def test_doesnt_delete_images_from_another_instance(self):
        timeutils.set_time_override()
        instance_id = self.job["metadata"]["instance_id"]
        self.nova_client.servers.get(mox.IsA(str)).AndReturn(MockServer())
        self.nova_client.servers.create_image(mox.IsA(str), mox.IsA(str), self.snapshot_meta).AndReturn(IMAGE_ID)
        self.nova_client.images.get(IMAGE_ID).AndReturn(MockImageStatus("ACTIVE"))
        mock_retention = MockRetention(3)
        self.nova_client.rax_scheduled_images_python_novaclient_ext.get(mox.IsA(str)).AndReturn(mock_retention)
        mock_server = MockServer(instance_id=instance_id)
        image_list = self._create_images_list(mock_server.id, 5)
        to_delete = image_list[3:]
        image_list.extend(self._create_images_list(uuidutils.generate_uuid(), 3))
        self.nova_client.images.list(detailed=True).AndReturn(image_list)
        # The image list happens to be in descending created order
        self.nova_client.images.delete(to_delete[0].id)
        self.nova_client.images.delete(to_delete[1].id)
        self._simple_prepare_worker_mock()
        self.mox.StubOutWithMock(utils, "generate_notification")
        utils.generate_notification(None, "qonos.job.run.start", mox.IsA(dict), mox.IsA(str))
        utils.generate_notification(None, "qonos.job.run.end", mox.IsA(dict), mox.IsA(str))
        self.mox.ReplayAll()

        processor = TestableSnapshotProcessor(self.nova_client)
        processor.init_processor(self.worker)
        processor.process_job(self.job)

        self.mox.VerifyAll()

开发者ID:hmakkapati，项目名称:qonos，代码行数:27，代码来源:test_snapshot.py

示例8: test_process_job_should_succeed_after_multiple_tries

    def test_process_job_should_succeed_after_multiple_tries(self):
        timeutils.set_time_override()
        self.nova_client.servers.get(mox.IsA(str)).AndReturn(MockServer())
        self.nova_client.servers.create_image(mox.IsA(str),
            mox.IsA(str), self.snapshot_meta).AndReturn(IMAGE_ID)
        self.nova_client.images.get(IMAGE_ID).AndReturn(
            MockImageStatus('QUEUED'))
        self.nova_client.images.get(IMAGE_ID).AndReturn(
            MockImageStatus('SAVING'))
        self.nova_client.images.get(IMAGE_ID).AndReturn(
            MockImageStatus('SAVING'))
        self.nova_client.images.get(IMAGE_ID).AndReturn(
            MockImageStatus('ACTIVE'))
        mock_retention = MockRetention()
        self.nova_client.rax_scheduled_images_python_novaclient_ext.\
            get(mox.IsA(str)).AndReturn(mock_retention)
        self._simple_prepare_worker_mock()
        self.mox.StubOutWithMock(utils, 'generate_notification')
        utils.generate_notification(None, 'qonos.job.run.start', mox.IsA(dict),
                                    mox.IsA(str))
        utils.generate_notification(None, 'qonos.job.run.end', mox.IsA(dict),
                                    mox.IsA(str))
        self.mox.ReplayAll()

        processor = TestableSnapshotProcessor(self.nova_client)
        processor.init_processor(self.worker)

        processor.process_job(self.job)

        self.mox.VerifyAll()

开发者ID:pperezrubio，项目名称:qonos，代码行数:30，代码来源:test_snapshot.py

示例9: _do_test_process_job_should_update_image_error

    def _do_test_process_job_should_update_image_error(self, error_status):
        base_time = timeutils.utcnow()
        time_seq = [
            base_time,
            base_time,
            base_time + datetime.timedelta(seconds=305),
            base_time + datetime.timedelta(seconds=605),
            base_time + datetime.timedelta(seconds=905),
            base_time + datetime.timedelta(seconds=1205),
            base_time + datetime.timedelta(seconds=1505),
            ]
        timeutils.set_time_override_seq(time_seq)

        job = copy.deepcopy(self.job)
        job['timeout'] = base_time + datetime.timedelta(minutes=60)

        self.nova_client.servers.get(mox.IsA(str)).AndReturn(MockServer())
        self.nova_client.servers.create_image(mox.IsA(str),
            mox.IsA(str), self.snapshot_meta).AndReturn(IMAGE_ID)
        self.nova_client.images.get(IMAGE_ID).AndReturn(
            MockImageStatus('QUEUED'))
        self.nova_client.images.get(IMAGE_ID).AndReturn(
            MockImageStatus('SAVING'))
        self.nova_client.images.get(IMAGE_ID).AndReturn(
            MockImageStatus('SAVING'))
        self.nova_client.images.get(IMAGE_ID).AndReturn(
            MockImageStatus('SAVING'))
        self.nova_client.images.get(IMAGE_ID).AndReturn(
            error_status)

        self._init_worker_mock()
        self.worker.update_job(fakes.JOB_ID, 'PROCESSING', timeout=None,
                               error_message=None)
        self.worker.update_job(fakes.JOB_ID, 'PROCESSING', timeout=None,
                               error_message=None)
        self.worker.update_job(fakes.JOB_ID, 'PROCESSING', timeout=None,
                               error_message=None)
        self.worker.update_job(fakes.JOB_ID, 'PROCESSING', timeout=None,
                               error_message=None)
        self.worker.update_job(fakes.JOB_ID, 'ERROR', timeout=None,
                               error_message=mox.IsA(str))

        self.mox.StubOutWithMock(utils, 'generate_notification')
        utils.generate_notification(None, 'qonos.job.run.start', mox.IsA(dict),
                                    mox.IsA(str))
        self.mox.ReplayAll()

        processor = TestableSnapshotProcessor(self.nova_client)
        processor.init_processor(self.worker)

        processor.process_job(job)

        self.mox.VerifyAll()

开发者ID:pperezrubio，项目名称:qonos，代码行数:53，代码来源:test_snapshot.py

示例10: test_process_job_should_cancel_if_schedule_deleted

    def test_process_job_should_cancel_if_schedule_deleted(self):
        self._init_qonos_client()
        self.mox.StubOutWithMock(utils, "generate_notification")
        utils.generate_notification(None, "qonos.job.run.start", mox.IsA(dict), mox.IsA(str))
        self.worker.update_job(fakes.JOB_ID, "CANCELLED", timeout=None, error_message=mox.IsA(str))
        self.mox.ReplayAll()

        processor = TestableSnapshotProcessor(self.nova_client)
        processor.init_processor(self.worker)

        processor.process_job(self.job)

        self.mox.VerifyAll()

开发者ID:hmakkapati，项目名称:qonos，代码行数:13，代码来源:test_snapshot.py

示例11: test_process_job_should_update_status_and_timestamp

    def test_process_job_should_update_status_and_timestamp(self):
        base_time = timeutils.utcnow()
        time_seq = [
            base_time,
            base_time,
            base_time + datetime.timedelta(seconds=305),
            base_time + datetime.timedelta(minutes=60, seconds=5),
            base_time + datetime.timedelta(minutes=60, seconds=305),
            ]
        timeutils.set_time_override_seq(time_seq)

        job = copy.deepcopy(self.job)
        job['timeout'] = base_time + datetime.timedelta(minutes=60)

        self.nova_client.servers.get(mox.IsA(str)).AndReturn(MockServer())
        self.nova_client.servers.create_image(mox.IsA(str),
            mox.IsA(str), self.snapshot_meta).AndReturn(IMAGE_ID)
        self.nova_client.images.get(IMAGE_ID).AndReturn(
            MockImageStatus('QUEUED'))
        self.nova_client.images.get(IMAGE_ID).AndReturn(
            MockImageStatus('SAVING'))
        self.nova_client.images.get(IMAGE_ID).AndReturn(
            MockImageStatus('SAVING'))
        self.nova_client.images.get(IMAGE_ID).AndReturn(
            MockImageStatus('ACTIVE'))
        mock_retention = MockRetention()
        self.nova_client.rax_scheduled_images_python_novaclient_ext.\
            get(mox.IsA(str)).AndReturn(mock_retention)
        self._init_worker_mock()
        self.worker.update_job(fakes.JOB_ID, 'PROCESSING',
                               timeout=mox.IsA(datetime.datetime),
                               error_message=None)
        self.worker.update_job(fakes.JOB_ID, 'PROCESSING', timeout=None,
                               error_message=None)
        self.mox.StubOutWithMock(utils, 'generate_notification')
        utils.generate_notification(None, 'qonos.job.run.start', mox.IsA(dict),
                                    mox.IsA(str))
        utils.generate_notification(None, 'qonos.job.run.end', mox.IsA(dict),
                                    mox.IsA(str))
        self.worker.get_qonos_client().AndReturn(self.qonos_client)
        self.qonos_client.delete_schedule(mox.IsA(str))
        self.worker.update_job(fakes.JOB_ID, 'DONE', timeout=None,
                               error_message=None)
        self.mox.ReplayAll()

        processor = TestableSnapshotProcessor(self.nova_client)
        processor.init_processor(self.worker)

        processor.process_job(job)

        self.mox.VerifyAll()

开发者ID:amalaba，项目名称:qonos，代码行数:51，代码来源:test_snapshot.py

示例12: test_process_job_should_cancel_if_no_instance_id

    def test_process_job_should_cancel_if_no_instance_id(self):
        self._init_qonos_client(schedule=MockSchedule())
        self.mox.StubOutWithMock(utils, "generate_notification")
        utils.generate_notification(None, "qonos.job.run.start", mox.IsA(dict), mox.IsA(str))
        self.worker.update_job(fakes.JOB_ID, "PROCESSING", timeout=mox.IsA(datetime.datetime), error_message=None)
        self.worker.update_job(fakes.JOB_ID, "CANCELLED", timeout=None, error_message=mox.IsA(str))
        self.mox.ReplayAll()

        processor = TestableSnapshotProcessor(self.nova_client)
        processor.init_processor(self.worker)
        del self.job["metadata"]["instance_id"]
        processor.process_job(self.job)

        self.mox.VerifyAll()

开发者ID:hmakkapati，项目名称:qonos，代码行数:14，代码来源:test_snapshot.py

示例13: test_process_job_should_cancel_if_instance_not_found

    def test_process_job_should_cancel_if_instance_not_found(self):
        self._init_qonos_client(schedule=MockSchedule())
        self.mox.StubOutWithMock(utils, "generate_notification")
        utils.generate_notification(None, "qonos.job.run.start", mox.IsA(dict), mox.IsA(str))
        self.nova_client.servers.get(mox.IsA(str)).AndReturn(MockServer())
        self.nova_client.servers.create_image(mox.IsA(str), mox.IsA(str), self.snapshot_meta).AndRaise(
            exceptions.NotFound("404")
        )
        self.worker.update_job(fakes.JOB_ID, "PROCESSING", timeout=mox.IsA(datetime.datetime), error_message=None)
        self.worker.update_job(fakes.JOB_ID, "CANCELLED", timeout=None, error_message=mox.IsA(str))
        self.worker.get_qonos_client().AndReturn(self.qonos_client)
        self.qonos_client.delete_schedule(mox.IsA(str))
        self.mox.ReplayAll()

        processor = TestableSnapshotProcessor(self.nova_client)
        processor.init_processor(self.worker)
        processor.process_job(self.job)

        self.mox.VerifyAll()

开发者ID:hmakkapati，项目名称:qonos，代码行数:19，代码来源:test_snapshot.py

示例14: test_process_job_should_succeed_immediately

    def test_process_job_should_succeed_immediately(self):
        timeutils.set_time_override()
        self.nova_client.servers.get(mox.IsA(str)).AndReturn(MockServer())
        self.nova_client.servers.create_image(mox.IsA(str), mox.IsA(str), self.snapshot_meta).AndReturn(IMAGE_ID)
        self.nova_client.images.get(IMAGE_ID).AndReturn(MockImageStatus("ACTIVE"))
        mock_retention = MockRetention()
        self.nova_client.rax_scheduled_images_python_novaclient_ext.get(mox.IsA(str)).AndReturn(mock_retention)
        self._simple_prepare_worker_mock()

        self.mox.StubOutWithMock(utils, "generate_notification")
        utils.generate_notification(None, "qonos.job.run.start", mox.IsA(dict), mox.IsA(str))
        utils.generate_notification(None, "qonos.job.run.end", mox.IsA(dict), mox.IsA(str))
        self.worker.get_qonos_client().AndReturn(self.qonos_client)
        self.qonos_client.delete_schedule(mox.IsA(str))
        self.mox.ReplayAll()

        processor = TestableSnapshotProcessor(self.nova_client)
        processor.init_processor(self.worker)

        processor.process_job(self.job)

        self.mox.VerifyAll()

开发者ID:hmakkapati，项目名称:qonos，代码行数:22，代码来源:test_snapshot.py

示例15: test_process_job_new_image_when_retrying_with_failed_image

    def test_process_job_new_image_when_retrying_with_failed_image(self):
        timeutils.set_time_override()
        self.job['metadata']['image_id'] = IMAGE_ID
        self.job['status'] = 'ERROR'

        #make an error checking the previous image
        self.nova_client.images.get(IMAGE_ID).AndRaise(
            Exception())

        self.nova_client.servers.get(mox.IsA(str)).AndReturn(MockServer())
        self.nova_client.servers.create_image(mox.IsA(str),
            mox.IsA(str), self.snapshot_meta).AndReturn(IMAGE_ID)
        self.nova_client.images.get(IMAGE_ID).AndReturn(
            MockImageStatus('ACTIVE'))
        mock_retention = MockRetention()
        self.nova_client.rax_scheduled_images_python_novaclient_ext.\
            get(mox.IsA(str)).AndReturn(mock_retention)
        self._simple_prepare_worker_mock()

        self.mox.StubOutWithMock(utils, 'generate_notification')
        utils.generate_notification(None, 'qonos.job.retry', mox.IsA(dict),
                                    mox.IsA(str))
        utils.generate_notification(None, 'qonos.job.run.end', mox.IsA(dict),
                                    mox.IsA(str))
        self.worker.get_qonos_client().AndReturn(self.qonos_client)
        self.qonos_client.delete_schedule(mox.IsA(str))
        self.worker.update_job(fakes.JOB_ID, 'DONE',
                error_message=None, timeout=None).AndReturn(None)

        self.mox.ReplayAll()

        processor = TestableSnapshotProcessor(self.nova_client)
        processor.init_processor(self.worker)

        processor.process_job(self.job)

        self.mox.VerifyAll()

开发者ID:amalaba，项目名称:qonos，代码行数:37，代码来源:test_snapshot.py

示例16: test_process_job_should_update_status_timestamp_no_retries

    def test_process_job_should_update_status_timestamp_no_retries(self):
        base_time = timeutils.utcnow()
        time_seq = [
            base_time,
            base_time,
            base_time + datetime.timedelta(minutes=5, seconds=5),
            base_time + datetime.timedelta(minutes=60, seconds=5),
            base_time + datetime.timedelta(minutes=120, seconds=5),
            base_time + datetime.timedelta(minutes=180, seconds=5),
            base_time + datetime.timedelta(minutes=240, seconds=5),
        ]
        timeutils.set_time_override_seq(time_seq)

        job = copy.deepcopy(self.job)
        job["timeout"] = base_time + datetime.timedelta(minutes=60)

        self.nova_client.servers.get(mox.IsA(str)).AndReturn(MockServer())
        self.nova_client.servers.create_image(mox.IsA(str), mox.IsA(str), self.snapshot_meta).AndReturn(IMAGE_ID)
        self.nova_client.images.get(IMAGE_ID).AndReturn(MockImageStatus("QUEUED"))
        self.nova_client.images.get(IMAGE_ID).MultipleTimes().AndReturn(MockImageStatus("SAVING"))

        self._init_worker_mock()
        self.worker.update_job(fakes.JOB_ID, "PROCESSING", timeout=mox.IsA(datetime.datetime), error_message=None)
        self.worker.update_job(fakes.JOB_ID, "PROCESSING", timeout=mox.IsA(datetime.datetime), error_message=None)
        self.worker.update_job(fakes.JOB_ID, "PROCESSING", timeout=mox.IsA(datetime.datetime), error_message=None)
        self.worker.update_job(fakes.JOB_ID, "TIMED_OUT", timeout=None, error_message=None)
        self.mox.StubOutWithMock(utils, "generate_notification")
        utils.generate_notification(None, "qonos.job.run.start", mox.IsA(dict), mox.IsA(str))
        self.mox.ReplayAll()

        processor = TestableSnapshotProcessor(self.nova_client)
        processor.init_processor(self.worker)

        processor.process_job(job)

        self.mox.VerifyAll()

开发者ID:hmakkapati，项目名称:qonos，代码行数:36，代码来源:test_snapshot.py

示例17: test_process_job_should_update_status_only

    def test_process_job_should_update_status_only(self):
        base_time = timeutils.utcnow()
        time_seq = [
            base_time,
            base_time,
            base_time + datetime.timedelta(seconds=305),
            base_time + datetime.timedelta(seconds=605),
            base_time + datetime.timedelta(seconds=905),
        ]
        timeutils.set_time_override_seq(time_seq)

        job = copy.deepcopy(self.job)
        job["timeout"] = base_time + datetime.timedelta(minutes=60)

        self.nova_client.servers.get(mox.IsA(str)).AndReturn(MockServer())
        self.nova_client.servers.create_image(mox.IsA(str), mox.IsA(str), self.snapshot_meta).AndReturn(IMAGE_ID)
        self.nova_client.images.get(IMAGE_ID).AndReturn(MockImageStatus("QUEUED"))
        self.nova_client.images.get(IMAGE_ID).AndReturn(MockImageStatus("SAVING"))
        self.nova_client.images.get(IMAGE_ID).AndReturn(MockImageStatus("SAVING"))
        self.nova_client.images.get(IMAGE_ID).AndReturn(MockImageStatus("ACTIVE"))
        mock_retention = MockRetention()
        self.nova_client.rax_scheduled_images_python_novaclient_ext.get(mox.IsA(str)).AndReturn(mock_retention)
        self._simple_prepare_worker_mock(2)
        self.mox.StubOutWithMock(utils, "generate_notification")
        utils.generate_notification(None, "qonos.job.run.start", mox.IsA(dict), mox.IsA(str))
        utils.generate_notification(None, "qonos.job.run.end", mox.IsA(dict), mox.IsA(str))
        self.worker.get_qonos_client().AndReturn(self.qonos_client)
        self.qonos_client.delete_schedule(mox.IsA(str))
        self.mox.ReplayAll()

        processor = TestableSnapshotProcessor(self.nova_client)
        processor.init_processor(self.worker)

        processor.process_job(job)

        self.mox.VerifyAll()

开发者ID:hmakkapati，项目名称:qonos，代码行数:36，代码来源:test_snapshot.py

示例18: test_process_job_should_cancel_if_no_instance_id

    def test_process_job_should_cancel_if_no_instance_id(self):
        self._init_qonos_client(schedule=dict())
        self.mox.StubOutWithMock(utils, 'generate_notification')
        utils.generate_notification(None, 'qonos.job.run.start', mox.IsA(dict),
                                    mox.IsA(str))
        self.worker.update_job(fakes.JOB_ID, 'PROCESSING',
                               timeout=mox.IsA(datetime.datetime),
                               error_message=None).\
            AndReturn({'status': 'PROCESSING', 'timeout': 'new_timeout'})
        self.worker.update_job(fakes.JOB_ID, 'CANCELLED', timeout=None,
                               error_message=mox.IsA(str)).\
            AndReturn({'status': 'CANCELLED', 'timeout': self.job['timeout']})
        expected_processing_payload = \
            {'job': {'status': 'PROCESSING',
                     'hard_timeout': self.job['hard_timeout'],
                     'created_at': self.job['created_at'],
                     'modified_at': self.job['modified_at'],
                     'retry_count': 1,
                     'schedule_id': '33333333-3333-3333-3333-33333333',
                     'worker_id': '11111111-1111-1111-1111-11111111',
                     'timeout': 'new_timeout',
                     'action': 'snapshot',
                     'id': '22222222-2222-2222-2222-22222222',
                     'tenant': '44444444-4444-4444-4444-44444444',
                     'metadata': {}}}

        expected_cancelled_payload = \
            {'job': {'status': 'CANCELLED',
                     'hard_timeout': self.job['hard_timeout'],
                     'created_at': self.job['created_at'],
                     'modified_at': self.job['modified_at'],
                     'retry_count': 1,
                     'schedule_id': '33333333-3333-3333-3333-33333333',
                     'worker_id': '11111111-1111-1111-1111-11111111',
                     'timeout': self.job['timeout'],
                     'action': 'snapshot',
                     'id': '22222222-2222-2222-2222-22222222',
                     'tenant': '44444444-4444-4444-4444-44444444',
                     'metadata': {}}}
        utils.generate_notification(None, 'qonos.job.update',
                                    expected_processing_payload,
                                    mox.IsA(str))
        utils.generate_notification(None, 'qonos.job.failed',
                                    expected_cancelled_payload,
                                    mox.IsA(str))

        self.mox.ReplayAll()

        processor = TestableSnapshotProcessor(self.nova_client)
        processor.init_processor(self.worker)
        del self.job['metadata']['instance_id']
        processor.process_job(self.job)

        self.mox.VerifyAll()

开发者ID:clefelhocz，项目名称:qonos，代码行数:54，代码来源:test_snapshot.py

示例19: test_process_job_should_succeed_immediately

    def test_process_job_should_succeed_immediately(self):
        timeutils.set_time_override()
        self.nova_client.servers.get(mox.IsA(str)).AndReturn(MockServer())
        self.nova_client.servers.create_image(mox.IsA(str),
            mox.IsA(str), self.snapshot_meta).AndReturn(IMAGE_ID)
        self.nova_client.images.get(IMAGE_ID).AndReturn(
            MockImageStatus('ACTIVE'))
        mock_retention = MockRetention()
        self.nova_client.rax_scheduled_images_python_novaclient_ext.\
            get(mox.IsA(str)).AndReturn(mock_retention)
        self._init_worker_mock(False)

        expected_processing_payload = \
            {'job': {'status': 'PROCESSING',
                     'hard_timeout': self.job['hard_timeout'],
                     'created_at': self.job['created_at'],
                     'modified_at': self.job['modified_at'],
                     'retry_count': 1,
                     'schedule_id': '33333333-3333-3333-3333-33333333',
                     'worker_id': '11111111-1111-1111-1111-11111111',
                     'timeout': self.job['timeout'],
                     'action': 'snapshot',
                     'id': '22222222-2222-2222-2222-22222222',
                     'tenant': '44444444-4444-4444-4444-44444444',
                     'metadata': {'instance_id':
                                  '55555555-5555-5555-5555-55555555'}}}

        self.mox.StubOutWithMock(utils, 'generate_notification')
        utils.generate_notification(None, 'qonos.job.run.start', mox.IsA(dict),
                                    mox.IsA(str))
        utils.generate_notification(None, 'qonos.job.update',
                                    expected_processing_payload,
                                    mox.IsA(str))

        utils.generate_notification(None, 'qonos.job.run.end', mox.IsA(dict),
                                    mox.IsA(str))

        self.worker.get_qonos_client().AndReturn(self.qonos_client)
        self.qonos_client.delete_schedule(mox.IsA(str))
        processor = TestableSnapshotProcessor(self.nova_client)
        self.worker.update_job(fakes.JOB_ID, 'DONE', timeout=None,
                               error_message=None).\
            AndReturn({'status': 'CANCELLED', 'timeout': self.job['timeout']})
        self.mox.ReplayAll()

        processor.init_processor(self.worker)
        processor.process_job(self.job)

        self.mox.VerifyAll()

开发者ID:frielp，项目名称:qonos，代码行数:49，代码来源:test_snapshot.py

示例20: send_notification

 def send_notification(self, event_type, payload, level='INFO'):
     utils.generate_notification(None, event_type, payload, level)

开发者ID:hmakkapati，项目名称:qonos，代码行数:2，代码来源:worker.py

注：本文中的qonos.common.utils.generate_notification函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。

鲜花

握手

雷人

路过

鸡蛋

该文章已有0人参与评论

请发表评论

全部评论

专题导读

More+

10-27 六六分期app的软件客服如何联系？(六六分期

11-06 可心卡盟:win10系统火狐flash插件崩溃怎么

11-06 亲亲特价:怎么删除回收站图标

11-06 济南大学虚拟社区:鲁大师节能降温的具体办

11-06 xlueops.exe:无线网络安装向导

11-06 女斗合众国:win7系统cf与主机连接不稳定怎

11-06 0xc000022-[cf烟雾头]cf怎么调烟雾头

11-06 qizideyouhuo:应用程序无法正常启动0xc0000

11-06 ipz-185:win7系统vcf文件怎么打开

11-06 傻哥蹦迪:win10系统s4怎么打开usb调试

11-06 八神浩树gtaste:回收站清空了怎么恢复

11-06 妖尾之黑色守护:win10系统电脑没有1440x900

11-06 校园至尊魔王小说:win7系统浏览网页时字体

11-06 女斗合众国:win10系统访问共享文件夹提示请

11-06 tokyo hot n0654:恢复win7系统默认字体一招

11-06 雨酷仙境:设置win7系统转移临时文件夹腾出

11-06 阿穆纳伊之杖:win7系统开始菜单在右边还原

11-06 tunespotting:win10系统火狐flash插件总是

11-06 甘尔葛分析师：计谋网站seo关键词暴涨有什

11-06 蔡贵霖: 计谋网站seo关键词暴涨有什么秘密

11-06 博益网首页:ao3网页版进入不了解决方法

11-06 漏斗子专栏: 网站数据分析小白易懂精华篇

11-06 见证双虹怎么做:win7系统开启telnet命令的

11-06 颾狐蝶蜋:系统资源不足无法完成请求的服务

11-06 国光中学校歌:提交网站到alexa查询详细步骤

11-06 西安有情天:静态网页和动态网页的区别

11-06 红木雅尚斋:外部链接构造对网站的好处

11-06 前官礼遇：防止域名劫持–增强域安全性的10

11-06 密传二转答案: 中文分词算法有哪些

11-06 金泉家园邮编:百度快照劫持的表现及应对方

Python utils.serialize_datetimes函数代码示例发布时间：2022-05-26

Python timeutils.utcnow函数代码示例发布时间：2022-05-26

Python util.grid_equal函数代码示例

1 Python 入门教程

Python入门教程 Python 是一种解释型、面向对象、动态数据类型的高级程序设计语言。 P

阅读：13789|2022-01-22

2 Python wikiutil.getFrontPage函数代码示例

Python wikiutil.getFrontPage函数代码示例

阅读：10177|2022-05-24

3 Python 简介

Python 简介 Python 是一个高层次的结合了解释性、编译性、互动性和面向对象的脚本

阅读：4076|2022-01-22

4 Python tests.group函数代码示例

Python tests.group函数代码示例

阅读：4039|2022-05-27

5 Python util.check_if_user_has_permission

Python util.check_if_user_has_permission函数代码示例

阅读：3832|2022-05-27

6 Python 操练实例98

Python 练习实例98 Python 100例题目：从键盘输入一个字符串，将小写字母全部转换成大

阅读：3508|2022-01-22

7 Python 环境搭建

Python 环境搭建本章节我们将向大家介绍如何在本地搭建 Python 开发环境。 Py

阅读：3029|2022-01-22

8 Python output.darkgreen函数代码示例

Python output.darkgreen函数代码示例

阅读：2646|2022-05-25

9 Python 基础语法

Python 基础语法 Python 语言与 Perl，C 和 Java 等语言有许多相似之处。但是，也

阅读：2636|2022-01-22

10 Python 中文编码

Python 中文编码前面章节中我们已经学会了如何用 Python 输出 Hello, World!，英文没

阅读：2293|2022-01-22

客服电话

电子邮件

Python utils.generate_notification函数代码示例

示例1: test_generate_notification

示例2: test_doesnt_delete_images_on_retention_error

示例3: create

示例4: test_process_job_should_not_create_on_job_error_image_ok

示例5: test_deletes_images_more_than_retention

示例6: test_process_job_should_cancel_if_schedule_deleted

示例7: test_doesnt_delete_images_from_another_instance

示例8: test_process_job_should_succeed_after_multiple_tries

示例9: _do_test_process_job_should_update_image_error

示例10: test_process_job_should_cancel_if_schedule_deleted

示例11: test_process_job_should_update_status_and_timestamp

示例12: test_process_job_should_cancel_if_no_instance_id

示例13: test_process_job_should_cancel_if_instance_not_found

示例14: test_process_job_should_succeed_immediately

示例15: test_process_job_new_image_when_retrying_with_failed_image

示例16: test_process_job_should_update_status_timestamp_no_retries

示例17: test_process_job_should_update_status_only

示例18: test_process_job_should_cancel_if_no_instance_id

示例19: test_process_job_should_succeed_immediately

示例20: send_notification

请发表评论

全部评论

上一篇：

下一篇：

Python util.grid_equal函数代码示例

Python util.get_worker_name函数代码示例

Python util.get_webmention_target函数代

Python util.get_uuid函数代码示例

Python util.get_type_by_name函数代码示例

Python util.grid_equal函数代码示例

Python util.get_worker_name函数代码示例

Python util.get_webmention_target函数代

Python util.get_uuid函数代码示例

Python util.get_type_by_name函数代码示例

Python util.get_stdout函数代码示例

关于我们

产品与服务

解决方案

139-2527-9053