tests: check clean_afterjobs method
This commit is contained in:
parent
83f7941808
commit
52feb63060
|
@ -1,6 +1,7 @@
|
|||
import json
|
||||
import re
|
||||
import sys
|
||||
import time
|
||||
import pickle
|
||||
import shutil
|
||||
import tempfile
|
||||
|
@ -20,6 +21,7 @@ from django.utils.encoding import force_text
|
|||
from django.utils.six import BytesIO, StringIO
|
||||
from quixote import cleanup
|
||||
from wcs.qommon import get_publisher_class
|
||||
from wcs.qommon.afterjobs import AfterJob
|
||||
from wcs.qommon.http_request import HTTPRequest
|
||||
from wcs.qommon.cron import CronJob
|
||||
|
||||
|
@ -264,3 +266,28 @@ def test_cron_command():
|
|||
get_publisher_class().cronjobs = []
|
||||
call_command('cron', job_name='job2')
|
||||
assert jobs == ['job2']
|
||||
|
||||
|
||||
def test_clean_afterjobs():
|
||||
pub = create_temporary_pub()
|
||||
|
||||
job = AfterJob(id='a')
|
||||
job.status = 'completed'
|
||||
job.creation_time = time.time() - 3 * 3600
|
||||
job.completion_time = time.time() - 3 * 3600
|
||||
job.store()
|
||||
|
||||
job = AfterJob(id='b')
|
||||
job.status = 'completed'
|
||||
job.creation_time = time.time()
|
||||
job.completion_time = time.time()
|
||||
job.store()
|
||||
|
||||
job = AfterJob(id='c')
|
||||
job.status = 'running'
|
||||
job.creation_time = time.time() - 3 * 86400
|
||||
job.store()
|
||||
|
||||
pub.clean_afterjobs()
|
||||
assert AfterJob.count() == 1
|
||||
assert AfterJob.select()[0].id == 'b'
|
||||
|
|
Loading…
Reference in New Issue