Coverage for portality / events / consumers / bg_job_finished_notify.py: 100%

1# ~~BGJobFinishedNotifiy:Consumer~~ 

2from portality.util import url_for 

3 

4from portality.events.consumer import EventConsumer 

5from portality import constants 

6from portality import models 

7from portality.lib import edges 

8from portality.bll import DOAJ 

9from portality.bll import exceptions 

10 

11 

12class BGJobFinishedNotify(EventConsumer): 

13 ID = "bg:job_finished:notify" 

14 

15 @classmethod 

16 def should_consume(cls, event): 

17 return event.id == constants.BACKGROUND_JOB_FINISHED and \ 

18 event.context.get("job") is not None 

19 

20 @classmethod 

21 def consume(cls, event): 

22 source = event.context.get("job") 

23 try: 

24 job = models.BackgroundJob(**source) 

25 except Exception as e: 

26 raise exceptions.NoSuchObjectException("Unable to construct a BackgroundJob object from the data supplied {x}".format(x=e)) 

27 

28 if job.user is None: 

29 return None 

30 

31 acc = models.Account.pull(job.user) 

32 if acc is None or not acc.has_role("admin"): 

33 return None 

34 

35 # ~~-> Notifications:Service ~~ 

36 svc = DOAJ.notificationsService() 

37 

38 # don't make the escaped query, as url_for is also going to escape it, and it will wind up double-escaped! 

39 string_id_query = edges.make_query_json(query_string=job.id) 

40 # note we're using the doaj url_for wrapper, not the flask url_for directly, due to the request context hack required 

41 url = url_for("admin.background_jobs_search", source=string_id_query) 

42 

43 notification = models.Notification() 

44 notification.who = acc.id 

45 notification.created_by = cls.ID 

46 notification.classification = constants.NOTIFICATION_CLASSIFICATION_FINISHED 

47 notification.long = svc.long_notification(cls.ID).format( 

48 job_id=job.id, 

49 action=job.action, 

50 status=job.status) 

51 notification.short = svc.short_notification(cls.ID) 

52 notification.action = url 

53 

54 svc.notify(notification) 

55 return notification