本文整理汇总了Python中qonos.common.timeutils.set_time_override函数的典型用法代码示例。如果您正苦于以下问题:Python set_time_override函数的具体用法?Python set_time_override怎么用?Python set_time_override使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了set_time_override函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: 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.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'))
self.nova_client.servers.get(mox.IsA(str)).AndReturn(
MockServer())
self.worker.update_job(fakes.JOB_ID, 'PROCESSING',
timeout=mox.IsA(datetime.datetime),
error_message=None)
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(fakes.JOB['job'])
self.mox.VerifyAll()
开发者ID:isethi,项目名称:qonos,代码行数:27,代码来源:test_snapshot.py
示例2: test_doesnt_delete_images_from_another_instance
def test_doesnt_delete_images_from_another_instance(self):
timeutils.set_time_override()
instance_id = fakes.JOB['job']['metadata']['instance_id']
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_server = MockServer(instance_id=instance_id, retention=3)
self.nova_client.servers.get(mox.IsA(str)).AndReturn(mock_server)
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.worker.update_job(fakes.JOB_ID, 'PROCESSING',
timeout=mox.IsA(datetime.datetime),
error_message=None)
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(fakes.JOB['job'])
self.mox.VerifyAll()
开发者ID:isethi,项目名称:qonos,代码行数:30,代码来源:test_snapshot.py
示例3: test_schedule_update_remove_metadata
def test_schedule_update_remove_metadata(self):
fixture = {
'id': str(uuid.uuid4()),
'tenant': str(uuid.uuid4()),
'action': 'snapshot',
'minute': 30,
'hour': 2,
'schedule_metadata': [
{
'key': 'instance_id',
'value': 'my_instance',
},
],
}
schedule = self.db_api.schedule_create(fixture)
fixture = {
'schedule_metadata': [],
}
timeutils.set_time_override()
timeutils.advance_time_seconds(2)
updated_schedule = self.db_api.schedule_update(schedule['id'], fixture)
timeutils.clear_time_override()
self.assertTrue(uuidutils.is_uuid_like(schedule['id']))
self.assertEqual(updated_schedule['tenant'], schedule['tenant'])
self.assertEqual(updated_schedule['action'], schedule['action'])
self.assertEqual(updated_schedule['minute'], schedule['minute'])
self.assertEqual(updated_schedule['hour'], schedule['hour'])
metadata = updated_schedule['schedule_metadata']
self.assertEqual(len(metadata), 0)
# updated child metadata collection doesn't update the parent schedule
self.assertEqual(updated_schedule['updated_at'],
schedule['updated_at'])
开发者ID:pperezrubio,项目名称:qonos,代码行数:34,代码来源:base.py
示例4: 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
示例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_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
示例7: 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.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:ameade,项目名称:qonos,代码行数:26,代码来源:test_snapshot.py
示例8: 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:ameade,项目名称:qonos,代码行数:27,代码来源:test_snapshot.py
示例9: test_process_job_should_continue_when_image_id_present
def test_process_job_should_continue_when_image_id_present(self):
timeutils.set_time_override()
self.job['metadata']['image_id'] = IMAGE_ID
self.job['status'] = 'PROCESSING'
# Note NO call to create_image is expected
self.nova_client.images.get(IMAGE_ID).AndReturn(
MockImageStatus('ACTIVE'))
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.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,代码行数:27,代码来源:test_snapshot.py
示例10: test_generate_image_name_weekly
def test_generate_image_name_weekly(self):
schedule = {'day_of_week': 1}
timeutils.set_time_override(datetime.datetime(2013, 3, 22, 22, 39, 27))
timestamp = '1363991967'
processor = TestableSnapshotProcessor(self.nova_client)
image_name = processor.generate_image_name(schedule, 'test')
self.assertEqual(image_name, 'Weekly-test-' + timestamp)
开发者ID:clefelhocz,项目名称:qonos,代码行数:7,代码来源:test_snapshot.py
示例11: 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
示例12: test_process_job_should_create_on_job_error_image_failed
def test_process_job_should_create_on_job_error_image_failed(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('KILLED'))
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.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,代码行数:31,代码来源:test_snapshot.py
示例13: test_process_job_should_create_on_job_error_image_failed
def test_process_job_should_create_on_job_error_image_failed(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("KILLED"))
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.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,代码行数:25,代码来源:test_snapshot.py
示例14: 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.update', mox.IsA(dict),
mox.IsA(str)).MultipleTimes()
utils.generate_notification(None, 'qonos.job.run.end', mox.IsA(dict),
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:clefelhocz,项目名称:qonos,代码行数:34,代码来源:test_snapshot.py
示例15: test_process_job_should_not_continue_when_in_bad_status
def test_process_job_should_not_continue_when_in_bad_status(self):
timeutils.set_time_override()
self.job['metadata']['image_id'] = IMAGE_ID
self.job['status'] = 'ERROR'
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.ReplayAll()
processor = TestableSnapshotProcessor(self.nova_client)
processor.init_processor(self.worker)
processor.process_job(self.job)
self.mox.VerifyAll()
开发者ID:pperezrubio,项目名称:qonos,代码行数:25,代码来源:test_snapshot.py
示例16: test_generate_image_name
def test_generate_image_name(self):
job = self.job_fixture("INSTANCE_ID")
timeutils.set_time_override(datetime.datetime(2013, 3, 22, 22, 39, 27))
timestamp = '1363991967'
with TestableSnapshotProcessor(job, None, []) as processor:
image_name = processor.generate_image_name("test")
self.assertEqual(image_name, 'Daily-test-' + timestamp)
开发者ID:broble,项目名称:qonos,代码行数:8,代码来源:test_snapshot_processor.py
示例17: test_jobs_cleanup_hard_timed_out
def test_jobs_cleanup_hard_timed_out(self):
jobs = self.db_api.job_get_all()
self.assertEqual(len(jobs), 2)
timeutils.set_time_override()
timeutils.advance_time_delta(datetime.timedelta(hours=4, minutes=1))
self.db_api._jobs_cleanup_hard_timed_out()
timeutils.clear_time_override()
jobs = self.db_api.job_get_all()
self.assertEqual(len(jobs), 0)
开发者ID:pperezrubio,项目名称:qonos,代码行数:9,代码来源:base.py
示例18: test_generate_image_name_long_server_name
def test_generate_image_name_long_server_name(self):
timeutils.set_time_override(datetime.datetime(2013, 3, 22, 22, 39, 27))
timestamp = "1363991967"
processor = TestableSnapshotProcessor(self.nova_client)
fake_server_name = "a" * 255
expected_server_name = "a" * (255 - len(timestamp) - len("Daily--"))
image_name = processor.generate_image_name(fake_server_name)
expected_image_name = "Daily-" + expected_server_name + "-" + timestamp
self.assertEqual(image_name, expected_image_name)
开发者ID:hmakkapati,项目名称:qonos,代码行数:9,代码来源:test_snapshot.py
示例19: test_get_schedules
def test_get_schedules(self):
timeutils.set_time_override()
start_time = timeutils.isotime()
timeutils.advance_time_seconds(30)
end_time = timeutils.isotime()
filter_args = {'next_run_after': start_time,
'next_run_before': end_time}
self.client.list_schedules(filter_args=filter_args).AndReturn([])
self.mox.ReplayAll()
self.scheduler.get_schedules(start_time, end_time)
self.mox.VerifyAll()
开发者ID:coreywright,项目名称:qonos,代码行数:12,代码来源:test_scheduler.py
示例20: test_process_job_should_exponentially_increates_timeout
def test_process_job_should_exponentially_increates_timeout(self):
status = MockImageStatus('ERROR')
job = copy.deepcopy(self.job)
self._do_test_process_job_should_update_image_error(status, job=job)
self._reset_mocks()
new_now = timeutils.utcnow() + datetime.timedelta(minutes=120)
timeutils.clear_time_override()
timeutils.set_time_override(new_now)
job['status'] = 'ERROR'
job['retry_count'] = 2
self._do_test_process_job_should_update_image_error(status,
include_create=False, include_queued=False, is_retry=True, job=job)
开发者ID:amalaba,项目名称:qonos,代码行数:12,代码来源:test_snapshot.py
注:本文中的qonos.common.timeutils.set_time_override函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论