MrMeeb
42696112cb
All checks were successful
Build-Release-Image / Build-Image (linux/arm64) (push) Successful in 3m40s
Build-Release-Image / Build-Image (linux/amd64) (push) Successful in 4m7s
Build-Release-Image / Merge-Images (push) Successful in 22s
Build-Release-Image / Create-Release (push) Successful in 15s
Build-Release-Image / Notify (push) Successful in 17s
73 lines
1.9 KiB
Python
73 lines
1.9 KiB
Python
import arrow
|
|
|
|
from app import config
|
|
from app.models import Job, JobState
|
|
from tasks.cleanup_old_jobs import cleanup_old_jobs
|
|
|
|
|
|
def test_cleanup_old_jobs():
|
|
Job.filter().delete()
|
|
now = arrow.now()
|
|
delete_ids = [
|
|
Job.create(
|
|
updated_at=now.shift(minutes=-1),
|
|
state=JobState.done.value,
|
|
name="",
|
|
payload="",
|
|
flush=True,
|
|
).id,
|
|
Job.create(
|
|
updated_at=now.shift(minutes=-1),
|
|
state=JobState.error.value,
|
|
name="",
|
|
payload="",
|
|
flush=True,
|
|
).id,
|
|
Job.create(
|
|
updated_at=now.shift(minutes=-1),
|
|
state=JobState.taken.value,
|
|
attempts=config.JOB_MAX_ATTEMPTS,
|
|
name="",
|
|
payload="",
|
|
flush=True,
|
|
).id,
|
|
]
|
|
|
|
keep_ids = [
|
|
Job.create(
|
|
updated_at=now.shift(minutes=+1),
|
|
state=JobState.done.value,
|
|
name="",
|
|
payload="",
|
|
flush=True,
|
|
).id,
|
|
Job.create(
|
|
updated_at=now.shift(minutes=+1),
|
|
state=JobState.error.value,
|
|
name="",
|
|
payload="",
|
|
flush=True,
|
|
).id,
|
|
Job.create(
|
|
updated_at=now.shift(minutes=+1),
|
|
state=JobState.taken.value,
|
|
attempts=config.JOB_MAX_ATTEMPTS,
|
|
name="",
|
|
payload="",
|
|
flush=True,
|
|
).id,
|
|
Job.create(
|
|
updated_at=now.shift(minutes=-1),
|
|
state=JobState.taken.value,
|
|
attempts=config.JOB_MAX_ATTEMPTS - 1,
|
|
name="",
|
|
payload="",
|
|
flush=True,
|
|
).id,
|
|
]
|
|
cleanup_old_jobs(now)
|
|
for delete_id in delete_ids:
|
|
assert Job.get(id=delete_id) is None
|
|
for keep_id in keep_ids:
|
|
assert Job.get(id=keep_id) is not None
|