Coverage for app/backend/src/tests/fixtures/sessions.py: 99%
311 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-19 15:47 +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.i18n import LocalizationContext
13from couchers.i18n.locales import DEFAULT_LOCALE
14from couchers.middleware.interceptors import (
15 CouchersMiddlewareInterceptor,
16 _try_get_and_update_user_details,
17 check_permissions,
18)
19from couchers.middleware.proto_annotations import get_proto_annotations
20from couchers.proto import (
21 account_pb2_grpc,
22 admin_pb2_grpc,
23 api_pb2_grpc,
24 auth_pb2_grpc,
25 blocking_pb2_grpc,
26 bugs_pb2_grpc,
27 communities_pb2_grpc,
28 conversations_pb2_grpc,
29 discussions_pb2_grpc,
30 donations_pb2_grpc,
31 editor_pb2_grpc,
32 events_pb2_grpc,
33 galleries_pb2_grpc,
34 gis_pb2_grpc,
35 groups_pb2_grpc,
36 iris_pb2_grpc,
37 jail_pb2_grpc,
38 media_pb2_grpc,
39 moderation_pb2_grpc,
40 notifications_pb2_grpc,
41 pages_pb2_grpc,
42 postal_verification_pb2_grpc,
43 public_pb2_grpc,
44 public_trips_pb2_grpc,
45 references_pb2_grpc,
46 reporting_pb2_grpc,
47 requests_pb2_grpc,
48 resources_pb2_grpc,
49 search_pb2_grpc,
50 stripe_pb2_grpc,
51 threads_pb2_grpc,
52)
53from couchers.servicers.account import Account, Iris
54from couchers.servicers.admin import Admin
55from couchers.servicers.api import API
56from couchers.servicers.auth import Auth
57from couchers.servicers.blocking import Blocking
58from couchers.servicers.bugs import Bugs
59from couchers.servicers.communities import Communities
60from couchers.servicers.conversations import Conversations
61from couchers.servicers.discussions import Discussions
62from couchers.servicers.donations import Donations, Stripe
63from couchers.servicers.editor import Editor
64from couchers.servicers.events import Events
65from couchers.servicers.galleries import Galleries
66from couchers.servicers.gis import GIS
67from couchers.servicers.groups import Groups
68from couchers.servicers.jail import Jail
69from couchers.servicers.media import Media, get_media_auth_interceptor
70from couchers.servicers.moderation import Moderation
71from couchers.servicers.notifications import Notifications
72from couchers.servicers.pages import Pages
73from couchers.servicers.postal_verification import PostalVerification
74from couchers.servicers.public import Public
75from couchers.servicers.public_trips import PublicTrips
76from couchers.servicers.references import References
77from couchers.servicers.reporting import Reporting
78from couchers.servicers.requests import Requests
79from couchers.servicers.resources import Resources
80from couchers.servicers.search import Search
81from couchers.servicers.threads import Threads
82from tests.fixtures import query_log
85class _MockCouchersContext:
86 @property
87 def headers(self):
88 return {}
90 def get_header(self, name):
91 return None
93 def set_cookies(self, cookies):
94 pass
97class CookieMetadataPlugin(grpc.AuthMetadataPlugin):
98 """
99 Injects the right `cookie: couchers-sesh=...` header into the metadata
100 """
102 def __init__(self, token: str):
103 self.token = token
105 def __call__(self, context, callback) -> None:
106 callback((("cookie", f"couchers-sesh={self.token}"),), None)
109class MetadataKeeperInterceptor(grpc.UnaryUnaryClientInterceptor):
110 def __init__(self):
111 self.latest_headers = {}
113 def intercept_unary_unary(self, continuation, client_call_details, request):
114 call = continuation(client_call_details, request)
115 self.latest_headers = dict(call.initial_metadata())
116 self.latest_header_raw = call.initial_metadata()
117 return call
120class QuerySpanInterceptor(grpc.ServerInterceptor):
121 """Attributes the queries a handler issues to the RPC being served.
123 Wraps only the returned handler, not the continuation, so the downstream middleware's auth lookups stay out of
124 the span. The wrapper runs on the thread that services the call, which is what query_log's thread-local needs.
125 """
127 def intercept_service(self, continuation, handler_call_details):
128 method = handler_call_details.method
129 # The downstream middleware authenticates inside its own intercept_service, so this span covers the auth
130 # lookups. They're a real per-call cost and worth seeing, just not conflated with the handler's own work.
131 with query_log.span("auth", method):
132 handler = continuation(handler_call_details)
133 if handler is None or handler.unary_unary is None: 133 ↛ 134line 133 didn't jump to line 134 because the condition on line 133 was never true
134 return handler
136 inner = handler.unary_unary
138 def wrapper(request, context):
139 with query_log.span("rpc", method):
140 return inner(request, context)
142 return grpc.unary_unary_rpc_method_handler(
143 wrapper,
144 request_deserializer=handler.request_deserializer,
145 response_serializer=handler.response_serializer,
146 )
149class FakeRpcError(grpc.RpcError):
150 def __init__(self, code: grpc.StatusCode, details: str):
151 self._code = code
152 self._details = details
154 def code(self) -> grpc.StatusCode:
155 return self._code
157 def details(self) -> str:
158 return self._details
161class MockGrpcContext:
162 """
163 Pure mock of grpc.ServicerContext for testing.
164 """
166 def __init__(self):
167 self._initial_metadata = []
168 self._invocation_metadata: list[tuple[str, str]] = []
170 def abort(self, code: grpc.StatusCode, details: str) -> NoReturn:
171 raise FakeRpcError(code, details)
173 def invocation_metadata(self) -> list[tuple[str, str]]:
174 return self._invocation_metadata
176 def send_initial_metadata(self, metadata):
177 self._initial_metadata.extend(metadata)
180class FakeChannel:
181 """
182 Mock gRPC channel for testing that orchestrates context creation.
184 This holds the test state (token) and creates proper CouchersContext
185 instances when handlers are invoked.
186 """
188 def __init__(self, token: str | None = None, *, locale: str | None = None):
189 self.handlers: dict[str, Any] = {}
190 self._token = token
191 self._locale = locale or DEFAULT_LOCALE
193 def add_generic_rpc_handlers(self, generic_rpc_handlers: Any):
194 _validate_generic_rpc_handlers(generic_rpc_handlers)
195 self.handlers.update(generic_rpc_handlers[0]._method_handlers)
197 def unary_unary(self, method, request_serializer, response_deserializer):
198 handler = self.handlers[method]
200 def fake_handler(request):
201 with query_log.span("auth", method):
202 auth_info = _try_get_and_update_user_details(
203 self._token,
204 is_api_key=False,
205 ip_address="127.0.0.1",
206 user_agent="Testing User-Agent",
207 sofa=None,
208 client_platform=None,
209 )
210 auth_level = get_proto_annotations().auth_level(method)
211 check_permissions(auth_info, auth_level)
213 # Do a full serialization cycle on the request and the
214 # response to catch accidental use of unserializable data.
215 request = handler.request_deserializer(request_serializer(request))
217 # Span covers the handler and its session but not the auth lookup above, matching the boundary
218 # couchers.middleware.perf uses in prod and the real-server sessions below.
219 with query_log.span("rpc", method), session_scope() as session:
220 context = make_interactive_context(
221 grpc_context=MockGrpcContext(),
222 user_id=auth_info.user_id if auth_info else None,
223 is_api_key=False,
224 token=self._token if auth_info else None,
225 localization=LocalizationContext(
226 locale=(auth_info and auth_info.ui_language_preference) or self._locale,
227 timezone=ZoneInfo((auth_info and auth_info.timezone) or "Etc/UTC"),
228 ),
229 sofa="test_sofa_cookie_value",
230 )
232 response = handler.unary_unary(request, context, session)
234 return response_deserializer(handler.response_serializer(response))
236 return fake_handler
239@contextmanager
240def run_server(grpc_channel_options=(), token: str | None = None):
241 with futures.ThreadPoolExecutor(1) as executor:
242 if token:
243 call_creds = grpc.metadata_call_credentials(CookieMetadataPlugin(token))
244 creds = grpc.composite_channel_credentials(grpc.local_channel_credentials(), call_creds)
245 else:
246 creds = grpc.local_channel_credentials()
248 srv = grpc.server(executor, interceptors=[QuerySpanInterceptor(), CouchersMiddlewareInterceptor()])
249 port = srv.add_secure_port("localhost:0", grpc.local_server_credentials())
250 srv.start()
252 try:
253 with grpc.secure_channel(f"localhost:{port}", creds, options=grpc_channel_options) as channel:
254 metadata_interceptor = MetadataKeeperInterceptor()
255 channel = grpc.intercept_channel(channel, metadata_interceptor)
256 yield srv, channel, metadata_interceptor
257 finally:
258 srv.stop(None).wait()
261# Sessions that start a real GRPC server.
262@contextmanager
263def auth_api_session(
264 grpc_channel_options=(),
265) -> Generator[tuple[auth_pb2_grpc.AuthStub, MetadataKeeperInterceptor]]:
266 """
267 Create an Auth API for testing
269 This needs to use the real server since it plays around with headers
270 """
271 with run_server(grpc_channel_options) as (server, channel, metadata_interceptor):
272 auth_pb2_grpc.add_AuthServicer_to_server(Auth(), server)
273 yield auth_pb2_grpc.AuthStub(channel), metadata_interceptor
276@contextmanager
277def real_api_session(token: str):
278 """
279 Create an API for testing, using TCP sockets, uses the token for auth
280 """
281 with run_server(token=token) as (server, channel, metadata_interceptor):
282 api_pb2_grpc.add_APIServicer_to_server(API(), server)
283 yield api_pb2_grpc.APIStub(channel)
286@contextmanager
287def real_admin_session(token: str):
288 """
289 Create an Admin service for testing, using TCP sockets, uses the token for auth
290 """
291 with run_server(token=token) as (server, channel, metadata_interceptor):
292 admin_pb2_grpc.add_AdminServicer_to_server(Admin(), server)
293 yield admin_pb2_grpc.AdminStub(channel)
296@contextmanager
297def real_editor_session(token: str):
298 """
299 Create an Editor service for testing, using TCP sockets, uses the token for auth
300 """
301 with run_server(token=token) as (server, channel, metadata_interceptor):
302 editor_pb2_grpc.add_EditorServicer_to_server(Editor(), server)
303 yield editor_pb2_grpc.EditorStub(channel)
306@contextmanager
307def real_moderation_session(token: str):
308 """
309 Create a Moderation service for testing, using TCP sockets, uses the token for auth
310 """
311 with run_server(token=token) as (server, channel, metadata_interceptor):
312 moderation_pb2_grpc.add_ModerationServicer_to_server(Moderation(), server)
313 yield moderation_pb2_grpc.ModerationStub(channel)
316@contextmanager
317def real_account_session(token: str):
318 """
319 Create an Account service for testing, using TCP sockets, uses the token for auth
320 """
321 with run_server(token=token) as (server, channel, metadata_interceptor):
322 account_pb2_grpc.add_AccountServicer_to_server(Account(), server)
323 yield account_pb2_grpc.AccountStub(channel)
326@contextmanager
327def real_jail_session(token: str):
328 """
329 Create a Jail service for testing, using TCP sockets, uses the token for auth
330 """
331 with run_server(token=token) as (server, channel, metadata_interceptor):
332 jail_pb2_grpc.add_JailServicer_to_server(Jail(), server)
333 yield jail_pb2_grpc.JailStub(channel)
336@contextmanager
337def real_stripe_session():
338 """
339 Create a Stripe service for testing, using TCP sockets
340 """
341 with run_server() as (server, channel, metadata_interceptor):
342 stripe_pb2_grpc.add_StripeServicer_to_server(Stripe(), server)
343 yield stripe_pb2_grpc.StripeStub(channel)
346@contextmanager
347def real_iris_session():
348 with run_server() as (server, channel, metadata_interceptor):
349 iris_pb2_grpc.add_IrisServicer_to_server(Iris(), server)
350 yield iris_pb2_grpc.IrisStub(channel)
353@contextmanager
354def real_bugs_session():
355 """
356 Bugs over a real server so requests can carry metadata (HTTP request headers)
357 and the response's initial metadata (HTTP response headers) can be asserted.
358 """
359 with run_server() as (server, channel, metadata_interceptor):
360 bugs_pb2_grpc.add_BugsServicer_to_server(Bugs(), server)
361 yield bugs_pb2_grpc.BugsStub(channel), metadata_interceptor
364@contextmanager
365def media_session(bearer_token: str):
366 """
367 Create a fresh Media API for testing, uses the bearer token for media auth
368 """
369 media_auth_interceptor = get_media_auth_interceptor(bearer_token)
371 with futures.ThreadPoolExecutor(1) as executor:
372 server = grpc.server(executor, interceptors=[media_auth_interceptor])
373 port = server.add_secure_port("localhost:0", grpc.local_server_credentials())
374 media_pb2_grpc.add_MediaServicer_to_server(Media(), server)
375 server.start()
377 call_creds = grpc.access_token_call_credentials(bearer_token)
378 comp_creds = grpc.composite_channel_credentials(grpc.local_channel_credentials(), call_creds)
380 try:
381 with grpc.secure_channel(f"localhost:{port}", comp_creds) as channel:
382 yield media_pb2_grpc.MediaStub(channel)
383 finally:
384 server.stop(None).wait()
387# Sessions that don't need to start a real GRPC server.
388# Note: these don't need to be context managers, but they are so that
389# we can switch to a real implementation if needed.
390@contextmanager
391def api_session(token: str):
392 """
393 Create an API for testing, uses the token for auth
394 """
395 channel = FakeChannel(token)
396 api_pb2_grpc.add_APIServicer_to_server(API(), channel)
397 yield api_pb2_grpc.APIStub(channel)
400@contextmanager
401def gis_session(token: str):
402 channel = FakeChannel(token)
403 gis_pb2_grpc.add_GISServicer_to_server(GIS(), channel)
404 yield gis_pb2_grpc.GISStub(channel)
407@contextmanager
408def public_session(token: str | None = None):
409 channel = FakeChannel(token)
410 public_pb2_grpc.add_PublicServicer_to_server(Public(), channel)
411 yield public_pb2_grpc.PublicStub(channel)
414@contextmanager
415def public_trips_session(token: str):
416 channel = FakeChannel(token)
417 public_trips_pb2_grpc.add_PublicTripsServicer_to_server(PublicTrips(), channel)
418 yield public_trips_pb2_grpc.PublicTripsStub(channel)
421@contextmanager
422def conversations_session(token: str):
423 """
424 Create a Conversations API for testing, uses the token for auth
425 """
426 channel = FakeChannel(token)
427 conversations_pb2_grpc.add_ConversationsServicer_to_server(Conversations(), channel)
428 yield conversations_pb2_grpc.ConversationsStub(channel)
431@contextmanager
432def requests_session(token: str):
433 """
434 Create a Requests API for testing, uses the token for auth
435 """
436 channel = FakeChannel(token)
437 requests_pb2_grpc.add_RequestsServicer_to_server(Requests(), channel)
438 yield requests_pb2_grpc.RequestsStub(channel)
441@contextmanager
442def threads_session(token: str):
443 channel = FakeChannel(token)
444 threads_pb2_grpc.add_ThreadsServicer_to_server(Threads(), channel)
445 yield threads_pb2_grpc.ThreadsStub(channel)
448@contextmanager
449def discussions_session(token: str):
450 channel = FakeChannel(token)
451 discussions_pb2_grpc.add_DiscussionsServicer_to_server(Discussions(), channel)
452 yield discussions_pb2_grpc.DiscussionsStub(channel)
455@contextmanager
456def donations_session(token: str):
457 channel = FakeChannel(token)
458 donations_pb2_grpc.add_DonationsServicer_to_server(Donations(), channel)
459 yield donations_pb2_grpc.DonationsStub(channel)
462@contextmanager
463def pages_session(token: str):
464 channel = FakeChannel(token)
465 pages_pb2_grpc.add_PagesServicer_to_server(Pages(), channel)
466 yield pages_pb2_grpc.PagesStub(channel)
469@contextmanager
470def communities_session(token: str):
471 channel = FakeChannel(token)
472 communities_pb2_grpc.add_CommunitiesServicer_to_server(Communities(), channel)
473 yield communities_pb2_grpc.CommunitiesStub(channel)
476@contextmanager
477def groups_session(token: str):
478 channel = FakeChannel(token)
479 groups_pb2_grpc.add_GroupsServicer_to_server(Groups(), channel)
480 yield groups_pb2_grpc.GroupsStub(channel)
483@contextmanager
484def blocking_session(token: str):
485 channel = FakeChannel(token)
486 blocking_pb2_grpc.add_BlockingServicer_to_server(Blocking(), channel)
487 yield blocking_pb2_grpc.BlockingStub(channel)
490@contextmanager
491def notifications_session(token: str):
492 channel = FakeChannel(token)
493 notifications_pb2_grpc.add_NotificationsServicer_to_server(Notifications(), channel)
494 yield notifications_pb2_grpc.NotificationsStub(channel)
497@contextmanager
498def account_session(token: str):
499 """
500 Create a Account API for testing, uses the token for auth
501 """
502 channel = FakeChannel(token)
503 account_pb2_grpc.add_AccountServicer_to_server(Account(), channel)
504 yield account_pb2_grpc.AccountStub(channel)
507@contextmanager
508def search_session(token: str):
509 """
510 Create a Search API for testing, uses the token for auth
511 """
512 channel = FakeChannel(token)
513 search_pb2_grpc.add_SearchServicer_to_server(Search(), channel)
514 yield search_pb2_grpc.SearchStub(channel)
517@contextmanager
518def references_session(token: str):
519 """
520 Create a References API for testing, uses the token for auth
521 """
522 channel = FakeChannel(token)
523 references_pb2_grpc.add_ReferencesServicer_to_server(References(), channel)
524 yield references_pb2_grpc.ReferencesStub(channel)
527@contextmanager
528def galleries_session(token: str):
529 """
530 Create a Galleries API for testing, uses the token for auth
531 """
532 channel = FakeChannel(token)
533 galleries_pb2_grpc.add_GalleriesServicer_to_server(Galleries(), channel)
534 yield galleries_pb2_grpc.GalleriesStub(channel)
537@contextmanager
538def reporting_session(token: str):
539 channel = FakeChannel(token)
540 reporting_pb2_grpc.add_ReportingServicer_to_server(Reporting(), channel)
541 yield reporting_pb2_grpc.ReportingStub(channel)
544@contextmanager
545def events_session(token: str):
546 channel = FakeChannel(token)
547 events_pb2_grpc.add_EventsServicer_to_server(Events(), channel)
548 yield events_pb2_grpc.EventsStub(channel)
551@contextmanager
552def postal_verification_session(token: str):
553 channel = FakeChannel(token)
554 postal_verification_pb2_grpc.add_PostalVerificationServicer_to_server(PostalVerification(), channel)
555 yield postal_verification_pb2_grpc.PostalVerificationStub(channel)
558@contextmanager
559def bugs_session(token: str | None = None):
560 channel = FakeChannel(token)
561 bugs_pb2_grpc.add_BugsServicer_to_server(Bugs(), channel)
562 yield bugs_pb2_grpc.BugsStub(channel)
565@contextmanager
566def resources_session(*, locale: str | None = None):
567 channel = FakeChannel(locale=locale)
568 resources_pb2_grpc.add_ResourcesServicer_to_server(Resources(), channel)
569 yield resources_pb2_grpc.ResourcesStub(channel)