Coverage for app/backend/src/tests/test_communities.py: 100%
722 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-03 11:56 +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, pages_pb2
28from couchers.tasks import enforce_community_memberships
29from couchers.utils import create_coordinate, create_polygon_lat_lng, now, to_multi
30from tests.fixtures.api import events as events_api
31from tests.fixtures.db import generate_user, get_user_id_and_token
32from tests.fixtures.misc import Moderator
33from tests.fixtures.sessions import (
34 auth_api_session,
35 communities_session,
36 discussions_session,
37 pages_session,
38)
39from tests.test_auth import get_session_cookie_tokens
40from tests.utils import local_date_time
42# For testing purposes, restrict ourselves to a 1D-world, consisting of "intervals" that have width 2, and coordinates
43# that are points at (x, 1).
44# 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
45# mostly fine
48def create_1d_polygon(lb: int, ub: int) -> WKBElement:
49 # given a lower bound and upper bound on x, creates the given interval
50 return create_polygon_lat_lng([[lb, 0], [lb, 2], [ub, 2], [ub, 0], [lb, 0]])
53def create_1d_point(x: int) -> WKBElement:
54 return create_coordinate(x, 1)
57def create_community(
58 session: Session,
59 interval_lb: int,
60 interval_ub: int,
61 name: str,
62 admins: list[User],
63 extra_members: list[User],
64 parent: Node | None,
65) -> Node:
66 node_type = CHILD_NODE_TYPE[parent.node_type if parent else None]
67 node = Node(
68 geom=to_multi(create_1d_polygon(interval_lb, interval_ub)),
69 parent_node_id=parent.id if parent else None,
70 node_type=node_type,
71 )
72 session.add(node)
73 session.flush()
74 cluster = Cluster(
75 name=f"{name}",
76 description=f"Description for {name}",
77 parent_node_id=node.id,
78 is_official_cluster=True,
79 )
80 session.add(cluster)
81 session.flush()
82 thread = Thread()
83 session.add(thread)
84 session.flush()
85 main_page = Page(
86 parent_node_id=cluster.parent_node_id,
87 creator_user_id=admins[0].id,
88 owner_cluster_id=cluster.id,
89 type=PageType.main_page,
90 thread_id=thread.id,
91 )
92 session.add(main_page)
93 session.flush()
94 page_version = PageVersion(
95 page_id=main_page.id,
96 editor_user_id=admins[0].id,
97 title=f"Main page for the {name} community",
98 content="There is nothing here yet...",
99 )
100 session.add(page_version)
101 for admin in admins:
102 cluster.cluster_subscriptions.append(
103 ClusterSubscription(
104 user_id=admin.id,
105 cluster_id=cluster.id,
106 role=ClusterRole.admin,
107 )
108 )
109 for member in extra_members:
110 cluster.cluster_subscriptions.append(
111 ClusterSubscription(
112 user_id=member.id,
113 cluster_id=cluster.id,
114 role=ClusterRole.member,
115 )
116 )
117 session.commit()
118 # other members will be added by enforce_community_memberships()
119 return node
122def create_group(
123 session: Session, name: str, admins: list[User], members: list[User], parent_community: Node | None
124) -> Cluster:
125 assert parent_community is not None
126 cluster = Cluster(
127 name=f"{name}",
128 description=f"Description for {name}",
129 parent_node_id=parent_community.id,
130 )
131 session.add(cluster)
132 session.flush()
133 thread = Thread()
134 session.add(thread)
135 session.flush()
136 main_page = Page(
137 parent_node_id=cluster.parent_node_id,
138 creator_user_id=admins[0].id,
139 owner_cluster_id=cluster.id,
140 type=PageType.main_page,
141 thread_id=thread.id,
142 )
143 session.add(main_page)
144 session.flush()
145 page_version = PageVersion(
146 page_id=main_page.id,
147 editor_user_id=admins[0].id,
148 title=f"Main page for the {name} community",
149 content="There is nothing here yet...",
150 )
151 session.add(page_version)
152 for admin in admins:
153 cluster.cluster_subscriptions.append(
154 ClusterSubscription(
155 user_id=admin.id,
156 cluster_id=cluster.id,
157 role=ClusterRole.admin,
158 )
159 )
160 for member in members:
161 cluster.cluster_subscriptions.append(
162 ClusterSubscription(
163 user_id=member.id,
164 cluster_id=cluster.id,
165 role=ClusterRole.member,
166 )
167 )
168 session.commit()
169 return cluster
172def create_place(token: str, title: str, content: str, address: str, x: float) -> None:
173 with pages_session(token) as api:
174 api.CreatePlace(
175 pages_pb2.CreatePlaceReq(
176 title=title,
177 content=content,
178 address=address,
179 location=pages_pb2.Coordinate(
180 lat=x,
181 lng=1,
182 ),
183 )
184 )
187def create_discussion(token: str, community_id: int | None, group_id: int | None, title: str, content: str) -> None:
188 # set group_id or community_id to None
189 with discussions_session(token) as api:
190 api.CreateDiscussion(
191 discussions_pb2.CreateDiscussionReq(
192 title=title,
193 content=content,
194 owner_community_id=community_id,
195 owner_group_id=group_id,
196 )
197 )
200def create_event(
201 token: str, community_id: int | None, group_id: int | None, title: str, content: str, start_td: timedelta
202) -> None:
203 res = events_api.create_event(
204 token=token,
205 title=title,
206 content=content,
207 start_date_time=local_date_time(now() + start_td),
208 end_date_time=local_date_time(now() + start_td + timedelta(hours=2)),
209 )
210 events_api.transfer_event(
211 token=token, event_id=res.event_id, new_owner_community_id=community_id, new_owner_group_id=group_id
212 )
215def get_community_id(session: Session, community_name: str) -> int:
216 return session.execute(
217 select(Cluster.parent_node_id).where(Cluster.is_official_cluster).where(Cluster.name == community_name)
218 ).scalar_one()
221def get_group_id(session: Session, group_name: str) -> int:
222 return session.execute(
223 select(Cluster.id).where(~Cluster.is_official_cluster).where(Cluster.name == group_name)
224 ).scalar_one()
227@pytest.fixture(scope="class")
228def testing_communities(db_class):
229 user1, token1 = generate_user(username="user1", geom=create_1d_point(1), geom_radius=0.1)
230 user2, token2 = generate_user(username="user2", geom=create_1d_point(2), geom_radius=0.1)
231 user3, token3 = generate_user(username="user3", geom=create_1d_point(3), geom_radius=0.1)
232 user4, token4 = generate_user(username="user4", geom=create_1d_point(8), geom_radius=0.1)
233 user5, token5 = generate_user(username="user5", geom=create_1d_point(6), geom_radius=0.1)
234 user6, token6 = generate_user(username="user6", geom=create_1d_point(65), geom_radius=0.1)
235 user7, token7 = generate_user(username="user7", geom=create_1d_point(80), geom_radius=0.1)
236 user8, token8 = generate_user(username="user8", geom=create_1d_point(51), geom_radius=0.1)
238 with session_scope() as session:
239 w = create_community(session, 0, 100, "Global", [user1, user3, user7], [], None)
240 c2 = create_community(session, 52, 100, "Country 2", [user6, user7], [], w)
241 c2r1 = create_community(session, 52, 71, "Country 2, Region 1", [user6], [user8], c2)
242 c2r1c1 = create_community(session, 53, 70, "Country 2, Region 1, City 1", [user8], [], c2r1)
243 c1 = create_community(session, 0, 50, "Country 1", [user1, user2], [], w)
244 c1r1 = create_community(session, 0, 10, "Country 1, Region 1", [user1, user2], [], c1)
245 c1r1c1 = create_community(session, 0, 5, "Country 1, Region 1, City 1", [user2], [], c1r1)
246 c1r1c2 = create_community(session, 7, 10, "Country 1, Region 1, City 2", [user4, user5], [user2], c1r1)
247 c1r2 = create_community(session, 20, 25, "Country 1, Region 2", [user2], [], c1)
248 c1r2c1 = create_community(session, 21, 23, "Country 1, Region 2, City 1", [user2], [], c1r2)
250 h = create_group(session, "Hitchhikers", [user1, user2], [user5, user8], w)
251 create_group(session, "Country 1, Region 1, Foodies", [user1], [user2, user4], c1r1)
252 create_group(session, "Country 1, Region 1, Skaters", [user2], [user1], c1r1)
253 create_group(session, "Country 1, Region 2, Foodies", [user2], [user4, user5], c1r2)
254 create_group(session, "Country 2, Region 1, Foodies", [user6], [user7], c2r1)
256 w_id = w.id
257 c1r1c2_id = c1r1c2.id
258 h_id = h.id
259 c1_id = c1.id
261 create_discussion(token1, w_id, None, "Discussion title 1", "Discussion content 1")
262 create_discussion(token3, w_id, None, "Discussion title 2", "Discussion content 2")
263 create_discussion(token3, w_id, None, "Discussion title 3", "Discussion content 3")
264 create_discussion(token3, w_id, None, "Discussion title 4", "Discussion content 4")
265 create_discussion(token3, w_id, None, "Discussion title 5", "Discussion content 5")
266 create_discussion(token3, w_id, None, "Discussion title 6", "Discussion content 6")
267 create_discussion(token4, c1r1c2_id, None, "Discussion title 7", "Discussion content 7")
268 create_discussion(token5, None, h_id, "Discussion title 8", "Discussion content 8")
269 create_discussion(token1, None, h_id, "Discussion title 9", "Discussion content 9")
270 create_discussion(token2, None, h_id, "Discussion title 10", "Discussion content 10")
271 create_discussion(token3, None, h_id, "Discussion title 11", "Discussion content 11")
272 create_discussion(token4, None, h_id, "Discussion title 12", "Discussion content 12")
273 create_discussion(token5, None, h_id, "Discussion title 13", "Discussion content 13")
274 create_discussion(token8, None, h_id, "Discussion title 14", "Discussion content 14")
276 create_event(token3, c1_id, None, "Event title 1", "Event content 1", timedelta(hours=1))
277 create_event(token1, c1_id, None, "Event title 2", "Event content 2", timedelta(hours=2))
278 create_event(token3, c1_id, None, "Event title 3", "Event content 3", timedelta(hours=3))
279 create_event(token1, c1_id, None, "Event title 4", "Event content 4", timedelta(hours=4))
280 create_event(token3, c1_id, None, "Event title 5", "Event content 5", timedelta(hours=5))
281 create_event(token1, c1_id, None, "Event title 6", "Event content 6", timedelta(hours=6))
282 create_event(token2, None, h_id, "Event title 7", "Event content 7", timedelta(hours=7))
283 create_event(token2, None, h_id, "Event title 8", "Event content 8", timedelta(hours=8))
284 create_event(token2, None, h_id, "Event title 9", "Event content 9", timedelta(hours=9))
285 create_event(token2, None, h_id, "Event title 10", "Event content 10", timedelta(hours=10))
286 create_event(token2, None, h_id, "Event title 11", "Event content 11", timedelta(hours=11))
287 create_event(token2, None, h_id, "Event title 12", "Event content 12", timedelta(hours=12))
289 # Approve all events for visibility (UMS starts events as SHADOWED)
290 mod_user, mod_token = generate_user(is_superuser=True)
291 mod = Moderator(mod_user, mod_token)
292 with session_scope() as session:
293 occurrence_ids = session.execute(select(EventOccurrence.id)).scalars().all()
294 discussion_ids = session.execute(select(Discussion.id)).scalars().all()
295 for oid in occurrence_ids:
296 mod.approve_event_occurrence(oid)
297 for did in discussion_ids:
298 mod.approve_discussion(did)
300 enforce_community_memberships()
302 create_place(token1, "Country 1, Region 1, Attraction", "Place content", "Somewhere in c1r1", 6)
303 create_place(token2, "Country 1, Region 1, City 1, Attraction 1", "Place content", "Somewhere in c1r1c1", 3)
304 create_place(token2, "Country 1, Region 1, City 1, Attraction 2", "Place content", "Somewhere in c1r1c1", 4)
305 create_place(token8, "Global, Attraction", "Place content", "Somewhere in w", 51.5)
306 create_place(token6, "Country 2, Region 1, Attraction", "Place content", "Somewhere in c2r1", 59)
308 refresh_materialized_views(empty_pb2.Empty())
310 yield
313class TestCommunities:
314 @staticmethod
315 def test_GetCommunity(testing_communities):
316 with session_scope() as session:
317 user1_id, token1 = get_user_id_and_token(session, "user1")
318 user2_id, token2 = get_user_id_and_token(session, "user2")
319 user6_id, token6 = get_user_id_and_token(session, "user6")
320 w_id = get_community_id(session, "Global")
321 c1_id = get_community_id(session, "Country 1")
322 c1r1_id = get_community_id(session, "Country 1, Region 1")
323 c1r1c1_id = get_community_id(session, "Country 1, Region 1, City 1")
324 c2_id = get_community_id(session, "Country 2")
326 with communities_session(token2) as api:
327 res = api.GetCommunity(
328 communities_pb2.GetCommunityReq(
329 community_id=w_id,
330 )
331 )
332 assert res.name == "Global"
333 assert res.slug == "global"
334 assert res.description == "Description for Global"
335 assert len(res.parents) == 1
336 assert res.parents[0].HasField("community")
337 assert res.parents[0].community.community_id == w_id
338 assert res.parents[0].community.name == "Global"
339 assert res.parents[0].community.slug == "global"
340 assert res.parents[0].community.description == "Description for Global"
341 assert res.main_page.type == pages_pb2.PAGE_TYPE_MAIN_PAGE
342 assert res.main_page.slug == "main-page-for-the-global-community"
343 assert res.main_page.last_editor_user_id == user1_id
344 assert res.main_page.creator_user_id == user1_id
345 assert res.main_page.owner_community_id == w_id
346 assert res.main_page.title == "Main page for the Global community"
347 assert res.main_page.content == "There is nothing here yet..."
348 assert not res.main_page.can_edit
349 assert not res.main_page.can_moderate
350 assert res.main_page.editor_user_ids == [user1_id]
351 assert res.member
352 assert not res.admin
353 assert res.member_count == 8
354 assert res.admin_count == 3
356 res = api.GetCommunity(
357 communities_pb2.GetCommunityReq(
358 community_id=c1r1c1_id,
359 )
360 )
361 assert res.community_id == c1r1c1_id
362 assert res.name == "Country 1, Region 1, City 1"
363 assert res.slug == "country-1-region-1-city-1"
364 assert res.description == "Description for Country 1, Region 1, City 1"
365 assert len(res.parents) == 4
366 assert res.parents[0].HasField("community")
367 assert res.parents[0].community.community_id == w_id
368 assert res.parents[0].community.name == "Global"
369 assert res.parents[0].community.slug == "global"
370 assert res.parents[0].community.description == "Description for Global"
371 assert res.parents[1].HasField("community")
372 assert res.parents[1].community.community_id == c1_id
373 assert res.parents[1].community.name == "Country 1"
374 assert res.parents[1].community.slug == "country-1"
375 assert res.parents[1].community.description == "Description for Country 1"
376 assert res.parents[2].HasField("community")
377 assert res.parents[2].community.community_id == c1r1_id
378 assert res.parents[2].community.name == "Country 1, Region 1"
379 assert res.parents[2].community.slug == "country-1-region-1"
380 assert res.parents[2].community.description == "Description for Country 1, Region 1"
381 assert res.parents[3].HasField("community")
382 assert res.parents[3].community.community_id == c1r1c1_id
383 assert res.parents[3].community.name == "Country 1, Region 1, City 1"
384 assert res.parents[3].community.slug == "country-1-region-1-city-1"
385 assert res.parents[3].community.description == "Description for Country 1, Region 1, City 1"
386 assert res.main_page.type == pages_pb2.PAGE_TYPE_MAIN_PAGE
387 assert res.main_page.slug == "main-page-for-the-country-1-region-1-city-1-community"
388 assert res.main_page.last_editor_user_id == user2_id
389 assert res.main_page.creator_user_id == user2_id
390 assert res.main_page.owner_community_id == c1r1c1_id
391 assert res.main_page.title == "Main page for the Country 1, Region 1, City 1 community"
392 assert res.main_page.content == "There is nothing here yet..."
393 assert res.main_page.can_edit
394 assert res.main_page.can_moderate
395 assert res.main_page.editor_user_ids == [user2_id]
396 assert res.member
397 assert res.admin
398 assert res.member_count == 3
399 assert res.admin_count == 1
401 res = api.GetCommunity(
402 communities_pb2.GetCommunityReq(
403 community_id=c2_id,
404 )
405 )
406 assert res.community_id == c2_id
407 assert res.name == "Country 2"
408 assert res.slug == "country-2"
409 assert res.description == "Description for Country 2"
410 assert len(res.parents) == 2
411 assert res.parents[0].HasField("community")
412 assert res.parents[0].community.community_id == w_id
413 assert res.parents[0].community.name == "Global"
414 assert res.parents[0].community.slug == "global"
415 assert res.parents[0].community.description == "Description for Global"
416 assert res.parents[1].HasField("community")
417 assert res.parents[1].community.community_id == c2_id
418 assert res.parents[1].community.name == "Country 2"
419 assert res.parents[1].community.slug == "country-2"
420 assert res.parents[1].community.description == "Description for Country 2"
421 assert res.main_page.type == pages_pb2.PAGE_TYPE_MAIN_PAGE
422 assert res.main_page.slug == "main-page-for-the-country-2-community"
423 assert res.main_page.last_editor_user_id == user6_id
424 assert res.main_page.creator_user_id == user6_id
425 assert res.main_page.owner_community_id == c2_id
426 assert res.main_page.title == "Main page for the Country 2 community"
427 assert res.main_page.content == "There is nothing here yet..."
428 assert not res.main_page.can_edit
429 assert not res.main_page.can_moderate
430 assert res.main_page.editor_user_ids == [user6_id]
431 assert not res.member
432 assert not res.admin
433 assert res.member_count == 2
434 assert res.admin_count == 2
436 @staticmethod
437 def test_ListCommunities(testing_communities):
438 with session_scope() as session:
439 user1_id, token1 = get_user_id_and_token(session, "user1")
440 c1_id = get_community_id(session, "Country 1")
441 c1r1_id = get_community_id(session, "Country 1, Region 1")
442 c1r2_id = get_community_id(session, "Country 1, Region 2")
444 with communities_session(token1) as api:
445 res = api.ListCommunities(
446 communities_pb2.ListCommunitiesReq(
447 community_id=c1_id,
448 )
449 )
450 assert [c.community_id for c in res.communities] == [c1r1_id, c1r2_id]
452 @staticmethod
453 def test_ListCommunities_all(testing_communities):
454 with session_scope() as session:
455 user1_id, token1 = get_user_id_and_token(session, "user1")
456 w_id = get_community_id(session, "Global")
457 c1_id = get_community_id(session, "Country 1")
458 c1r1_id = get_community_id(session, "Country 1, Region 1")
459 c1r1c1_id = get_community_id(session, "Country 1, Region 1, City 1")
460 c1r1c2_id = get_community_id(session, "Country 1, Region 1, City 2")
461 c1r2_id = get_community_id(session, "Country 1, Region 2")
462 c1r2c1_id = get_community_id(session, "Country 1, Region 2, City 1")
463 c2_id = get_community_id(session, "Country 2")
464 c2r1_id = get_community_id(session, "Country 2, Region 1")
465 c2r1c1_id = get_community_id(session, "Country 2, Region 1, City 1")
467 # Fetch all communities ordered by name
468 with communities_session(token1) as api:
469 res = api.ListCommunities(
470 communities_pb2.ListCommunitiesReq(
471 page_size=5,
472 )
473 )
474 assert [c.community_id for c in res.communities] == [c1_id, c1r1_id, c1r1c1_id, c1r1c2_id, c1r2_id]
475 res = api.ListCommunities(
476 communities_pb2.ListCommunitiesReq(
477 page_size=2,
478 page_token=res.next_page_token,
479 )
480 )
481 assert [c.community_id for c in res.communities] == [c1r2c1_id, c2_id]
482 res = api.ListCommunities(
483 communities_pb2.ListCommunitiesReq(
484 page_size=5,
485 page_token=res.next_page_token,
486 )
487 )
488 assert [c.community_id for c in res.communities] == [c2r1_id, c2r1c1_id, w_id]
490 @staticmethod
491 def test_ListUserCommunities(testing_communities):
492 with session_scope() as session:
493 user2_id, token2 = get_user_id_and_token(session, "user2")
494 w_id = get_community_id(session, "Global")
495 c1_id = get_community_id(session, "Country 1")
496 c1r1_id = get_community_id(session, "Country 1, Region 1")
497 c1r1c1_id = get_community_id(session, "Country 1, Region 1, City 1")
498 c1r1c2_id = get_community_id(session, "Country 1, Region 1, City 2")
499 c1r2_id = get_community_id(session, "Country 1, Region 2")
500 c1r2c1_id = get_community_id(session, "Country 1, Region 2, City 1")
502 # Fetch user2's communities from user2's account
503 with communities_session(token2) as api:
504 res = api.ListUserCommunities(communities_pb2.ListUserCommunitiesReq())
505 assert [c.community_id for c in res.communities] == [
506 c1r1c1_id,
507 c1r1c2_id,
508 c1r2c1_id,
509 c1r1_id,
510 c1r2_id,
511 c1_id,
512 w_id,
513 ]
515 # paginated, with pages crossing node type boundaries
516 res = api.ListUserCommunities(communities_pb2.ListUserCommunitiesReq(page_size=2))
517 assert [c.community_id for c in res.communities] == [c1r1c1_id, c1r1c2_id]
518 res = api.ListUserCommunities(
519 communities_pb2.ListUserCommunitiesReq(page_size=2, page_token=res.next_page_token)
520 )
521 assert [c.community_id for c in res.communities] == [c1r2c1_id, c1r1_id]
522 res = api.ListUserCommunities(
523 communities_pb2.ListUserCommunitiesReq(page_size=2, page_token=res.next_page_token)
524 )
525 assert [c.community_id for c in res.communities] == [c1r2_id, c1_id]
526 res = api.ListUserCommunities(
527 communities_pb2.ListUserCommunitiesReq(page_size=2, page_token=res.next_page_token)
528 )
529 assert [c.community_id for c in res.communities] == [w_id]
530 assert not res.next_page_token
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 c1r1c1_id,
550 c1r1c2_id,
551 c1r2c1_id,
552 c1r1_id,
553 c1r2_id,
554 c1_id,
555 w_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, fast_passwords):
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] == [c1r1c2_id, c1r1_id, c1_id, w_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