Coverage for app/backend/src/tests/test_communities.py: 100%
724 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-23 04:10 +0000
1from datetime import timedelta
3import grpc
4import pytest
5from geoalchemy2 import WKBElement
6from google.protobuf import empty_pb2, wrappers_pb2
7from sqlalchemy import select
8from sqlalchemy.orm import Session
10from couchers.db import is_user_in_node_geography, session_scope
11from couchers.helpers.clusters import CHILD_NODE_TYPE
12from couchers.materialized_views import refresh_materialized_views
13from couchers.models import (
14 Cluster,
15 ClusterRole,
16 ClusterSubscription,
17 Discussion,
18 EventOccurrence,
19 Node,
20 Page,
21 PageType,
22 PageVersion,
23 SignupFlow,
24 Thread,
25 User,
26)
27from couchers.proto import api_pb2, auth_pb2, communities_pb2, discussions_pb2, events_pb2, pages_pb2
28from couchers.tasks import enforce_community_memberships
29from couchers.utils import Timestamp_from_datetime, create_coordinate, create_polygon_lat_lng, now, to_multi
30from tests.fixtures.db import generate_user, get_user_id_and_token
31from tests.fixtures.misc import Moderator
32from tests.fixtures.sessions import (
33 auth_api_session,
34 communities_session,
35 discussions_session,
36 events_session,
37 pages_session,
38)
39from tests.test_auth import get_session_cookie_tokens
42@pytest.fixture(autouse=True)
43def _(testconfig):
44 pass
47# For testing purposes, restrict ourselves to a 1D-world, consisting of "intervals" that have width 2, and coordinates
48# that are points at (x, 1).
49# we'll stick to EPSG4326, even though it's not ideal, so don't use too large values, but it's around the equator, so
50# mostly fine
53def create_1d_polygon(lb: int, ub: int) -> WKBElement:
54 # given a lower bound and upper bound on x, creates the given interval
55 return create_polygon_lat_lng([[lb, 0], [lb, 2], [ub, 2], [ub, 0], [lb, 0]])
58def create_1d_point(x: int) -> WKBElement:
59 return create_coordinate(x, 1)
62def create_community(
63 session: Session,
64 interval_lb: int,
65 interval_ub: int,
66 name: str,
67 admins: list[User],
68 extra_members: list[User],
69 parent: Node | None,
70) -> Node:
71 node_type = CHILD_NODE_TYPE[parent.node_type if parent else None]
72 node = Node(
73 geom=to_multi(create_1d_polygon(interval_lb, interval_ub)),
74 parent_node_id=parent.id if parent else None,
75 node_type=node_type,
76 )
77 session.add(node)
78 session.flush()
79 cluster = Cluster(
80 name=f"{name}",
81 description=f"Description for {name}",
82 parent_node_id=node.id,
83 is_official_cluster=True,
84 )
85 session.add(cluster)
86 session.flush()
87 thread = Thread()
88 session.add(thread)
89 session.flush()
90 main_page = Page(
91 parent_node_id=cluster.parent_node_id,
92 creator_user_id=admins[0].id,
93 owner_cluster_id=cluster.id,
94 type=PageType.main_page,
95 thread_id=thread.id,
96 )
97 session.add(main_page)
98 session.flush()
99 page_version = PageVersion(
100 page_id=main_page.id,
101 editor_user_id=admins[0].id,
102 title=f"Main page for the {name} community",
103 content="There is nothing here yet...",
104 )
105 session.add(page_version)
106 for admin in admins:
107 cluster.cluster_subscriptions.append(
108 ClusterSubscription(
109 user_id=admin.id,
110 cluster_id=cluster.id,
111 role=ClusterRole.admin,
112 )
113 )
114 for member in extra_members:
115 cluster.cluster_subscriptions.append(
116 ClusterSubscription(
117 user_id=member.id,
118 cluster_id=cluster.id,
119 role=ClusterRole.member,
120 )
121 )
122 session.commit()
123 # other members will be added by enforce_community_memberships()
124 return node
127def create_group(
128 session: Session, name: str, admins: list[User], members: list[User], parent_community: Node | None
129) -> Cluster:
130 assert parent_community is not None
131 cluster = Cluster(
132 name=f"{name}",
133 description=f"Description for {name}",
134 parent_node_id=parent_community.id,
135 )
136 session.add(cluster)
137 session.flush()
138 thread = Thread()
139 session.add(thread)
140 session.flush()
141 main_page = Page(
142 parent_node_id=cluster.parent_node_id,
143 creator_user_id=admins[0].id,
144 owner_cluster_id=cluster.id,
145 type=PageType.main_page,
146 thread_id=thread.id,
147 )
148 session.add(main_page)
149 session.flush()
150 page_version = PageVersion(
151 page_id=main_page.id,
152 editor_user_id=admins[0].id,
153 title=f"Main page for the {name} community",
154 content="There is nothing here yet...",
155 )
156 session.add(page_version)
157 for admin in admins:
158 cluster.cluster_subscriptions.append(
159 ClusterSubscription(
160 user_id=admin.id,
161 cluster_id=cluster.id,
162 role=ClusterRole.admin,
163 )
164 )
165 for member in members:
166 cluster.cluster_subscriptions.append(
167 ClusterSubscription(
168 user_id=member.id,
169 cluster_id=cluster.id,
170 role=ClusterRole.member,
171 )
172 )
173 session.commit()
174 return cluster
177def create_place(token: str, title: str, content: str, address: str, x: float) -> None:
178 with pages_session(token) as api:
179 api.CreatePlace(
180 pages_pb2.CreatePlaceReq(
181 title=title,
182 content=content,
183 address=address,
184 location=pages_pb2.Coordinate(
185 lat=x,
186 lng=1,
187 ),
188 )
189 )
192def create_discussion(token: str, community_id: int | None, group_id: int | None, title: str, content: str) -> None:
193 # set group_id or community_id to None
194 with discussions_session(token) as api:
195 api.CreateDiscussion(
196 discussions_pb2.CreateDiscussionReq(
197 title=title,
198 content=content,
199 owner_community_id=community_id,
200 owner_group_id=group_id,
201 )
202 )
205def create_event(
206 token: str, community_id: int | None, group_id: int | None, title: str, content: str, start_td: timedelta
207) -> None:
208 with events_session(token) as api:
209 res = api.CreateEvent(
210 events_pb2.CreateEventReq(
211 title=title,
212 content=content,
213 location=events_pb2.EventLocation(
214 address="Near Null Island",
215 lat=0.1,
216 lng=0.2,
217 ),
218 start_time=Timestamp_from_datetime(now() + start_td),
219 end_time=Timestamp_from_datetime(now() + start_td + timedelta(hours=2)),
220 timezone="UTC",
221 )
222 )
223 api.TransferEvent(
224 events_pb2.TransferEventReq(
225 event_id=res.event_id,
226 new_owner_community_id=community_id,
227 new_owner_group_id=group_id,
228 )
229 )
232def get_community_id(session: Session, community_name: str) -> int:
233 return session.execute(
234 select(Cluster.parent_node_id).where(Cluster.is_official_cluster).where(Cluster.name == community_name)
235 ).scalar_one()
238def get_group_id(session: Session, group_name: str) -> int:
239 return session.execute(
240 select(Cluster.id).where(~Cluster.is_official_cluster).where(Cluster.name == group_name)
241 ).scalar_one()
244@pytest.fixture(scope="class")
245def testing_communities(db_class, testconfig):
246 user1, token1 = generate_user(username="user1", geom=create_1d_point(1), geom_radius=0.1)
247 user2, token2 = generate_user(username="user2", geom=create_1d_point(2), geom_radius=0.1)
248 user3, token3 = generate_user(username="user3", geom=create_1d_point(3), geom_radius=0.1)
249 user4, token4 = generate_user(username="user4", geom=create_1d_point(8), geom_radius=0.1)
250 user5, token5 = generate_user(username="user5", geom=create_1d_point(6), geom_radius=0.1)
251 user6, token6 = generate_user(username="user6", geom=create_1d_point(65), geom_radius=0.1)
252 user7, token7 = generate_user(username="user7", geom=create_1d_point(80), geom_radius=0.1)
253 user8, token8 = generate_user(username="user8", geom=create_1d_point(51), geom_radius=0.1)
255 with session_scope() as session:
256 w = create_community(session, 0, 100, "Global", [user1, user3, user7], [], None)
257 c2 = create_community(session, 52, 100, "Country 2", [user6, user7], [], w)
258 c2r1 = create_community(session, 52, 71, "Country 2, Region 1", [user6], [user8], c2)
259 c2r1c1 = create_community(session, 53, 70, "Country 2, Region 1, City 1", [user8], [], c2r1)
260 c1 = create_community(session, 0, 50, "Country 1", [user1, user2], [], w)
261 c1r1 = create_community(session, 0, 10, "Country 1, Region 1", [user1, user2], [], c1)
262 c1r1c1 = create_community(session, 0, 5, "Country 1, Region 1, City 1", [user2], [], c1r1)
263 c1r1c2 = create_community(session, 7, 10, "Country 1, Region 1, City 2", [user4, user5], [user2], c1r1)
264 c1r2 = create_community(session, 20, 25, "Country 1, Region 2", [user2], [], c1)
265 c1r2c1 = create_community(session, 21, 23, "Country 1, Region 2, City 1", [user2], [], c1r2)
267 h = create_group(session, "Hitchhikers", [user1, user2], [user5, user8], w)
268 create_group(session, "Country 1, Region 1, Foodies", [user1], [user2, user4], c1r1)
269 create_group(session, "Country 1, Region 1, Skaters", [user2], [user1], c1r1)
270 create_group(session, "Country 1, Region 2, Foodies", [user2], [user4, user5], c1r2)
271 create_group(session, "Country 2, Region 1, Foodies", [user6], [user7], c2r1)
273 w_id = w.id
274 c1r1c2_id = c1r1c2.id
275 h_id = h.id
276 c1_id = c1.id
278 create_discussion(token1, w_id, None, "Discussion title 1", "Discussion content 1")
279 create_discussion(token3, w_id, None, "Discussion title 2", "Discussion content 2")
280 create_discussion(token3, w_id, None, "Discussion title 3", "Discussion content 3")
281 create_discussion(token3, w_id, None, "Discussion title 4", "Discussion content 4")
282 create_discussion(token3, w_id, None, "Discussion title 5", "Discussion content 5")
283 create_discussion(token3, w_id, None, "Discussion title 6", "Discussion content 6")
284 create_discussion(token4, c1r1c2_id, None, "Discussion title 7", "Discussion content 7")
285 create_discussion(token5, None, h_id, "Discussion title 8", "Discussion content 8")
286 create_discussion(token1, None, h_id, "Discussion title 9", "Discussion content 9")
287 create_discussion(token2, None, h_id, "Discussion title 10", "Discussion content 10")
288 create_discussion(token3, None, h_id, "Discussion title 11", "Discussion content 11")
289 create_discussion(token4, None, h_id, "Discussion title 12", "Discussion content 12")
290 create_discussion(token5, None, h_id, "Discussion title 13", "Discussion content 13")
291 create_discussion(token8, None, h_id, "Discussion title 14", "Discussion content 14")
293 create_event(token3, c1_id, None, "Event title 1", "Event content 1", timedelta(hours=1))
294 create_event(token1, c1_id, None, "Event title 2", "Event content 2", timedelta(hours=2))
295 create_event(token3, c1_id, None, "Event title 3", "Event content 3", timedelta(hours=3))
296 create_event(token1, c1_id, None, "Event title 4", "Event content 4", timedelta(hours=4))
297 create_event(token3, c1_id, None, "Event title 5", "Event content 5", timedelta(hours=5))
298 create_event(token1, c1_id, None, "Event title 6", "Event content 6", timedelta(hours=6))
299 create_event(token2, None, h_id, "Event title 7", "Event content 7", timedelta(hours=7))
300 create_event(token2, None, h_id, "Event title 8", "Event content 8", timedelta(hours=8))
301 create_event(token2, None, h_id, "Event title 9", "Event content 9", timedelta(hours=9))
302 create_event(token2, None, h_id, "Event title 10", "Event content 10", timedelta(hours=10))
303 create_event(token2, None, h_id, "Event title 11", "Event content 11", timedelta(hours=11))
304 create_event(token2, None, h_id, "Event title 12", "Event content 12", timedelta(hours=12))
306 # Approve all events for visibility (UMS starts events as SHADOWED)
307 mod_user, mod_token = generate_user(is_superuser=True)
308 mod = Moderator(mod_user, mod_token)
309 with session_scope() as session:
310 occurrence_ids = session.execute(select(EventOccurrence.id)).scalars().all()
311 discussion_ids = session.execute(select(Discussion.id)).scalars().all()
312 for oid in occurrence_ids:
313 mod.approve_event_occurrence(oid)
314 for did in discussion_ids:
315 mod.approve_discussion(did)
317 enforce_community_memberships()
319 create_place(token1, "Country 1, Region 1, Attraction", "Place content", "Somewhere in c1r1", 6)
320 create_place(token2, "Country 1, Region 1, City 1, Attraction 1", "Place content", "Somewhere in c1r1c1", 3)
321 create_place(token2, "Country 1, Region 1, City 1, Attraction 2", "Place content", "Somewhere in c1r1c1", 4)
322 create_place(token8, "Global, Attraction", "Place content", "Somewhere in w", 51.5)
323 create_place(token6, "Country 2, Region 1, Attraction", "Place content", "Somewhere in c2r1", 59)
325 refresh_materialized_views(empty_pb2.Empty())
327 yield
330class TestCommunities:
331 @staticmethod
332 def test_GetCommunity(testing_communities):
333 with session_scope() as session:
334 user1_id, token1 = get_user_id_and_token(session, "user1")
335 user2_id, token2 = get_user_id_and_token(session, "user2")
336 user6_id, token6 = get_user_id_and_token(session, "user6")
337 w_id = get_community_id(session, "Global")
338 c1_id = get_community_id(session, "Country 1")
339 c1r1_id = get_community_id(session, "Country 1, Region 1")
340 c1r1c1_id = get_community_id(session, "Country 1, Region 1, City 1")
341 c2_id = get_community_id(session, "Country 2")
343 with communities_session(token2) as api:
344 res = api.GetCommunity(
345 communities_pb2.GetCommunityReq(
346 community_id=w_id,
347 )
348 )
349 assert res.name == "Global"
350 assert res.slug == "global"
351 assert res.description == "Description for Global"
352 assert len(res.parents) == 1
353 assert res.parents[0].HasField("community")
354 assert res.parents[0].community.community_id == w_id
355 assert res.parents[0].community.name == "Global"
356 assert res.parents[0].community.slug == "global"
357 assert res.parents[0].community.description == "Description for Global"
358 assert res.main_page.type == pages_pb2.PAGE_TYPE_MAIN_PAGE
359 assert res.main_page.slug == "main-page-for-the-global-community"
360 assert res.main_page.last_editor_user_id == user1_id
361 assert res.main_page.creator_user_id == user1_id
362 assert res.main_page.owner_community_id == w_id
363 assert res.main_page.title == "Main page for the Global community"
364 assert res.main_page.content == "There is nothing here yet..."
365 assert not res.main_page.can_edit
366 assert not res.main_page.can_moderate
367 assert res.main_page.editor_user_ids == [user1_id]
368 assert res.member
369 assert not res.admin
370 assert res.member_count == 8
371 assert res.admin_count == 3
373 res = api.GetCommunity(
374 communities_pb2.GetCommunityReq(
375 community_id=c1r1c1_id,
376 )
377 )
378 assert res.community_id == c1r1c1_id
379 assert res.name == "Country 1, Region 1, City 1"
380 assert res.slug == "country-1-region-1-city-1"
381 assert res.description == "Description for Country 1, Region 1, City 1"
382 assert len(res.parents) == 4
383 assert res.parents[0].HasField("community")
384 assert res.parents[0].community.community_id == w_id
385 assert res.parents[0].community.name == "Global"
386 assert res.parents[0].community.slug == "global"
387 assert res.parents[0].community.description == "Description for Global"
388 assert res.parents[1].HasField("community")
389 assert res.parents[1].community.community_id == c1_id
390 assert res.parents[1].community.name == "Country 1"
391 assert res.parents[1].community.slug == "country-1"
392 assert res.parents[1].community.description == "Description for Country 1"
393 assert res.parents[2].HasField("community")
394 assert res.parents[2].community.community_id == c1r1_id
395 assert res.parents[2].community.name == "Country 1, Region 1"
396 assert res.parents[2].community.slug == "country-1-region-1"
397 assert res.parents[2].community.description == "Description for Country 1, Region 1"
398 assert res.parents[3].HasField("community")
399 assert res.parents[3].community.community_id == c1r1c1_id
400 assert res.parents[3].community.name == "Country 1, Region 1, City 1"
401 assert res.parents[3].community.slug == "country-1-region-1-city-1"
402 assert res.parents[3].community.description == "Description for Country 1, Region 1, City 1"
403 assert res.main_page.type == pages_pb2.PAGE_TYPE_MAIN_PAGE
404 assert res.main_page.slug == "main-page-for-the-country-1-region-1-city-1-community"
405 assert res.main_page.last_editor_user_id == user2_id
406 assert res.main_page.creator_user_id == user2_id
407 assert res.main_page.owner_community_id == c1r1c1_id
408 assert res.main_page.title == "Main page for the Country 1, Region 1, City 1 community"
409 assert res.main_page.content == "There is nothing here yet..."
410 assert res.main_page.can_edit
411 assert res.main_page.can_moderate
412 assert res.main_page.editor_user_ids == [user2_id]
413 assert res.member
414 assert res.admin
415 assert res.member_count == 3
416 assert res.admin_count == 1
418 res = api.GetCommunity(
419 communities_pb2.GetCommunityReq(
420 community_id=c2_id,
421 )
422 )
423 assert res.community_id == c2_id
424 assert res.name == "Country 2"
425 assert res.slug == "country-2"
426 assert res.description == "Description for Country 2"
427 assert len(res.parents) == 2
428 assert res.parents[0].HasField("community")
429 assert res.parents[0].community.community_id == w_id
430 assert res.parents[0].community.name == "Global"
431 assert res.parents[0].community.slug == "global"
432 assert res.parents[0].community.description == "Description for Global"
433 assert res.parents[1].HasField("community")
434 assert res.parents[1].community.community_id == c2_id
435 assert res.parents[1].community.name == "Country 2"
436 assert res.parents[1].community.slug == "country-2"
437 assert res.parents[1].community.description == "Description for Country 2"
438 assert res.main_page.type == pages_pb2.PAGE_TYPE_MAIN_PAGE
439 assert res.main_page.slug == "main-page-for-the-country-2-community"
440 assert res.main_page.last_editor_user_id == user6_id
441 assert res.main_page.creator_user_id == user6_id
442 assert res.main_page.owner_community_id == c2_id
443 assert res.main_page.title == "Main page for the Country 2 community"
444 assert res.main_page.content == "There is nothing here yet..."
445 assert not res.main_page.can_edit
446 assert not res.main_page.can_moderate
447 assert res.main_page.editor_user_ids == [user6_id]
448 assert not res.member
449 assert not res.admin
450 assert res.member_count == 2
451 assert res.admin_count == 2
453 @staticmethod
454 def test_ListCommunities(testing_communities):
455 with session_scope() as session:
456 user1_id, token1 = get_user_id_and_token(session, "user1")
457 c1_id = get_community_id(session, "Country 1")
458 c1r1_id = get_community_id(session, "Country 1, Region 1")
459 c1r2_id = get_community_id(session, "Country 1, Region 2")
461 with communities_session(token1) as api:
462 res = api.ListCommunities(
463 communities_pb2.ListCommunitiesReq(
464 community_id=c1_id,
465 )
466 )
467 assert [c.community_id for c in res.communities] == [c1r1_id, c1r2_id]
469 @staticmethod
470 def test_ListCommunities_all(testing_communities):
471 with session_scope() as session:
472 user1_id, token1 = get_user_id_and_token(session, "user1")
473 w_id = get_community_id(session, "Global")
474 c1_id = get_community_id(session, "Country 1")
475 c1r1_id = get_community_id(session, "Country 1, Region 1")
476 c1r1c1_id = get_community_id(session, "Country 1, Region 1, City 1")
477 c1r1c2_id = get_community_id(session, "Country 1, Region 1, City 2")
478 c1r2_id = get_community_id(session, "Country 1, Region 2")
479 c1r2c1_id = get_community_id(session, "Country 1, Region 2, City 1")
480 c2_id = get_community_id(session, "Country 2")
481 c2r1_id = get_community_id(session, "Country 2, Region 1")
482 c2r1c1_id = get_community_id(session, "Country 2, Region 1, City 1")
484 # Fetch all communities ordered by name
485 with communities_session(token1) as api:
486 res = api.ListCommunities(
487 communities_pb2.ListCommunitiesReq(
488 page_size=5,
489 )
490 )
491 assert [c.community_id for c in res.communities] == [c1_id, c1r1_id, c1r1c1_id, c1r1c2_id, c1r2_id]
492 res = api.ListCommunities(
493 communities_pb2.ListCommunitiesReq(
494 page_size=2,
495 page_token=res.next_page_token,
496 )
497 )
498 assert [c.community_id for c in res.communities] == [c1r2c1_id, c2_id]
499 res = api.ListCommunities(
500 communities_pb2.ListCommunitiesReq(
501 page_size=5,
502 page_token=res.next_page_token,
503 )
504 )
505 assert [c.community_id for c in res.communities] == [c2r1_id, c2r1c1_id, w_id]
507 @staticmethod
508 def test_ListUserCommunities(testing_communities):
509 with session_scope() as session:
510 user2_id, token2 = get_user_id_and_token(session, "user2")
511 w_id = get_community_id(session, "Global")
512 c1_id = get_community_id(session, "Country 1")
513 c1r1_id = get_community_id(session, "Country 1, Region 1")
514 c1r1c1_id = get_community_id(session, "Country 1, Region 1, City 1")
515 c1r1c2_id = get_community_id(session, "Country 1, Region 1, City 2")
516 c1r2_id = get_community_id(session, "Country 1, Region 2")
517 c1r2c1_id = get_community_id(session, "Country 1, Region 2, City 1")
519 # Fetch user2's communities from user2's account
520 with communities_session(token2) as api:
521 res = api.ListUserCommunities(communities_pb2.ListUserCommunitiesReq())
522 assert [c.community_id for c in res.communities] == [
523 c1r1c1_id,
524 c1r1c2_id,
525 c1r2c1_id,
526 c1r1_id,
527 c1r2_id,
528 c1_id,
529 w_id,
530 ]
532 # paginated, with pages crossing node type boundaries
533 res = api.ListUserCommunities(communities_pb2.ListUserCommunitiesReq(page_size=2))
534 assert [c.community_id for c in res.communities] == [c1r1c1_id, c1r1c2_id]
535 res = api.ListUserCommunities(
536 communities_pb2.ListUserCommunitiesReq(page_size=2, page_token=res.next_page_token)
537 )
538 assert [c.community_id for c in res.communities] == [c1r2c1_id, c1r1_id]
539 res = api.ListUserCommunities(
540 communities_pb2.ListUserCommunitiesReq(page_size=2, page_token=res.next_page_token)
541 )
542 assert [c.community_id for c in res.communities] == [c1r2_id, c1_id]
543 res = api.ListUserCommunities(
544 communities_pb2.ListUserCommunitiesReq(page_size=2, page_token=res.next_page_token)
545 )
546 assert [c.community_id for c in res.communities] == [w_id]
547 assert not res.next_page_token
549 @staticmethod
550 def test_ListOtherUserCommunities(testing_communities):
551 with session_scope() as session:
552 user1_id, token1 = get_user_id_and_token(session, "user1")
553 user2_id, token2 = get_user_id_and_token(session, "user2")
554 w_id = get_community_id(session, "Global")
555 c1_id = get_community_id(session, "Country 1")
556 c1r1_id = get_community_id(session, "Country 1, Region 1")
557 c1r1c1_id = get_community_id(session, "Country 1, Region 1, City 1")
558 c1r1c2_id = get_community_id(session, "Country 1, Region 1, City 2")
559 c1r2_id = get_community_id(session, "Country 1, Region 2")
560 c1r2c1_id = get_community_id(session, "Country 1, Region 2, City 1")
562 # Fetch user2's communities from user1's account
563 with communities_session(token1) as api:
564 res = api.ListUserCommunities(communities_pb2.ListUserCommunitiesReq(user_id=user2_id))
565 assert [c.community_id for c in res.communities] == [
566 c1r1c1_id,
567 c1r1c2_id,
568 c1r2c1_id,
569 c1r1_id,
570 c1r2_id,
571 c1_id,
572 w_id,
573 ]
575 @staticmethod
576 def test_ListGroups(testing_communities):
577 with session_scope() as session:
578 user1_id, token1 = get_user_id_and_token(session, "user1")
579 user5_id, token5 = get_user_id_and_token(session, "user5")
580 w_id = get_community_id(session, "Global")
581 hitchhikers_id = get_group_id(session, "Hitchhikers")
582 c1r1_id = get_community_id(session, "Country 1, Region 1")
583 foodies_id = get_group_id(session, "Country 1, Region 1, Foodies")
584 skaters_id = get_group_id(session, "Country 1, Region 1, Skaters")
586 with communities_session(token1) as api:
587 res = api.ListGroups(
588 communities_pb2.ListGroupsReq(
589 community_id=c1r1_id,
590 )
591 )
592 assert [g.group_id for g in res.groups] == [foodies_id, skaters_id]
594 with communities_session(token5) as api:
595 res = api.ListGroups(
596 communities_pb2.ListGroupsReq(
597 community_id=w_id,
598 )
599 )
600 assert len(res.groups) == 1
601 assert res.groups[0].group_id == hitchhikers_id
603 @staticmethod
604 def test_ListAdmins(testing_communities):
605 with session_scope() as session:
606 user1_id, token1 = get_user_id_and_token(session, "user1")
607 user3_id, token3 = get_user_id_and_token(session, "user3")
608 user4_id, token4 = get_user_id_and_token(session, "user4")
609 user5_id, token5 = get_user_id_and_token(session, "user5")
610 user7_id, token7 = get_user_id_and_token(session, "user7")
611 w_id = get_community_id(session, "Global")
612 c1r1c2_id = get_community_id(session, "Country 1, Region 1, City 2")
614 with communities_session(token1) as api:
615 res = api.ListAdmins(
616 communities_pb2.ListAdminsReq(
617 community_id=w_id,
618 )
619 )
620 assert res.admin_user_ids == [user1_id, user3_id, user7_id]
622 res = api.ListAdmins(
623 communities_pb2.ListAdminsReq(
624 community_id=c1r1c2_id,
625 )
626 )
627 assert res.admin_user_ids == [user4_id, user5_id]
629 @staticmethod
630 def test_AddAdmin(testing_communities):
631 with session_scope() as session:
632 user4_id, token4 = get_user_id_and_token(session, "user4")
633 user5_id, _ = get_user_id_and_token(session, "user5")
634 user2_id, _ = get_user_id_and_token(session, "user2")
635 user8_id, token8 = get_user_id_and_token(session, "user8")
636 node_id = get_community_id(session, "Country 1, Region 1, City 2")
638 with communities_session(token8) as api:
639 with pytest.raises(grpc.RpcError) as err:
640 api.AddAdmin(communities_pb2.AddAdminReq(community_id=node_id, user_id=user2_id))
641 assert err.value.code() == grpc.StatusCode.FAILED_PRECONDITION
642 assert err.value.details() == "You're not allowed to moderate that community"
644 with communities_session(token4) as api:
645 res = api.ListAdmins(communities_pb2.ListAdminsReq(community_id=node_id))
646 assert res.admin_user_ids == [user4_id, user5_id]
648 with pytest.raises(grpc.RpcError) as err:
649 api.AddAdmin(communities_pb2.AddAdminReq(community_id=node_id, user_id=user8_id))
650 assert err.value.code() == grpc.StatusCode.FAILED_PRECONDITION
651 assert err.value.details() == "That user is not in the community."
653 with pytest.raises(grpc.RpcError) as err:
654 api.AddAdmin(communities_pb2.AddAdminReq(community_id=node_id, user_id=user5_id))
655 assert err.value.code() == grpc.StatusCode.FAILED_PRECONDITION
656 assert err.value.details() == "That user is already an admin."
658 api.AddAdmin(communities_pb2.AddAdminReq(community_id=node_id, user_id=user2_id))
659 res = api.ListAdmins(communities_pb2.ListAdminsReq(community_id=node_id))
660 assert res.admin_user_ids == [user2_id, user4_id, user5_id]
661 # Cleanup because database changes do not roll back
662 api.RemoveAdmin(communities_pb2.RemoveAdminReq(community_id=node_id, user_id=user2_id))
664 @staticmethod
665 def test_RemoveAdmin(testing_communities):
666 with session_scope() as session:
667 user4_id, token4 = get_user_id_and_token(session, "user4")
668 user5_id, _ = get_user_id_and_token(session, "user5")
669 user2_id, _ = get_user_id_and_token(session, "user2")
670 user8_id, token8 = get_user_id_and_token(session, "user8")
671 node_id = get_community_id(session, "Country 1, Region 1, City 2")
673 with communities_session(token8) as api:
674 with pytest.raises(grpc.RpcError) as err:
675 api.AddAdmin(communities_pb2.AddAdminReq(community_id=node_id, user_id=user2_id))
676 assert err.value.code() == grpc.StatusCode.FAILED_PRECONDITION
677 assert err.value.details() == "You're not allowed to moderate that community"
679 with communities_session(token4) as api:
680 res = api.ListAdmins(communities_pb2.ListAdminsReq(community_id=node_id))
681 assert res.admin_user_ids == [user4_id, user5_id]
683 with pytest.raises(grpc.RpcError) as err:
684 api.RemoveAdmin(communities_pb2.RemoveAdminReq(community_id=node_id, user_id=user8_id))
685 assert err.value.code() == grpc.StatusCode.FAILED_PRECONDITION
686 assert err.value.details() == "That user is not in the community."
688 with pytest.raises(grpc.RpcError) as err:
689 api.RemoveAdmin(communities_pb2.RemoveAdminReq(community_id=node_id, user_id=user2_id))
690 assert err.value.code() == grpc.StatusCode.FAILED_PRECONDITION
691 assert err.value.details() == "That user is not an admin."
693 api.RemoveAdmin(communities_pb2.RemoveAdminReq(community_id=node_id, user_id=user5_id))
694 res = api.ListAdmins(communities_pb2.ListAdminsReq(community_id=node_id))
695 assert res.admin_user_ids == [user4_id]
696 # Cleanup because database changes do not roll back
697 api.AddAdmin(communities_pb2.AddAdminReq(community_id=node_id, user_id=user5_id))
699 @staticmethod
700 def test_ListMembers(testing_communities):
701 with session_scope() as session:
702 user1_id, token1 = get_user_id_and_token(session, "user1")
703 user2_id, token2 = get_user_id_and_token(session, "user2")
704 user3_id, token3 = get_user_id_and_token(session, "user3")
705 user4_id, token4 = get_user_id_and_token(session, "user4")
706 user5_id, token5 = get_user_id_and_token(session, "user5")
707 user6_id, token6 = get_user_id_and_token(session, "user6")
708 user7_id, token7 = get_user_id_and_token(session, "user7")
709 user8_id, token8 = get_user_id_and_token(session, "user8")
710 w_id = get_community_id(session, "Global")
711 c1r1c2_id = get_community_id(session, "Country 1, Region 1, City 2")
713 with communities_session(token1) as api:
714 res = api.ListMembers(
715 communities_pb2.ListMembersReq(
716 community_id=w_id,
717 )
718 )
719 assert res.member_user_ids == [
720 user8_id,
721 user7_id,
722 user6_id,
723 user5_id,
724 user4_id,
725 user3_id,
726 user2_id,
727 user1_id,
728 ]
730 res = api.ListMembers(
731 communities_pb2.ListMembersReq(
732 community_id=c1r1c2_id,
733 )
734 )
735 assert res.member_user_ids == [user5_id, user4_id, user2_id]
737 @staticmethod
738 def test_ListNearbyUsers(testing_communities):
739 with session_scope() as session:
740 user1_id, token1 = get_user_id_and_token(session, "user1")
741 user2_id, token2 = get_user_id_and_token(session, "user2")
742 user3_id, token3 = get_user_id_and_token(session, "user3")
743 user4_id, token4 = get_user_id_and_token(session, "user4")
744 user5_id, token5 = get_user_id_and_token(session, "user5")
745 user6_id, token6 = get_user_id_and_token(session, "user6")
746 user7_id, token7 = get_user_id_and_token(session, "user7")
747 user8_id, token8 = get_user_id_and_token(session, "user8")
748 w_id = get_community_id(session, "Global")
749 c1r1c2_id = get_community_id(session, "Country 1, Region 1, City 2")
751 with communities_session(token1) as api:
752 res = api.ListNearbyUsers(
753 communities_pb2.ListNearbyUsersReq(
754 community_id=w_id,
755 )
756 )
757 assert res.nearby_user_ids == [
758 user1_id,
759 user2_id,
760 user3_id,
761 user4_id,
762 user5_id,
763 user6_id,
764 user7_id,
765 user8_id,
766 ]
768 res = api.ListNearbyUsers(
769 communities_pb2.ListNearbyUsersReq(
770 community_id=c1r1c2_id,
771 )
772 )
773 assert res.nearby_user_ids == [user4_id]
775 @staticmethod
776 def test_ListDiscussions(testing_communities):
777 with session_scope() as session:
778 user1_id, token1 = get_user_id_and_token(session, "user1")
779 w_id = get_community_id(session, "Global")
780 c1r1c2_id = get_community_id(session, "Country 1, Region 1, City 2")
782 with communities_session(token1) as api:
783 res = api.ListDiscussions(
784 communities_pb2.ListDiscussionsReq(
785 community_id=w_id,
786 page_size=3,
787 )
788 )
789 assert [d.title for d in res.discussions] == [
790 "Discussion title 6",
791 "Discussion title 5",
792 "Discussion title 4",
793 ]
794 for d in res.discussions:
795 assert d.thread.thread_id > 0
796 assert d.thread.num_responses == 0
798 res = api.ListDiscussions(
799 communities_pb2.ListDiscussionsReq(
800 community_id=w_id,
801 page_token=res.next_page_token,
802 page_size=2,
803 )
804 )
805 assert [d.title for d in res.discussions] == [
806 "Discussion title 3",
807 "Discussion title 2",
808 ]
809 for d in res.discussions:
810 assert d.thread.thread_id > 0
811 assert d.thread.num_responses == 0
813 res = api.ListDiscussions(
814 communities_pb2.ListDiscussionsReq(
815 community_id=w_id,
816 page_token=res.next_page_token,
817 page_size=2,
818 )
819 )
820 assert [d.title for d in res.discussions] == [
821 "Discussion title 1",
822 ]
823 for d in res.discussions:
824 assert d.thread.thread_id > 0
825 assert d.thread.num_responses == 0
827 res = api.ListDiscussions(
828 communities_pb2.ListDiscussionsReq(
829 community_id=c1r1c2_id,
830 )
831 )
832 assert [d.title for d in res.discussions] == [
833 "Discussion title 7",
834 ]
835 for d in res.discussions:
836 assert d.thread.thread_id > 0
837 assert d.thread.num_responses == 0
839 @staticmethod
840 def test_is_user_in_node_geography(testing_communities):
841 with session_scope() as session:
842 c1_id = get_community_id(session, "Country 1")
844 user1_id, _ = get_user_id_and_token(session, "user1")
845 user2_id, _ = get_user_id_and_token(session, "user2")
846 user3_id, _ = get_user_id_and_token(session, "user3")
847 user4_id, _ = get_user_id_and_token(session, "user4")
848 user5_id, _ = get_user_id_and_token(session, "user5")
850 # All these users should be in Country 1's geography
851 assert is_user_in_node_geography(session, user1_id, c1_id)
852 assert is_user_in_node_geography(session, user2_id, c1_id)
853 assert is_user_in_node_geography(session, user3_id, c1_id)
854 assert is_user_in_node_geography(session, user4_id, c1_id)
855 assert is_user_in_node_geography(session, user5_id, c1_id)
857 @staticmethod
858 def test_ListEvents(testing_communities):
859 with session_scope() as session:
860 user1_id, token1 = get_user_id_and_token(session, "user1")
861 c1_id = get_community_id(session, "Country 1")
863 with communities_session(token1) as api:
864 res = api.ListEvents(
865 communities_pb2.ListEventsReq(
866 community_id=c1_id,
867 page_size=3,
868 )
869 )
870 assert [d.title for d in res.events] == [
871 "Event title 1",
872 "Event title 2",
873 "Event title 3",
874 ]
876 res = api.ListEvents(
877 communities_pb2.ListEventsReq(
878 community_id=c1_id,
879 page_token=res.next_page_token,
880 page_size=2,
881 )
882 )
883 assert [d.title for d in res.events] == [
884 "Event title 4",
885 "Event title 5",
886 ]
888 res = api.ListEvents(
889 communities_pb2.ListEventsReq(
890 community_id=c1_id,
891 page_token=res.next_page_token,
892 page_size=2,
893 )
894 )
895 assert [d.title for d in res.events] == [
896 "Event title 6",
897 ]
898 assert not res.next_page_token
900 @staticmethod
901 def test_empty_query_aborts(testing_communities):
902 with session_scope() as session:
903 _, token = get_user_id_and_token(session, "user1")
905 with communities_session(token) as api:
906 with pytest.raises(grpc.RpcError) as err:
907 api.SearchCommunities(communities_pb2.SearchCommunitiesReq(query=" "))
908 assert err.value.code() == grpc.StatusCode.INVALID_ARGUMENT
909 assert err.value.details() == "Query must be at least 3 characters long."
911 @staticmethod
912 def test_min_length_lt_3_aborts(testing_communities):
913 """
914 len(query) < 3 → return INVALID_ARGUMENT: query_too_short
915 """
916 with session_scope() as session:
917 _, token = get_user_id_and_token(session, "user1")
919 with communities_session(token) as api:
920 with pytest.raises(grpc.RpcError) as err:
921 api.SearchCommunities(communities_pb2.SearchCommunitiesReq(query="zz", page_size=5))
922 assert err.value.code() == grpc.StatusCode.INVALID_ARGUMENT
923 assert err.value.details() == "Query must be at least 3 characters long."
925 @staticmethod
926 def test_typo_matches_existing_name(testing_communities):
927 """
928 Word_similarity should match a simple typo in community name.
929 """
930 with session_scope() as session:
931 _, token = get_user_id_and_token(session, "user1")
932 c1_id = get_community_id(session, "Country 1")
934 with communities_session(token) as api:
935 res = api.SearchCommunities(communities_pb2.SearchCommunitiesReq(query="Coutri 1", page_size=5))
936 ids = [c.community_id for c in res.communities]
937 assert c1_id in ids
939 @staticmethod
940 def test_word_similarity_matches_partial_word(testing_communities):
941 """
942 Query 'city' should match 'Country 1, Region 1, City 1'.
943 """
944 with session_scope() as session:
945 _, token = get_user_id_and_token(session, "user1")
946 city1_id = get_community_id(session, "Country 1, Region 1, City 1") # переименовал для ясности
948 with communities_session(token) as api:
949 res = api.SearchCommunities(communities_pb2.SearchCommunitiesReq(query="city", page_size=5))
950 ids = [c.community_id for c in res.communities]
951 assert city1_id in ids
953 @staticmethod
954 def test_results_sorted_by_similarity(testing_communities):
955 """
956 Results should be ordered by similarity score (best match first).
957 For query 'Country 1, Region', the full region name should rank higher
958 than deeper descendants like 'City 1'.
959 """
960 with session_scope() as session:
961 _, token = get_user_id_and_token(session, "user1")
962 region_id = get_community_id(session, "Country 1, Region 1")
963 city_id = get_community_id(session, "Country 1, Region 1, City 1")
965 with communities_session(token) as api:
966 res = api.SearchCommunities(communities_pb2.SearchCommunitiesReq(query="Country 1, Region", page_size=5))
967 ids = [c.community_id for c in res.communities]
969 assert region_id in ids
970 assert city_id in ids
971 assert ids.index(region_id) < ids.index(city_id)
973 @staticmethod
974 def test_no_results_returns_empty(testing_communities):
975 """
976 For a nonsense query that shouldn't meet the similarity threshold, return empty list.
977 """
978 with session_scope() as session:
979 _, token = get_user_id_and_token(session, "user1")
981 with communities_session(token) as api:
982 res = api.SearchCommunities(communities_pb2.SearchCommunitiesReq(query="qwertyuiopasdf", page_size=5))
983 assert res.communities == []
985 @staticmethod
986 def test_ListAllCommunities(testing_communities):
987 """
988 Test that ListAllCommunities returns all communities with proper hierarchy information.
989 """
990 with session_scope() as session:
991 user1_id, token1 = get_user_id_and_token(session, "user1")
992 user2_id, token2 = get_user_id_and_token(session, "user2")
993 user6_id, token6 = get_user_id_and_token(session, "user6")
994 w_id = get_community_id(session, "Global")
995 c1_id = get_community_id(session, "Country 1")
996 c1r1_id = get_community_id(session, "Country 1, Region 1")
997 c1r1c1_id = get_community_id(session, "Country 1, Region 1, City 1")
998 c1r1c2_id = get_community_id(session, "Country 1, Region 1, City 2")
999 c1r2_id = get_community_id(session, "Country 1, Region 2")
1000 c1r2c1_id = get_community_id(session, "Country 1, Region 2, City 1")
1001 c2_id = get_community_id(session, "Country 2")
1002 c2r1_id = get_community_id(session, "Country 2, Region 1")
1003 c2r1c1_id = get_community_id(session, "Country 2, Region 1, City 1")
1005 # Test with user1 who is a member of multiple communities
1006 with communities_session(token1) as api:
1007 res = api.ListAllCommunities(communities_pb2.ListAllCommunitiesReq())
1009 # Should return all 10 communities
1010 assert len(res.communities) == 10
1012 # Get all community IDs
1013 community_ids = [c.community_id for c in res.communities]
1014 assert set(community_ids) == {
1015 w_id,
1016 c1_id,
1017 c1r1_id,
1018 c1r1c1_id,
1019 c1r1c2_id,
1020 c1r2_id,
1021 c1r2c1_id,
1022 c2_id,
1023 c2r1_id,
1024 c2r1c1_id,
1025 }
1027 # Check that each community has the required fields
1028 for community in res.communities:
1029 assert community.community_id > 0
1030 assert len(community.name) > 0
1031 assert len(community.slug) > 0
1032 assert community.member_count > 0
1033 # member field should be a boolean
1034 assert isinstance(community.member, bool)
1035 # parents should be present for hierarchical ordering
1036 assert len(community.parents) >= 1
1037 # created timestamp should be present
1038 assert community.HasField("created")
1039 assert community.created.seconds > 0
1041 # Find specific communities and verify their data
1042 global_community = next(c for c in res.communities if c.community_id == w_id)
1043 assert global_community.name == "Global"
1044 assert global_community.slug == "global"
1045 assert global_community.member # user1 is a member
1046 assert global_community.member_count == 8
1047 assert len(global_community.parents) == 1 # Only itself
1049 c1r1c1_community = next(c for c in res.communities if c.community_id == c1r1c1_id)
1050 assert c1r1c1_community.name == "Country 1, Region 1, City 1"
1051 assert c1r1c1_community.slug == "country-1-region-1-city-1"
1052 assert c1r1c1_community.member # user1 is a member
1053 assert c1r1c1_community.member_count == 3
1054 assert len(c1r1c1_community.parents) == 4 # Global, Country 1, Region 1, City 1
1055 # Verify parent hierarchy
1056 assert c1r1c1_community.parents[0].community.community_id == w_id
1057 assert c1r1c1_community.parents[1].community.community_id == c1_id
1058 assert c1r1c1_community.parents[2].community.community_id == c1r1_id
1059 assert c1r1c1_community.parents[3].community.community_id == c1r1c1_id
1061 # Test with user6 who has different community memberships
1062 with communities_session(token6) as api:
1063 res = api.ListAllCommunities(communities_pb2.ListAllCommunitiesReq())
1065 # Should still return all 10 communities
1066 assert len(res.communities) == 10
1068 # Find Country 2 community - user6 should be a member
1069 c2_community = next(c for c in res.communities if c.community_id == c2_id)
1070 assert c2_community.member # user6 is a member
1071 assert c2_community.member_count == 2
1073 # Find Country 1 - user6 should NOT be a member
1074 c1_community = next(c for c in res.communities if c.community_id == c1_id)
1075 assert not c1_community.member # user6 is not a member
1077 # Global - user6 should be a member
1078 global_community = next(c for c in res.communities if c.community_id == w_id)
1079 assert global_community.member # user6 is a member
1081 @staticmethod
1082 def test_ListRecentCommunities(testing_communities, monkeypatch):
1083 """
1084 ListRecentCommunities returns newest-first communities across the whole tree,
1085 honouring page_size.
1086 """
1087 with session_scope() as session:
1088 _, token = get_user_id_and_token(session, "user1")
1089 # communities are created in this order in the fixture, so creation
1090 # time ascends down the list
1091 w_id = get_community_id(session, "Global")
1092 c2_id = get_community_id(session, "Country 2")
1093 c2r1_id = get_community_id(session, "Country 2, Region 1")
1094 c2r1c1_id = get_community_id(session, "Country 2, Region 1, City 1")
1095 c1_id = get_community_id(session, "Country 1")
1096 c1r1_id = get_community_id(session, "Country 1, Region 1")
1097 c1r1c1_id = get_community_id(session, "Country 1, Region 1, City 1")
1098 c1r1c2_id = get_community_id(session, "Country 1, Region 1, City 2")
1099 c1r2_id = get_community_id(session, "Country 1, Region 2")
1100 c1r2c1_id = get_community_id(session, "Country 1, Region 2, City 1")
1102 newest_first = [
1103 c1r2c1_id,
1104 c1r2_id,
1105 c1r1c2_id,
1106 c1r1c1_id,
1107 c1r1_id,
1108 c1_id,
1109 c2r1c1_id,
1110 c2r1_id,
1111 c2_id,
1112 w_id,
1113 ]
1115 with communities_session(token) as api:
1116 res = api.ListRecentCommunities(communities_pb2.ListRecentCommunitiesReq(page_size=3))
1117 assert [c.community_id for c in res.communities] == newest_first[:3]
1118 for community in res.communities:
1119 assert community.HasField("created")
1120 assert community.created.seconds > 0
1122 # default page size returns all communities for this fixture
1123 res = api.ListRecentCommunities(communities_pb2.ListRecentCommunitiesReq())
1124 assert [c.community_id for c in res.communities] == newest_first
1126 # page_size is clamped to MAX_PAGINATION_LENGTH on the server
1127 monkeypatch.setattr("couchers.servicers.communities.MAX_PAGINATION_LENGTH", 4)
1128 res = api.ListRecentCommunities(communities_pb2.ListRecentCommunitiesReq(page_size=1000))
1129 assert [c.community_id for c in res.communities] == newest_first[:4]
1130 # and also applies to the default when the request omits page_size
1131 res = api.ListRecentCommunities(communities_pb2.ListRecentCommunitiesReq())
1132 assert [c.community_id for c in res.communities] == newest_first[:4]
1135def test_JoinCommunity_and_LeaveCommunity(testing_communities):
1136 # these are separate as they mutate the database
1137 with session_scope() as session:
1138 # at x=1, inside c1 (country 1)
1139 user1_id, token1 = get_user_id_and_token(session, "user1")
1140 # at x=51, not inside c1
1141 user8_id, token8 = get_user_id_and_token(session, "user8")
1142 c1_id = get_community_id(session, "Country 1")
1144 with communities_session(token1) as api:
1145 assert api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c1_id)).member
1147 # user1 is already part of c1, cannot join
1148 with pytest.raises(grpc.RpcError) as e:
1149 res = api.JoinCommunity(
1150 communities_pb2.JoinCommunityReq(
1151 community_id=c1_id,
1152 )
1153 )
1154 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
1155 assert e.value.details() == "You're already in that community."
1157 assert api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c1_id)).member
1159 # user1 is inside c1, cannot leave
1160 with pytest.raises(grpc.RpcError) as e:
1161 res = api.LeaveCommunity(
1162 communities_pb2.LeaveCommunityReq(
1163 community_id=c1_id,
1164 )
1165 )
1166 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
1167 assert (
1168 e.value.details()
1169 == "Your location on your profile is within this community, so you cannot leave it. However, you can adjust your notifications in your account settings."
1170 )
1172 assert api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c1_id)).member
1174 with communities_session(token8) as api:
1175 assert not api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c1_id)).member
1177 # user8 is not in c1 yet, cannot leave
1178 with pytest.raises(grpc.RpcError) as e:
1179 res = api.LeaveCommunity(
1180 communities_pb2.LeaveCommunityReq(
1181 community_id=c1_id,
1182 )
1183 )
1184 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
1185 assert e.value.details() == "You're not in that community."
1187 assert not api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c1_id)).member
1189 # user8 is not in c1 and not part, can join
1190 res = api.JoinCommunity(
1191 communities_pb2.JoinCommunityReq(
1192 community_id=c1_id,
1193 )
1194 )
1196 assert api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c1_id)).member
1198 # user8 is not in c1 and but now part, can't join again
1199 with pytest.raises(grpc.RpcError) as e:
1200 res = api.JoinCommunity(
1201 communities_pb2.JoinCommunityReq(
1202 community_id=c1_id,
1203 )
1204 )
1205 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
1206 assert e.value.details() == "You're already in that community."
1208 assert api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c1_id)).member
1210 # user8 is not in c1 yet, but part of it, can leave
1211 res = api.LeaveCommunity(
1212 communities_pb2.LeaveCommunityReq(
1213 community_id=c1_id,
1214 )
1215 )
1216 assert not api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c1_id)).member
1219def test_LeaveCommunity_regression(db):
1220 # See github issue #1444, repro:
1221 # 1. Join more than one community
1222 # 2. Leave one of them
1223 # 3. You are no longer in any community
1224 # admin
1225 user1, token1 = generate_user(username="user1", geom=create_1d_point(200), geom_radius=0.1)
1226 # joiner/leaver
1227 user2, token2 = generate_user(username="user2", geom=create_1d_point(201), geom_radius=0.1)
1229 with session_scope() as session:
1230 c0 = create_community(session, 0, 100, "Community 0", [user1], [], None)
1231 c1 = create_community(session, 0, 50, "Community 1", [user1], [], c0)
1232 c2 = create_community(session, 0, 10, "Community 2", [user1], [], c0)
1233 c0_id = c0.id
1234 c1_id = c1.id
1235 c2_id = c2.id
1237 enforce_community_memberships()
1239 with communities_session(token1) as api:
1240 assert api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c0_id)).member
1241 assert api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c1_id)).member
1242 assert api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c2_id)).member
1244 with communities_session(token2) as api:
1245 # first check we're not in any communities
1246 assert not api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c0_id)).member
1247 assert not api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c1_id)).member
1248 assert not api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c2_id)).member
1250 # join some communities
1251 api.JoinCommunity(communities_pb2.JoinCommunityReq(community_id=c1_id))
1252 api.JoinCommunity(communities_pb2.JoinCommunityReq(community_id=c2_id))
1254 # check memberships
1255 assert not api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c0_id)).member
1256 assert api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c1_id)).member
1257 assert api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c2_id)).member
1259 # leave just c2
1260 api.LeaveCommunity(communities_pb2.LeaveCommunityReq(community_id=c2_id))
1262 # check memberships
1263 assert not api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c0_id)).member
1264 assert api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c1_id)).member
1265 assert not api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c2_id)).member
1268def test_enforce_community_memberships_for_user(testing_communities):
1269 """
1270 Make sure the user is added to the right communities on signup
1271 """
1272 with auth_api_session() as (auth_api, metadata_interceptor):
1273 res = auth_api.SignupFlow(
1274 auth_pb2.SignupFlowReq(
1275 basic=auth_pb2.SignupBasic(name="testing", email="email@couchers.org.invalid"),
1276 account=auth_pb2.SignupAccount(
1277 username="frodo",
1278 password="a very insecure password",
1279 birthdate="1970-01-01",
1280 gender="Bot",
1281 hosting_status=api_pb2.HOSTING_STATUS_CAN_HOST,
1282 city="Country 1, Region 1, City 2",
1283 # lat=8, lng=1 is equivalent to creating this coordinate with create_coordinate(8)
1284 lat=8,
1285 lng=1,
1286 radius=500,
1287 accept_tos=True,
1288 ),
1289 feedback=auth_pb2.ContributorForm(),
1290 accept_community_guidelines=wrappers_pb2.BoolValue(value=True),
1291 motivations=auth_pb2.SignupMotivations(motivations=["surfing"]),
1292 )
1293 )
1294 with session_scope() as session:
1295 email_token = (
1296 session.execute(select(SignupFlow).where(SignupFlow.flow_token == res.flow_token)).scalar_one().email_token
1297 )
1298 with auth_api_session() as (auth_api, metadata_interceptor):
1299 res = auth_api.SignupFlow(auth_pb2.SignupFlowReq(email_token=email_token))
1300 user_id = res.auth_res.user_id
1302 # now check the user is in the right communities
1303 with session_scope() as session:
1304 w_id = get_community_id(session, "Global")
1305 c1_id = get_community_id(session, "Country 1")
1306 c1r1_id = get_community_id(session, "Country 1, Region 1")
1307 c1r1c2_id = get_community_id(session, "Country 1, Region 1, City 2")
1309 token, _ = get_session_cookie_tokens(metadata_interceptor)
1311 with communities_session(token) as api:
1312 res = api.ListUserCommunities(communities_pb2.ListUserCommunitiesReq())
1313 assert [c.community_id for c in res.communities] == [c1r1c2_id, c1r1_id, c1_id, w_id]
1316# TODO: requires transferring of content
1318# def test_ListPlaces(db, testing_communities):
1319# pass
1321# def test_ListGuides(db, testing_communities):
1322# pass
1324# def test_ListEvents(db, testing_communities):
1325# pass