Coverage for app/backend/src/tests/fixtures/api/events.py: 98%
118 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-19 17:13 +0000
1from contextlib import contextmanager
2from datetime import date, datetime, time, timedelta
3from typing import cast
4from zoneinfo import ZoneInfo
6import grpc
7from google.protobuf import wrappers_pb2
8from sqlalchemy import select
10from couchers.db import session_scope
11from couchers.models import EventOccurrence, EventRecurrence
12from couchers.proto import events_pb2, events_pb2_grpc
13from couchers.proto.google.api import httpbody_pb2
14from couchers.servicers.events import Events
15from couchers.utils import today
16from tests.fixtures.misc import Moderator
17from tests.fixtures.sessions import FakeChannel
19# Event location constants which timezone_areas.sql-fake has regions for.
20# Neutral default location. Resolves to Etc/GMT with real dataset and Etc/UTC with fake dataset.
21EVENT_LOCATION_DEFAULT_UTC = events_pb2.EventLocation(address="Near Null Island", lat=0.1, lng=0.2)
22EVENT_LOCATION_HELSINKI = events_pb2.EventLocation(address="Helsinki, Finland", lat=60.192059, lng=24.945831)
23EVENT_LOCATION_NEW_YORK = events_pb2.EventLocation(address="New York, New York", lat=40.712776, lng=-74.005974)
26def _date_time_to_iso8601(value: tuple[date, time]) -> str:
27 return f"{value[0].isoformat()}T{value[1].isoformat()}"
30def _one_hour_after(value: tuple[date, time]) -> tuple[date, time]:
31 end_datetime = datetime.combine(value[0], value[1]) + timedelta(hours=1)
32 return end_datetime.date(), end_datetime.time()
35def _string_value(value: str | None) -> wrappers_pb2.StringValue | None:
36 return wrappers_pb2.StringValue(value=value) if value is not None else None
39def _int64_value(value: int | None) -> wrappers_pb2.Int64Value | None:
40 return wrappers_pb2.Int64Value(value=value) if value is not None else None
43@contextmanager
44def _events_session(token: str):
45 channel = FakeChannel(token)
46 events_pb2_grpc.add_EventsServicer_to_server(Events(), cast(grpc.Server, channel))
47 yield events_pb2_grpc.EventsStub(cast(grpc.Channel, channel))
50def create_event(
51 *,
52 token: str,
53 title: str | None = "Dummy Title",
54 content: str | None = "Dummy content.",
55 photo_key: str | None = None,
56 parent_community_id: int | None = None,
57 location: events_pb2.EventLocation | None = EVENT_LOCATION_DEFAULT_UTC,
58 # Force callers to provide local date/times, not datetimes with timezones,
59 # since the API accepts local times and resolves the timezone from the location.
60 start_date_time: tuple[date, time] | None = None,
61 end_date_time: tuple[date, time] | None = None,
62 approve_by: Moderator | None = None,
63) -> events_pb2.Event:
64 if start_date_time is None:
65 start_date_time = (today() + timedelta(days=1), time(12, 0))
66 if end_date_time is None:
67 end_date_time = _one_hour_after(start_date_time)
69 with _events_session(token) as session:
70 event: events_pb2.Event = session.CreateEvent(
71 events_pb2.CreateEventReq(
72 title=title,
73 content=content,
74 photo_key=photo_key,
75 parent_community_id=parent_community_id,
76 location=location,
77 start_datetime_iso8601_local=_date_time_to_iso8601(start_date_time),
78 end_datetime_iso8601_local=_date_time_to_iso8601(end_date_time),
79 )
80 )
82 if approve_by is not None:
83 approve_by.approve_event_occurrence(event.event_id)
85 return event
88def schedule_event(
89 *,
90 token: str,
91 event_id: int,
92 content: str | None = "Dummy content.",
93 photo_key: str | None = None,
94 location: events_pb2.EventLocation | None = EVENT_LOCATION_DEFAULT_UTC,
95 # Force callers to provide local date/times, not datetimes with timezones,
96 # since the API accepts local times and resolves the timezone from the location.
97 start_date_time: tuple[date, time],
98 end_date_time: tuple[date, time] | None = None,
99 approve_by: Moderator | None = None,
100) -> events_pb2.Event:
101 if end_date_time is None: 101 ↛ 102line 101 didn't jump to line 102 because the condition on line 101 was never true
102 end_date_time = _one_hour_after(start_date_time)
104 with _events_session(token) as session:
105 event: events_pb2.Event = session.ScheduleEvent(
106 events_pb2.ScheduleEventReq(
107 event_id=event_id,
108 content=content,
109 photo_key=photo_key,
110 location=location,
111 start_datetime_iso8601_local=_date_time_to_iso8601(start_date_time),
112 end_datetime_iso8601_local=_date_time_to_iso8601(end_date_time),
113 )
114 )
116 if approve_by is not None:
117 approve_by.approve_event_occurrence(event.event_id)
119 return event
122def update_event(
123 *,
124 token: str,
125 event_id: int,
126 # None leaves the field unchanged.
127 title: str | None = None,
128 content: str | None = None,
129 photo_key: str | None = None,
130 location: events_pb2.EventLocation | None = None,
131 start_date_time: tuple[date, time] | None = None,
132 end_date_time: tuple[date, time] | None = None,
133 update_all_future: bool = False,
134 should_notify: bool = False,
135) -> events_pb2.Event:
136 with _events_session(token) as session:
137 event: events_pb2.Event = session.UpdateEvent(
138 events_pb2.UpdateEventReq(
139 event_id=event_id,
140 update_all_future=update_all_future,
141 title=_string_value(title),
142 content=_string_value(content),
143 photo_key=_string_value(photo_key),
144 location=location,
145 start_datetime_iso8601_local=_string_value(
146 _date_time_to_iso8601(start_date_time) if start_date_time is not None else None
147 ),
148 end_datetime_iso8601_local=_string_value(
149 _date_time_to_iso8601(end_date_time) if end_date_time is not None else None
150 ),
151 should_notify=should_notify,
152 )
153 )
154 return event
157def get_event(*, token: str, event_id: int) -> events_pb2.Event:
158 with _events_session(token) as session:
159 event: events_pb2.Event = session.GetEvent(events_pb2.GetEventReq(event_id=event_id))
160 return event
163def get_event_calendar_file(*, token: str, event_id: int) -> httpbody_pb2.HttpBody:
164 with _events_session(token) as session:
165 res: httpbody_pb2.HttpBody = session.GetEventCalendarFile(events_pb2.GetEventCalendarFileReq(event_id=event_id))
166 return res
169def transfer_event(
170 *, token: str, event_id: int, new_owner_community_id: int | None = None, new_owner_group_id: int | None = None
171) -> events_pb2.Event:
172 with _events_session(token) as session:
173 event: events_pb2.Event = session.TransferEvent(
174 events_pb2.TransferEventReq(
175 event_id=event_id,
176 new_owner_community_id=new_owner_community_id,
177 new_owner_group_id=new_owner_group_id,
178 )
179 )
180 return event
183def request_community_invite(*, token: str, event_id: int) -> None:
184 with _events_session(token) as session:
185 session.RequestCommunityInvite(events_pb2.RequestCommunityInviteReq(event_id=event_id))
188def set_event_subscription(*, token: str, event_id: int, subscribe: bool) -> events_pb2.Event:
189 with _events_session(token) as session:
190 event: events_pb2.Event = session.SetEventSubscription(
191 events_pb2.SetEventSubscriptionReq(event_id=event_id, subscribe=subscribe)
192 )
193 return event
196def list_event_subscribers(
197 *, token: str, event_id: int, page_size: int | None = None, page_token: str | None = None
198) -> events_pb2.ListEventSubscribersRes:
199 with _events_session(token) as session:
200 res: events_pb2.ListEventSubscribersRes = session.ListEventSubscribers(
201 events_pb2.ListEventSubscribersReq(event_id=event_id, page_size=page_size, page_token=page_token)
202 )
203 return res
206def set_event_attendance(
207 *, token: str, event_id: int, attendance_state: events_pb2.AttendanceState.ValueType
208) -> events_pb2.Event:
209 with _events_session(token) as session:
210 event: events_pb2.Event = session.SetEventAttendance(
211 events_pb2.SetEventAttendanceReq(event_id=event_id, attendance_state=attendance_state)
212 )
213 return event
216def list_event_attendees(
217 *, token: str, event_id: int, page_size: int | None = None, page_token: str | None = None
218) -> events_pb2.ListEventAttendeesRes:
219 with _events_session(token) as session:
220 res: events_pb2.ListEventAttendeesRes = session.ListEventAttendees(
221 events_pb2.ListEventAttendeesReq(event_id=event_id, page_size=page_size, page_token=page_token)
222 )
223 return res
226def invite_event_organizer(*, token: str, event_id: int, user_id: int) -> None:
227 with _events_session(token) as session:
228 session.InviteEventOrganizer(
229 events_pb2.InviteEventOrganizerReq(
230 event_id=event_id,
231 user_id=user_id,
232 )
233 )
236def remove_event_organizer(*, token: str, event_id: int, user_id: int | None = None) -> None:
237 with _events_session(token) as session:
238 session.RemoveEventOrganizer(
239 events_pb2.RemoveEventOrganizerReq(
240 event_id=event_id,
241 user_id=_int64_value(user_id),
242 )
243 )
246def list_event_organizers(
247 *, token: str, event_id: int, page_size: int | None = None, page_token: str | None = None
248) -> events_pb2.ListEventOrganizersRes:
249 with _events_session(token) as session:
250 res: events_pb2.ListEventOrganizersRes = session.ListEventOrganizers(
251 events_pb2.ListEventOrganizersReq(event_id=event_id, page_size=page_size, page_token=page_token)
252 )
253 return res
256def list_event_occurrences(
257 *,
258 token: str,
259 event_id: int,
260 past: bool = False,
261 include_cancelled: bool = False,
262 page_size: int | None = None,
263 page_token: str | None = None,
264) -> events_pb2.ListEventOccurrencesRes:
265 with _events_session(token) as session:
266 res: events_pb2.ListEventOccurrencesRes = session.ListEventOccurrences(
267 events_pb2.ListEventOccurrencesReq(
268 event_id=event_id,
269 past=past,
270 include_cancelled=include_cancelled,
271 page_size=page_size,
272 page_token=page_token,
273 )
274 )
275 return res
278def list_all_events(
279 *,
280 token: str,
281 past: bool = False,
282 include_cancelled: bool = False,
283 page_size: int | None = None,
284 page_token: str | None = None,
285) -> events_pb2.ListAllEventsRes:
286 with _events_session(token) as session:
287 res: events_pb2.ListAllEventsRes = session.ListAllEvents(
288 events_pb2.ListAllEventsReq(
289 past=past, include_cancelled=include_cancelled, page_size=page_size, page_token=page_token
290 )
291 )
292 return res
295def list_my_events(
296 *,
297 token: str,
298 attending: bool = False,
299 exclude_attending: bool = False,
300 subscribed: bool = False,
301 organizing: bool = False,
302 include_cancelled: bool = False,
303 my_communities: bool = False,
304 my_communities_exclude_global: bool = False,
305 page_size: int | None = None,
306 page_token: str | None = None,
307 page_number: int | None = None,
308) -> events_pb2.ListMyEventsRes:
309 with _events_session(token) as session:
310 res: events_pb2.ListMyEventsRes = session.ListMyEvents(
311 events_pb2.ListMyEventsReq(
312 attending=attending,
313 exclude_attending=exclude_attending,
314 subscribed=subscribed,
315 organizing=organizing,
316 include_cancelled=include_cancelled,
317 my_communities=my_communities,
318 my_communities_exclude_global=my_communities_exclude_global,
319 page_size=page_size,
320 page_token=page_token,
321 page_number=page_number,
322 )
323 )
324 return res
327def cancel_event(*, token: str, event_id: int) -> None:
328 with _events_session(token) as session:
329 session.CancelEvent(events_pb2.CancelEventReq(event_id=event_id))
332# "x" prefix because this is not yet an API, but must be accomplished through database manipulation
333def x_make_event_recurring(
334 event: events_pb2.Event,
335 *,
336 rrule_interval: int = 1,
337 ends_on_date: date,
338) -> None:
339 with session_scope() as session:
340 occurrence = session.execute(select(EventOccurrence).where(EventOccurrence.id == event.event_id)).scalar_one()
342 timezone = ZoneInfo(occurrence.timezone)
343 start_datetime_in_tz = occurrence.start_time.astimezone(timezone)
344 end_datetime_in_tz = occurrence.end_time.astimezone(timezone)
345 day_delta = (end_datetime_in_tz.date() - start_datetime_in_tz.date()).days
346 session.add(
347 EventRecurrence(
348 event_id=occurrence.event_id,
349 content=occurrence.content,
350 # TODO(#9845): Initialize photo_key. Implies adding admin usage tracking infra.
351 geom=occurrence.geom,
352 address=occurrence.address,
353 timezone=occurrence.timezone,
354 dtstart_date=start_datetime_in_tz.date(),
355 start_time=start_datetime_in_tz.time(),
356 day_delta=day_delta,
357 end_time=end_datetime_in_tz.time(),
358 rrule_interval=rrule_interval,
359 last_scheduled_date=start_datetime_in_tz.date(),
360 ends_on_date=ends_on_date,
361 )
362 )