Coverage for app/backend/src/tests/test_pages.py: 100%

521 statements  

« prev     ^ index     » next       coverage.py v7.16.1, created at 2026-09-19 15:47 +0000

1from datetime import timedelta 

2 

3import grpc 

4import pytest 

5from google.protobuf import wrappers_pb2 

6 

7from couchers.crypto import random_hex 

8from couchers.db import session_scope 

9from couchers.models import ( 

10 Cluster, 

11 ClusterRole, 

12 ClusterSubscription, 

13 Node, 

14 NodeType, 

15 Page, 

16 PageType, 

17 PageVersion, 

18 Thread, 

19 Upload, 

20) 

21from couchers.proto import pages_pb2 

22from couchers.utils import create_polygon_lat_lng, now, to_aware_datetime, to_multi 

23from tests.fixtures.db import generate_user 

24from tests.fixtures.sessions import pages_session 

25from tests.fixtures.timewarp import FrozenTimewarp 

26from tests.test_communities import create_community 

27 

28 

29def test_create_place_errors(db): 

30 user, token = generate_user() 

31 with session_scope() as session: 

32 create_community(session, 0, 2, "Root node", [user], [], None) 

33 

34 with pages_session(token) as api: 

35 with pytest.raises(grpc.RpcError) as e: 

36 api.CreatePlace( 

37 pages_pb2.CreatePlaceReq( 

38 title=None, 

39 content="dummy content", 

40 address="dummy address", 

41 location=pages_pb2.Coordinate( 

42 lat=1, 

43 lng=1, 

44 ), 

45 ) 

46 ) 

47 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT 

48 assert e.value.details() == "Missing page title." 

49 

50 with pages_session(token) as api: 

51 with pytest.raises(grpc.RpcError) as e: 

52 api.CreatePlace( 

53 pages_pb2.CreatePlaceReq( 

54 title="dummy title", 

55 content=None, 

56 address="dummy address", 

57 location=pages_pb2.Coordinate( 

58 lat=1, 

59 lng=1, 

60 ), 

61 ) 

62 ) 

63 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT 

64 assert e.value.details() == "Missing page text." 

65 

66 with pages_session(token) as api: 

67 with pytest.raises(grpc.RpcError) as e: 

68 api.CreatePlace( 

69 pages_pb2.CreatePlaceReq( 

70 title="dummy title", 

71 content="dummy content", 

72 address=None, 

73 location=pages_pb2.Coordinate( 

74 lat=1, 

75 lng=1, 

76 ), 

77 ) 

78 ) 

79 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT 

80 assert e.value.details() == "Missing page address." 

81 

82 with pages_session(token) as api: 

83 with pytest.raises(grpc.RpcError) as e: 

84 api.CreatePlace( 

85 pages_pb2.CreatePlaceReq( 

86 title="dummy title", 

87 content="dummy content", 

88 address="dummy address", 

89 ) 

90 ) 

91 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT 

92 assert e.value.details() == "Missing page geo-location." 

93 

94 

95def test_create_guide_errors(db): 

96 user, token = generate_user() 

97 with session_scope() as session: 

98 c_id = create_community(session, 0, 2, "Root node", [user], [], None).id 

99 

100 with pages_session(token) as api: 

101 with pytest.raises(grpc.RpcError) as e: 

102 api.CreateGuide( 

103 pages_pb2.CreateGuideReq( 

104 title=None, 

105 content="dummy content", 

106 parent_community_id=c_id, 

107 ) 

108 ) 

109 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT 

110 assert e.value.details() == "Missing page title." 

111 

112 with pages_session(token) as api: 

113 with pytest.raises(grpc.RpcError) as e: 

114 api.CreateGuide( 

115 pages_pb2.CreateGuideReq( 

116 title="dummy title", 

117 content=None, 

118 parent_community_id=c_id, 

119 ) 

120 ) 

121 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT 

122 assert e.value.details() == "Missing page text." 

123 

124 with pages_session(token) as api: 

125 with pytest.raises(grpc.RpcError) as e: 

126 api.CreateGuide( 

127 pages_pb2.CreateGuideReq( 

128 title="dummy title", 

129 content="dummy content", 

130 ) 

131 ) 

132 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT 

133 assert e.value.details() == "Missing page parent." 

134 

135 with pages_session(token) as api: 

136 with pytest.raises(grpc.RpcError) as e: 

137 api.CreateGuide( 

138 pages_pb2.CreateGuideReq( 

139 title="dummy title", 

140 content="dummy content", 

141 parent_community_id=9999, 

142 ) 

143 ) 

144 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT 

145 assert e.value.details() == "Community not found." 

146 

147 with pages_session(token) as api: 

148 with pytest.raises(grpc.RpcError) as e: 

149 api.CreateGuide( 

150 pages_pb2.CreateGuideReq( 

151 title="dummy title", 

152 content="dummy content", 

153 address="dummy address", 

154 parent_community_id=c_id, 

155 ) 

156 ) 

157 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT 

158 assert e.value.details() == "You need to either supply an address and location or neither for a guide." 

159 

160 with pages_session(token) as api: 

161 with pytest.raises(grpc.RpcError) as e: 

162 api.CreateGuide( 

163 pages_pb2.CreateGuideReq( 

164 title="dummy title", 

165 content="dummy content", 

166 location=pages_pb2.Coordinate( 

167 lat=1, 

168 lng=1, 

169 ), 

170 parent_community_id=c_id, 

171 ) 

172 ) 

173 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT 

174 assert e.value.details() == "You need to either supply an address and location or neither for a guide." 

175 

176 

177def test_create_page_place(db, frozen_timewarp): 

178 user, token = generate_user() 

179 with session_scope() as session: 

180 c_id = create_community(session, 0, 2, "Root node", [user], [], None).id 

181 

182 with pages_session(token) as api: 

183 time_before = now() 

184 res = api.CreatePlace( 

185 pages_pb2.CreatePlaceReq( 

186 title="dummy !#¤%&/-*' title", 

187 content="dummy content", 

188 address="dummy address", 

189 location=pages_pb2.Coordinate( 

190 lat=1, 

191 lng=1, 

192 ), 

193 ) 

194 ) 

195 

196 assert res.title == "dummy !#¤%&/-*' title" 

197 assert res.type == pages_pb2.PAGE_TYPE_PLACE 

198 assert res.content == "dummy content" 

199 assert res.address == "dummy address" 

200 assert res.location.lat == 1 

201 assert res.location.lng == 1 

202 assert res.slug == "dummy-title" 

203 assert time_before <= to_aware_datetime(res.created) <= now() 

204 assert time_before <= to_aware_datetime(res.last_edited) <= now() 

205 assert res.last_editor_user_id == user.id 

206 assert res.creator_user_id == user.id 

207 assert res.owner_user_id == user.id 

208 assert not res.owner_community_id 

209 assert not res.owner_group_id 

210 assert res.editor_user_ids == [user.id] 

211 assert res.can_edit 

212 assert res.can_moderate 

213 

214 

215def test_create_page_guide(db, frozen_timewarp): 

216 user, token = generate_user() 

217 with session_scope() as session: 

218 c_id = create_community(session, 0, 2, "Root node", [user], [], None).id 

219 

220 with pages_session(token) as api: 

221 time_before = now() 

222 res = api.CreateGuide( 

223 pages_pb2.CreateGuideReq( 

224 title="dummy title", 

225 content="dummy content", 

226 address="dummy address", 

227 location=pages_pb2.Coordinate( 

228 lat=1, 

229 lng=1, 

230 ), 

231 parent_community_id=c_id, 

232 ) 

233 ) 

234 

235 assert res.title == "dummy title" 

236 assert res.type == pages_pb2.PAGE_TYPE_GUIDE 

237 assert res.content == "dummy content" 

238 assert res.address == "dummy address" 

239 assert res.location.lat == 1 

240 assert res.location.lng == 1 

241 assert res.slug == "dummy-title" 

242 assert time_before <= to_aware_datetime(res.created) <= now() 

243 assert time_before <= to_aware_datetime(res.last_edited) <= now() 

244 assert res.last_editor_user_id == user.id 

245 assert res.creator_user_id == user.id 

246 assert res.owner_user_id == user.id 

247 assert not res.owner_community_id 

248 assert not res.owner_group_id 

249 assert res.editor_user_ids == [user.id] 

250 assert res.can_edit 

251 assert res.can_moderate 

252 

253 with pages_session(token) as api: 

254 time_before = now() 

255 res = api.CreateGuide( 

256 pages_pb2.CreateGuideReq( 

257 title="dummy title", 

258 content="dummy content", 

259 parent_community_id=c_id, 

260 ) 

261 ) 

262 

263 assert res.title == "dummy title" 

264 assert res.type == pages_pb2.PAGE_TYPE_GUIDE 

265 assert res.content == "dummy content" 

266 assert not res.address 

267 assert not res.HasField("location") 

268 assert res.slug == "dummy-title" 

269 assert time_before <= to_aware_datetime(res.created) <= now() 

270 assert time_before <= to_aware_datetime(res.last_edited) <= now() 

271 assert res.last_editor_user_id == user.id 

272 assert res.creator_user_id == user.id 

273 assert res.owner_user_id == user.id 

274 assert not res.owner_community_id 

275 assert not res.owner_group_id 

276 assert res.editor_user_ids == [user.id] 

277 assert res.can_edit 

278 assert res.can_moderate 

279 

280 

281def test_get_page(db, frozen_timewarp): 

282 user1, token1 = generate_user() 

283 user2, token2 = generate_user() 

284 with session_scope() as session: 

285 c_id = create_community(session, 0, 2, "Root node", [user1], [], None).id 

286 

287 with pages_session(token1) as api: 

288 time_before_create = now() 

289 page_id = api.CreatePlace( 

290 pages_pb2.CreatePlaceReq( 

291 title="dummy title", 

292 content="dummy content", 

293 address="dummy address", 

294 location=pages_pb2.Coordinate( 

295 lat=1, 

296 lng=1, 

297 ), 

298 ) 

299 ).page_id 

300 

301 with pages_session(token2) as api: 

302 time_before_get = now() 

303 res = api.GetPage(pages_pb2.GetPageReq(page_id=page_id)) 

304 assert res.title == "dummy title" 

305 assert res.content == "dummy content" 

306 assert res.address == "dummy address" 

307 assert res.location.lat == 1 

308 assert res.location.lng == 1 

309 assert res.slug == "dummy-title" 

310 assert time_before_create <= to_aware_datetime(res.created) <= time_before_get 

311 assert time_before_create <= to_aware_datetime(res.last_edited) <= time_before_get 

312 assert res.last_editor_user_id == user1.id 

313 assert res.creator_user_id == user1.id 

314 assert res.owner_user_id == user1.id 

315 assert not res.owner_community_id 

316 assert not res.owner_group_id 

317 assert res.editor_user_ids == [user1.id] 

318 assert not res.can_edit 

319 assert not res.can_moderate 

320 assert res.thread.thread_id > 0 

321 assert res.thread.num_responses == 0 

322 

323 

324def test_update_page(db, frozen_timewarp: FrozenTimewarp): 

325 user, token = generate_user() 

326 with session_scope() as session: 

327 c_id = create_community(session, 0, 2, "Root node", [user], [], None).id 

328 

329 with pages_session(token) as api: 

330 time_before_create = now() 

331 page_id = api.CreatePlace( 

332 pages_pb2.CreatePlaceReq( 

333 title="dummy title", 

334 content="dummy content", 

335 address="dummy address", 

336 location=pages_pb2.Coordinate( 

337 lat=1, 

338 lng=1, 

339 ), 

340 ) 

341 ).page_id 

342 

343 # the clock is stopped, so each edit needs the test to move it on for last_edited to change 

344 frozen_timewarp.advance(timedelta(minutes=1)) 

345 time_before_update = now() 

346 api.UpdatePage( 

347 pages_pb2.UpdatePageReq( 

348 page_id=page_id, 

349 title=wrappers_pb2.StringValue(value="test title"), 

350 ) 

351 ) 

352 

353 time_before_get = now() 

354 res = api.GetPage(pages_pb2.GetPageReq(page_id=page_id)) 

355 assert res.title == "test title" 

356 assert res.content == "dummy content" 

357 assert res.address == "dummy address" 

358 assert res.location.lat == 1 

359 assert res.location.lng == 1 

360 assert res.slug == "test-title" 

361 assert time_before_create <= to_aware_datetime(res.created) <= time_before_update 

362 assert time_before_update <= to_aware_datetime(res.last_edited) <= time_before_get 

363 assert to_aware_datetime(res.created) <= to_aware_datetime(res.last_edited) 

364 assert res.last_editor_user_id == user.id 

365 assert res.creator_user_id == user.id 

366 assert res.owner_user_id == user.id 

367 assert not res.owner_community_id 

368 assert not res.owner_group_id 

369 assert res.editor_user_ids == [user.id] 

370 assert res.can_edit 

371 assert res.can_moderate 

372 

373 frozen_timewarp.advance(timedelta(minutes=1)) 

374 time_before_second_update = now() 

375 api.UpdatePage( 

376 pages_pb2.UpdatePageReq( 

377 page_id=page_id, 

378 content=wrappers_pb2.StringValue(value="test content"), 

379 ) 

380 ) 

381 

382 time_before_second_get = now() 

383 res = api.GetPage(pages_pb2.GetPageReq(page_id=page_id)) 

384 assert res.title == "test title" 

385 assert res.content == "test content" 

386 assert res.address == "dummy address" 

387 assert res.location.lat == 1 

388 assert res.location.lng == 1 

389 assert res.slug == "test-title" 

390 assert time_before_create <= to_aware_datetime(res.created) <= time_before_update 

391 assert time_before_second_update <= to_aware_datetime(res.last_edited) <= time_before_second_get 

392 assert to_aware_datetime(res.created) <= to_aware_datetime(res.last_edited) 

393 assert res.last_editor_user_id == user.id 

394 assert res.creator_user_id == user.id 

395 assert res.owner_user_id == user.id 

396 assert not res.owner_community_id 

397 assert not res.owner_group_id 

398 assert res.editor_user_ids == [user.id] 

399 assert res.can_edit 

400 assert res.can_moderate 

401 assert res.thread.thread_id > 0 

402 assert res.thread.num_responses == 0 

403 

404 frozen_timewarp.advance(timedelta(minutes=1)) 

405 time_before_third_update = now() 

406 api.UpdatePage( 

407 pages_pb2.UpdatePageReq( 

408 page_id=page_id, 

409 address=wrappers_pb2.StringValue(value="test address"), 

410 ) 

411 ) 

412 

413 time_before_third_get = now() 

414 res = api.GetPage(pages_pb2.GetPageReq(page_id=page_id)) 

415 assert res.title == "test title" 

416 assert res.content == "test content" 

417 assert res.address == "test address" 

418 assert res.location.lat == 1 

419 assert res.location.lng == 1 

420 assert res.slug == "test-title" 

421 assert time_before_create <= to_aware_datetime(res.created) <= time_before_update 

422 assert time_before_third_update <= to_aware_datetime(res.last_edited) <= time_before_third_get 

423 assert to_aware_datetime(res.created) <= to_aware_datetime(res.last_edited) 

424 assert res.last_editor_user_id == user.id 

425 assert res.creator_user_id == user.id 

426 assert res.owner_user_id == user.id 

427 assert not res.owner_community_id 

428 assert not res.owner_group_id 

429 assert res.editor_user_ids == [user.id] 

430 assert res.can_edit 

431 assert res.can_moderate 

432 assert res.thread.thread_id > 0 

433 assert res.thread.num_responses == 0 

434 

435 frozen_timewarp.advance(timedelta(minutes=1)) 

436 time_before_fourth_update = now() 

437 api.UpdatePage( 

438 pages_pb2.UpdatePageReq( 

439 page_id=page_id, 

440 location=pages_pb2.Coordinate( 

441 lat=3, 

442 lng=1.222, 

443 ), 

444 ) 

445 ) 

446 

447 time_before_fourth_get = now() 

448 res = api.GetPage(pages_pb2.GetPageReq(page_id=page_id)) 

449 assert res.title == "test title" 

450 assert res.content == "test content" 

451 assert res.address == "test address" 

452 assert res.location.lat == 3 

453 assert res.location.lng == 1.222 

454 assert res.slug == "test-title" 

455 assert time_before_create <= to_aware_datetime(res.created) <= time_before_update 

456 assert time_before_fourth_update <= to_aware_datetime(res.last_edited) <= time_before_fourth_get 

457 assert to_aware_datetime(res.created) <= to_aware_datetime(res.last_edited) 

458 assert res.last_editor_user_id == user.id 

459 assert res.creator_user_id == user.id 

460 assert res.owner_user_id == user.id 

461 assert not res.owner_community_id 

462 assert not res.owner_group_id 

463 assert res.editor_user_ids == [user.id] 

464 assert res.can_edit 

465 assert res.can_moderate 

466 assert res.thread.thread_id > 0 

467 assert res.thread.num_responses == 0 

468 

469 

470def test_owner_not_moderator(db): 

471 """ 

472 You can be the owner of content yet not have moderation rights 

473 """ 

474 user1, token1 = generate_user() 

475 user2, token2 = generate_user() 

476 with session_scope() as session: 

477 c_id = create_community(session, 0, 2, "Root node", [user1], [], None).id 

478 

479 # user2 makes page, is owner but not moderator, so can edit, not moderate 

480 with pages_session(token2) as api: 

481 res = api.CreatePlace( 

482 pages_pb2.CreatePlaceReq( 

483 title="dummy title", 

484 content="dummy content", 

485 address="dummy address", 

486 location=pages_pb2.Coordinate( 

487 lat=1, 

488 lng=1, 

489 ), 

490 ) 

491 ) 

492 assert res.title == "dummy title" 

493 assert res.content == "dummy content" 

494 assert res.address == "dummy address" 

495 assert res.location.lat == 1 

496 assert res.location.lng == 1 

497 assert res.slug == "dummy-title" 

498 assert res.last_editor_user_id == user2.id 

499 assert res.creator_user_id == user2.id 

500 assert res.owner_user_id == user2.id 

501 assert not res.owner_community_id 

502 assert not res.owner_group_id 

503 assert res.editor_user_ids == [user2.id] 

504 assert res.can_edit 

505 assert not res.can_moderate 

506 

507 page_id = res.page_id 

508 

509 # user1 is not owner so can't edit but can moderate 

510 with pages_session(token1) as api: 

511 res = api.GetPage(pages_pb2.GetPageReq(page_id=page_id)) 

512 assert res.title == "dummy title" 

513 assert res.content == "dummy content" 

514 assert res.address == "dummy address" 

515 assert res.location.lat == 1 

516 assert res.location.lng == 1 

517 assert res.slug == "dummy-title" 

518 assert res.last_editor_user_id == user2.id 

519 assert res.creator_user_id == user2.id 

520 assert res.owner_user_id == user2.id 

521 assert not res.owner_community_id 

522 assert not res.owner_group_id 

523 assert res.editor_user_ids == [user2.id] 

524 assert res.can_edit 

525 assert res.can_moderate 

526 assert res.thread.thread_id > 0 

527 assert res.thread.num_responses == 0 

528 

529 

530def test_update_page_errors(db): 

531 user, token = generate_user() 

532 with session_scope() as session: 

533 c_id = create_community(session, 0, 2, "Root node", [user], [], None).id 

534 

535 with pages_session(token) as api: 

536 page_id = api.CreatePlace( 

537 pages_pb2.CreatePlaceReq( 

538 title="dummy title", 

539 content="dummy content", 

540 address="dummy address", 

541 location=pages_pb2.Coordinate( 

542 lat=1, 

543 lng=1, 

544 ), 

545 ) 

546 ).page_id 

547 

548 with pytest.raises(grpc.RpcError) as e: 

549 api.UpdatePage( 

550 pages_pb2.UpdatePageReq( 

551 page_id=page_id, 

552 title=wrappers_pb2.StringValue(value=""), 

553 ) 

554 ) 

555 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT 

556 assert e.value.details() == "Missing page title." 

557 

558 with pytest.raises(grpc.RpcError) as e: 

559 api.UpdatePage( 

560 pages_pb2.UpdatePageReq( 

561 page_id=page_id, 

562 content=wrappers_pb2.StringValue(value=""), 

563 ) 

564 ) 

565 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT 

566 assert e.value.details() == "Missing page text." 

567 

568 with pytest.raises(grpc.RpcError) as e: 

569 api.UpdatePage( 

570 pages_pb2.UpdatePageReq( 

571 page_id=page_id, 

572 address=wrappers_pb2.StringValue(value=""), 

573 ) 

574 ) 

575 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT 

576 assert e.value.details() == "Missing page address." 

577 

578 

579def test_page_transfer(db): 

580 # transfers the pages 

581 user1, token1 = generate_user() 

582 # admin of the community/group 

583 user2, token2 = generate_user() 

584 # member of the community/group, shouldn't ever have edit access to anything 

585 user3, token3 = generate_user() 

586 with session_scope() as session: 

587 # create a community 

588 node = Node( 

589 geom=to_multi(create_polygon_lat_lng([[0, 0], [0, 2], [2, 2], [2, 0], [0, 0]])), node_type=NodeType.world 

590 ) 

591 session.add(node) 

592 session.flush() 

593 community_cluster = Cluster( 

594 name="Testing Community", 

595 description="Description for testing community", 

596 parent_node_id=node.id, 

597 is_official_cluster=True, 

598 ) 

599 session.add(community_cluster) 

600 session.flush() 

601 thread = Thread() 

602 session.add(thread) 

603 session.flush() 

604 main_page = Page( 

605 parent_node_id=community_cluster.parent_node_id, 

606 creator_user_id=user2.id, 

607 owner_cluster_id=community_cluster.id, 

608 type=PageType.main_page, 

609 thread_id=thread.id, 

610 ) 

611 session.add(main_page) 

612 session.flush() 

613 session.add( 

614 PageVersion( 

615 page_id=main_page.id, 

616 editor_user_id=user2.id, 

617 title="Main page for the testing community", 

618 content="Empty.", 

619 ) 

620 ) 

621 community_cluster.cluster_subscriptions.append( 

622 ClusterSubscription( 

623 user_id=user2.id, 

624 cluster_id=community_cluster.id, 

625 role=ClusterRole.admin, 

626 ) 

627 ) 

628 community_cluster.cluster_subscriptions.append( 

629 ClusterSubscription( 

630 user_id=user3.id, 

631 cluster_id=community_cluster.id, 

632 role=ClusterRole.member, 

633 ) 

634 ) 

635 

636 # create a group 

637 group_cluster = Cluster( 

638 name="Testing Group", 

639 description="Description for testing group", 

640 parent_node_id=node.id, 

641 ) 

642 session.add(group_cluster) 

643 session.flush() 

644 thread = Thread() 

645 session.add(thread) 

646 session.flush() 

647 main_page = Page( 

648 parent_node_id=group_cluster.parent_node_id, 

649 creator_user_id=user2.id, 

650 owner_cluster_id=group_cluster.id, 

651 type=PageType.main_page, 

652 thread_id=thread.id, 

653 ) 

654 session.add(main_page) 

655 session.flush() 

656 session.add( 

657 PageVersion( 

658 page_id=main_page.id, 

659 editor_user_id=user2.id, 

660 title="Main page for the testing community", 

661 content="Empty.", 

662 ) 

663 ) 

664 group_cluster.cluster_subscriptions.append( 

665 ClusterSubscription( 

666 user_id=user2.id, 

667 cluster_id=group_cluster.id, 

668 role=ClusterRole.admin, 

669 ) 

670 ) 

671 group_cluster.cluster_subscriptions.append( 

672 ClusterSubscription( 

673 user_id=user3.id, 

674 cluster_id=group_cluster.id, 

675 role=ClusterRole.member, 

676 ) 

677 ) 

678 session.flush() 

679 

680 community_id = node.id 

681 community_cluster_id = community_cluster.id 

682 group_id = group_cluster.id 

683 

684 with pages_session(token1) as api: 

685 create_page_req = pages_pb2.CreatePlaceReq( 

686 title="title", 

687 content="content", 

688 address="address", 

689 location=pages_pb2.Coordinate( 

690 lat=1, 

691 lng=1, 

692 ), 

693 ) 

694 

695 # transfer should work fine to a community 

696 page1 = api.CreatePlace(create_page_req) 

697 assert page1.owner_user_id == user1.id 

698 assert page1.can_edit 

699 assert not page1.can_moderate 

700 

701 with pages_session(token2) as api: 

702 assert api.GetPage(pages_pb2.GetPageReq(page_id=page1.page_id)).can_edit 

703 assert api.GetPage(pages_pb2.GetPageReq(page_id=page1.page_id)).can_moderate 

704 

705 with pages_session(token3) as api: 

706 assert not api.GetPage(pages_pb2.GetPageReq(page_id=page1.page_id)).can_edit 

707 assert not api.GetPage(pages_pb2.GetPageReq(page_id=page1.page_id)).can_moderate 

708 

709 with pages_session(token1) as api: 

710 page1 = api.TransferPage( 

711 pages_pb2.TransferPageReq( 

712 page_id=page1.page_id, 

713 new_owner_community_id=community_id, 

714 ) 

715 ) 

716 assert page1.owner_community_id == community_id 

717 assert not page1.can_edit 

718 assert not page1.can_moderate 

719 

720 with pages_session(token2) as api: 

721 assert api.GetPage(pages_pb2.GetPageReq(page_id=page1.page_id)).can_edit 

722 assert api.GetPage(pages_pb2.GetPageReq(page_id=page1.page_id)).can_moderate 

723 

724 with pages_session(token3) as api: 

725 assert not api.GetPage(pages_pb2.GetPageReq(page_id=page1.page_id)).can_edit 

726 assert not api.GetPage(pages_pb2.GetPageReq(page_id=page1.page_id)).can_moderate 

727 

728 with pages_session(token1) as api: 

729 # now we're no longer the owner, can't transfer 

730 page1 = api.GetPage(pages_pb2.GetPageReq(page_id=page1.page_id)) 

731 with pytest.raises(grpc.RpcError) as e: 

732 api.TransferPage( 

733 pages_pb2.TransferPageReq( 

734 page_id=page1.page_id, 

735 new_owner_group_id=group_id, 

736 ) 

737 ) 

738 assert e.value.code() == grpc.StatusCode.PERMISSION_DENIED 

739 assert e.value.details() == "You're not allowed to transfer that page." 

740 page1 = api.GetPage(pages_pb2.GetPageReq(page_id=page1.page_id)) 

741 assert page1.owner_community_id == community_id 

742 assert not page1.can_edit 

743 assert not page1.can_moderate 

744 

745 with pages_session(token2) as api: 

746 assert api.GetPage(pages_pb2.GetPageReq(page_id=page1.page_id)).can_edit 

747 assert api.GetPage(pages_pb2.GetPageReq(page_id=page1.page_id)).can_moderate 

748 

749 with pages_session(token3) as api: 

750 assert not api.GetPage(pages_pb2.GetPageReq(page_id=page1.page_id)).can_edit 

751 assert not api.GetPage(pages_pb2.GetPageReq(page_id=page1.page_id)).can_moderate 

752 

753 with pages_session(token1) as api: 

754 # try a new page, just for fun 

755 page2 = api.CreatePlace(create_page_req) 

756 assert page2.owner_user_id == user1.id 

757 

758 page2 = api.TransferPage( 

759 pages_pb2.TransferPageReq( 

760 page_id=page2.page_id, 

761 new_owner_community_id=community_id, 

762 ) 

763 ) 

764 assert page2.owner_community_id == community_id 

765 

766 with pages_session(token2) as api: 

767 assert api.GetPage(pages_pb2.GetPageReq(page_id=page2.page_id)).can_edit 

768 assert api.GetPage(pages_pb2.GetPageReq(page_id=page2.page_id)).can_moderate 

769 

770 with pages_session(token3) as api: 

771 assert not api.GetPage(pages_pb2.GetPageReq(page_id=page2.page_id)).can_edit 

772 assert not api.GetPage(pages_pb2.GetPageReq(page_id=page2.page_id)).can_moderate 

773 

774 with pages_session(token1) as api: 

775 # can't transfer a page to an official cluster, only through community 

776 page3 = api.CreatePlace(create_page_req) 

777 assert page3.owner_user_id == user1.id 

778 

779 with pytest.raises(grpc.RpcError) as e: 

780 api.TransferPage( 

781 pages_pb2.TransferPageReq( 

782 page_id=page3.page_id, 

783 new_owner_community_id=community_cluster_id, 

784 ) 

785 ) 

786 assert e.value.code() == grpc.StatusCode.NOT_FOUND 

787 assert e.value.details() == "Group or community not found." 

788 page3 = api.GetPage(pages_pb2.GetPageReq(page_id=page3.page_id)) 

789 assert page3.owner_user_id == user1.id 

790 

791 # can transfer to group 

792 page4 = api.CreatePlace(create_page_req) 

793 assert page4.owner_user_id == user1.id 

794 assert page4.can_edit 

795 assert not page4.can_moderate 

796 

797 with pages_session(token2) as api: 

798 assert api.GetPage(pages_pb2.GetPageReq(page_id=page4.page_id)).can_edit 

799 assert api.GetPage(pages_pb2.GetPageReq(page_id=page4.page_id)).can_moderate 

800 

801 with pages_session(token3) as api: 

802 assert not api.GetPage(pages_pb2.GetPageReq(page_id=page4.page_id)).can_edit 

803 assert not api.GetPage(pages_pb2.GetPageReq(page_id=page4.page_id)).can_moderate 

804 

805 with pages_session(token1) as api: 

806 page4 = api.TransferPage( 

807 pages_pb2.TransferPageReq( 

808 page_id=page4.page_id, 

809 new_owner_group_id=group_id, 

810 ) 

811 ) 

812 assert page4.owner_group_id == group_id 

813 assert not page4.can_edit 

814 assert not page4.can_moderate 

815 

816 with pages_session(token2) as api: 

817 assert api.GetPage(pages_pb2.GetPageReq(page_id=page4.page_id)).can_edit 

818 assert api.GetPage(pages_pb2.GetPageReq(page_id=page4.page_id)).can_moderate 

819 

820 with pages_session(token3) as api: 

821 assert not api.GetPage(pages_pb2.GetPageReq(page_id=page4.page_id)).can_edit 

822 assert not api.GetPage(pages_pb2.GetPageReq(page_id=page4.page_id)).can_moderate 

823 

824 with pages_session(token1) as api: 

825 # now we're no longer the owner, can't transfer 

826 with pytest.raises(grpc.RpcError) as e: 

827 api.TransferPage( 

828 pages_pb2.TransferPageReq( 

829 page_id=page4.page_id, 

830 new_owner_community_id=community_id, 

831 ) 

832 ) 

833 assert e.value.code() == grpc.StatusCode.PERMISSION_DENIED 

834 assert e.value.details() == "You're not allowed to transfer that page." 

835 page4 = api.GetPage(pages_pb2.GetPageReq(page_id=page4.page_id)) 

836 assert page4.owner_group_id == group_id 

837 

838 

839def test_page_photo(db): 

840 user, token = generate_user() 

841 with session_scope() as session: 

842 c_id = create_community(session, 0, 2, "Root node", [user], [], None).id 

843 

844 with pages_session(token) as api: 

845 key1 = random_hex(32) 

846 filename1 = random_hex(32) 

847 

848 with session_scope() as session: 

849 session.add( 

850 Upload( 

851 key=key1, 

852 filename=filename1, 

853 creator_user_id=user.id, 

854 ) 

855 ) 

856 

857 res = api.CreatePlace( 

858 pages_pb2.CreatePlaceReq( 

859 title="dummy title", 

860 content="dummy content", 

861 photo_key=key1, 

862 address="dummy address", 

863 location=pages_pb2.Coordinate( 

864 lat=1, 

865 lng=1, 

866 ), 

867 ) 

868 ) 

869 

870 assert filename1 in res.photo_url 

871 

872 # can't create with non-existent photo 

873 with pytest.raises(grpc.RpcError) as e: 

874 api.CreatePlace( 

875 pages_pb2.CreatePlaceReq( 

876 title="dummy title", 

877 content="dummy content", 

878 photo_key="nonexisten", 

879 address="dummy address", 

880 location=pages_pb2.Coordinate( 

881 lat=1, 

882 lng=1, 

883 ), 

884 ) 

885 ) 

886 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT 

887 assert e.value.details() == "Photo not found." 

888 

889 # can create page with no photo 

890 res = api.CreatePlace( 

891 pages_pb2.CreatePlaceReq( 

892 title="dummy title", 

893 content="dummy content", 

894 address="dummy address", 

895 location=pages_pb2.Coordinate( 

896 lat=1, 

897 lng=1, 

898 ), 

899 ) 

900 ) 

901 

902 assert res.photo_url == "" 

903 

904 page_id = res.page_id 

905 

906 # can't set it to non-existent stuff 

907 with pytest.raises(grpc.RpcError) as e: 

908 api.UpdatePage( 

909 pages_pb2.UpdatePageReq( 

910 page_id=page_id, 

911 photo_key=wrappers_pb2.StringValue(value="non-existent id"), 

912 ) 

913 ) 

914 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT 

915 assert e.value.details() == "Photo not found." 

916 

917 key2 = random_hex(32) 

918 filename2 = random_hex(32) 

919 

920 with session_scope() as session: 

921 session.add( 

922 Upload( 

923 key=key2, 

924 filename=filename2, 

925 creator_user_id=user.id, 

926 ) 

927 ) 

928 

929 res = api.UpdatePage( 

930 pages_pb2.UpdatePageReq( 

931 page_id=page_id, 

932 photo_key=wrappers_pb2.StringValue(value=key2), 

933 ) 

934 ) 

935 

936 assert filename2 in res.photo_url 

937 

938 # can unset it 

939 

940 res = api.UpdatePage( 

941 pages_pb2.UpdatePageReq( 

942 page_id=page_id, 

943 photo_key=wrappers_pb2.StringValue(value=""), 

944 ) 

945 ) 

946 

947 assert res.photo_url == "" 

948 

949 

950def test_list_user_places(db): 

951 user1, token1 = generate_user() 

952 with session_scope() as session: 

953 c_id = create_community(session, 0, 2, "Root node", [user1], [], None).id 

954 

955 with pages_session(token1) as api: 

956 place1_id = api.CreatePlace( 

957 pages_pb2.CreatePlaceReq( 

958 title="dummy title", 

959 content="dummy content", 

960 address="dummy address", 

961 location=pages_pb2.Coordinate( 

962 lat=1, 

963 lng=1, 

964 ), 

965 ) 

966 ).page_id 

967 

968 place2_id = api.CreatePlace( 

969 pages_pb2.CreatePlaceReq( 

970 title="dummy title 2", 

971 content="dummy content 2", 

972 address="dummy address 2", 

973 location=pages_pb2.Coordinate( 

974 lat=1, 

975 lng=1, 

976 ), 

977 ) 

978 ).page_id 

979 

980 with pages_session(token1) as api: 

981 res = api.ListUserPlaces(pages_pb2.ListUserPlacesReq()) 

982 assert [p.page_id for p in res.places] == [place1_id, place2_id] 

983 

984 

985def test_list_other_user_places(db): 

986 user1, token1 = generate_user() 

987 user2, token2 = generate_user() 

988 with session_scope() as session: 

989 c_id = create_community(session, 0, 2, "Root node", [user1], [], None).id 

990 

991 with pages_session(token1) as api: 

992 place1_id = api.CreatePlace( 

993 pages_pb2.CreatePlaceReq( 

994 title="dummy title", 

995 content="dummy content", 

996 address="dummy address", 

997 location=pages_pb2.Coordinate( 

998 lat=1, 

999 lng=1, 

1000 ), 

1001 ) 

1002 ).page_id 

1003 

1004 place2_id = api.CreatePlace( 

1005 pages_pb2.CreatePlaceReq( 

1006 title="dummy title 2", 

1007 content="dummy content 2", 

1008 address="dummy address 2", 

1009 location=pages_pb2.Coordinate( 

1010 lat=1, 

1011 lng=1, 

1012 ), 

1013 ) 

1014 ).page_id 

1015 

1016 with pages_session(token2) as api: 

1017 res = api.ListUserPlaces(pages_pb2.ListUserPlacesReq(user_id=user1.id)) 

1018 assert [p.page_id for p in res.places] == [place1_id, place2_id] 

1019 

1020 

1021def test_list_user_guides(db): 

1022 user1, token1 = generate_user() 

1023 with session_scope() as session: 

1024 c_id = create_community(session, 0, 2, "Root node", [user1], [], None).id 

1025 

1026 with pages_session(token1) as api: 

1027 guide1_id = api.CreateGuide( 

1028 pages_pb2.CreateGuideReq( 

1029 title="dummy title", 

1030 content="dummy content", 

1031 address="dummy address", 

1032 location=pages_pb2.Coordinate( 

1033 lat=1, 

1034 lng=1, 

1035 ), 

1036 parent_community_id=c_id, 

1037 ) 

1038 ).page_id 

1039 

1040 guide2_id = api.CreateGuide( 

1041 pages_pb2.CreateGuideReq( 

1042 title="dummy title 2", 

1043 content="dummy content 2", 

1044 address="dummy address 2", 

1045 location=pages_pb2.Coordinate( 

1046 lat=1, 

1047 lng=1, 

1048 ), 

1049 parent_community_id=c_id, 

1050 ) 

1051 ).page_id 

1052 

1053 with pages_session(token1) as api: 

1054 res = api.ListUserGuides(pages_pb2.ListUserGuidesReq()) 

1055 assert [p.page_id for p in res.guides] == [guide1_id, guide2_id] 

1056 

1057 

1058def test_list_other_user_guides(db): 

1059 user1, token1 = generate_user() 

1060 user2, token2 = generate_user() 

1061 with session_scope() as session: 

1062 c_id = create_community(session, 0, 2, "Root node", [user1], [], None).id 

1063 

1064 with pages_session(token1) as api: 

1065 guide1_id = api.CreateGuide( 

1066 pages_pb2.CreateGuideReq( 

1067 title="dummy title", 

1068 content="dummy content", 

1069 address="dummy address", 

1070 location=pages_pb2.Coordinate( 

1071 lat=1, 

1072 lng=1, 

1073 ), 

1074 parent_community_id=c_id, 

1075 ) 

1076 ).page_id 

1077 

1078 guide2_id = api.CreateGuide( 

1079 pages_pb2.CreateGuideReq( 

1080 title="dummy title 2", 

1081 content="dummy content 2", 

1082 address="dummy address 2", 

1083 location=pages_pb2.Coordinate( 

1084 lat=1, 

1085 lng=1, 

1086 ), 

1087 parent_community_id=c_id, 

1088 ) 

1089 ).page_id 

1090 

1091 with pages_session(token2) as api: 

1092 res = api.ListUserGuides(pages_pb2.ListUserGuidesReq(user_id=user1.id)) 

1093 assert [p.page_id for p in res.guides] == [guide1_id, guide2_id]