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 c1 = create_community(session, 0, 50, "Country 1", [user1, user2], [], w) 

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

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

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

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

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

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

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

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

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 with communities_session(token1) as api: 

445 res = api.ListCommunities( 

446 communities_pb2.ListCommunitiesReq( 

447 page_size=5, 

448 ) 

449 ) 

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

451 res = api.ListCommunities( 

452 communities_pb2.ListCommunitiesReq( 

453 page_size=2, 

454 page_token=res.next_page_token, 

455 ) 

456 ) 

457 assert [c.community_id for c in res.communities] == [c1r2_id, c1r2c1_id] 

458 res = api.ListCommunities( 

459 communities_pb2.ListCommunitiesReq( 

460 page_size=5, 

461 page_token=res.next_page_token, 

462 ) 

463 ) 

464 assert [c.community_id for c in res.communities] == [c2_id, c2r1_id, c2r1c1_id] 

465 

466 @staticmethod 

467 def test_ListUserCommunities(testing_communities): 

468 with session_scope() as session: 

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

470 w_id = get_community_id(session, "Global") 

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

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

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

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

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

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

477 

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

479 with communities_session(token2) as api: 

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

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

482 w_id, 

483 c1_id, 

484 c1r1_id, 

485 c1r1c1_id, 

486 c1r1c2_id, 

487 c1r2_id, 

488 c1r2c1_id, 

489 ] 

490 

491 @staticmethod 

492 def test_ListOtherUserCommunities(testing_communities): 

493 with session_scope() as session: 

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

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

496 w_id = get_community_id(session, "Global") 

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

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

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

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

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

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

503 

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

505 with communities_session(token1) as api: 

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

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

508 w_id, 

509 c1_id, 

510 c1r1_id, 

511 c1r1c1_id, 

512 c1r1c2_id, 

513 c1r2_id, 

514 c1r2c1_id, 

515 ] 

516 

517 @staticmethod 

518 def test_ListGroups(testing_communities): 

519 with session_scope() as session: 

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

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

522 w_id = get_community_id(session, "Global") 

523 hitchhikers_id = get_group_id(session, "Hitchhikers") 

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

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

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

527 

528 with communities_session(token1) as api: 

529 res = api.ListGroups( 

530 communities_pb2.ListGroupsReq( 

531 community_id=c1r1_id, 

532 ) 

533 ) 

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

535 

536 with communities_session(token5) as api: 

537 res = api.ListGroups( 

538 communities_pb2.ListGroupsReq( 

539 community_id=w_id, 

540 ) 

541 ) 

542 assert len(res.groups) == 1 

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

544 

545 @staticmethod 

546 def test_ListAdmins(testing_communities): 

547 with session_scope() as session: 

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

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

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

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

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

553 w_id = get_community_id(session, "Global") 

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

555 

556 with communities_session(token1) as api: 

557 res = api.ListAdmins( 

558 communities_pb2.ListAdminsReq( 

559 community_id=w_id, 

560 ) 

561 ) 

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

563 

564 res = api.ListAdmins( 

565 communities_pb2.ListAdminsReq( 

566 community_id=c1r1c2_id, 

567 ) 

568 ) 

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

570 

571 @staticmethod 

572 def test_AddAdmin(testing_communities): 

573 with session_scope() as session: 

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

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

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

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

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

579 

580 with communities_session(token8) as api: 

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

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

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

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

585 

586 with communities_session(token4) as api: 

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

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

589 

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

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

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

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

594 

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

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

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

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

599 

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

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

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

603 # Cleanup because database changes do not roll back 

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

605 

606 @staticmethod 

607 def test_RemoveAdmin(testing_communities): 

608 with session_scope() as session: 

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

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

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

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

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

614 

615 with communities_session(token8) as api: 

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

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

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

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

620 

621 with communities_session(token4) as api: 

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

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

624 

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

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

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

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

629 

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

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

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

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

634 

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

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

637 assert res.admin_user_ids == [user4_id] 

638 # Cleanup because database changes do not roll back 

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

640 

641 @staticmethod 

642 def test_ListMembers(testing_communities): 

643 with session_scope() as session: 

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

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

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

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

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

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

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

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

652 w_id = get_community_id(session, "Global") 

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

654 

655 with communities_session(token1) as api: 

656 res = api.ListMembers( 

657 communities_pb2.ListMembersReq( 

658 community_id=w_id, 

659 ) 

660 ) 

661 assert res.member_user_ids == [ 

662 user1_id, 

663 user2_id, 

664 user3_id, 

665 user4_id, 

666 user5_id, 

667 user6_id, 

668 user7_id, 

669 user8_id, 

670 ] 

671 

672 res = api.ListMembers( 

673 communities_pb2.ListMembersReq( 

674 community_id=c1r1c2_id, 

675 )