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

1# from flask import url_for 

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 consumes(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 

30 

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

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

33 return 

34 

35 svc = DOAJ.notificationsService() 

36 

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

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

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

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

41 

42 notification = models.Notification() 

43 notification.who = acc.id 

44 notification.created_by = cls.ID 

45 notification.classification = constants.NOTIFICATION_CLASSIFICATION_FINISHED 

46 notification.long = svc.long_notification(cls.ID).format(job_id=job.id, action=job.action, status=job.status) 

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

48 notification.action = url 

49 

50 svc.notify(notification)