Coverage for app/backend/src/tests/fixtures/sessions.py: 99%
302 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-22 18:39 +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 dashboard_pb2_grpc,
31 discussions_pb2_grpc,
32 donations_pb2_grpc,
33 editor_pb2_grpc,
34 events_pb2_grpc,
35 galleries_pb2_grpc,
36 gis_pb2_grpc,
37 groups_pb2_grpc,
38 iris_pb2_grpc,
39 jail_pb2_grpc,
40 media_pb2_grpc,
41 moderation_pb2_grpc,
42 notifications_pb2_grpc,
43 pages_pb2_grpc,
44 postal_verification_pb2_grpc,
45 public_pb2_grpc,
46 public_trips_pb2_grpc,
47 references_pb2_grpc,
48 reporting_pb2_grpc,
49 requests_pb2_grpc,
50 resources_pb2_grpc,
51 search_pb2_grpc,
52 stripe_pb2_grpc,
53 threads_pb2_grpc,
54)
55from couchers.servicers.account import Account, Iris
56from couchers.servicers.admin import Admin
57from couchers.servicers.api import API
58from couchers.servicers.auth import Auth
59from couchers.servicers.blocking import Blocking
60from couchers.servicers.bugs import Bugs
61from couchers.servicers.communities import Communities
62from couchers.servicers.conversations import Conversations
63from couchers.servicers.dashboard import Dashboard
64from couchers.servicers.discussions import Discussions
65from couchers.servicers.donations import Donations, Stripe
66from couchers.servicers.editor import Editor
67from couchers.servicers.events import Events
68from couchers.servicers.galleries import Galleries
69from couchers.servicers.gis import GIS
70from couchers.servicers.groups import Groups
71from couchers.servicers.jail import Jail
72from couchers.servicers.media import Media, get_media_auth_interceptor
73from couchers.servicers.moderation import Moderation
74from couchers.servicers.notifications import Notifications
75from couchers.servicers.pages import Pages
76from couchers.servicers.postal_verification import PostalVerification
77from couchers.servicers.public import Public
78from couchers.servicers.public_trips import PublicTrips
79from couchers.servicers.references import References
80from couchers.servicers.reporting import Reporting
81from couchers.servicers.requests import Requests
82from couchers.servicers.resources import Resources
83from couchers.servicers.search import Search
84from couchers.servicers.threads import Threads
87class _MockCouchersContext:
88 @property
89 def headers(self):
90 return {}
92 def get_header(self, name):
93 return None
96class CookieMetadataPlugin(grpc.AuthMetadataPlugin):
97 """
98 Injects the right `cookie: couchers-sesh=...` header into the metadata
99 """
101 def __init__(self, token: str):
102 self.token = token
104 def __call__(self, context, callback) -> None:
105 callback((("cookie", f"couchers-sesh={self.token}"),), None)
108class MetadataKeeperInterceptor(grpc.UnaryUnaryClientInterceptor):
109 def __init__(self):
110 self.latest_headers = {}
112 def intercept_unary_unary(self, continuation, client_call_details, request):
113 call = continuation(client_call_details, request)
114 self.latest_headers = dict(call.initial_metadata())
115 self.latest_header_raw = call.initial_metadata()
116 return call
119class FakeRpcError(grpc.RpcError):
120 def __init__(self, code: grpc.StatusCode, details: str):
121 self._code = code
122 self._details = details
124 def code(self) -> grpc.StatusCode:
125 return self._code
127 def details(self) -> str:
128 return self._details
131class MockGrpcContext:
132 """
133 Pure mock of grpc.ServicerContext for testing.
134 """
136 def __init__(self):
137 self._initial_metadata = []
138 self._invocation_metadata: list[tuple[str, str]] = []
140 def abort(self, code: grpc.StatusCode, details: str) -> NoReturn:
141 raise FakeRpcError(code, details)
143 def invocation_metadata(self) -> list[tuple[str, str]]:
144 return self._invocation_metadata
146 def send_initial_metadata(self, metadata):
147 self._initial_metadata.extend(metadata)
150class FakeChannel:
151 """
152 Mock gRPC channel for testing that orchestrates context creation.
154 This holds the test state (token) and creates proper CouchersContext
155 instances when handlers are invoked.
156 """
158 def __init__(self, token: str | None = None, *, locale: str | None = None):
159 self.handlers: dict[str, Any] = {}
160 self._token = token
161 self._locale = locale or DEFAULT_LOCALE
162 self._pool = get_descriptor_pool()
164 def add_generic_rpc_handlers(self, generic_rpc_handlers: Any):
165 _validate_generic_rpc_handlers(generic_rpc_handlers)
166 self.handlers.update(generic_rpc_handlers[0]._method_handlers)
168 def unary_unary(self, method, request_serializer, response_deserializer):
169 handler = self.handlers[method]
171 def fake_handler(request):
172 auth_info = _try_get_and_update_user_details(
173 self._token,
174 is_api_key=False,
175 ip_address="127.0.0.1",
176 user_agent="Testing User-Agent",
177 sofa=None,
178 client_platform=None,
179 )
180 auth_level = find_auth_level(self._pool, method)
181 check_permissions(auth_info, auth_level)
183 # Do a full serialization cycle on the request and the
184 # response to catch accidental use of unserializable data.
185 request = handler.request_deserializer(request_serializer(request))
187 with session_scope() as session:
188 context = make_interactive_context(
189 grpc_context=MockGrpcContext(),
190 user_id=auth_info.user_id if auth_info else None,
191 is_api_key=False,
192 token=self._token if auth_info else None,
193 localization=LocalizationContext(
194 locale=(auth_info and auth_info.ui_language_preference) or self._locale,
195 timezone=ZoneInfo((auth_info and auth_info.timezone) or "Etc/UTC"),
196 ),
197 sofa="test_sofa_cookie_value",
198 )
200 response = handler.unary_unary(request, context, session)
202 return response_deserializer(handler.response_serializer(response))
204 return fake_handler
207@contextmanager
208def run_server(grpc_channel_options=(), token: str | None = None):
209 with futures.ThreadPoolExecutor(1) as executor:
210 if token:
211 call_creds = grpc.metadata_call_credentials(CookieMetadataPlugin(token))
212 creds = grpc.composite_channel_credentials(grpc.local_channel_credentials(), call_creds)
213 else:
214 creds = grpc.local_channel_credentials()
216 srv = grpc.server(executor, interceptors=[CouchersMiddlewareInterceptor()])
217 port = srv.add_secure_port("localhost:0", grpc.local_server_credentials())
218 srv.start()
220 try:
221 with grpc.secure_channel(f"localhost:{port}", creds, options=grpc_channel_options) as channel:
222 metadata_interceptor = MetadataKeeperInterceptor()
223 channel = grpc.intercept_channel(channel, metadata_interceptor)
224 yield srv, channel, metadata_interceptor
225 finally:
226 srv.stop(None).wait()
229# Sessions that start a real GRPC server.
230@contextmanager
231def auth_api_session(
232 grpc_channel_options=(),
233) -> Generator[tuple[auth_pb2_grpc.AuthStub, MetadataKeeperInterceptor]]:
234 """
235 Create an Auth API for testing
237 This needs to use the real server since it plays around with headers
238 """
239 with run_server(grpc_channel_options) as (server, channel, metadata_interceptor):
240 auth_pb2_grpc.add_AuthServicer_to_server(Auth(), server)
241 yield auth_pb2_grpc.AuthStub(channel), metadata_interceptor
244@contextmanager
245def real_api_session(token: str):
246 """
247 Create an API for testing, using TCP sockets, uses the token for auth
248 """
249 with run_server(token=token) as (server, channel, metadata_interceptor):
250 api_pb2_grpc.add_APIServicer_to_server(API(), server)
251 yield api_pb2_grpc.APIStub(channel)
254@contextmanager
255def real_admin_session(token: str):
256 """
257 Create an Admin service for testing, using TCP sockets, uses the token for auth
258 """
259 with run_server(token=token) as (server, channel, metadata_interceptor):
260 admin_pb2_grpc.add_AdminServicer_to_server(Admin(), server)
261 yield admin_pb2_grpc.AdminStub(channel)
264@contextmanager
265def real_editor_session(token: str):
266 """
267 Create an Editor service for testing, using TCP sockets, uses the token for auth
268 """
269 with run_server(token=token) as (server, channel, metadata_interceptor):
270 editor_pb2_grpc.add_EditorServicer_to_server(Editor(), server)
271 yield editor_pb2_grpc.EditorStub(channel)
274@contextmanager
275def real_moderation_session(token: str):
276 """
277 Create a Moderation service for testing, using TCP sockets, uses the token for auth
278 """
279 with run_server(token=token) as (server, channel, metadata_interceptor):
280 moderation_pb2_grpc.add_ModerationServicer_to_server(Moderation(), server)
281 yield moderation_pb2_grpc.ModerationStub(channel)
284@contextmanager
285def real_account_session(token: str):
286 """
287 Create an Account service for testing, using TCP sockets, uses the token for auth
288 """
289 with run_server(token=token) as (server, channel, metadata_interceptor):
290 account_pb2_grpc.add_AccountServicer_to_server(Account(), server)
291 yield account_pb2_grpc.AccountStub(channel)
294@contextmanager
295def real_jail_session(token: str):
296 """
297 Create a Jail service for testing, using TCP sockets, uses the token for auth
298 """
299 with run_server(token=token) as (server, channel, metadata_interceptor):
300 jail_pb2_grpc.add_JailServicer_to_server(Jail(), server)
301 yield jail_pb2_grpc.JailStub(channel)
304@contextmanager
305def real_stripe_session():
306 """
307 Create a Stripe service for testing, using TCP sockets
308 """
309 with run_server() as (server, channel, metadata_interceptor):
310 stripe_pb2_grpc.add_StripeServicer_to_server(Stripe(), server)
311 yield stripe_pb2_grpc.StripeStub(channel)
314@contextmanager
315def real_iris_session():
316 with run_server() as (server, channel, metadata_interceptor):
317 iris_pb2_grpc.add_IrisServicer_to_server(Iris(), server)
318 yield iris_pb2_grpc.IrisStub(channel)
321@contextmanager
322def real_bugs_session():
323 """
324 Bugs over a real server so requests can carry metadata (HTTP request headers)
325 and the response's initial metadata (HTTP response headers) can be asserted.
326 """
327 with run_server() as (server, channel, metadata_interceptor):
328 bugs_pb2_grpc.add_BugsServicer_to_server(Bugs(), server)
329 yield bugs_pb2_grpc.BugsStub(channel), metadata_interceptor
332@contextmanager
333def media_session(bearer_token: str):
334 """
335 Create a fresh Media API for testing, uses the bearer token for media auth
336 """
337 media_auth_interceptor = get_media_auth_interceptor(bearer_token)
339 with futures.ThreadPoolExecutor(1) as executor:
340 server = grpc.server(executor, interceptors=[media_auth_interceptor])
341 port = server.add_secure_port("localhost:0", grpc.local_server_credentials())
342 media_pb2_grpc.add_MediaServicer_to_server(Media(), server)
343 server.start()
345 call_creds = grpc.access_token_call_credentials(bearer_token)
346 comp_creds = grpc.composite_channel_credentials(grpc.local_channel_credentials(), call_creds)
348 try:
349 with grpc.secure_channel(f"localhost:{port}", comp_creds) as channel:
350 yield media_pb2_grpc.MediaStub(channel)
351 finally:
352 server.stop(None).wait()
355# Sessions that don't need to start a real GRPC server.
356# Note: these don't need to be context managers, but they are so that
357# we can switch to a real implementation if needed.
358@contextmanager
359def api_session(token: str):
360 """
361 Create an API for testing, uses the token for auth
362 """
363 channel = FakeChannel(token)
364 api_pb2_grpc.add_APIServicer_to_server(API(), channel)
365 yield api_pb2_grpc.APIStub(channel)
368@contextmanager
369def gis_session(token: str):
370 channel = FakeChannel(token)
371 gis_pb2_grpc.add_GISServicer_to_server(GIS(), channel)
372 yield gis_pb2_grpc.GISStub(channel)
375@contextmanager
376def public_session():
377 channel = FakeChannel()
378 public_pb2_grpc.add_PublicServicer_to_server(Public(), channel)
379 yield public_pb2_grpc.PublicStub(channel)
382@contextmanager
383def public_trips_session(token: str):
384 channel = FakeChannel(token)
385 public_trips_pb2_grpc.add_PublicTripsServicer_to_server(PublicTrips(), channel)
386 yield public_trips_pb2_grpc.PublicTripsStub(channel)
389@contextmanager
390def conversations_session(token: str):
391 """
392 Create a Conversations API for testing, uses the token for auth
393 """
394 channel = FakeChannel(token)
395 conversations_pb2_grpc.add_ConversationsServicer_to_server(Conversations(), channel)
396 yield conversations_pb2_grpc.ConversationsStub(channel)
399@contextmanager
400def requests_session(token: str):
401 """
402 Create a Requests API for testing, uses the token for auth
403 """
404 channel = FakeChannel(token)
405 requests_pb2_grpc.add_RequestsServicer_to_server(Requests(), channel)
406 yield requests_pb2_grpc.RequestsStub(channel)
409@contextmanager
410def threads_session(token: str):
411 channel = FakeChannel(token)
412 threads_pb2_grpc.add_ThreadsServicer_to_server(Threads(), channel)
413 yield threads_pb2_grpc.ThreadsStub(channel)
416@contextmanager
417def discussions_session(token: str):
418 channel = FakeChannel(token)
419 discussions_pb2_grpc.add_DiscussionsServicer_to_server(Discussions(), channel)
420 yield discussions_pb2_grpc.DiscussionsStub(channel)
423@contextmanager
424def dashboard_session(token: str):
425 channel = FakeChannel(token)
426 dashboard_pb2_grpc.add_DashboardServicer_to_server(Dashboard(), channel)
427 yield dashboard_pb2_grpc.DashboardStub(channel)
430@contextmanager
431def donations_session(token: str):
432 channel = FakeChannel(token)
433 donations_pb2_grpc.add_DonationsServicer_to_server(Donations(), channel)
434 yield donations_pb2_grpc.DonationsStub(channel)
437@contextmanager
438def pages_session(token: str):
439 channel = FakeChannel(token)
440 pages_pb2_grpc.add_PagesServicer_to_server(Pages(), channel)
441 yield pages_pb2_grpc.PagesStub(channel)
444@contextmanager
445def communities_session(token: str):
446 channel = FakeChannel(token)
447 communities_pb2_grpc.add_CommunitiesServicer_to_server(Communities(), channel)
448 yield communities_pb2_grpc.CommunitiesStub(channel)
451@contextmanager
452def groups_session(token: str):
453 channel = FakeChannel(token)
454 groups_pb2_grpc.add_GroupsServicer_to_server(Groups(), channel)
455 yield groups_pb2_grpc.GroupsStub(channel)
458@contextmanager
459def blocking_session(token: str):
460 channel = FakeChannel(token)
461 blocking_pb2_grpc.add_BlockingServicer_to_server(Blocking(), channel)
462 yield blocking_pb2_grpc.BlockingStub(channel)
465@contextmanager
466def notifications_session(token: str):
467 channel = FakeChannel(token)
468 notifications_pb2_grpc.add_NotificationsServicer_to_server(Notifications(), channel)
469 yield notifications_pb2_grpc.NotificationsStub(channel)
472@contextmanager
473def account_session(token: str):
474 """
475 Create a Account API for testing, uses the token for auth
476 """
477 channel = FakeChannel(token)
478 account_pb2_grpc.add_AccountServicer_to_server(Account(), channel)
479 yield account_pb2_grpc.AccountStub(channel)
482@contextmanager
483def search_session(token: str):
484 """
485 Create a Search API for testing, uses the token for auth
486 """
487 channel = FakeChannel(token)
488 search_pb2_grpc.add_SearchServicer_to_server(Search(), channel)
489 yield search_pb2_grpc.SearchStub(channel)
492@contextmanager
493def references_session(token: str):
494 """
495 Create a References API for testing, uses the token for auth
496 """
497 channel = FakeChannel(token)
498 references_pb2_grpc.add_ReferencesServicer_to_server(References(), channel)
499 yield references_pb2_grpc.ReferencesStub(channel)
502@contextmanager
503def galleries_session(token: str):
504 """
505 Create a Galleries API for testing, uses the token for auth
506 """
507 channel = FakeChannel(token)
508 galleries_pb2_grpc.add_GalleriesServicer_to_server(Galleries(), channel)
509 yield galleries_pb2_grpc.GalleriesStub(channel)
512@contextmanager
513def reporting_session(token: str):
514 channel = FakeChannel(token)
515 reporting_pb2_grpc.add_ReportingServicer_to_server(Reporting(), channel)
516 yield reporting_pb2_grpc.ReportingStub(channel)
519@contextmanager
520def events_session(token: str):
521 channel = FakeChannel(token)
522 events_pb2_grpc.add_EventsServicer_to_server(Events(), channel)
523 yield events_pb2_grpc.EventsStub(channel)
526@contextmanager
527def postal_verification_session(token: str):
528 channel = FakeChannel(token)
529 postal_verification_pb2_grpc.add_PostalVerificationServicer_to_server(PostalVerification(), channel)
530 yield postal_verification_pb2_grpc.PostalVerificationStub(channel)
533@contextmanager
534def bugs_session(token: str | None = None):
535 channel = FakeChannel(token)
536 bugs_pb2_grpc.add_BugsServicer_to_server(Bugs(), channel)
537 yield bugs_pb2_grpc.BugsStub(channel)
540@contextmanager
541def resources_session(*, locale: str | None = None):
542 channel = FakeChannel(locale=locale)
543 resources_pb2_grpc.add_ResourcesServicer_to_server(Resources(), channel)
544 yield resources_pb2_grpc.ResourcesStub(channel)