-
Notifications
You must be signed in to change notification settings - Fork 16
/
background.py
74 lines (55 loc) · 1.94 KB
/
background.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
from copy import deepcopy
from portality import models
from portality.background import BackgroundTask
from portality.lib import dates
from portality.models import BackgroundJob
class BackgroundFixtureFactory(object):
@classmethod
def example(cls):
return deepcopy(BACKGROUND_JOB)
@classmethod
def get_task(cls, user=None, run_fail=False, cleanup_fail=False):
job = models.BackgroundJob()
if user is not None:
job.user = user
return MockBackgroundTask(job, run_fail=run_fail, cleanup_fail=cleanup_fail)
class MockBackgroundTask(BackgroundTask):
def __init__(self, background_job, run_fail=False, cleanup_fail=False):
super(MockBackgroundTask, self).__init__(background_job)
self.run_fail = run_fail
self.cleanup_fail = cleanup_fail
def run(self):
if self.run_fail:
raise Exception("Mock Run Failed")
def cleanup(self):
if self.cleanup_fail:
raise Exception("Mock Cleanup Failed")
BACKGROUND_JOB = {
"id": "123456789",
"created_date": "2001-01-01T00:00:00Z",
"last_updated": "2001-01-02T00:00:00Z",
"status": "queued",
"user": "testuser",
"queue_id": "abcdef",
"action": "status_change",
"params": {},
"reference": {},
"audit": [
{"message": "created job", "timestamp": "2001-01-01T00:00:00Z"}
]
}
def save_mock_bgjob(action=None, status=None, created_before_sec=0, is_save=True,
queue_id=None):
bgjob = BackgroundJob()
if action:
from portality.tasks.journal_csv import JournalCSVBackgroundTask
bgjob.action = JournalCSVBackgroundTask.__action__
if status:
bgjob._set_with_struct("status", status)
if created_before_sec != 0:
bgjob.set_created(dates.format(dates.before_now(created_before_sec)))
if queue_id:
bgjob.queue_id = queue_id
if is_save:
bgjob.save(blocking=True)
return bgjob