Coverage for app/backend/src/tests/test_calendar_events.py: 97%

1import re 

2from datetime import timedelta 

3 

4import ics 

5import pytest 

6 

7from couchers.email.calendar_events import create_host_request_calendar, create_host_request_cancellation_calendar 

8from couchers.i18n.context import LocalizationContext 

9from couchers.proto import conversations_pb2, requests_pb2 

10from couchers.proto.requests_pb2 import HostRequest 

11from couchers.utils import today 

12from tests.fixtures.db import generate_user 

13from tests.fixtures.misc import EmailCollector, Moderator 

14from tests.fixtures.sessions import requests_session 

15from tests.test_requests import valid_request_text 

16 

17 

18@pytest.fixture(autouse=True) 

19def _(testconfig): 

20 pass 

21 

22 

23def test_initial_ics_content(): 

24 host_request = HostRequest( 

25 host_request_id=42, from_date="2000-01-01", to_date="2000-01-02", hosting_city="New York" 

26 ) 

27 

28 ics: str = create_host_request_calendar( 

29 host_request, other_name="Bob", hosting=True, loc_context=LocalizationContext.en_utc() 

30 ).serialize() 

31 assert _normalize_ics(ics) == _normalize_ics(""" 

32BEGIN:VCALENDAR 

33VERSION:2.0 

34PRODID:-//Couchers.org//Couchers//EN 

35BEGIN:VEVENT 

36SEQUENCE:0 

37DTSTART;VALUE=DATE:20000101 

38DTEND;VALUE=DATE:20000103 

39DESCRIPTION:<stripped>/42 

40LOCATION:New York 

41SUMMARY:Hosting Bob 

42UID:host_request.42@<stripped> 

43URL:<stripped>/42 

44END:VEVENT 

45METHOD:PUBLISH 

46END:VCALENDAR 

47 """) 

48 

49 

50def test_cancellation_ics_content(): 

51 host_request = HostRequest( 

52 host_request_id=42, from_date="2000-01-01", to_date="2000-01-02", hosting_city="New York" 

53 ) 

54 

55 ics: str = create_host_request_cancellation_calendar( 

56 host_request, other_name="Bob", hosting=True, loc_context=LocalizationContext.en_utc() 

57 ).serialize() 

58 assert _normalize_ics(ics) == _normalize_ics(""" 

59BEGIN:VCALENDAR 

60VERSION:2.0 

61PRODID:-//Couchers.org//Couchers//EN 

62BEGIN:VEVENT 

63SEQUENCE:1 

64DTSTART;VALUE=DATE:20000101 

65DTEND;VALUE=DATE:20000103 

66DESCRIPTION:<stripped>/42 

67LOCATION:New York 

68STATUS:CANCELLED 

69SUMMARY:Cancelled: Hosting Bob 

70UID:host_request.42@<stripped> 

71URL:<stripped>/42 

72END:VEVENT 

73METHOD:PUBLISH 

74END:VCALENDAR 

75 """) 

76 

77 

78def _normalize_ics(ics: str) -> str: 

79 # Normalize whitespace: 

80 # - The ics library produces '\r\n', in-code literals are '\n' 

81 # - In-code literals have start/end newlines and indentation. 

82 ics = ics.replace("\r\n", "\n").strip() 

83 

84 # Strip the domain in the UID, which depends on environment variables 

85 ics = re.sub( 

86 r"^UID:.*@(.*)$", lambda match: match[0].removesuffix(match[1]) + "<stripped>", ics, flags=re.MULTILINE 

87 ) 

88 

89 # Strip the domain in the URL and DESCRIPTION, which depends on environment variables 

90 ics = re.sub(r"^(DESCRIPTION|URL):(.*)/(\d+)", r"\1:<stripped>/\3", ics, flags=re.MULTILINE) 

91 

92 return ics 

93 

94 

95def test_host_request_attachments(db, email_collector: EmailCollector, moderator: Moderator): 

96 host, host_token = generate_user(complete_profile=True) 

97 surfer, surfer_token = generate_user(complete_profile=True) 

98 

99 from_date = today() + timedelta(days=2) 

100 to_date = today() + timedelta(days=3) 

101 

102 # Send the host request, no calendar attachment yet 

103 with requests_session(surfer_token) as api: 

104 hr_id = api.CreateHostRequest( 

105 requests_pb2.CreateHostRequestReq( 

106 host_user_id=host.id, 

107 from_date=from_date.isoformat(), 

108 to_date=to_date.isoformat(), 

109 text=valid_request_text("can i stay plz"), 

110 ) 

111 ).host_request_id 

112 

113 moderator.approve_host_request(hr_id) 

114 

115 email = email_collector.pop_for_recipient(host.email, last=True) 

116 assert "request" in email.subject and surfer.name in email.subject 

117 assert not email.attachments 

118 

119 # Host accepts, surfer gets a calendar attachment 

120 with requests_session(host_token) as api: 

121 api.RespondHostRequest( 

122 requests_pb2.RespondHostRequestReq( 

123 host_request_id=hr_id, 

124 status=conversations_pb2.HOST_REQUEST_STATUS_ACCEPTED, 

125 text="Accepting host request", 

126 ) 

127 ) 

128 

129 email = email_collector.pop_for_recipient(surfer.email, last=True) 

130 assert "accept" in email.subject and host.name in email.subject 

131 ics_event = _get_email_ics_attachment_calendar_event(email) 

132 assert _get_ics_event_sequence(ics_event) == 0 

133 assert not ics_event.status 

134 assert ics_event.begin.date() == from_date 

135 assert ics_event.end.date() == (to_date + timedelta(days=1)) 

136 assert ics_event.name == f"Surfing with {host.name}" 

137 assert ics_event.location == host.city 

138 

139 # Surfer confirms, host gets a calendar attachment 

140 with requests_session(surfer_token) as api: 

141 api.RespondHostRequest( 

142 requests_pb2.RespondHostRequestReq( 

143 host_request_id=hr_id, 

144 status=conversations_pb2.HOST_REQUEST_STATUS_CONFIRMED, 

145 text="Confirming host request", 

146 ) 

147 ) 

148 

149 email = email_collector.pop_for_recipient(host.email, last=True) 

150 assert "confirm" in email.subject and surfer.name in email.subject 

151 ics_event = _get_email_ics_attachment_calendar_event(email) 

152 assert _get_ics_event_sequence(ics_event) == 0 

153 assert not ics_event.status 

154 assert ics_event.name == f"Hosting {surfer.name}" 

155 

156 # Surfer cancels, host gets a calendar attachment 

157 with requests_session(surfer_token) as api: 

158 api.RespondHostRequest( 

159 requests_pb2.RespondHostRequestReq( 

160 host_request_id=hr_id, 

161 status=conversations_pb2.HOST_REQUEST_STATUS_CANCELLED, 

162 text="Cancelling host request", 

163 ) 

164 ) 

165 

166 email = email_collector.pop_for_recipient(host.email, last=True) 

167 assert "cancel" in email.subject and surfer.name in email.subject 

168 ics_event = _get_email_ics_attachment_calendar_event(email) 

169 assert _get_ics_event_sequence(ics_event) == 1 

170 assert ics_event.status == "CANCELLED" 

171 

172 

173def test_host_request_attachments_disabled(db, email_collector: EmailCollector, feature_flags, moderator: Moderator): 

174 feature_flags.set("email_ics_attachments_enabled", False) 

175 

176 host, host_token = generate_user(complete_profile=True) 

177 surfer, surfer_token = generate_user(complete_profile=True) 

178 

179 from_date = today() + timedelta(days=2) 

180 to_date = today() + timedelta(days=3) 

181 

182 with requests_session(surfer_token) as api: 

183 hr_id = api.CreateHostRequest( 

184 requests_pb2.CreateHostRequestReq( 

185 host_user_id=host.id, 

186 from_date=from_date.isoformat(), 

187 to_date=to_date.isoformat(), 

188 text=valid_request_text("can i stay plz"), 

189 ) 

190 ).host_request_id 

191 

192 moderator.approve_host_request(hr_id) 

193 

194 email_collector.pop_for_recipient(host.email, last=True) 

195 

196 # Host accepts: normally the surfer would get a calendar attachment, but the flag is off 

197 with requests_session(host_token) as api: 

198 api.RespondHostRequest( 

199 requests_pb2.RespondHostRequestReq( 

200 host_request_id=hr_id, 

201 status=conversations_pb2.HOST_REQUEST_STATUS_ACCEPTED, 

202 text="Accepting host request", 

203 ) 

204 ) 

205 

206 email = email_collector.pop_for_recipient(surfer.email, last=True) 

207 assert "accept" in email.subject and host.name in email.subject 

208 assert not email.attachments 

209 

210 

211def _get_email_ics_attachment_calendar_event(e) -> ics.Event: 

212 assert len(e.attachments or []) == 1 

213 ics_attachment = e.attachments[0] 

214 assert ics_attachment.content_type.startswith("text/calendar") 

215 ics_calendar = ics.Calendar(ics_attachment.data.decode("utf-8")) 

216 assert f"method={ics_calendar.method}" in ics_attachment.content_type 

217 assert len(ics_calendar.events) == 1 

218 return next(iter(ics_calendar.events)) 

219 

220 

221def _get_ics_event_sequence(event: ics.Event) -> int | None: 

222 for x in event.extra: 222 ↛ 225line 222 didn't jump to line 225 because the loop on line 222 didn't complete

223 if x.name == "SEQUENCE": 223 ↛ 222line 223 didn't jump to line 222 because the condition on line 223 was always true

224 return int(x.value) 

225 return None