Coverage for postrfp / web / hooks / mail_delivered.py: 100%

1import logging 

2from datetime import datetime 

3 

4from postrfp.model import EmailNotification 

5from postrfp.shared import utils 

6from .webapp import WebhookApp, NotificationHook, PostmarkVars 

7 

8log = logging.getLogger(__name__) 

9 

10 

11def postmark_isodate(datestring: str): 

12 """ 

13 The microseconds portion of postmark api's datestring seems to 

14 have 7, not 6, digits 

15 2014-08-01T13:28:10.2735391-04:00 

16 should be 

17 2014-08-01T13:28:10.273539-04:00 

18 """ 

19 if len(datestring) == 33: 

20 datestring = datestring[0:26] + datestring[27:] 

21 return datetime.fromisoformat(datestring) 

22 

23 

24class DeliveredHook(NotificationHook): 

25 new_status = EmailNotification.Status.delivered 

26 

27 def update_notification_record(self, data: PostmarkVars, en: EmailNotification): 

28 try: 

29 new_date = postmark_isodate(data.api_dict["DeliveredAt"]) 

30 except ValueError: 

31 new_date = utils.utcnow() 

32 

33 en.delivered_date = new_date 

34 

35 

36class BounceHook(NotificationHook): 

37 new_status = EmailNotification.Status.bounced 

38 

39 def update_notification_record(self, data: PostmarkVars, en: EmailNotification): 

40 bounce_type = data.api_dict.get("Name", "Bounce") 

41 bounce_description = data.api_dict.get("Description", "Details not provided") 

42 msg = f"{bounce_type}: {bounce_description} \nAt: {utils.utcnow()}" 

43 if en.error is None: 

44 en.error = msg 

45 else: 

46 en.error = en.error + "\n\n" + msg 

47 

48 

49WebhookApp.register_route("/delivered", DeliveredHook()) 

50WebhookApp.register_route("/bounced", BounceHook())