test(backups): test that the job has run

This commit is contained in:
Houkime 2023-05-08 12:43:11 +00:00 committed by Inex Code
parent 2a87eb80f9
commit ebff2b308a
2 changed files with 27 additions and 2 deletions

View file

@ -133,9 +133,26 @@ class Jobs:
key = _redis_log_key_from_uuid(job.uid) key = _redis_log_key_from_uuid(job.uid)
if redis.exists(key): if redis.exists(key):
assert redis.type(key) == "list" assert redis.type(key) == "list"
redis.lpush(key, str(status)) redis.lpush(key, status.value)
redis.expire(key, 10) redis.expire(key, 10)
@staticmethod
def status_updates(job: Job) -> typing.List[JobStatus]:
result = []
redis = RedisPool().get_connection()
key = _redis_log_key_from_uuid(job.uid)
if not redis.exists(key):
return []
status_strings = redis.lrange(key, 0, -1)
for status in status_strings:
try:
result.append(JobStatus[status])
except KeyError as e:
raise ValueError("impossible job status: " + status) from e
return result
@staticmethod @staticmethod
def update( def update(
job: Job, job: Job,

View file

@ -227,6 +227,12 @@ def assert_job_finished(job_type, count):
assert finished_types.count(job_type) == count assert finished_types.count(job_type) == count
def assert_job_has_run(job_type):
finished_jobs = [job for job in Jobs.get_jobs() if job.status is JobStatus.FINISHED]
job = [job for job in finished_jobs if job.type_id == job_type][0]
assert JobStatus.RUNNING in Jobs.status_updates(job)
def test_backup_service_task(backups, dummy_service): def test_backup_service_task(backups, dummy_service):
handle = start_backup(dummy_service) handle = start_backup(dummy_service)
handle(blocking=True) handle(blocking=True)
@ -235,7 +241,9 @@ def test_backup_service_task(backups, dummy_service):
assert len(snaps) == 1 assert len(snaps) == 1
id = dummy_service.get_id() id = dummy_service.get_id()
assert_job_finished(f"services.{id}.backup", count=1) job_type_id = f"services.{id}.backup"
assert_job_finished(job_type_id, count=1)
assert_job_has_run(job_type_id)
def test_restore_snapshot_task(backups, dummy_service): def test_restore_snapshot_task(backups, dummy_service):