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.i18n import LocalizationContext 

12from couchers.jobs.enqueue import queue_job 

13from couchers.metrics import emails_counter 

14from couchers.proto.internal import jobs_pb2 

15from couchers.templating import Jinja2Template 

16from couchers.utils import now 

17 

18 

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

20 """ 

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

22 """ 

23 

24 # Import here to avoid circular dependency 

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

26 

27 queue_job( 

28 session, 

29 job=send_email, 

30 payload=payload, 

31 priority=5, 

32 ) 

33 

34 emails_counter.inc() 

35 

36 

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

38 _queue_email(session, payload) 

39 

40 

41def queue_userless_email( 

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

43) -> None: 

44 """ 

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

46 

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

48 """ 

49 

50 loc_context = context.localization 

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

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

53 

54 footer = EmailFooter(timezone_name=loc_context.localized_timezone, copyright_year=now().year, unsubscribe_info=None) 

55 rendered = render_email(email, footer, loc_context) 

56 

57 queue_email( 

58 session, 

59 jobs_pb2.SendEmailPayload( 

60 sender_name=config.NOTIFICATION_EMAIL_SENDER, 

61 sender_email=config.NOTIFICATION_EMAIL_ADDRESS, 

62 recipient=recipient, 

63 subject=config.NOTIFICATION_PREFIX + rendered.subject, 

64 plain=rendered.body_plaintext, 

65 html=rendered.body_html, 

66 html_related_parts=rendered.html_image_parts, 

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

68 ), 

69 ) 

70 

71 

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

73 

74 

75def queue_system_email(session: Session, recipient: str, template_name: str, template_args: dict[str, Any]) -> None: 

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

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

78 

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

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

81 

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

83 

84 queue_email( 

85 session, 

86 jobs_pb2.SendEmailPayload( 

87 sender_name=config.NOTIFICATION_EMAIL_SENDER, 

88 sender_email=config.NOTIFICATION_EMAIL_ADDRESS, 

89 recipient=recipient, 

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

91 plain=plain, 

92 source_data=template_name, 

93 ), 

94 )