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