Coverage for app/backend/src/tests/fixtures/sessions.py: 99%
296 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-19 00:32 +0000
1from collections.abc import Generator
2from concurrent import futures
3from contextlib import contextmanager
4from typing import Any, NoReturn
5from zoneinfo import ZoneInfo
7import grpc
8from grpc._server import _validate_generic_rpc_handlers
10from couchers.context import make_interactive_context
11from couchers.db import session_scope
12from couchers.descriptor_pool import get_descriptor_pool
13from couchers.i18n import LocalizationContext
14from couchers.i18n.locales import DEFAULT_LOCALE
15from couchers.interceptors import (
16 CouchersMiddlewareInterceptor,
17 _try_get_and_update_user_details,
18 check_permissions,
19 find_auth_level,
20)
21from couchers.proto import (
22 account_pb2_grpc,
23 admin_pb2_grpc,
24 api_pb2_grpc,
25 auth_pb2_grpc,
26 blocking_pb2_grpc,
27 bugs_pb2_grpc,
28 communities_pb2_grpc,
29 conversations_pb2_grpc,
30 discussions_pb2_grpc,
31 donations_pb2_grpc,
32 editor_pb2_grpc,
33 events_pb2_grpc,
34 galleries_pb2_grpc,
35 gis_pb2_grpc,
36 groups_pb2_grpc,
37 iris_pb2_grpc,
38 jail_pb2_grpc,
39 media_pb2_grpc,
40 moderation_pb2_grpc,
41 notifications_pb2_grpc,
42 pages_pb2_grpc,
43 postal_verification_pb2_grpc,
44 public_pb2_grpc,
45 public_trips_pb2_grpc,
46 references_pb2_grpc,
47 reporting_pb2_grpc,
48 requests_pb2_grpc,
49 resources_pb2_grpc,
50 search_pb2_grpc,
51 stripe_pb2_grpc,
52 threads_pb2_grpc,
53)
54from couchers.servicers.account import Account, Iris
55from couchers.servicers.admin import Admin
56from couchers.servicers.api import API
57from couchers.servicers.auth import Auth
58from couchers.servicers.blocking import Blocking
59from couchers.servicers.bugs import Bugs
60from couchers.servicers.communities import Communities
61from couchers.servicers.conversations import Conversations
62from couchers.servicers.discussions import Discussions
63from couchers.servicers.donations import Donations, Stripe
64from couchers.servicers.editor import Editor
65from couchers.servicers.events import Events
66from couchers.servicers.galleries import Galleries
67from couchers.servicers.gis import GIS
68from couchers.servicers.groups import Groups
69from couchers.servicers.jail import Jail
70from couchers.servicers.media import Media, get_media_auth_interceptor
71from couchers.servicers.moderation import Moderation
72from couchers.servicers.notifications import Notifications
73from couchers.servicers.pages import Pages
74from couchers.servicers.postal_verification import PostalVerification
75from couchers.servicers.public import Public
76from couchers.servicers.public_trips import PublicTrips
77from couchers.servicers.references import References
78from couchers.servicers.reporting import Reporting
79from couchers.servicers.requests import Requests
80from couchers.servicers.resources import Resources
81from couchers.servicers.search import Search
82from couchers.servicers.threads import Threads
85class _MockCouchersContext:
86 @property
87 def headers(self):
88 return {}
90 def get_header(self, name):
91 return None
94class CookieMetadataPlugin(grpc.AuthMetadataPlugin):
95 """
96 Injects the right `cookie: couchers-sesh=...` header into the metadata
97 """
99 def __init__(self, token: str):
100 self.token = token
102 def __call__(self, context, callback) -> None:
103 callback((("cookie", f"couchers-sesh={self.token}"),), None)
106class MetadataKeeperInterceptor(grpc.UnaryUnaryClientInterceptor):
107 def __init__(self):
108 self.latest_headers = {}
110 def intercept_unary_unary(self, continuation, client_call_details, request):
111 call = continuation(client_call_details, request)
112 self.latest_headers = dict(call.initial_metadata())
113 self.latest_header_raw = call.initial_metadata()
114 return call
117class FakeRpcError(grpc.RpcError):
118 def __init__(self, code: grpc.StatusCode, details: str):
119 self._code = code
120 self._details = details
122 def code(self) -> grpc.StatusCode:
123 return self._code
125 def details(self) -> str:
126 return self._details
129class MockGrpcContext:
130 """
131 Pure mock of grpc.ServicerContext for testing.
132 """
134 def __init__(self):
135 self._initial_metadata = []
136 self._invocation_metadata: list[tuple[str, str]] = []
138 def abort(self, code: grpc.StatusCode, details: str) -> NoReturn:
139 raise FakeRpcError(code, details)
141 def invocation_metadata(self) -> list[tuple[str, str]]:
142 return self._invocation_metadata
144 def send_initial_metadata(self, metadata):
145 self._initial_metadata.extend(metadata)
148class FakeChannel:
149 """
150 Mock gRPC channel for testing that orchestrates context creation.
152 This holds the test state (token) and creates proper CouchersContext
153 instances when handlers are invoked.
154 """
156 def __init__(self, token: str | None = None, *, locale: str | None = None):
157 self.handlers: dict[str, Any] = {}
158 self._token = token
159 self._locale = locale or DEFAULT_LOCALE
160 self._pool = get_descriptor_pool()
162 def add_generic_rpc_handlers(self, generic_rpc_handlers: Any):
163 _validate_generic_rpc_handlers(generic_rpc_handlers)
164 self.handlers.update(generic_rpc_handlers[0]._method_handlers)
166 def unary_unary(self, method, request_serializer, response_deserializer):
167 handler = self.handlers[method]
169 def fake_handler(request):
170 auth_info = _try_get_and_update_user_details(
171 self._token,
172 is_api_key=False,
173 ip_address="127.0.0.1",
174 user_agent="Testing User-Agent",
175 sofa=None,
176 client_platform=None,
177 )
178 auth_level = find_auth_level(self._pool, method)
179 check_permissions(auth_info, auth_level)
181 # Do a full serialization cycle on the request and the
182 # response to catch accidental use of unserializable data.
183 request = handler.request_deserializer(request_serializer(request))
185 with session_scope() as session:
186 context = make_interactive_context(
187 grpc_context=MockGrpcContext(),
188 user_id=auth_info.user_id if auth_info else None,
189 is_api_key=False,
190 token=self._token if auth_info else None,
191 localization=LocalizationContext(
192 locale=(auth_info and auth_info.ui_language_preference) or self._locale,
193 timezone=ZoneInfo((auth_info and auth_info.timezone) or "Etc/UTC"),
194 ),
195 sofa="test_sofa_cookie_value",
196 )
198 response = handler.unary_unary(request, context, session)
200 return response_deserializer(handler.response_serializer(response))
202 return fake_handler
205@contextmanager
206def run_server(grpc_channel_options=(), token: str | None = None):
207 with futures.ThreadPoolExecutor(1) as executor:
208 if token:
209 call_creds = grpc.metadata_call_credentials(CookieMetadataPlugin(token))
210 creds = grpc.composite_channel_credentials(grpc.local_channel_credentials(), call_creds)
211 else:
212 creds = grpc.local_channel_credentials()
214 srv = grpc.server(executor, interceptors=[CouchersMiddlewareInterceptor()])
215 port = srv.add_secure_port("localhost:0", grpc.local_server_credentials())
216 srv.start()
218 try:
219 with grpc.secure_channel(f"localhost:{port}", creds, options=grpc_channel_options) as channel:
220 metadata_interceptor = MetadataKeeperInterceptor()
221 channel = grpc.intercept_channel(channel, metadata_interceptor)
222 yield srv, channel, metadata_interceptor
223 finally:
224 srv.stop(None).wait()
227# Sessions that start a real GRPC server.
228@contextmanager
229def auth_api_session(
230 grpc_channel_options=(),
231) -> Generator[tuple[auth_pb2_grpc.AuthStub, MetadataKeeperInterceptor]]:
232 """
233 Create an Auth API for testing
235 This needs to use the real server since it plays around with headers
236 """
237 with run_server(grpc_channel_options) as (server, channel, metadata_interceptor):
238 auth_pb2_grpc.add_AuthServicer_to_server(Auth(), server)
239 yield auth_pb2_grpc.AuthStub(channel), metadata_interceptor
242@contextmanager
243def real_api_session(token: str):
244 """
245 Create an API for testing, using TCP sockets, uses the token for auth
246 """
247 with run_server(token=token) as (server, channel, metadata_interceptor):
248 api_pb2_grpc.add_APIServicer_to_server(API(), server)
249 yield api_pb2_grpc.APIStub(channel)
252@contextmanager
253def real_admin_session(token: str):
254 """
255 Create an Admin service for testing, using TCP sockets, uses the token for auth
256 """
257 with run_server(token=token) as (server, channel, metadata_interceptor):
258 admin_pb2_grpc.add_AdminServicer_to_server(Admin(), server)
259 yield admin_pb2_grpc.AdminStub(channel)
262@contextmanager
263def real_editor_session(token: str):
264 """
265 Create an Editor service for testing, using TCP sockets, uses the token for auth
266 """
267 with run_server(token=token) as (server, channel, metadata_interceptor):
268 editor_pb2_grpc.add_EditorServicer_to_server(Editor(), server)
269 yield editor_pb2_grpc.EditorStub(channel)
272@contextmanager
273def real_moderation_session(token: str):
274 """
275 Create a Moderation service for testing, using TCP sockets, uses the token for auth
276 """
277 with run_server(token=token) as (server, channel, metadata_interceptor):
278 moderation_pb2_grpc.add_ModerationServicer_to_server(Moderation(), server)
279 yield moderation_pb2_grpc.ModerationStub(channel)
282@contextmanager
283def real_account_session(token: str):
284 """
285 Create an Account service for testing, using TCP sockets, uses the token for auth
286 """
287 with run_server(token=token) as (server, channel, metadata_interceptor):
288 account_pb2_grpc.add_AccountServicer_to_server(Account(), server)
289 yield account_pb2_grpc.AccountStub(channel)
292@contextmanager
293def real_jail_session(token: str):
294 """
295 Create a Jail service for testing, using TCP sockets, uses the token for auth
296 """
297 with run_server(token=token) as (server, channel, metadata_interceptor):
298 jail_pb2_grpc.add_JailServicer_to_server(Jail(), server)
299 yield jail_pb2_grpc.JailStub(channel)
302@contextmanager
303def real_stripe_session():
304 """
305 Create a Stripe service for testing, using TCP sockets
306 """
307 with run_server() as (server, channel, metadata_interceptor):
308 stripe_pb2_grpc.add_StripeServicer_to_server(Stripe(), server)
309 yield stripe_pb2_grpc.StripeStub(channel)
312@contextmanager
313def real_iris_session():
314 with run_server() as (server, channel, metadata_interceptor):
315 iris_pb2_grpc.add_IrisServicer_to_server(Iris(), server)
316 yield iris_pb2_grpc.IrisStub(channel)
319@contextmanager
320def real_bugs_session():
321 """
322 Bugs over a real server so requests can carry metadata (HTTP request headers)
323 and the response's initial metadata (HTTP response headers) can be asserted.
324 """
325 with run_server() as (server, channel, metadata_interceptor):
326 bugs_pb2_grpc.add_BugsServicer_to_server(Bugs(), server)
327 yield bugs_pb2_grpc.BugsStub(channel), metadata_interceptor
330@contextmanager
331def media_session(bearer_token: str):
332 """
333 Create a fresh Media API for testing, uses the bearer token for media auth
334 """
335 media_auth_interceptor = get_media_auth_interceptor(bearer_token)
337 with futures.ThreadPoolExecutor(1) as executor:
338 server = grpc.server(executor, interceptors=[media_auth_interceptor])
339 port = server.add_secure_port("localhost:0", grpc.local_server_credentials())
340 media_pb2_grpc.add_MediaServicer_to_server(Media(), server)
341 server.start()
343 call_creds = grpc.access_token_call_credentials(bearer_token)
344 comp_creds = grpc.composite_channel_credentials(grpc.local_channel_credentials(), call_creds)
346 try:
347 with grpc.secure_channel(f"localhost:{port}", comp_creds) as channel:
348 yield media_pb2_grpc.MediaStub(channel)
349 finally:
350 server.stop(None).wait()
353# Sessions that don't need to start a real GRPC server.
354# Note: these don't need to be context managers, but they are so that
355# we can switch to a real implementation if needed.
356@contextmanager
357def api_session(token: str):
358 """
359 Create an API for testing, uses the token for auth
360 """
361 channel = FakeChannel(token)
362 api_pb2_grpc.add_APIServicer_to_server(API(), channel)
363 yield api_pb2_grpc.APIStub(channel)
366@contextmanager
367def gis_session(token: str):
368 channel = FakeChannel(token)
369 gis_pb2_grpc.add_GISServicer_to_server(GIS(), channel)
370 yield gis_pb2_grpc.GISStub(channel)
373@contextmanager
374def public_session():
375 channel = FakeChannel()
376 public_pb2_grpc.add_PublicServicer_to_server(Public(), channel)
377 yield public_pb2_grpc.PublicStub(channel)
380@contextmanager
381def public_trips_session(token: str):
382 channel = FakeChannel(token)
383 public_trips_pb2_grpc.add_PublicTripsServicer_to_server(PublicTrips(), channel)
384 yield public_trips_pb2_grpc.PublicTripsStub(channel)
387@contextmanager
388def conversations_session(token: str):
389 """
390 Create a Conversations API for testing, uses the token for auth
391 """
392 channel = FakeChannel(token)
393 conversations_pb2_grpc.add_ConversationsServicer_to_server(Conversations(), channel)
394 yield conversations_pb2_grpc.ConversationsStub(channel)
397@contextmanager
398def requests_session(token: str):
399 """
400 Create a Requests API for testing, uses the token for auth
401 """
402 channel = FakeChannel(token)
403 requests_pb2_grpc.add_RequestsServicer_to_server(Requests(), channel)
404 yield requests_pb2_grpc.RequestsStub(channel)
407@contextmanager
408def threads_session(token: str):
409 channel = FakeChannel(token)
410 threads_pb2_grpc.add_ThreadsServicer_to_server(Threads(), channel)
411 yield threads_pb2_grpc.ThreadsStub(channel)
414@contextmanager
415def discussions_session(token: str):
416 channel = FakeChannel(token)
417 discussions_pb2_grpc.add_DiscussionsServicer_to_server(Discussions(), channel)
418 yield discussions_pb2_grpc.DiscussionsStub(channel)
421@contextmanager
422def donations_session(token: str):
423 channel = FakeChannel(token)
424 donations_pb2_grpc.add_DonationsServicer_to_server(Donations(), channel)
425 yield donations_pb2_grpc.DonationsStub(channel)
428@contextmanager
429def pages_session(token: str):
430 channel = FakeChannel(token)
431 pages_pb2_grpc.add_PagesServicer_to_server(Pages(), channel)
432 yield pages_pb2_grpc.PagesStub(channel)
435@contextmanager
436def communities_session(token: str):
437 channel = FakeChannel(token)
438 communities_pb2_grpc.add_CommunitiesServicer_to_server(Communities(), channel)
439 yield communities_pb2_grpc.CommunitiesStub(channel)
442@contextmanager
443def groups_session(token: str):
444 channel = FakeChannel(token)
445 groups_pb2_grpc.add_GroupsServicer_to_server(Groups(), channel)
446 yield groups_pb2_grpc.GroupsStub(channel)
449@contextmanager
450def blocking_session(token: str):
451 channel = FakeChannel(token)
452 blocking_pb2_grpc.add_BlockingServicer_to_server(Blocking(), channel)
453 yield blocking_pb2_grpc.BlockingStub(channel)
456@contextmanager
457def notifications_session(token: str):
458 channel = FakeChannel(token)
459 notifications_pb2_grpc.add_NotificationsServicer_to_server(Notifications(), channel)
460 yield notifications_pb2_grpc.NotificationsStub(channel)
463@contextmanager
464def account_session(token: str):
465 """
466 Create a Account API for testing, uses the token for auth
467 """
468 channel = FakeChannel(token)
469 account_pb2_grpc.add_AccountServicer_to_server(Account(), channel)
470 yield account_pb2_grpc.AccountStub(channel)
473@contextmanager
474def search_session(token: str):
475 """
476 Create a Search API for testing, uses the token for auth
477 """
478 channel = FakeChannel(token)
479 search_pb2_grpc.add_SearchServicer_to_server(Search(), channel)
480 yield search_pb2_grpc.SearchStub(channel)
483@contextmanager
484def references_session(token: str):
485 """
486 Create a References API for testing, uses the token for auth
487 """
488 channel = FakeChannel(token)
489 references_pb2_grpc.add_ReferencesServicer_to_server(References(), channel)
490 yield references_pb2_grpc.ReferencesStub(channel)
493@contextmanager
494def galleries_session(token: str):
495 """
496 Create a Galleries API for testing, uses the token for auth
497 """
498 channel = FakeChannel(token)
499 galleries_pb2_grpc.add_GalleriesServicer_to_server(Galleries(), channel)
500 yield galleries_pb2_grpc.GalleriesStub(channel)
503@contextmanager
504def reporting_session(token: str):
505 channel = FakeChannel(token)
506 reporting_pb2_grpc.add_ReportingServicer_to_server(Reporting(), channel)
507 yield reporting_pb2_grpc.ReportingStub(channel)
510@contextmanager
511def events_session(token: str):
512 channel = FakeChannel(token)
513 events_pb2_grpc.add_EventsServicer_to_server(Events(), channel)
514 yield events_pb2_grpc.EventsStub(channel)
517@contextmanager
518def postal_verification_session(token: str):
519 channel = FakeChannel(token)
520 postal_verification_pb2_grpc.add_PostalVerificationServicer_to_server(PostalVerification(), channel)
521 yield postal_verification_pb2_grpc.PostalVerificationStub(channel)
524@contextmanager
525def bugs_session(token: str | None = None):
526 channel = FakeChannel(token)
527 bugs_pb2_grpc.add_BugsServicer_to_server(Bugs(), channel)
528 yield bugs_pb2_grpc.BugsStub(channel)
531@contextmanager
532def resources_session(*, locale: str | None = None):
533 channel = FakeChannel(locale=locale)
534 resources_pb2_grpc.add_ResourcesServicer_to_server(Resources(), channel)
535 yield resources_pb2_grpc.ResourcesStub(channel)