Coverage for app/backend/src/tests/test_groups.py: 100%
252 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-19 15:47 +0000
1import grpc
2import pytest
4from couchers.db import session_scope
5from couchers.proto import groups_pb2, pages_pb2
6from couchers.tasks import enforce_community_memberships
7from tests.fixtures.db import generate_user, get_user_id_and_token
8from tests.fixtures.sessions import groups_session
9from tests.test_communities import ( # noqa
10 create_1d_point,
11 create_community,
12 create_group,
13 get_community_id,
14 get_group_id,
15 testing_communities,
16)
19class TestGroups:
20 @staticmethod
21 def test_GetGroup(testing_communities):
22 # implicitly tests visibility and blocking, since all groups have invisible, blocked, and blocking member and admin
23 with session_scope() as session:
24 user2_id, token2 = get_user_id_and_token(session, "user2")
25 w_id = get_community_id(session, "Global")
26 c1_id = get_community_id(session, "Country 1")
27 c2_id = get_community_id(session, "Country 2")
28 c1r2_id = get_community_id(session, "Country 1, Region 2")
29 c2r1_id = get_community_id(session, "Country 2, Region 1")
30 hitchhikers_id = get_group_id(session, "Hitchhikers")
31 c1r2foodies_id = get_group_id(session, "Country 1, Region 2, Foodies")
32 c2r1foodies_id = get_group_id(session, "Country 2, Region 1, Foodies")
34 with groups_session(token2) as api:
35 res = api.GetGroup(
36 groups_pb2.GetGroupReq(
37 group_id=hitchhikers_id,
38 )
39 )
40 assert res.group_id == hitchhikers_id
41 assert res.name == "Hitchhikers"
42 assert res.slug == "hitchhikers"
43 assert res.description == "Description for Hitchhikers"
44 assert len(res.parents) == 2
45 assert res.parents[0].HasField("community")
46 assert res.parents[0].community.community_id == w_id
47 assert res.parents[0].community.name == "Global"
48 assert res.parents[0].community.slug == "global"
49 assert res.parents[0].community.description == "Description for Global"
50 assert res.parents[1].HasField("group")
51 assert res.parents[1].group.group_id == hitchhikers_id
52 assert res.parents[1].group.name == "Hitchhikers"
53 assert res.parents[1].group.slug == "hitchhikers"
54 assert res.parents[1].group.description == "Description for Hitchhikers"
55 assert res.main_page.type == pages_pb2.PAGE_TYPE_MAIN_PAGE
56 assert res.main_page.slug == "main-page-for-the-hitchhikers-community"
57 assert res.main_page.last_editor_user_id == 1
58 assert res.main_page.creator_user_id == 1
59 assert res.main_page.owner_group_id == hitchhikers_id
60 assert res.main_page.title == "Main page for the Hitchhikers community"
61 assert res.main_page.content == "There is nothing here yet..."
62 assert res.main_page.can_edit
63 assert not res.main_page.can_moderate
64 assert res.main_page.editor_user_ids == [1]
65 assert res.member
66 assert res.admin
67 assert res.member_count == 4
68 assert res.admin_count == 2
70 res = api.GetGroup(
71 groups_pb2.GetGroupReq(
72 group_id=c1r2foodies_id,
73 )
74 )
75 assert res.group_id == c1r2foodies_id
76 assert res.name == "Country 1, Region 2, Foodies"
77 assert res.slug == "country-1-region-2-foodies"
78 assert res.description == "Description for Country 1, Region 2, Foodies"
79 assert len(res.parents) == 4
80 assert res.parents[0].HasField("community")
81 assert res.parents[0].community.community_id == w_id
82 assert res.parents[0].community.name == "Global"
83 assert res.parents[0].community.slug == "global"
84 assert res.parents[0].community.description == "Description for Global"
85 assert res.parents[1].HasField("community")
86 assert res.parents[1].community.community_id == c1_id
87 assert res.parents[1].community.name == "Country 1"
88 assert res.parents[1].community.slug == "country-1"
89 assert res.parents[1].community.description == "Description for Country 1"
90 assert res.parents[2].HasField("community")
91 assert res.parents[2].community.community_id == c1r2_id
92 assert res.parents[2].community.name == "Country 1, Region 2"
93 assert res.parents[2].community.slug == "country-1-region-2"
94 assert res.parents[2].community.description == "Description for Country 1, Region 2"
95 assert res.parents[3].HasField("group")
96 assert res.parents[3].group.group_id == c1r2foodies_id
97 assert res.parents[3].group.name == "Country 1, Region 2, Foodies"
98 assert res.parents[3].group.slug == "country-1-region-2-foodies"
99 assert res.parents[3].group.description == "Description for Country 1, Region 2, Foodies"
100 assert res.main_page.type == pages_pb2.PAGE_TYPE_MAIN_PAGE
101 assert res.main_page.slug == "main-page-for-the-country-1-region-2-foodies-community"
102 assert res.main_page.last_editor_user_id == 2
103 assert res.main_page.creator_user_id == 2
104 assert res.main_page.owner_group_id == c1r2foodies_id
105 assert res.main_page.title == "Main page for the Country 1, Region 2, Foodies community"
106 assert res.main_page.content == "There is nothing here yet..."
107 assert res.main_page.can_edit
108 assert res.main_page.can_moderate
109 assert res.main_page.editor_user_ids == [2]
110 assert res.member
111 assert res.admin
112 assert res.member_count == 3
113 assert res.admin_count == 1
115 res = api.GetGroup(
116 groups_pb2.GetGroupReq(
117 group_id=c2r1foodies_id,
118 )
119 )
120 assert res.group_id == c2r1foodies_id
121 assert res.name == "Country 2, Region 1, Foodies"
122 assert res.slug == "country-2-region-1-foodies"
123 assert res.description == "Description for Country 2, Region 1, Foodies"
124 assert len(res.parents) == 4
125 assert res.parents[0].HasField("community")
126 assert res.parents[0].community.community_id == w_id
127 assert res.parents[0].community.name == "Global"
128 assert res.parents[0].community.slug == "global"
129 assert res.parents[0].community.description == "Description for Global"
130 assert res.parents[1].HasField("community")
131 assert res.parents[1].community.community_id == c2_id
132 assert res.parents[1].community.name == "Country 2"
133 assert res.parents[1].community.slug == "country-2"
134 assert res.parents[1].community.description == "Description for Country 2"
135 assert res.parents[2].HasField("community")
136 assert res.parents[2].community.community_id == c2r1_id
137 assert res.parents[2].community.name == "Country 2, Region 1"
138 assert res.parents[2].community.slug == "country-2-region-1"
139 assert res.parents[2].community.description == "Description for Country 2, Region 1"
140 assert res.parents[3].HasField("group")
141 assert res.parents[3].group.group_id == c2r1foodies_id
142 assert res.parents[3].group.name == "Country 2, Region 1, Foodies"
143 assert res.parents[3].group.slug == "country-2-region-1-foodies"
144 assert res.parents[3].group.description == "Description for Country 2, Region 1, Foodies"
145 assert res.main_page.type == pages_pb2.PAGE_TYPE_MAIN_PAGE
146 assert res.main_page.slug == "main-page-for-the-country-2-region-1-foodies-community"
147 assert res.main_page.last_editor_user_id == 6
148 assert res.main_page.creator_user_id == 6
149 assert res.main_page.owner_group_id == c2r1foodies_id
150 assert res.main_page.title == "Main page for the Country 2, Region 1, Foodies community"
151 assert res.main_page.content == "There is nothing here yet..."
152 assert not res.main_page.can_edit
153 assert not res.main_page.can_moderate
154 assert res.main_page.editor_user_ids == [6]
155 assert not res.member
156 assert not res.admin
157 assert res.member_count == 2
158 assert res.admin_count == 1
160 @staticmethod
161 def test_ListAdmins(testing_communities):
162 # implicitly tests visibility and blocking, since all groups have invisible, blocked, and blocking admin
163 with session_scope() as session:
164 user1_id, token1 = get_user_id_and_token(session, "user1")
165 user2_id, token2 = get_user_id_and_token(session, "user2")
166 hitchhikers_id = get_group_id(session, "Hitchhikers")
167 c1r2foodies_id = get_group_id(session, "Country 1, Region 2, Foodies")
169 with groups_session(token1) as api:
170 res = api.ListAdmins(
171 groups_pb2.ListAdminsReq(
172 group_id=hitchhikers_id,
173 )
174 )
175 assert res.admin_user_ids == [user1_id, user2_id]
177 res = api.ListAdmins(
178 groups_pb2.ListAdminsReq(
179 group_id=c1r2foodies_id,
180 )
181 )
182 assert res.admin_user_ids == [user2_id]
184 @staticmethod
185 def test_ListMembers(testing_communities):
186 # implicitly tests visibility and blocking, since all groups have invisible, blocked, and blocking member
187 with session_scope() as session:
188 user1_id, token1 = get_user_id_and_token(session, "user1")
189 user2_id, token2 = get_user_id_and_token(session, "user2")
190 user4_id, token4 = get_user_id_and_token(session, "user4")
191 user5_id, token5 = get_user_id_and_token(session, "user5")
192 user8_id, token8 = get_user_id_and_token(session, "user8")
193 hitchhikers_id = get_group_id(session, "Hitchhikers")
194 c1r2foodies_id = get_group_id(session, "Country 1, Region 2, Foodies")
196 with groups_session(token1) as api:
197 res = api.ListMembers(
198 groups_pb2.ListMembersReq(
199 group_id=hitchhikers_id,
200 )
201 )
202 assert res.member_user_ids == [user1_id, user2_id, user5_id, user8_id]
204 res = api.ListMembers(
205 groups_pb2.ListMembersReq(
206 group_id=c1r2foodies_id,
207 )
208 )
209 assert res.member_user_ids == [user2_id, user4_id, user5_id]
211 @staticmethod
212 def test_ListDiscussions(testing_communities):
213 with session_scope() as session:
214 user1_id, token1 = get_user_id_and_token(session, "user1")
215 hitchhikers_id = get_group_id(session, "Hitchhikers")
217 with groups_session(token1) as api:
218 res = api.ListDiscussions(
219 groups_pb2.ListDiscussionsReq(
220 group_id=hitchhikers_id,
221 page_size=5,
222 )
223 )
224 assert [d.title for d in res.discussions] == [
225 "Discussion title 8",
226 "Discussion title 9",
227 "Discussion title 10",
228 "Discussion title 11",
229 "Discussion title 12",
230 ]
231 for d in res.discussions:
232 assert d.thread.thread_id > 0
233 assert d.thread.num_responses == 0
235 res = api.ListDiscussions(
236 groups_pb2.ListDiscussionsReq(
237 group_id=hitchhikers_id,
238 page_token=res.next_page_token,
239 page_size=5,
240 )
241 )
242 assert [d.title for d in res.discussions] == [
243 "Discussion title 13",
244 "Discussion title 14",
245 ]
246 for d in res.discussions:
247 assert d.thread.thread_id > 0
248 assert d.thread.num_responses == 0
250 @staticmethod
251 def test_ListUserGroups(testing_communities):
252 with session_scope() as session:
253 user1_id, token1 = get_user_id_and_token(session, "user1")
254 hitchhikers_id = get_group_id(session, "Hitchhikers")
255 foodies_id = get_group_id(session, "Country 1, Region 1, Foodies")
256 skaters_id = get_group_id(session, "Country 1, Region 1, Skaters")
258 # List user1's groups from user1's account
259 with groups_session(token1) as api:
260 res = api.ListUserGroups(groups_pb2.ListUserGroupsReq())
261 assert [g.group_id for g in res.groups] == [hitchhikers_id, foodies_id, skaters_id]
263 @staticmethod
264 def test_ListOtherUserGroups(testing_communities):
265 with session_scope() as session:
266 user1_id, token1 = get_user_id_and_token(session, "user1")
267 user2_id, token2 = get_user_id_and_token(session, "user2")
268 hitchhikers_id = get_group_id(session, "Hitchhikers")
269 foodies_id = get_group_id(session, "Country 1, Region 1, Foodies")
270 skaters_id = get_group_id(session, "Country 1, Region 1, Skaters")
272 # List user1's groups from user2's account
273 with groups_session(token2) as api:
274 res = api.ListUserGroups(groups_pb2.ListUserGroupsReq(user_id=user1_id))
275 assert [g.group_id for g in res.groups] == [hitchhikers_id, foodies_id, skaters_id]
277 @staticmethod
278 def test_ListEvents(testing_communities):
279 with session_scope() as session:
280 user7_id, token7 = get_user_id_and_token(session, "user7")
281 hitchhikers_id = get_group_id(session, "Hitchhikers")
283 with groups_session(token7) as api:
284 res = api.ListEvents(
285 groups_pb2.ListEventsReq(
286 group_id=hitchhikers_id,
287 page_size=3,
288 )
289 )
290 assert [d.title for d in res.events] == [
291 "Event title 7",
292 "Event title 8",
293 "Event title 9",
294 ]
296 res = api.ListEvents(
297 groups_pb2.ListEventsReq(
298 group_id=hitchhikers_id,
299 page_token=res.next_page_token,
300 page_size=2,
301 )
302 )
303 assert [d.title for d in res.events] == [
304 "Event title 10",
305 "Event title 11",
306 ]
308 res = api.ListEvents(
309 groups_pb2.ListEventsReq(
310 group_id=hitchhikers_id,
311 page_token=res.next_page_token,
312 page_size=2,
313 )
314 )
315 assert [d.title for d in res.events] == [
316 "Event title 12",
317 ]
318 assert not res.next_page_token
321def test_JoinGroup_and_LeaveGroup(testing_communities):
322 # these tests are separate from above as they mutate the database
323 with session_scope() as session:
324 user_id, token = get_user_id_and_token(session, "user3")
325 h_id = get_group_id(session, "Hitchhikers")
327 with groups_session(token) as api:
328 # not in group at start
329 assert not api.GetGroup(groups_pb2.GetGroupReq(group_id=h_id)).member
331 # can't leave
332 with pytest.raises(grpc.RpcError) as e:
333 res = api.LeaveGroup(
334 groups_pb2.LeaveGroupReq(
335 group_id=h_id,
336 )
337 )
338 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
339 assert e.value.details() == "You're not in that group."
341 # didn't magically join
342 assert not api.GetGroup(groups_pb2.GetGroupReq(group_id=h_id)).member
344 # but can join
345 res = api.JoinGroup(
346 groups_pb2.JoinGroupReq(
347 group_id=h_id,
348 )
349 )
351 # should be there now
352 assert api.GetGroup(groups_pb2.GetGroupReq(group_id=h_id)).member
354 # can't join again
355 with pytest.raises(grpc.RpcError) as e:
356 res = api.JoinGroup(
357 groups_pb2.JoinGroupReq(
358 group_id=h_id,
359 )
360 )
361 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
362 assert e.value.details() == "You're already in that group."
364 # didn't magically leave
365 assert api.GetGroup(groups_pb2.GetGroupReq(group_id=h_id)).member
367 # now we can leave though
368 res = api.LeaveGroup(
369 groups_pb2.LeaveGroupReq(
370 group_id=h_id,
371 )
372 )
374 # managed to leave
375 assert not api.GetGroup(groups_pb2.GetGroupReq(group_id=h_id)).member
378def test_LeaveGroup_regression(db):
379 # see test_LeaveCommunity_regression
381 # admin
382 user1, token1 = generate_user(username="user1", geom=create_1d_point(200), geom_radius=0.1)
383 # joiner/leaver
384 user2, token2 = generate_user(username="user2", geom=create_1d_point(201), geom_radius=0.1)
386 with session_scope() as session:
387 c0 = create_community(session, 0, 100, "Community 0", [user1], [], None)
388 g1 = create_group(session, "Group 1", [user1], [], c0)
389 g2 = create_group(session, "Group 2", [user1], [], c0)
390 g1_id = g1.id
391 g2_id = g2.id
393 enforce_community_memberships()
395 with groups_session(token1) as api:
396 assert api.GetGroup(groups_pb2.GetGroupReq(group_id=g1_id)).member
397 assert api.GetGroup(groups_pb2.GetGroupReq(group_id=g2_id)).member
399 with groups_session(token2) as api:
400 # first check we're not in any groups
401 assert not api.GetGroup(groups_pb2.GetGroupReq(group_id=g1_id)).member
402 assert not api.GetGroup(groups_pb2.GetGroupReq(group_id=g2_id)).member
404 # join some groups
405 api.JoinGroup(groups_pb2.JoinGroupReq(group_id=g1_id))
406 api.JoinGroup(groups_pb2.JoinGroupReq(group_id=g2_id))
408 # check memberships
409 assert api.GetGroup(groups_pb2.GetGroupReq(group_id=g1_id)).member
410 assert api.GetGroup(groups_pb2.GetGroupReq(group_id=g2_id)).member
412 # leave just g2
413 api.LeaveGroup(groups_pb2.LeaveGroupReq(group_id=g2_id))
415 # check memberships
416 assert api.GetGroup(groups_pb2.GetGroupReq(group_id=g1_id)).member
417 assert not api.GetGroup(groups_pb2.GetGroupReq(group_id=g2_id)).member
420# TODO: also requires implementing content transfer functionality
421# Note: allegedly groups cannot contain content other than discussions!
423# def test_ListPlaces(db, testing_communities):
424# pass
426# def test_ListGuides(db, testing_communities):
427# pass
429# def test_ListEvents(db, testing_communities):
430# pass