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