Coverage for rfpy/web/hooks/webapp.py: 98%

1import logging 

2from typing import Dict, Callable 

3 

4from webob import Request, Response 

5from webob.exc import HTTPNotAcceptable, HTTPMethodNotAllowed 

6from sqlalchemy.orm.exc import NoResultFound 

7 

8from rfpy.model import EmailNotification 

9from rfpy.web.base import WSGIApp 

10 

11 

12log = logging.getLogger(__name__) 

13 

14 

15class WebhookApp(WSGIApp): 

16 routes: Dict[str, Callable] = {} 

17 

18 def validate_user(self, request): 

19 pass 

20 

21 def build_sux(self): 

22 pass 

23 

24 @classmethod 

25 def register_route(cls, path: str, func: Callable): 

26 cls.routes[path.lstrip("/")] = func 

27 

28 

29class PostmarkVars: 

30 def __init__(self, data: dict): 

31 self.message_id = data["MessageID"] 

32 self.recipient = data["Recipient"] 

33 self.tag = data.get("Tag", None) 

34 self.api_dict = data 

35 

36 

37class NotificationHook: 

38 """Base class for handling Postmark webhook POST requests""" 

39 

40 new_status: EmailNotification.Status 

41 

42 def __call__(self, request: Request): 

43 try: 

44 session = request.session 

45 data = self.extract(request) 

46 try: 

47 em = self.lookup(session, data.message_id) 

48 except HTTPNotAcceptable as hna: 

49 return hna 

50 self.check_emails_match(data, em) 

51 self.update_notification_record(data, em) 

52 em.delivery_status = self.new_status 

53 

54 session.commit() 

55 

56 log.info("Updated notification %s to status %s", em, self.new_status) 

57 

58 except NoResultFound: 

59 log.warning( 

60 "EmailNotification with message ID %s not found", data.message_id 

61 ) 

62 

63 except Exception: 

64 log.exception("Error handling webhook from postmark") 

65 

66 finally: 

67 session.close() 

68 return Response("Thanks so much. That was delightful.") 

69 

70 def extract(self, request: Request) -> PostmarkVars: 

71 if request.method.upper() != "POST": 

72 raise HTTPMethodNotAllowed( 

73 "Webhook for email delivery events - use POST to update" 

74 ) 

75 return PostmarkVars(request.json) 

76 

77 def lookup(self, session, message_id: str) -> EmailNotification: 

78 return ( 

79 session.query(EmailNotification) 

80 .filter(EmailNotification.message_id == message_id) 

81 .one() 

82 ) 

83 

84 def check_emails_match(self, data: PostmarkVars, em: EmailNotification): 

85 if data.recipient != em.email: 

86 m = "Recipient email %s for MessageId %s does not match recorded email %s" 

87 log.warning(m, data.recipient, data.message_id, em.email) 

88 

89 def update_notification_record(self, data: PostmarkVars, em: EmailNotification): 

90 raise NotImplementedError