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