Coverage for app/backend/src/tests/test_communities.py: 100%
715 statements
« prev ^ index » next coverage.py v7.14.2, created at 2026-06-21 09:29 +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 offline_information=events_pb2.OfflineEventInformation(
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 w_id,
524 c1_id,
525 c1r1_id,
526 c1r1c1_id,
527 c1r1c2_id,
528 c1r2_id,
529 c1r2c1_id,
530 ]
532 @staticmethod
533 def test_ListOtherUserCommunities(testing_communities):
534 with session_scope() as session:
535 user1_id, token1 = get_user_id_and_token(session, "user1")
536 user2_id, token2 = get_user_id_and_token(session, "user2")
537 w_id = get_community_id(session, "Global")
538 c1_id = get_community_id(session, "Country 1")
539 c1r1_id = get_community_id(session, "Country 1, Region 1")
540 c1r1c1_id = get_community_id(session, "Country 1, Region 1, City 1")
541 c1r1c2_id = get_community_id(session, "Country 1, Region 1, City 2")
542 c1r2_id = get_community_id(session, "Country 1, Region 2")
543 c1r2c1_id = get_community_id(session, "Country 1, Region 2, City 1")
545 # Fetch user2's communities from user1's account
546 with communities_session(token1) as api:
547 res = api.ListUserCommunities(communities_pb2.ListUserCommunitiesReq(user_id=user2_id))
548 assert [c.community_id for c in res.communities] == [
549 w_id,
550 c1_id,
551 c1r1_id,
552 c1r1c1_id,
553 c1r1c2_id,
554 c1r2_id,
555 c1r2c1_id,
556 ]
558 @staticmethod
559 def test_ListGroups(testing_communities):
560 with session_scope() as session:
561 user1_id, token1 = get_user_id_and_token(session, "user1")
562 user5_id, token5 = get_user_id_and_token(session, "user5")
563 w_id = get_community_id(session, "Global")
564 hitchhikers_id = get_group_id(session, "Hitchhikers")
565 c1r1_id = get_community_id(session, "Country 1, Region 1")
566 foodies_id = get_group_id(session, "Country 1, Region 1, Foodies")
567 skaters_id = get_group_id(session, "Country 1, Region 1, Skaters")
569 with communities_session(token1) as api:
570 res = api.ListGroups(
571 communities_pb2.ListGroupsReq(
572 community_id=c1r1_id,
573 )
574 )
575 assert [g.group_id for g in res.groups] == [foodies_id, skaters_id]
577 with communities_session(token5) as api:
578 res = api.ListGroups(
579 communities_pb2.ListGroupsReq(
580 community_id=w_id,
581 )
582 )
583 assert len(res.groups) == 1
584 assert res.groups[0].group_id == hitchhikers_id
586 @staticmethod
587 def test_ListAdmins(testing_communities):
588 with session_scope() as session:
589 user1_id, token1 = get_user_id_and_token(session, "user1")
590 user3_id, token3 = get_user_id_and_token(session, "user3")
591 user4_id, token4 = get_user_id_and_token(session, "user4")
592 user5_id, token5 = get_user_id_and_token(session, "user5")
593 user7_id, token7 = get_user_id_and_token(session, "user7")
594 w_id = get_community_id(session, "Global")
595 c1r1c2_id = get_community_id(session, "Country 1, Region 1, City 2")
597 with communities_session(token1) as api:
598 res = api.ListAdmins(
599 communities_pb2.ListAdminsReq(
600 community_id=w_id,
601 )
602 )
603 assert res.admin_user_ids == [user1_id, user3_id, user7_id]
605 res = api.ListAdmins(
606 communities_pb2.ListAdminsReq(
607 community_id=c1r1c2_id,
608 )
609 )
610 assert res.admin_user_ids == [user4_id, user5_id]
612 @staticmethod
613 def test_AddAdmin(testing_communities):
614 with session_scope() as session:
615 user4_id, token4 = get_user_id_and_token(session, "user4")
616 user5_id, _ = get_user_id_and_token(session, "user5")
617 user2_id, _ = get_user_id_and_token(session, "user2")
618 user8_id, token8 = get_user_id_and_token(session, "user8")
619 node_id = get_community_id(session, "Country 1, Region 1, City 2")
621 with communities_session(token8) as api:
622 with pytest.raises(grpc.RpcError) as err:
623 api.AddAdmin(communities_pb2.AddAdminReq(community_id=node_id, user_id=user2_id))
624 assert err.value.code() == grpc.StatusCode.FAILED_PRECONDITION
625 assert err.value.details() == "You're not allowed to moderate that community"
627 with communities_session(token4) as api:
628 res = api.ListAdmins(communities_pb2.ListAdminsReq(community_id=node_id))
629 assert res.admin_user_ids == [user4_id, user5_id]
631 with pytest.raises(grpc.RpcError) as err:
632 api.AddAdmin(communities_pb2.AddAdminReq(community_id=node_id, user_id=user8_id))
633 assert err.value.code() == grpc.StatusCode.FAILED_PRECONDITION
634 assert err.value.details() == "That user is not in the community."
636 with pytest.raises(grpc.RpcError) as err:
637 api.AddAdmin(communities_pb2.AddAdminReq(community_id=node_id, user_id=user5_id))
638 assert err.value.code() == grpc.StatusCode.FAILED_PRECONDITION
639 assert err.value.details() == "That user is already an admin."
641 api.AddAdmin(communities_pb2.AddAdminReq(community_id=node_id, user_id=user2_id))
642 res = api.ListAdmins(communities_pb2.ListAdminsReq(community_id=node_id))
643 assert res.admin_user_ids == [user2_id, user4_id, user5_id]
644 # Cleanup because database changes do not roll back
645 api.RemoveAdmin(communities_pb2.RemoveAdminReq(community_id=node_id, user_id=user2_id))
647 @staticmethod
648 def test_RemoveAdmin(testing_communities):
649 with session_scope() as session:
650 user4_id, token4 = get_user_id_and_token(session, "user4")
651 user5_id, _ = get_user_id_and_token(session, "user5")
652 user2_id, _ = get_user_id_and_token(session, "user2")
653 user8_id, token8 = get_user_id_and_token(session, "user8")
654 node_id = get_community_id(session, "Country 1, Region 1, City 2")
656 with communities_session(token8) as api:
657 with pytest.raises(grpc.RpcError) as err:
658 api.AddAdmin(communities_pb2.AddAdminReq(community_id=node_id, user_id=user2_id))
659 assert err.value.code() == grpc.StatusCode.FAILED_PRECONDITION
660 assert err.value.details() == "You're not allowed to moderate that community"
662 with communities_session(token4) as api:
663 res = api.ListAdmins(communities_pb2.ListAdminsReq(community_id=node_id))
664 assert res.admin_user_ids == [user4_id, user5_id]
666 with pytest.raises(grpc.RpcError) as err:
667 api.RemoveAdmin(communities_pb2.RemoveAdminReq(community_id=node_id, user_id=user8_id))
668 assert err.value.code() == grpc.StatusCode.FAILED_PRECONDITION
669 assert err.value.details() == "That user is not in the community."
671 with pytest.raises(grpc.RpcError) as err:
672 api.RemoveAdmin(communities_pb2.RemoveAdminReq(community_id=node_id, user_id=user2_id))
673 assert err.value.code() == grpc.StatusCode.FAILED_PRECONDITION
674 assert err.value.details() == "That user is not an admin."
676 api.RemoveAdmin(communities_pb2.RemoveAdminReq(community_id=node_id, user_id=user5_id))
677 res = api.ListAdmins(communities_pb2.ListAdminsReq(community_id=node_id))
678 assert res.admin_user_ids == [user4_id]
679 # Cleanup because database changes do not roll back
680 api.AddAdmin(communities_pb2.AddAdminReq(community_id=node_id, user_id=user5_id))
682 @staticmethod
683 def test_ListMembers(testing_communities):
684 with session_scope() as session:
685 user1_id, token1 = get_user_id_and_token(session, "user1")
686 user2_id, token2 = get_user_id_and_token(session, "user2")
687 user3_id, token3 = get_user_id_and_token(session, "user3")
688 user4_id, token4 = get_user_id_and_token(session, "user4")
689 user5_id, token5 = get_user_id_and_token(session, "user5")
690 user6_id, token6 = get_user_id_and_token(session, "user6")
691 user7_id, token7 = get_user_id_and_token(session, "user7")
692 user8_id, token8 = get_user_id_and_token(session, "user8")
693 w_id = get_community_id(session, "Global")
694 c1r1c2_id = get_community_id(session, "Country 1, Region 1, City 2")
696 with communities_session(token1) as api:
697 res = api.ListMembers(
698 communities_pb2.ListMembersReq(
699 community_id=w_id,
700 )
701 )
702 assert res.member_user_ids == [
703 user8_id,
704 user7_id,
705 user6_id,
706 user5_id,
707 user4_id,
708 user3_id,
709 user2_id,
710 user1_id,
711 ]
713 res = api.ListMembers(
714 communities_pb2.ListMembersReq(
715 community_id=c1r1c2_id,
716 )
717 )
718 assert res.member_user_ids == [user5_id, user4_id, user2_id]
720 @staticmethod
721 def test_ListNearbyUsers(testing_communities):
722 with session_scope() as session:
723 user1_id, token1 = get_user_id_and_token(session, "user1")
724 user2_id, token2 = get_user_id_and_token(session, "user2")
725 user3_id, token3 = get_user_id_and_token(session, "user3")
726 user4_id, token4 = get_user_id_and_token(session, "user4")
727 user5_id, token5 = get_user_id_and_token(session, "user5")
728 user6_id, token6 = get_user_id_and_token(session, "user6")
729 user7_id, token7 = get_user_id_and_token(session, "user7")
730 user8_id, token8 = get_user_id_and_token(session, "user8")
731 w_id = get_community_id(session, "Global")
732 c1r1c2_id = get_community_id(session, "Country 1, Region 1, City 2")
734 with communities_session(token1) as api:
735 res = api.ListNearbyUsers(
736 communities_pb2.ListNearbyUsersReq(
737 community_id=w_id,
738 )
739 )
740 assert res.nearby_user_ids == [
741 user1_id,
742 user2_id,
743 user3_id,
744 user4_id,
745 user5_id,
746 user6_id,
747 user7_id,
748 user8_id,
749 ]
751 res = api.ListNearbyUsers(
752 communities_pb2.ListNearbyUsersReq(
753 community_id=c1r1c2_id,
754 )
755 )
756 assert res.nearby_user_ids == [user4_id]
758 @staticmethod
759 def test_ListDiscussions(testing_communities):
760 with session_scope() as session:
761 user1_id, token1 = get_user_id_and_token(session, "user1")
762 w_id = get_community_id(session, "Global")
763 c1r1c2_id = get_community_id(session, "Country 1, Region 1, City 2")
765 with communities_session(token1) as api:
766 res = api.ListDiscussions(
767 communities_pb2.ListDiscussionsReq(
768 community_id=w_id,
769 page_size=3,
770 )
771 )
772 assert [d.title for d in res.discussions] == [
773 "Discussion title 6",
774 "Discussion title 5",
775 "Discussion title 4",
776 ]
777 for d in res.discussions:
778 assert d.thread.thread_id > 0
779 assert d.thread.num_responses == 0
781 res = api.ListDiscussions(
782 communities_pb2.ListDiscussionsReq(
783 community_id=w_id,
784 page_token=res.next_page_token,
785 page_size=2,
786 )
787 )
788 assert [d.title for d in res.discussions] == [
789 "Discussion title 3",
790 "Discussion title 2",
791 ]
792 for d in res.discussions:
793 assert d.thread.thread_id > 0
794 assert d.thread.num_responses == 0
796 res = api.ListDiscussions(
797 communities_pb2.ListDiscussionsReq(
798 community_id=w_id,
799 page_token=res.next_page_token,
800 page_size=2,
801 )
802 )
803 assert [d.title for d in res.discussions] == [
804 "Discussion title 1",
805 ]
806 for d in res.discussions:
807 assert d.thread.thread_id > 0
808 assert d.thread.num_responses == 0
810 res = api.ListDiscussions(
811 communities_pb2.ListDiscussionsReq(
812 community_id=c1r1c2_id,
813 )
814 )
815 assert [d.title for d in res.discussions] == [
816 "Discussion title 7",
817 ]
818 for d in res.discussions:
819 assert d.thread.thread_id > 0
820 assert d.thread.num_responses == 0
822 @staticmethod
823 def test_is_user_in_node_geography(testing_communities):
824 with session_scope() as session:
825 c1_id = get_community_id(session, "Country 1")
827 user1_id, _ = get_user_id_and_token(session, "user1")
828 user2_id, _ = get_user_id_and_token(session, "user2")
829 user3_id, _ = get_user_id_and_token(session, "user3")
830 user4_id, _ = get_user_id_and_token(session, "user4")
831 user5_id, _ = get_user_id_and_token(session, "user5")
833 # All these users should be in Country 1's geography
834 assert is_user_in_node_geography(session, user1_id, c1_id)
835 assert is_user_in_node_geography(session, user2_id, c1_id)
836 assert is_user_in_node_geography(session, user3_id, c1_id)
837 assert is_user_in_node_geography(session, user4_id, c1_id)
838 assert is_user_in_node_geography(session, user5_id, c1_id)
840 @staticmethod
841 def test_ListEvents(testing_communities):
842 with session_scope() as session:
843 user1_id, token1 = get_user_id_and_token(session, "user1")
844 c1_id = get_community_id(session, "Country 1")
846 with communities_session(token1) as api:
847 res = api.ListEvents(
848 communities_pb2.ListEventsReq(
849 community_id=c1_id,
850 page_size=3,
851 )
852 )
853 assert [d.title for d in res.events] == [
854 "Event title 1",
855 "Event title 2",
856 "Event title 3",
857 ]
859 res = api.ListEvents(
860 communities_pb2.ListEventsReq(
861 community_id=c1_id,
862 page_token=res.next_page_token,
863 page_size=2,
864 )
865 )
866 assert [d.title for d in res.events] == [
867 "Event title 4",
868 "Event title 5",
869 ]
871 res = api.ListEvents(
872 communities_pb2.ListEventsReq(
873 community_id=c1_id,
874 page_token=res.next_page_token,
875 page_size=2,
876 )
877 )
878 assert [d.title for d in res.events] == [
879 "Event title 6",
880 ]
881 assert not res.next_page_token
883 @staticmethod
884 def test_empty_query_aborts(testing_communities):
885 with session_scope() as session:
886 _, token = get_user_id_and_token(session, "user1")
888 with communities_session(token) as api:
889 with pytest.raises(grpc.RpcError) as err:
890 api.SearchCommunities(communities_pb2.SearchCommunitiesReq(query=" "))
891 assert err.value.code() == grpc.StatusCode.INVALID_ARGUMENT
892 assert err.value.details() == "Query must be at least 3 characters long."
894 @staticmethod
895 def test_min_length_lt_3_aborts(testing_communities):
896 """
897 len(query) < 3 → return INVALID_ARGUMENT: query_too_short
898 """
899 with session_scope() as session:
900 _, token = get_user_id_and_token(session, "user1")
902 with communities_session(token) as api:
903 with pytest.raises(grpc.RpcError) as err:
904 api.SearchCommunities(communities_pb2.SearchCommunitiesReq(query="zz", page_size=5))
905 assert err.value.code() == grpc.StatusCode.INVALID_ARGUMENT
906 assert err.value.details() == "Query must be at least 3 characters long."
908 @staticmethod
909 def test_typo_matches_existing_name(testing_communities):
910 """
911 Word_similarity should match a simple typo in community name.
912 """
913 with session_scope() as session:
914 _, token = get_user_id_and_token(session, "user1")
915 c1_id = get_community_id(session, "Country 1")
917 with communities_session(token) as api:
918 res = api.SearchCommunities(communities_pb2.SearchCommunitiesReq(query="Coutri 1", page_size=5))
919 ids = [c.community_id for c in res.communities]
920 assert c1_id in ids
922 @staticmethod
923 def test_word_similarity_matches_partial_word(testing_communities):
924 """
925 Query 'city' should match 'Country 1, Region 1, City 1'.
926 """
927 with session_scope() as session:
928 _, token = get_user_id_and_token(session, "user1")
929 city1_id = get_community_id(session, "Country 1, Region 1, City 1") # переименовал для ясности
931 with communities_session(token) as api:
932 res = api.SearchCommunities(communities_pb2.SearchCommunitiesReq(query="city", page_size=5))
933 ids = [c.community_id for c in res.communities]
934 assert city1_id in ids
936 @staticmethod
937 def test_results_sorted_by_similarity(testing_communities):
938 """
939 Results should be ordered by similarity score (best match first).
940 For query 'Country 1, Region', the full region name should rank higher
941 than deeper descendants like 'City 1'.
942 """
943 with session_scope() as session:
944 _, token = get_user_id_and_token(session, "user1")
945 region_id = get_community_id(session, "Country 1, Region 1")
946 city_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="Country 1, Region", page_size=5))
950 ids = [c.community_id for c in res.communities]
952 assert region_id in ids
953 assert city_id in ids
954 assert ids.index(region_id) < ids.index(city_id)
956 @staticmethod
957 def test_no_results_returns_empty(testing_communities):
958 """
959 For a nonsense query that shouldn't meet the similarity threshold, return empty list.
960 """
961 with session_scope() as session:
962 _, token = get_user_id_and_token(session, "user1")
964 with communities_session(token) as api:
965 res = api.SearchCommunities(communities_pb2.SearchCommunitiesReq(query="qwertyuiopasdf", page_size=5))
966 assert res.communities == []
968 @staticmethod
969 def test_ListAllCommunities(testing_communities):
970 """
971 Test that ListAllCommunities returns all communities with proper hierarchy information.
972 """
973 with session_scope() as session:
974 user1_id, token1 = get_user_id_and_token(session, "user1")
975 user2_id, token2 = get_user_id_and_token(session, "user2")
976 user6_id, token6 = get_user_id_and_token(session, "user6")
977 w_id = get_community_id(session, "Global")
978 c1_id = get_community_id(session, "Country 1")
979 c1r1_id = get_community_id(session, "Country 1, Region 1")
980 c1r1c1_id = get_community_id(session, "Country 1, Region 1, City 1")
981 c1r1c2_id = get_community_id(session, "Country 1, Region 1, City 2")
982 c1r2_id = get_community_id(session, "Country 1, Region 2")
983 c1r2c1_id = get_community_id(session, "Country 1, Region 2, City 1")
984 c2_id = get_community_id(session, "Country 2")
985 c2r1_id = get_community_id(session, "Country 2, Region 1")
986 c2r1c1_id = get_community_id(session, "Country 2, Region 1, City 1")
988 # Test with user1 who is a member of multiple communities
989 with communities_session(token1) as api:
990 res = api.ListAllCommunities(communities_pb2.ListAllCommunitiesReq())
992 # Should return all 10 communities
993 assert len(res.communities) == 10
995 # Get all community IDs
996 community_ids = [c.community_id for c in res.communities]
997 assert set(community_ids) == {
998 w_id,
999 c1_id,
1000 c1r1_id,
1001 c1r1c1_id,
1002 c1r1c2_id,
1003 c1r2_id,
1004 c1r2c1_id,
1005 c2_id,
1006 c2r1_id,
1007 c2r1c1_id,
1008 }
1010 # Check that each community has the required fields
1011 for community in res.communities:
1012 assert community.community_id > 0
1013 assert len(community.name) > 0
1014 assert len(community.slug) > 0
1015 assert community.member_count > 0
1016 # member field should be a boolean
1017 assert isinstance(community.member, bool)
1018 # parents should be present for hierarchical ordering
1019 assert len(community.parents) >= 1
1020 # created timestamp should be present
1021 assert community.HasField("created")
1022 assert community.created.seconds > 0
1024 # Find specific communities and verify their data
1025 global_community = next(c for c in res.communities if c.community_id == w_id)
1026 assert global_community.name == "Global"
1027 assert global_community.slug == "global"
1028 assert global_community.member # user1 is a member
1029 assert global_community.member_count == 8
1030 assert len(global_community.parents) == 1 # Only itself
1032 c1r1c1_community = next(c for c in res.communities if c.community_id == c1r1c1_id)
1033 assert c1r1c1_community.name == "Country 1, Region 1, City 1"
1034 assert c1r1c1_community.slug == "country-1-region-1-city-1"
1035 assert c1r1c1_community.member # user1 is a member
1036 assert c1r1c1_community.member_count == 3
1037 assert len(c1r1c1_community.parents) == 4 # Global, Country 1, Region 1, City 1
1038 # Verify parent hierarchy
1039 assert c1r1c1_community.parents[0].community.community_id == w_id
1040 assert c1r1c1_community.parents[1].community.community_id == c1_id
1041 assert c1r1c1_community.parents[2].community.community_id == c1r1_id
1042 assert c1r1c1_community.parents[3].community.community_id == c1r1c1_id
1044 # Test with user6 who has different community memberships
1045 with communities_session(token6) as api:
1046 res = api.ListAllCommunities(communities_pb2.ListAllCommunitiesReq())
1048 # Should still return all 10 communities
1049 assert len(res.communities) == 10
1051 # Find Country 2 community - user6 should be a member
1052 c2_community = next(c for c in res.communities if c.community_id == c2_id)
1053 assert c2_community.member # user6 is a member
1054 assert c2_community.member_count == 2
1056 # Find Country 1 - user6 should NOT be a member
1057 c1_community = next(c for c in res.communities if c.community_id == c1_id)
1058 assert not c1_community.member # user6 is not a member
1060 # Global - user6 should be a member
1061 global_community = next(c for c in res.communities if c.community_id == w_id)
1062 assert global_community.member # user6 is a member
1064 @staticmethod
1065 def test_ListRecentCommunities(testing_communities, monkeypatch):
1066 """
1067 ListRecentCommunities returns newest-first communities across the whole tree,
1068 honouring page_size.
1069 """
1070 with session_scope() as session:
1071 _, token = get_user_id_and_token(session, "user1")
1072 # communities are created in this order in the fixture, so creation
1073 # time ascends down the list
1074 w_id = get_community_id(session, "Global")
1075 c2_id = get_community_id(session, "Country 2")
1076 c2r1_id = get_community_id(session, "Country 2, Region 1")
1077 c2r1c1_id = get_community_id(session, "Country 2, Region 1, City 1")
1078 c1_id = get_community_id(session, "Country 1")
1079 c1r1_id = get_community_id(session, "Country 1, Region 1")
1080 c1r1c1_id = get_community_id(session, "Country 1, Region 1, City 1")
1081 c1r1c2_id = get_community_id(session, "Country 1, Region 1, City 2")
1082 c1r2_id = get_community_id(session, "Country 1, Region 2")
1083 c1r2c1_id = get_community_id(session, "Country 1, Region 2, City 1")
1085 newest_first = [
1086 c1r2c1_id,
1087 c1r2_id,
1088 c1r1c2_id,
1089 c1r1c1_id,
1090 c1r1_id,
1091 c1_id,
1092 c2r1c1_id,
1093 c2r1_id,
1094 c2_id,
1095 w_id,
1096 ]
1098 with communities_session(token) as api:
1099 res = api.ListRecentCommunities(communities_pb2.ListRecentCommunitiesReq(page_size=3))
1100 assert [c.community_id for c in res.communities] == newest_first[:3]
1101 for community in res.communities:
1102 assert community.HasField("created")
1103 assert community.created.seconds > 0
1105 # default page size returns all communities for this fixture
1106 res = api.ListRecentCommunities(communities_pb2.ListRecentCommunitiesReq())
1107 assert [c.community_id for c in res.communities] == newest_first
1109 # page_size is clamped to MAX_PAGINATION_LENGTH on the server
1110 monkeypatch.setattr("couchers.servicers.communities.MAX_PAGINATION_LENGTH", 4)
1111 res = api.ListRecentCommunities(communities_pb2.ListRecentCommunitiesReq(page_size=1000))
1112 assert [c.community_id for c in res.communities] == newest_first[:4]
1113 # and also applies to the default when the request omits page_size
1114 res = api.ListRecentCommunities(communities_pb2.ListRecentCommunitiesReq())
1115 assert [c.community_id for c in res.communities] == newest_first[:4]
1118def test_JoinCommunity_and_LeaveCommunity(testing_communities):
1119 # these are separate as they mutate the database
1120 with session_scope() as session:
1121 # at x=1, inside c1 (country 1)
1122 user1_id, token1 = get_user_id_and_token(session, "user1")
1123 # at x=51, not inside c1
1124 user8_id, token8 = get_user_id_and_token(session, "user8")
1125 c1_id = get_community_id(session, "Country 1")
1127 with communities_session(token1) as api:
1128 assert api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c1_id)).member
1130 # user1 is already part of c1, cannot join
1131 with pytest.raises(grpc.RpcError) as e:
1132 res = api.JoinCommunity(
1133 communities_pb2.JoinCommunityReq(
1134 community_id=c1_id,
1135 )
1136 )
1137 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
1138 assert e.value.details() == "You're already in that community."
1140 assert api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c1_id)).member
1142 # user1 is inside c1, cannot leave
1143 with pytest.raises(grpc.RpcError) as e:
1144 res = api.LeaveCommunity(
1145 communities_pb2.LeaveCommunityReq(
1146 community_id=c1_id,
1147 )
1148 )
1149 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
1150 assert (
1151 e.value.details()
1152 == "Your location on your profile is within this community, so you cannot leave it. However, you can adjust your notifications in your account settings."
1153 )
1155 assert api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c1_id)).member
1157 with communities_session(token8) as api:
1158 assert not api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c1_id)).member
1160 # user8 is not in c1 yet, cannot leave
1161 with pytest.raises(grpc.RpcError) as e:
1162 res = api.LeaveCommunity(
1163 communities_pb2.LeaveCommunityReq(
1164 community_id=c1_id,
1165 )
1166 )
1167 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
1168 assert e.value.details() == "You're not in that community."
1170 assert not api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c1_id)).member
1172 # user8 is not in c1 and not part, can join
1173 res = api.JoinCommunity(
1174 communities_pb2.JoinCommunityReq(
1175 community_id=c1_id,
1176 )
1177 )
1179 assert api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c1_id)).member
1181 # user8 is not in c1 and but now part, can't join again
1182 with pytest.raises(grpc.RpcError) as e:
1183 res = api.JoinCommunity(
1184 communities_pb2.JoinCommunityReq(
1185 community_id=c1_id,
1186 )
1187 )
1188 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
1189 assert e.value.details() == "You're already in that community."
1191 assert api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c1_id)).member
1193 # user8 is not in c1 yet, but part of it, can leave
1194 res = api.LeaveCommunity(
1195 communities_pb2.LeaveCommunityReq(
1196 community_id=c1_id,
1197 )
1198 )
1199 assert not api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c1_id)).member
1202def test_LeaveCommunity_regression(db):
1203 # See github issue #1444, repro:
1204 # 1. Join more than one community
1205 # 2. Leave one of them
1206 # 3. You are no longer in any community
1207 # admin
1208 user1, token1 = generate_user(username="user1", geom=create_1d_point(200), geom_radius=0.1)
1209 # joiner/leaver
1210 user2, token2 = generate_user(username="user2", geom=create_1d_point(201), geom_radius=0.1)
1212 with session_scope() as session:
1213 c0 = create_community(session, 0, 100, "Community 0", [user1], [], None)
1214 c1 = create_community(session, 0, 50, "Community 1", [user1], [], c0)
1215 c2 = create_community(session, 0, 10, "Community 2", [user1], [], c0)
1216 c0_id = c0.id
1217 c1_id = c1.id
1218 c2_id = c2.id
1220 enforce_community_memberships()
1222 with communities_session(token1) as api:
1223 assert api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c0_id)).member
1224 assert api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c1_id)).member
1225 assert api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c2_id)).member
1227 with communities_session(token2) as api:
1228 # first check we're not in any communities
1229 assert not api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c0_id)).member
1230 assert not api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c1_id)).member
1231 assert not api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c2_id)).member
1233 # join some communities
1234 api.JoinCommunity(communities_pb2.JoinCommunityReq(community_id=c1_id))
1235 api.JoinCommunity(communities_pb2.JoinCommunityReq(community_id=c2_id))
1237 # check memberships
1238 assert not api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c0_id)).member
1239 assert api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c1_id)).member
1240 assert api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c2_id)).member
1242 # leave just c2
1243 api.LeaveCommunity(communities_pb2.LeaveCommunityReq(community_id=c2_id))
1245 # check memberships
1246 assert not api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c0_id)).member
1247 assert api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c1_id)).member
1248 assert not api.GetCommunity(communities_pb2.GetCommunityReq(community_id=c2_id)).member
1251def test_enforce_community_memberships_for_user(testing_communities):
1252 """
1253 Make sure the user is added to the right communities on signup
1254 """
1255 with auth_api_session() as (auth_api, metadata_interceptor):
1256 res = auth_api.SignupFlow(
1257 auth_pb2.SignupFlowReq(
1258 basic=auth_pb2.SignupBasic(name="testing", email="email@couchers.org.invalid"),
1259 account=auth_pb2.SignupAccount(
1260 username="frodo",
1261 password="a very insecure password",
1262 birthdate="1970-01-01",
1263 gender="Bot",
1264 hosting_status=api_pb2.HOSTING_STATUS_CAN_HOST,
1265 city="Country 1, Region 1, City 2",
1266 # lat=8, lng=1 is equivalent to creating this coordinate with create_coordinate(8)
1267 lat=8,
1268 lng=1,
1269 radius=500,
1270 accept_tos=True,
1271 ),
1272 feedback=auth_pb2.ContributorForm(),
1273 accept_community_guidelines=wrappers_pb2.BoolValue(value=True),
1274 motivations=auth_pb2.SignupMotivations(motivations=["surfing"]),
1275 )
1276 )
1277 with session_scope() as session:
1278 email_token = (
1279 session.execute(select(SignupFlow).where(SignupFlow.flow_token == res.flow_token)).scalar_one().email_token
1280 )
1281 with auth_api_session() as (auth_api, metadata_interceptor):
1282 res = auth_api.SignupFlow(auth_pb2.SignupFlowReq(email_token=email_token))
1283 user_id = res.auth_res.user_id
1285 # now check the user is in the right communities
1286 with session_scope() as session:
1287 w_id = get_community_id(session, "Global")
1288 c1_id = get_community_id(session, "Country 1")
1289 c1r1_id = get_community_id(session, "Country 1, Region 1")
1290 c1r1c2_id = get_community_id(session, "Country 1, Region 1, City 2")
1292 token, _ = get_session_cookie_tokens(metadata_interceptor)
1294 with communities_session(token) as api:
1295 res = api.ListUserCommunities(communities_pb2.ListUserCommunitiesReq())
1296 assert [c.community_id for c in res.communities] == [w_id, c1_id, c1r1_id, c1r1c2_id]
1299# TODO: requires transferring of content
1301# def test_ListPlaces(db, testing_communities):
1302# pass
1304# def test_ListGuides(db, testing_communities):
1305# pass
1307# def test_ListEvents(db, testing_communities):
1308# pass