Coverage for app/backend/src/tests/test_email_zz.py: 100%

1import re 

2 

3from sqlalchemy import select 

4 

5from couchers.context import make_logged_out_context 

6from couchers.crypto import random_hex 

7from couchers.db import session_scope 

8from couchers.email.zz import ( 

9 OPEN_PIXEL_GIF, 

10 add_open_pixel, 

11 add_zz_to_html_links, 

12 generate_zz, 

13 open_pixel_url, 

14) 

15from couchers.i18n import LocalizationContext 

16from couchers.models import Notification, NotificationTopicAction, SignupFlow 

17from couchers.models.logging import EventLog 

18from couchers.proto import api_pb2, bugs_pb2 

19from couchers.tasks import send_signup_email 

20from tests.fixtures.db import generate_user, get_friend_relationship 

21from tests.fixtures.misc import EmailCollector, Moderator 

22from tests.fixtures.sessions import api_session, real_bugs_session 

23 

24 

25def test_add_zz_to_html_links(): 

26 html = ( 

27 '<a href="http://localhost:3000/quick-link?payload=x&amp;sig=y">a</a>' 

28 '<a href="http://localhost:3000/dashboard#top">b</a>' 

29 '<a href="https://example.com/">c</a>' 

30 '<a href="http://localhost:3000.evil.com/x">d</a>' 

31 '<img src="http://localhost:3000/logo512.png">' 

32 "<p>http://localhost:3000/user/bob</p>" 

33 ) 

34 assert add_zz_to_html_links(html, "abc") == ( 

35 '<a href="http://localhost:3000/quick-link?payload=x&amp;sig=y&amp;zz=abc">a</a>' 

36 '<a href="http://localhost:3000/dashboard?zz=abc#top">b</a>' 

37 '<a href="https://example.com/">c</a>' 

38 '<a href="http://localhost:3000.evil.com/x">d</a>' 

39 '<img src="http://localhost:3000/logo512.png">' 

40 "<p>http://localhost:3000/user/bob</p>" 

41 ) 

42 

43 

44def test_add_open_pixel(): 

45 html = add_open_pixel("<html><body><p>hi</p></body></html>", "abc") 

46 assert html.startswith("<html><body><p>hi</p><img ") 

47 assert html.endswith("></body></html>") 

48 assert f'src="{open_pixel_url("abc")}"' in html 

49 

50 

51def test_notification_email_gets_zz(db, email_collector: EmailCollector, moderator: Moderator, feature_flags): 

52 feature_flags.set("email_zz_links_enabled", True) 

53 to_user, _ = generate_user() 

54 from_user, from_token = generate_user() 

55 

56 with api_session(from_token) as api: 

57 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=to_user.id)) 

58 friend_relationship = get_friend_relationship(from_user, to_user) 

59 assert friend_relationship 

60 moderator.approve_friend_request(friend_relationship.id) 

61 

62 email = email_collector.pop_for_recipient(to_user.email, last=True) 

63 

64 with session_scope() as session: 

65 notification_id = session.execute( 

66 select(Notification.id) 

67 .where(Notification.user_id == to_user.id) 

68 .where(Notification.topic_action == NotificationTopicAction.friend_request__create) 

69 ).scalar_one() 

70 properties = session.execute( 

71 select(EventLog.properties).where(EventLog.event_type == "email.sent").where(EventLog.user_id == to_user.id) 

72 ).scalar_one() 

73 

74 zz = properties["zz"] 

75 assert properties == {"zz": zz, "source_data": email.source_data, "notification_id": notification_id} 

76 

77 hrefs = re.findall(r'href="(http://localhost:3000[^"]*)"', email.html) 

78 assert hrefs 

79 assert all(f"zz={zz}" in href for href in hrefs) 

80 assert "zz=" not in email.plain 

81 assert "/email/open/" not in email.html 

82 

83 

84def test_userless_email_gets_open_pixel_without_zz_links(db, email_collector: EmailCollector, feature_flags): 

85 feature_flags.set("email_open_pixel_enabled", True) 

86 recipient = f"{random_hex(12)}@couchers.org.invalid" 

87 

88 with session_scope() as session: 

89 send_signup_email( 

90 make_logged_out_context(LocalizationContext.en_utc()), 

91 session, 

92 SignupFlow(name="Frodo", email=recipient, flow_token=""), 

93 ) 

94 

95 email = email_collector.pop_for_recipient(recipient, last=True) 

96 

97 with session_scope() as session: 

98 event = session.execute(select(EventLog).where(EventLog.event_type == "email.sent")).scalar_one() 

99 assert event.user_id is None 

100 zz = event.properties["zz"] 

101 

102 assert "zz=" not in email.html 

103 assert "zz=" not in email.plain 

104 assert f'src="{open_pixel_url(zz)}"' in email.html 

105 

106 

107def test_record_email_open(db): 

108 zz = generate_zz() 

109 

110 with real_bugs_session() as (bugs, metadata_interceptor): 

111 res = bugs.RecordEmailOpen(bugs_pb2.RecordEmailOpenReq(zz=zz)) 

112 

113 assert res.content_type == "image/gif" 

114 assert res.data == OPEN_PIXEL_GIF 

115 assert metadata_interceptor.latest_headers["cache-control"] == "no-store" 

116 

117 with session_scope() as session: 

118 properties = session.execute( 

119 select(EventLog.properties).where(EventLog.event_type == "email.opened") 

120 ).scalar_one() 

121 assert properties == {"zz": zz} 

122 

123 

124def test_record_email_open_ignores_malformed_zz(db): 

125 with real_bugs_session() as (bugs, _): 

126 res = bugs.RecordEmailOpen(bugs_pb2.RecordEmailOpenReq(zz="not an zz")) 

127 

128 assert res.data == OPEN_PIXEL_GIF 

129 

130 with session_scope() as session: 

131 assert not session.execute(select(EventLog).where(EventLog.event_type == "email.opened")).scalars().all()