Coverage for app/backend/src/couchers/email/queuing.py: 95%

1from pathlib import Path 

2from typing import Any 

3 

4import yaml 

5from sqlalchemy.orm.session import Session 

6 

7from couchers.config import config 

8from couchers.context import CouchersContext 

9from couchers.email.blocks import EmailBase, EmailFooter 

10from couchers.email.rendering import render_email 

11from couchers.email.zz import add_open_pixel, add_zz_to_html_links, generate_zz 

12from couchers.event_log import log_event 

13from couchers.i18n import LocalizationContext 

14from couchers.jobs.enqueue import queue_job 

15from couchers.metrics import emails_counter 

16from couchers.proto.internal import jobs_pb2 

17from couchers.templating import Jinja2Template 

18from couchers.utils import now 

19 

20 

21def _queue_email(session: Session, payload: jobs_pb2.SendEmailPayload) -> None: 

22 """ 

23 This indirection is so that this can be easily mocked. Not sure how to do it better :( 

24 """ 

25 

26 # Import here to avoid circular dependency 

27 from couchers.jobs.handlers import send_email # noqa: PLC0415 

28 

29 queue_job( 

30 session, 

31 job=send_email, 

32 payload=payload, 

33 priority=15, 

34 ) 

35 

36 emails_counter.inc() 

37 

38 

39def queue_email(session: Session, payload: jobs_pb2.SendEmailPayload) -> None: 

40 _queue_email(session, payload) 

41 

42 

43def queue_email_with_zz(context: CouchersContext, session: Session, payload: jobs_pb2.SendEmailPayload) -> None: 

44 """ 

45 Queues an email, adding a zz id to its html links (see couchers/email/zz.py) and logging it as sent. 

46 """ 

47 zz = generate_zz() 

48 if payload.html: 48 ↛ 54line 48 didn't jump to line 54 because the condition on line 48 was always true

49 if context.get_boolean_value("email_zz_links_enabled", default=False): 

50 payload.html = add_zz_to_html_links(payload.html, zz) 

51 if context.get_boolean_value("email_open_pixel_enabled", default=False): 

52 payload.html = add_open_pixel(payload.html, zz) 

53 

54 properties: dict[str, str | int] = {"zz": zz, "source_data": payload.source_data} 

55 if payload.notification_id: 

56 properties["notification_id"] = payload.notification_id 

57 log_event(context, session, "email.sent", properties) 

58 

59 queue_email(session, payload) 

60 

61 

62def queue_userless_email( 

63 context: CouchersContext, session: Session, recipient: str, email: EmailBase, source_data_header: str 

64) -> None: 

65 """ 

66 This is a simplified version of couchers.notifications.background._send_email_notification 

67 

68 It's for the few security emails where we don't have a user to email but send directly to an email address. 

69 """ 

70 

71 loc_context = context.localization 

72 if not context.get_boolean_value("notification_translations_enabled", default=True): 72 ↛ 73line 72 didn't jump to line 73 because the condition on line 72 was never true

73 loc_context = LocalizationContext(locale="en", timezone=loc_context.timezone) 

74 

75 footer = EmailFooter(copyright_year=now().year, unsubscribe_info=None) 

76 rendered = render_email(email, footer, loc_context) 

77 

78 queue_email_with_zz( 

79 context, 

80 session, 

81 jobs_pb2.SendEmailPayload( 

82 sender_name=config.NOTIFICATION_EMAIL_SENDER, 

83 sender_email=config.NOTIFICATION_EMAIL_ADDRESS, 

84 recipient=recipient, 

85 subject=config.NOTIFICATION_PREFIX + rendered.subject, 

86 plain=rendered.body_plaintext, 

87 html=rendered.body_html, 

88 html_related_parts=rendered.html_image_parts, 

89 source_data=f"{source_data_header}; version={config.VERSION}", 

90 ), 

91 ) 

92 

93 

94_system_email_templates_dir = Path(__file__).parent / ".." / ".." / ".." / "templates" / "system" 

95 

96 

97def queue_system_email( 

98 session: Session, 

99 recipient: str, 

100 template_name: str, 

101 template_args: dict[str, Any], 

102 attachments: list[jobs_pb2.EmailPart] | None = None, 

103) -> None: 

104 source = (_system_email_templates_dir / f"{template_name}.md").read_text(encoding="utf8") 

105 _, frontmatter_source, text_source = source.split("---", 2) 

106 

107 rendered_frontmatter = Jinja2Template(source=frontmatter_source, html=False).render(template_args) 

108 frontmatter = yaml.load(rendered_frontmatter, Loader=yaml.FullLoader) 

109 

110 plain = Jinja2Template(source=text_source.strip(), html=False).render(template_args) 

111 

112 queue_email( 

113 session, 

114 jobs_pb2.SendEmailPayload( 

115 sender_name=config.NOTIFICATION_EMAIL_SENDER, 

116 sender_email=config.NOTIFICATION_EMAIL_ADDRESS, 

117 recipient=recipient, 

118 subject=config.NOTIFICATION_PREFIX + frontmatter["subject"], 

119 plain=plain, 

120 source_data=template_name, 

121 attachments=attachments or [], 

122 ), 

123 )