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