Coverage for src/tests/test_groups.py: 100%

255 statements  

« prev     ^ index     » next       coverage.py v7.5.0, created at 2024-07-22 17:19 +0000

1import grpc 

2import pytest 

3 

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 

17 

18 

19@pytest.fixture(autouse=True) 

20def _(testconfig): 

21 pass 

22 

23 

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") 

38 

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 

74 

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 

119 

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 

164 

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") 

173 

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] 

181 

182 res = api.ListAdmins( 

183 groups_pb2.ListAdminsReq( 

184 group_id=c1r2foodies_id, 

185 ) 

186 ) 

187 assert res.admin_user_ids == [user2_id] 

188 

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") 

200 

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] 

208 

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] 

215 

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") 

221 

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 

239 

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 

254 

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") 

262 

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] 

267 

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") 

276 

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] 

281 

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") 

287 

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 ] 

300 

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 ] 

312 

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 

324 

325 

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") 

331 

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 

335 

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 

345 

346 # didn't magically join 

347 assert not api.GetGroup(groups_pb2.GetGroupReq(group_id=h_id)).member 

348 

349 # but can join 

350 res = api.JoinGroup( 

351 groups_pb2.JoinGroupReq( 

352 group_id=h_id, 

353 ) 

354 ) 

355 

356 # should be there now 

357 assert api.GetGroup(groups_pb2.GetGroupReq(group_id=h_id)).member 

358 

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 

368 

369 # didn't magically leave 

370 assert api.GetGroup(groups_pb2.GetGroupReq(group_id=h_id)).member 

371 

372 # now we can leave though 

373 res = api.LeaveGroup( 

374 groups_pb2.LeaveGroupReq( 

375 group_id=h_id, 

376 ) 

377 ) 

378 

379 # managed to leave 

380 assert not api.GetGroup(groups_pb2.GetGroupReq(group_id=h_id)).member 

381 

382 

383def test_LeaveGroup_regression(db): 

384 # see test_LeaveCommunity_regression 

385 

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) 

390 

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 

397 

398 enforce_community_memberships() 

399 

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 

403 

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 

408 

409 # join some groups 

410 api.JoinGroup(groups_pb2.JoinGroupReq(group_id=g1_id)) 

411 api.JoinGroup(groups_pb2.JoinGroupReq(group_id=g2_id)) 

412 

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 

416 

417 # leave just g2 

418 api.LeaveGroup(groups_pb2.LeaveGroupReq(group_id=g2_id)) 

419 

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 

423 

424 

425# TODO: also requires implementing content transfer functionality 

426# Note: allegedly groups cannot contain content other than discussions! 

427 

428# def test_ListPlaces(db, testing_communities): 

429# pass 

430 

431# def test_ListGuides(db, testing_communities): 

432# pass 

433 

434# def test_ListEvents(db, testing_communities): 

435# pass