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

1from datetime import timedelta 

2 

3import grpc 

4import pytest 

5from google.protobuf import wrappers_pb2 

6 

7from couchers import errors 

8from couchers.db import session_scope 

9from couchers.materialized_views import refresh_materialized_views 

10from couchers.models import ( 

11 Cluster, 

12 ClusterRole, 

13 ClusterSubscription, 

14 Node, 

15 Page, 

16 PageType, 

17 PageVersion, 

18 SignupFlow, 

19 Thread, 

20) 

21from couchers.sql import couchers_select as select 

22from couchers.tasks import enforce_community_memberships 

23from couchers.utils import Timestamp_from_datetime, create_coordinate, create_polygon_lat_lng, now, to_multi 

24from proto import api_pb2, auth_pb2, communities_pb2, discussions_pb2, events_pb2, pages_pb2 

25from tests.test_auth import get_session_cookie_tokens 

26from tests.test_fixtures import ( # noqa 

27 auth_api_session, 

28 communities_session, 

29 db, 

30 discussions_session, 

31 events_session, 

32 generate_user, 

33 get_user_id_and_token, 

34 pages_session, 

35 recreate_database, 

36 testconfig, 

37) 

38 

39 

40@pytest.fixture(autouse=True) 

41def _(testconfig): 

42 pass 

43 

44 

45# For testing purposes, restrict ourselves to a 1D-world, consisting of "intervals" that have width 2, and coordinates 

46# that are points at (x, 1). 

47# 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 

48# mostly fine 

49 

50 

51def create_1d_polygon(lb, ub): 

52 # given a lower bound and upper bound on x, creates the given interval 

53 return create_polygon_lat_lng([[lb, 0], [lb, 2], [ub, 2], [ub, 0], [lb, 0]]) 

54 

55 

56def create_1d_point(x): 

57 return create_coordinate(x, 1) 

58 

59 

60def create_community(session, interval_lb, interval_ub, name, admins, extra_members, parent): 

61 node = Node( 

62 geom=to_multi(create_1d_polygon(interval_lb, interval_ub)), 

63 parent_node=parent, 

64 ) 

65 session.add(node) 

66 cluster = Cluster( 

67 name=f"{name}", 

68 description=f"Description for {name}", 

69 parent_node=node, 

70 is_official_cluster=True, 

71 ) 

72 session.add(cluster) 

73 main_page = Page( 

74 parent_node=cluster.parent_node, 

75 creator_user_id=admins[0].id, 

76 owner_cluster=cluster, 

77 type=PageType.main_page, 

78 thread=Thread(), 

79 ) 

80 session.add(main_page) 

81 page_version = PageVersion( 

82 page=main_page, 

83 editor_user_id=admins[0].id, 

84 title=f"Main page for the {name} community", 

85 content="There is nothing here yet...", 

86 ) 

87 session.add(page_version) 

88 for admin in admins: 

89 cluster.cluster_subscriptions.append( 

90 ClusterSubscription( 

91 user_id=admin.id, 

92 role=ClusterRole.admin, 

93 ) 

94 ) 

95 for member in extra_members: 

96 cluster.cluster_subscriptions.append( 

97 ClusterSubscription( 

98 user_id=member.id, 

99 role=ClusterRole.member, 

100 ) 

101 ) 

102 session.commit() 

103 # other members will be added by enforce_community_memberships() 

104 return node 

105 

106 

107def create_group(session, name, admins, members, parent_community): 

108 cluster = Cluster( 

109 name=f"{name}", 

110 description=f"Description for {name}", 

111 parent_node=parent_community, 

112 ) 

113 session.add(cluster) 

114 main_page = Page( 

115 parent_node=cluster.parent_node, 

116 creator_user=admins[0], 

117 owner_cluster=cluster, 

118 type=PageType.main_page, 

119 thread=Thread(), 

120 ) 

121 session.add(main_page) 

122 page_version = PageVersion( 

123 page=main_page, 

124 editor_user=admins[0], 

125 title=f"Main page for the {name} community", 

126 content="There is nothing here yet...", 

127 ) 

128 session.add(page_version) 

129 for admin in admins: 

130 cluster.cluster_subscriptions.append( 

131 ClusterSubscription( 

132 user=admin, 

133 role=ClusterRole.admin, 

134 ) 

135 ) 

136 for member in members: 

137 cluster.cluster_subscriptions.append( 

138 ClusterSubscription( 

139 user=member, 

140 role=ClusterRole.member, 

141 ) 

142 ) 

143 session.commit() 

144 return cluster 

145 

146 

147def create_place(token, title, content, address, x): 

148 with pages_session(token) as api: 

149 res = api.CreatePlace( 

150 pages_pb2.CreatePlaceReq( 

151 title=title, 

152 content=content, 

153 address=address, 

154 location=pages_pb2.Coordinate( 

155 lat=x, 

156 lng=1, 

157 ), 

158 ) 

159 ) 

160 

161 

162def create_discussion(token, community_id, group_id, title, content): 

163 # set group_id or community_id to None 

164 with discussions_session(token) as api: 

165 res = api.CreateDiscussion( 

166 discussions_pb2.CreateDiscussionReq( 

167 title=title, 

168 content=content, 

169 owner_community_id=community_id, 

170 owner_group_id=group_id, 

171 ) 

172 ) 

173 

174 

175def create_event(token, community_id, group_id, title, content, start_td): 

176 with events_session(token) as api: 

177 res = api.CreateEvent( 

178 events_pb2.CreateEventReq( 

179 title=title, 

180 content=content, 

181 offline_information=events_pb2.OfflineEventInformation( 

182 address="Near Null Island", 

183 lat=0.1, 

184 lng=0.2, 

185 ), 

186 start_time=Timestamp_from_datetime(now() + start_td), 

187 end_time=Timestamp_from_datetime(now() + start_td + timedelta(hours=2)), 

188 timezone="UTC", 

189 ) 

190 ) 

191 api.TransferEvent( 

192 events_pb2.TransferEventReq( 

193 event_id=res.event_id, 

194 new_owner_community_id=community_id, 

195 new_owner_group_id=group_id, 

196 ) 

197 ) 

198 

199 

200def get_community_id(session, community_name): 

201 return ( 

202 session.execute(select(Cluster).where(Cluster.is_official_cluster).where(Cluster.name == community_name)) 

203 .scalar_one() 

204 .parent_node_id 

205 ) 

206 

207 

208def get_group_id(session, group_name): 

209 return ( 

210 session.execute(select(Cluster).where(~Cluster.is_official_cluster).where(Cluster.name == group_name)) 

211 .scalar_one() 

212 .id 

213 ) 

214 

215 

216@pytest.fixture(scope="class") 

217def testing_communities(testconfig): 

218 recreate_database() 

219 user1, token1 = generate_user(username="user1", geom=create_1d_point(1), geom_radius=0.1) 

220 user2, token2 = generate_user(username="user2", geom=create_1d_point(2), geom_radius=0.1) 

221 user3, token3 = generate_user(username="user3", geom=create_1d_point(3), geom_radius=0.1) 

222 user4, token4 = generate_user(username="user4", geom=create_1d_point(8), geom_radius=0.1) 

223 user5, token5 = generate_user(username="user5", geom=create_1d_point(6), geom_radius=0.1) 

224 user6, token6 = generate_user(username="user6", geom=create_1d_point(65), geom_radius=0.1) 

225 user7, token7 = generate_user(username="user7", geom=create_1d_point(80), geom_radius=0.1) 

226 user8, token8 = generate_user(username="user8", geom=create_1d_point(51), geom_radius=0.1) 

227 

228 with session_scope() as session: 

229 w = create_community(session, 0, 100, "Global", [user1, user3, user7], [], None) 

230 c2 = create_community(session, 52, 100, "Country 2", [user6, user7], [], w) 

231 c2r1 = create_community(session, 52, 71, "Country 2, Region 1", [user6], [user8], c2) 

232 c2r1c1 = create_community(session, 53, 70, "Country 2, Region 1, City 1", [user8], [], c2r1) 

233 c1 = create_community(session, 0, 50, "Country 1", [user1, user2], [], w) 

234 c1r1 = create_community(session, 0, 10, "Country 1, Region 1", [user1, user2], [], c1) 

235 c1r1c1 = create_community(session, 0, 5, "Country 1, Region 1, City 1", [user2], [], c1r1) 

236 c1r1c2 = create_community(session, 7, 10, "Country 1, Region 1, City 2", [user4, user5], [user2], c1r1) 

237 c1r2 = create_community(session, 20, 25, "Country 1, Region 2", [user2], [], c1) 

238 c1r2c1 = create_community(session, 21, 23, "Country 1, Region 2, City 1", [user2], [], c1r2) 

239 

240 h = create_group(session, "Hitchhikers", [user1, user2], [user5, user8], w) 

241 create_group(session, "Country 1, Region 1, Foodies", [user1], [user2, user4], c1r1) 

242 create_group(session, "Country 1, Region 1, Skaters", [user2], [user1], c1r1) 

243 create_group(session, "Country 1, Region 2, Foodies", [user2], [user4, user5], c1r2) 

244 create_group(session, "Country 2, Region 1, Foodies", [user6], [user7], c2r1) 

245 

246 w_id = w.id 

247 c1r1c2_id = c1r1c2.id 

248 h_id = h.id 

249 c1_id = c1.id 

250 

251 create_discussion(token1, w_id, None, "Discussion title 1", "Discussion content 1") 

252 create_discussion(token3, w_id, None, "Discussion title 2", "Discussion content 2") 

253 create_discussion(token3, w_id, None, "Discussion title 3", "Discussion content 3") 

254 create_discussion(token3, w_id, None, "Discussion title 4", "Discussion content 4") 

255 create_discussion(token3, w_id, None, "Discussion title 5", "Discussion content 5") 

256 create_discussion(token3, w_id, None, "Discussion title 6", "Discussion content 6") 

257 create_discussion(token4, c1r1c2_id, None, "Discussion title 7", "Discussion content 7") 

258 create_discussion(token5, None, h_id, "Discussion title 8", "Discussion content 8") 

259 create_discussion(token1, None, h_id, "Discussion title 9", "Discussion content 9") 

260 create_discussion(token2, None, h_id, "Discussion title 10", "Discussion content 10") 

261 create_discussion(token3, None, h_id, "Discussion title 11", "Discussion content 11") 

262 create_discussion(token4, None, h_id, "Discussion title 12", "Discussion content 12") 

263 create_discussion(token5, None, h_id, "Discussion title 13", "Discussion content 13") 

264 create_discussion(token8, None, h_id, "Discussion title 14", "Discussion content 14") 

265 

266 create_event(token3, c1_id, None, "Event title 1", "Event content 1", timedelta(hours=1)) 

267 create_event(token1, c1_id, None, "Event title 2", "Event content 2", timedelta(hours=2)) 

268 create_event(token3, c1_id, None, "Event title 3", "Event content 3", timedelta(hours=3)) 

269 create_event(token1, c1_id, None, "Event title 4", "Event content 4", timedelta(hours=4)) 

270 create_event(token3, c1_id, None, "Event title 5", "Event content 5", timedelta(hours=5)) 

271 create_event(token1, c1_id, None, "Event title 6", "Event content 6", timedelta(hours=6)) 

272 create_event(token2, None, h_id, "Event title 7", "Event content 7", timedelta(hours=7)) 

273 create_event(token2, None, h_id, "Event title 8", "Event content 8", timedelta(hours=8)) 

274 create_event(token2, None, h_id, "Event title 9", "Event content 9", timedelta(hours=9)) 

275 create_event(token2, None, h_id, "Event title 10", "Event content 10", timedelta(hours=10)) 

276 create_event(token2, None, h_id, "Event title 11", "Event content 11", timedelta(hours=11)) 

277 create_event(token2, None, h_id, "Event title 12", "Event content 12", timedelta(hours=12)) 

278 

279 enforce_community_memberships() 

280 

281 create_place(token1, "Country 1, Region 1, Attraction", "Place content", "Somewhere in c1r1", 6) 

282 create_place(token2, "Country 1, Region 1, City 1, Attraction 1", "Place content", "Somewhere in c1r1c1", 3) 

283 create_place(token2, "Country 1, Region 1, City 1, Attraction 2", "Place content", "Somewhere in c1r1c1", 4) 

284 create_place(token8, "Global, Attraction", "Place content", "Somewhere in w", 51.5) 

285 create_place(token6, "Country 2, Region 1, Attraction", "Place content", "Somewhere in c2r1", 59) 

286 

287 refresh_materialized_views(None) 

288 

289 yield 

290 

291 

292class TestCommunities: 

293 @staticmethod 

294 def test_GetCommunity(testing_communities): 

295 with session_scope() as session: 

296 user2_id, token2 = get_user_id_and_token(session, "user2") 

297 w_id = get_community_id(session, "Global") 

298 c1_id = get_community_id(session, "Country 1") 

299 c1r1_id = get_community_id(session, "Country 1, Region 1") 

300 c1r1c1_id = get_community_id(session, "Country 1, Region 1, City 1") 

301 c2_id = get_community_id(session, "Country 2") 

302 

303 with communities_session(token2) as api: 

304 res = api.GetCommunity( 

305 communities_pb2.GetCommunityReq( 

306 community_id=w_id, 

307 ) 

308 ) 

309 assert res.name == "Global" 

310 assert res.slug == "global" 

311 assert res.description == "Description for Global" 

312 assert len(res.parents) == 1 

313 assert res.parents[0].HasField("community") 

314 assert res.parents[0].community.community_id == w_id 

315 assert res.parents[0].community.name == "Global" 

316 assert res.parents[0].community.slug == "global" 

317 assert res.parents[0].community.description == "Description for Global" 

318 assert res.main_page.type == pages_pb2.PAGE_TYPE_MAIN_PAGE 

319 assert res.main_page.slug == "main-page-for-the-global-community" 

320 assert res.main_page.last_editor_user_id == 1 

321 assert res.main_page.creator_user_id == 1 

322 assert res.main_page.owner_community_id == w_id 

323 assert res.main_page.title == "Main page for the Global community" 

324 assert res.main_page.content == "There is nothing here yet..." 

325 assert not res.main_page.can_edit 

326 assert not res.main_page.can_moderate 

327 assert res.main_page.editor_user_ids == [1] 

328 assert res.member 

329 assert not res.admin 

330 assert res.member_count == 8 

331 assert res.admin_count == 3 

332 

333 res = api.GetCommunity( 

334 communities_pb2.GetCommunityReq( 

335 community_id=c1r1c1_id, 

336 ) 

337 ) 

338 assert res.community_id == c1r1c1_id 

339 assert res.name == "Country 1, Region 1, City 1" 

340 assert res.slug == "country-1-region-1-city-1" 

341 assert res.description == "Description for Country 1, Region 1, City 1" 

342 assert len(res.parents) == 4 

343 assert res.parents[0].HasField("community") 

344 assert res.parents[0].community.community_id == w_id 

345 assert res.parents[0].community.name == "Global" 

346 assert res.parents[0].community.slug == "global" 

347 assert res.parents[0].community.description == "Description for Global" 

348 assert res.parents[1].HasField("community") 

349 assert res.parents[1].community.community_id == c1_id 

350 assert res.parents[1].community.name == "Country 1" 

351 assert res.parents[1].community.slug == "country-1" 

352 assert res.parents[1].community.description == "Description for Country 1" 

353 assert res.parents[2].HasField("community") 

354 assert res.parents[2].community.community_id == c1r1_id 

355 assert res.parents[2].community.name == "Country 1, Region 1" 

356 assert res.parents[2].community.slug == "country-1-region-1" 

357 assert res.parents[2].community.description == "Description for Country 1, Region 1" 

358 assert res.parents[3].HasField("community") 

359 assert res.parents[3].community.community_id == c1r1c1_id 

360 assert res.parents[3].community.name == "Country 1, Region 1, City 1" 

361 assert res.parents[3].community.slug == "country-1-region-1-city-1" 

362 assert res.parents[3].community.description == "Description for Country 1, Region 1, City 1" 

363 assert res.main_page.type == pages_pb2.PAGE_TYPE_MAIN_PAGE 

364 assert res.main_page.slug == "main-page-for-the-country-1-region-1-city-1-community" 

365 assert res.main_page.last_editor_user_id == 2 

366 assert res.main_page.creator_user_id == 2 

367 assert res.main_page.owner_community_id == c1r1c1_id 

368 assert res.main_page.title == "Main page for the Country 1, Region 1, City 1 community" 

369 assert res.main_page.content == "There is nothing here yet..." 

370 assert res.main_page.can_edit 

371 assert res.main_page.can_moderate 

372 assert res.main_page.editor_user_ids == [2] 

373 assert res.member 

374 assert res.admin 

375 assert res.member_count == 3 

376 assert res.admin_count == 1 

377 

378 res = api.GetCommunity( 

379 communities_pb2.GetCommunityReq( 

380 community_id=c2_id, 

381 ) 

382 ) 

383 assert res.community_id == c2_id 

384 assert res.name == "Country 2" 

385 assert res.slug == "country-2" 

386 assert res.description == "Description for Country 2" 

387 assert len(res.parents) == 2 

388 assert res.parents[0].HasField("community") 

389 assert res.parents[0].community.community_id == w_id 

390 assert res.parents[0].community.name == "Global" 

391 assert res.parents[0].community.slug == "global" 

392 assert res.parents[0].community.description == "Description for Global" 

393 assert res.parents[1].HasField("community") 

394 assert res.parents[1].community.community_id == c2_id 

395 assert res.parents[1].community.name == "Country 2" 

396 assert res.parents[1].community.slug == "country-2" 

397 assert res.parents[1].community.description == "Description for Country 2" 

398 assert res.main_page.type == pages_pb2.PAGE_TYPE_MAIN_PAGE 

399 assert res.main_page.slug == "main-page-for-the-country-2-community" 

400 assert res.main_page.last_editor_user_id == 6 

401 assert res.main_page.creator_user_id == 6 

402 assert res.main_page.owner_community_id == c2_id 

403 assert res.main_page.title == "Main page for the Country 2 community" 

404 assert res.main_page.content == "There is nothing here yet..." 

405 assert not res.main_page.can_edit 

406 assert not res.main_page.can_moderate 

407 assert res.main_page.editor_user_ids == [6] 

408 assert not res.member 

409 assert not res.admin 

410 assert res.member_count == 2 

411 assert res.admin_count == 2 

412 

413 @staticmethod 

414 def test_ListCommunities(testing_communities): 

415 with session_scope() as session: 

416 user1_id, token1 = get_user_id_and_token(session, "user1") 

417 c1_id = get_community_id(session, "Country 1") 

418 c1r1_id = get_community_id(session, "Country 1, Region 1") 

419 c1r2_id = get_community_id(session, "Country 1, Region 2") 

420 

421 with communities_session(token1) as api: 

422 res = api.ListCommunities( 

423 communities_pb2.ListCommunitiesReq( 

424 community_id=c1_id, 

425 ) 

426 ) 

427 assert [c.community_id for c in res.communities] == [c1r1_id, c1r2_id] 

428 

429 @staticmethod 

430 def test_ListCommunities_all(testing_communities): 

431 with session_scope() as session: 

432 user1_id, token1 = get_user_id_and_token(session, "user1") 

433 w_id = get_community_id(session, "Global") 

434 c1_id = get_community_id(session, "Country 1") 

435 c1r1_id = get_community_id(session, "Country 1, Region 1") 

436 c1r1c1_id = get_community_id(session, "Country 1, Region 1, City 1") 

437 c1r1c2_id = get_community_id(session, "Country 1, Region 1, City 2") 

438 c1r2_id = get_community_id(session, "Country 1, Region 2") 

439 c1r2c1_id = get_community_id(session, "Country 1, Region 2, City 1") 

440 c2_id = get_community_id(session, "Country 2") 

441 c2r1_id = get_community_id(session, "Country 2, Region 1") 

442 c2r1c1_id = get_community_id(session, "Country 2, Region 1, City 1") 

443 

444 # Fetch all communities ordered by name 

445 with communities_session(token1) as api: 

446 res = api.ListCommunities( 

447 communities_pb2.ListCommunitiesReq( 

448 page_size=5, 

449 ) 

450 ) 

451 assert [c.community_id for c in res.communities] == [c1_id, c1r1_id, c1r1c1_id, c1r1c2_id, c1r2_id] 

452 res = api.ListCommunities( 

453 communities_pb2.ListCommunitiesReq( 

454 page_size=2, 

455 page_token=res.next_page_token, 

456 ) 

457 ) 

458 assert [c.community_id for c in res.communities] == [c1r2c1_id, c2_id] 

459 res = api.ListCommunities( 

460 communities_pb2.ListCommunitiesReq( 

461 page_size=5, 

462 page_token=res.next_page_token, 

463 ) 

464 ) 

465 assert [c.community_id for c in res.communities] == [c2r1_id, c2r1c1_id, w_id] 

466 

467 @staticmethod 

468 def test_ListUserCommunities(testing_communities): 

469 with session_scope() as session: 

470 user2_id, token2 = get_user_id_and_token(session, "user2") 

471 w_id = get_community_id(session, "Global") 

472 c1_id = get_community_id(session, "Country 1") 

473 c1r1_id = get_community_id(session, "Country 1, Region 1") 

474 c1r1c1_id = get_community_id(session, "Country 1, Region 1, City 1") 

475 c1r1c2_id = get_community_id(session, "Country 1, Region 1, City 2") 

476 c1r2_id = get_community_id(session, "Country 1, Region 2") 

477 c1r2c1_id = get_community_id(session, "Country 1, Region 2, City 1") 

478 

479 # Fetch user2's communities from user2's account 

480 with communities_session(token2) as api: 

481 res = api.ListUserCommunities(communities_pb2.ListUserCommunitiesReq()) 

482 assert [c.community_id for c in res.communities] == [ 

483 w_id, 

484 c1_id, 

485 c1r1_id, 

486 c1r1c1_id, 

487 c1r1c2_id, 

488 c1r2_id, 

489 c1r2c1_id, 

490 ] 

491 

492 @staticmethod 

493 def test_ListOtherUserCommunities(testing_communities): 

494 with session_scope() as session: 

495 user1_id, token1 = get_user_id_and_token(session, "user1") 

496 user2_id, token2 = get_user_id_and_token(session, "user2") 

497 w_id = get_community_id(session, "Global") 

498 c1_id = get_community_id(session, "Country 1") 

499 c1r1_id = get_community_id(session, "Country 1, Region 1") 

500 c1r1c1_id = get_community_id(session, "Country 1, Region 1, City 1") 

501 c1r1c2_id = get_community_id(session, "Country 1, Region 1, City 2") 

502 c1r2_id = get_community_id(session, "Country 1, Region 2") 

503 c1r2c1_id = get_community_id(session, "Country 1, Region 2, City 1") 

504 

505 # Fetch user2's communities from user1's account 

506 with communities_session(token1) as api: 

507 res = api.ListUserCommunities(communities_pb2.ListUserCommunitiesReq(user_id=user2_id)) 

508 assert [c.community_id for c in res.communities] == [ 

509 w_id, 

510 c1_id, 

511 c1r1_id, 

512 c1r1c1_id, 

513 c1r1c2_id, 

514 c1r2_id, 

515 c1r2c1_id, 

516 ] 

517 

518 @staticmethod 

519 def test_ListGroups(testing_communities): 

520 with session_scope() as session: 

521 user1_id, token1 = get_user_id_and_token(session, "user1") 

522 user5_id, token5 = get_user_id_and_token(session, "user5") 

523 w_id = get_community_id(session, "Global") 

524 hitchhikers_id = get_group_id(session, "Hitchhikers") 

525 c1r1_id = get_community_id(session, "Country 1, Region 1") 

526 foodies_id = get_group_id(session, "Country 1, Region 1, Foodies") 

527 skaters_id = get_group_id(session, "Country 1, Region 1, Skaters") 

528 

529 with communities_session(token1) as api: 

530 res = api.ListGroups( 

531 communities_pb2.ListGroupsReq( 

532 community_id=c1r1_id, 

533 ) 

534 ) 

535 assert [g.group_id for g in res.groups] == [foodies_id, skaters_id] 

536 

537 with communities_session(token5) as api: 

538 res = api.ListGroups( 

539 communities_pb2.ListGroupsReq( 

540 community_id=w_id, 

541 ) 

542 ) 

543 assert len(res.groups) == 1 

544 assert res.groups[0].group_id == hitchhikers_id 

545 

546 @staticmethod 

547 def test_ListAdmins(testing_communities): 

548 with session_scope() as session: 

549 user1_id, token1 = get_user_id_and_token(session, "user1") 

550 user3_id, token3 = get_user_id_and_token(session, "user3") 

551 user4_id, token4 = get_user_id_and_token(session, "user4") 

552 user5_id, token5 = get_user_id_and_token(session, "user5") 

553 user7_id, token7 = get_user_id_and_token(session, "user7") 

554 w_id = get_community_id(session, "Global") 

555 c1r1c2_id = get_community_id(session, "Country 1, Region 1, City 2") 

556 

557 with communities_session(token1) as api: 

558 res = api.ListAdmins( 

559 communities_pb2.ListAdminsReq( 

560 community_id=w_id, 

561 ) 

562 ) 

563 assert res.admin_user_ids == [user1_id, user3_id, user7_id] 

564 

565 res = api.ListAdmins( 

566 communities_pb2.ListAdminsReq( 

567 community_id=c1r1c2_id, 

568 ) 

569 ) 

570 assert res.admin_user_ids == [user4_id, user5_id] 

571 

572 @staticmethod 

573 def test_AddAdmin(testing_communities): 

574 with session_scope() as session: 

575 user4_id, token4 = get_user_id_and_token(session, "user4") 

576 user5_id, _ = get_user_id_and_token(session, "user5") 

577 user2_id, _ = get_user_id_and_token(session, "user2") 

578 user8_id, token8 = get_user_id_and_token(session, "user8") 

579 node_id = get_community_id(session, "Country 1, Region 1, City 2") 

580 

581 with communities_session(token8) as api: 

582 with pytest.raises(grpc.RpcError) as err: 

583 api.AddAdmin(communities_pb2.AddAdminReq(community_id=node_id, user_id=user2_id)) 

584 assert err.value.code() == grpc.StatusCode.FAILED_PRECONDITION 

585 assert err.value.details() == errors.NODE_MODERATE_PERMISSION_DENIED 

586 

587 with communities_session(token4) as api: 

588 res = api.ListAdmins(communities_pb2.ListAdminsReq(community_id=node_id)) 

589 assert res.admin_user_ids == [user4_id, user5_id] 

590 

591 with pytest.raises(grpc.RpcError) as err: 

592 api.AddAdmin(communities_pb2.AddAdminReq(community_id=node_id, user_id=user8_id)) 

593 assert err.value.code() == grpc.StatusCode.FAILED_PRECONDITION 

594 assert err.value.details() == errors.USER_NOT_MEMBER 

595 

596 with pytest.raises(grpc.RpcError) as err: 

597 api.AddAdmin(communities_pb2.AddAdminReq(community_id=node_id, user_id=user5_id)) 

598 assert err.value.code() == grpc.StatusCode.FAILED_PRECONDITION 

599 assert err.value.details() == errors.USER_ALREADY_ADMIN 

600 

601 api.AddAdmin(communities_pb2.AddAdminReq(community_id=node_id, user_id=user2_id)) 

602 res = api.ListAdmins(communities_pb2.ListAdminsReq(community_id=node_id)) 

603 assert res.admin_user_ids == [user2_id, user4_id, user5_id] 

604 # Cleanup because database changes do not roll back 

605 api.RemoveAdmin(communities_pb2.RemoveAdminReq(community_id=node_id, user_id=user2_id)) 

606 

607 @staticmethod 

608 def test_RemoveAdmin(testing_communities): 

609 with session_scope() as session: 

610 user4_id, token4 = get_user_id_and_token(session, "user4") 

611 user5_id, _ = get_user_id_and_token(session, "user5") 

612 user2_id, _ = get_user_id_and_token(session, "user2") 

613 user8_id, token8 = get_user_id_and_token(session, "user8") 

614 node_id = get_community_id(session, "Country 1, Region 1, City 2") 

615 

616 with communities_session(token8) as api: 

617 with pytest.raises(grpc.RpcError) as err: 

618 api.AddAdmin(communities_pb2.AddAdminReq(community_id=node_id, user_id=user2_id)) 

619 assert err.value.code() == grpc.StatusCode.FAILED_PRECONDITION 

620 assert err.value.details() == errors.NODE_MODERATE_PERMISSION_DENIED 

621 

622 with communities_session(token4) as api: 

623 res = api.ListAdmins(communities_pb2.ListAdminsReq(community_id=node_id)) 

624 assert res.admin_user_ids == [user4_id, user5_id] 

625 

626 with pytest.raises(grpc.RpcError) as err: 

627 api.RemoveAdmin(communities_pb2.RemoveAdminReq(community_id=node_id, user_id=user8_id)) 

628 assert err.value.code() == grpc.StatusCode.FAILED_PRECONDITION 

629 assert err.value.details() == errors.USER_NOT_MEMBER 

630 

631 with pytest.raises(grpc.RpcError) as err: 

632 api.RemoveAdmin(communities_pb2.RemoveAdminReq(community_id=node_id, user_id=user2_id)) 

633 assert err.value.code() == grpc.StatusCode.FAILED_PRECONDITION 

634 assert err.value.details() == errors.USER_NOT_ADMIN 

635 

636 api.RemoveAdmin(communities_pb2.RemoveAdminReq(community_id=node_id, user_id=user5_id)) 

637 res = api.ListAdmins(communities_pb2.ListAdminsReq(community_id=node_id)) 

638 assert res.admin_user_ids == [user4_id] 

639 # Cleanup because database changes do not roll back 

640 api.AddAdmin(communities_pb2.AddAdminReq(community_id=node_id, user_id=user5_id)) 

641 

642 @staticmethod 

643 def test_ListMembers(testing_communities): 

644 with session_scope() as session: 

645 user1_id, token1 = get_user_id_and_token(session, "user1") 

646 user2_id, token2 = get_user_id_and_token(session, "user2") 

647 user3_id, token3 = get_user_id_and_token(session, "user3") 

648 user4_id, token4 = get_user_id_and_token(session, "user4") 

649 user5_id, token5 = get_user_id_and_token(session, "user5") 

650 user6_id, token6 = get_user_id_and_token(session, "user6") 

651 user7_id, token7 = get_user_id_and_token(session, "user7") 

652 user8_id, token8 = get_user_id_and_token(session, "user8") 

653 w_id = get_community_id(session, "Global") 

654 c1r1c2_id = get_community_id(session, "Country 1, Region 1, City 2") 

655 

656 with communities_session(token1) as api: 

657 res = api.ListMembers( 

658 communities_pb2.ListMembersReq( 

659 community_id=w_id, 

660 ) 

661 ) 

662 assert res.member_user_ids == [ 

663 user1_id, 

664 user2_id, 

665 user3_id, 

666 user4_id, 

667 user5_id, 

668 user6_id, 

669 user7_id, 

670 user8_id, 

671 ] 

672 

673 res = api.ListMembers( 

674 communities_pb2.ListMembersReq( 

675 community_id=c1r1c2_id,