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

1from datetime import date, timedelta 

2from unittest.mock import patch 

3 

4import grpc 

5import pytest 

6from sqlalchemy import select 

7 

8from couchers.db import session_scope 

9from couchers.models import Cluster, ClusterRole, ClusterSubscription, Node, NodeType, User 

10from couchers.models.public_trips import PublicTrip, PublicTripStatus 

11from couchers.proto import public_trips_pb2 

12from couchers.utils import create_polygon_lat_lng, now, to_multi, today 

13from tests.fixtures.db import generate_user 

14from tests.fixtures.sessions import public_trips_session 

15 

16 

17@pytest.fixture(autouse=True) 

18def _(testconfig): 

19 pass 

20 

21 

22# 150+ utf-16 code units to satisfy PUBLIC_TRIP_DESCRIPTION_MIN_LENGTH_UTF16. 

23VALID_DESCRIPTION = ( 

24 "Visiting the area for a week for a music festival. I love meeting new people " 

25 "and would really appreciate local tips. Happy to help with tasks in exchange." 

26) 

27 

28 

29def _make_node(node_type: NodeType = NodeType.locality, small_community_features_enabled: bool = True) -> int: 

30 # Polygon inside the fake Europe/Helsinki timezone area so node.timezone resolves. 

31 with session_scope() as session: 

32 node = Node( 

33 geom=to_multi(create_polygon_lat_lng([[60, 24], [60, 26], [62, 26], [62, 24], [60, 24]])), 

34 node_type=node_type, 

35 ) 

36 session.add(node) 

37 session.flush() 

38 cluster = Cluster( 

39 name="Test community", 

40 description="Test", 

41 parent_node_id=node.id, 

42 is_official_cluster=True, 

43 small_community_features_enabled=small_community_features_enabled, 

44 ) 

45 session.add(cluster) 

46 session.flush() 

47 return node.id 

48 

49 

50def _make_node_admin(user_id: int, node_id: int): 

51 with session_scope() as session: 

52 cluster_id = session.execute( 

53 select(Cluster.id).where(Cluster.parent_node_id == node_id).where(Cluster.is_official_cluster) 

54 ).scalar_one() 

55 session.add(ClusterSubscription(cluster_id=cluster_id, user_id=user_id, role=ClusterRole.admin)) 

56 

57 

58def _create_trip_directly( 

59 user_id: int, 

60 node_id: int, 

61 from_date, 

62 to_date, 

63 *, 

64 description: str = "Looking for a host!", 

65 status=None, 

66 same_gender_only: bool = False, 

67) -> int: 

68 with session_scope() as session: 

69 trip = PublicTrip( 

70 user_id=user_id, 

71 node_id=node_id, 

72 from_date=from_date, 

73 to_date=to_date, 

74 description=description, 

75 status=status or PublicTripStatus.searching_for_host, 

76 same_gender_only=same_gender_only, 

77 ) 

78 session.add(trip) 

79 session.flush() 

80 return trip.id 

81 

82 

83def test_create_public_trip(db): 

84 user, token = generate_user() 

85 node_id = _make_node() 

86 

87 from_date = today() + timedelta(days=5) 

88 to_date = today() + timedelta(days=10) 

89 

90 with public_trips_session(token) as api: 

91 res = api.CreatePublicTrip( 

92 public_trips_pb2.CreatePublicTripReq( 

93 node_id=node_id, 

94 from_date=from_date.isoformat(), 

95 to_date=to_date.isoformat(), 

96 description=VALID_DESCRIPTION, 

97 ) 

98 ) 

99 

100 assert res.trip_id > 0 

101 assert res.user.user_id == user.id 

102 assert res.node_id == node_id 

103 assert res.node_slug == "test-community" 

104 assert res.from_date == from_date.isoformat() 

105 assert res.to_date == to_date.isoformat() 

106 assert res.description == VALID_DESCRIPTION 

107 assert res.status == public_trips_pb2.PUBLIC_TRIP_STATUS_SEARCHING_FOR_HOST 

108 

109 with session_scope() as session: 

110 trip = session.execute(select(PublicTrip).where(PublicTrip.id == res.trip_id)).scalar_one() 

111 assert trip.user_id == user.id 

112 assert trip.node_id == node_id 

113 assert trip.status == PublicTripStatus.searching_for_host 

114 

115 

116def test_create_public_trip_incomplete_profile(db): 

117 _, token = generate_user(complete_profile=False) 

118 node_id = _make_node() 

119 

120 with public_trips_session(token) as api: 

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

122 api.CreatePublicTrip( 

123 public_trips_pb2.CreatePublicTripReq( 

124 node_id=node_id, 

125 from_date=(today() + timedelta(days=5)).isoformat(), 

126 to_date=(today() + timedelta(days=10)).isoformat(), 

127 description="Visiting town!", 

128 ) 

129 ) 

130 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION 

131 assert e.value.details() == "You have to complete your profile before you can create a public trip." 

132 

133 

134def test_create_public_trip_community_not_found(db): 

135 _, token = generate_user() 

136 

137 with public_trips_session(token) as api: 

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

139 api.CreatePublicTrip( 

140 public_trips_pb2.CreatePublicTripReq( 

141 node_id=999999, 

142 from_date=(today() + timedelta(days=5)).isoformat(), 

143 to_date=(today() + timedelta(days=10)).isoformat(), 

144 description="Visiting town!", 

145 ) 

146 ) 

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

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

149 

150 

151def test_create_public_trip_not_enabled(db): 

152 _, token = generate_user() 

153 node_id = _make_node(small_community_features_enabled=False) 

154 

155 with public_trips_session(token) as api: 

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

157 api.CreatePublicTrip( 

158 public_trips_pb2.CreatePublicTripReq( 

159 node_id=node_id, 

160 from_date=(today() + timedelta(days=5)).isoformat(), 

161 to_date=(today() + timedelta(days=10)).isoformat(), 

162 description="Visiting town!", 

163 ) 

164 ) 

165 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION 

166 assert e.value.details() == "Public trips are not enabled in this community." 

167 

168 

169@pytest.mark.parametrize( 

170 "node_type", 

171 [NodeType.region, NodeType.subregion, NodeType.locality, NodeType.sublocality], 

172) 

173def test_create_public_trip_allows_region_and_narrower(db, node_type): 

174 _, token = generate_user() 

175 node_id = _make_node(node_type=node_type) 

176 

177 with public_trips_session(token) as api: 

178 res = api.CreatePublicTrip( 

179 public_trips_pb2.CreatePublicTripReq( 

180 node_id=node_id, 

181 from_date=(today() + timedelta(days=5)).isoformat(), 

182 to_date=(today() + timedelta(days=10)).isoformat(), 

183 description=VALID_DESCRIPTION, 

184 ) 

185 ) 

186 assert res.trip_id > 0 

187 

188 

189def test_create_public_trip_in_past_uses_node_timezone(db): 

190 # Default user geom resolves to America/New_York; the node's geom is in Europe/Helsinki. 

191 # Simulate a moment where Helsinki has already rolled into the next day (2026-01-16) 

192 # while NYC is still on 2026-01-15. A from_date of 2026-01-15 is "today" in NYC but 

193 # "yesterday" in Helsinki, and must be rejected because the check uses the node's tz. 

194 _, token = generate_user() 

195 node_id = _make_node() 

196 

197 fake_today_by_tz = {"America/New_York": date(2026, 1, 15), "Europe/Helsinki": date(2026, 1, 16)} 

198 

199 with patch( 

200 "couchers.servicers.public_trips.today_in_timezone", 

201 side_effect=lambda tz: fake_today_by_tz[tz], 

202 ): 

203 with public_trips_session(token) as api: 

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

205 api.CreatePublicTrip( 

206 public_trips_pb2.CreatePublicTripReq( 

207 node_id=node_id, 

208 from_date="2026-01-15", 

209 to_date="2026-01-20", 

210 description=VALID_DESCRIPTION, 

211 ) 

212 ) 

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

214 

215 

216def test_create_public_trip_date_errors(db): 

217 _, token = generate_user() 

218 node_id = _make_node() 

219 

220 with public_trips_session(token) as api: 

221 # from_date in the past 

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

223 api.CreatePublicTrip( 

224 public_trips_pb2.CreatePublicTripReq( 

225 node_id=node_id, 

226 from_date=(today() - timedelta(days=2)).isoformat(), 

227 to_date=(today() + timedelta(days=1)).isoformat(), 

228 description="Visiting town!", 

229 ) 

230 ) 

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

232 

233 # from_date after to_date 

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

235 api.CreatePublicTrip( 

236 public_trips_pb2.CreatePublicTripReq( 

237 node_id=node_id, 

238 from_date=(today() + timedelta(days=10)).isoformat(), 

239 to_date=(today() + timedelta(days=5)).isoformat(), 

240 description="Visiting town!", 

241 ) 

242 ) 

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

244 

245 # empty description 

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

247 api.CreatePublicTrip( 

248 public_trips_pb2.CreatePublicTripReq( 

249 node_id=node_id, 

250 from_date=(today() + timedelta(days=5)).isoformat(), 

251 to_date=(today() + timedelta(days=10)).isoformat(), 

252 description=" ", 

253 ) 

254 ) 

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

256 

257 # invalid date format 

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

259 api.CreatePublicTrip( 

260 public_trips_pb2.CreatePublicTripReq( 

261 node_id=node_id, 

262 from_date="not-a-date", 

263 to_date=(today() + timedelta(days=10)).isoformat(), 

264 description="Visiting town!", 

265 ) 

266 ) 

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

268 

269 

270def test_create_public_trip_overlap(db): 

271 user, token = generate_user() 

272 node_id = _make_node() 

273 

274 _create_trip_directly( 

275 user.id, 

276 node_id, 

277 today() + timedelta(days=5), 

278 today() + timedelta(days=10), 

279 ) 

280 

281 with public_trips_session(token) as api: 

282 # overlapping dates should fail 

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

284 api.CreatePublicTrip( 

285 public_trips_pb2.CreatePublicTripReq( 

286 node_id=node_id, 

287 from_date=(today() + timedelta(days=8)).isoformat(), 

288 to_date=(today() + timedelta(days=12)).isoformat(), 

289 description=VALID_DESCRIPTION, 

290 ) 

291 ) 

292 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION 

293 

294 # non-overlapping dates should succeed 

295 res = api.CreatePublicTrip( 

296 public_trips_pb2.CreatePublicTripReq( 

297 node_id=node_id, 

298 from_date=(today() + timedelta(days=20)).isoformat(), 

299 to_date=(today() + timedelta(days=25)).isoformat(), 

300 description=VALID_DESCRIPTION, 

301 ) 

302 ) 

303 assert res.trip_id > 0 

304 

305 

306def test_create_public_trip_closed_trip_allows_new_overlap(db): 

307 user, token = generate_user() 

308 node_id = _make_node() 

309 

310 _create_trip_directly( 

311 user.id, 

312 node_id, 

313 today() + timedelta(days=5), 

314 today() + timedelta(days=10), 

315 status=PublicTripStatus.closed, 

316 ) 

317 

318 with public_trips_session(token) as api: 

319 # closed trips shouldn't block new overlapping ones 

320 res = api.CreatePublicTrip( 

321 public_trips_pb2.CreatePublicTripReq( 

322 node_id=node_id, 

323 from_date=(today() + timedelta(days=7)).isoformat(), 

324 to_date=(today() + timedelta(days=12)).isoformat(), 

325 description=VALID_DESCRIPTION, 

326 ) 

327 ) 

328 assert res.trip_id > 0 

329 

330 

331def test_get_public_trip(db): 

332 user, token = generate_user() 

333 node_id = _make_node() 

334 trip_id = _create_trip_directly(user.id, node_id, today() + timedelta(days=5), today() + timedelta(days=10)) 

335 

336 with public_trips_session(token) as api: 

337 res = api.GetPublicTrip(public_trips_pb2.GetPublicTripReq(trip_id=trip_id)) 

338 assert res.trip_id == trip_id 

339 assert res.user.user_id == user.id 

340 assert res.node_slug == "test-community" 

341 

342 

343def test_get_public_trip_not_found(db): 

344 _, token = generate_user() 

345 with public_trips_session(token) as api: 

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

347 api.GetPublicTrip(public_trips_pb2.GetPublicTripReq(trip_id=999999)) 

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

349 

350 

351def test_list_public_trips(db): 

352 traveler, _ = generate_user() 

353 _, host_token = generate_user() 

354 node_id = _make_node() 

355 

356 trip1 = _create_trip_directly(traveler.id, node_id, today() + timedelta(days=5), today() + timedelta(days=10)) 

357 trip2 = _create_trip_directly(traveler.id, node_id, today() + timedelta(days=20), today() + timedelta(days=25)) 

358 

359 with public_trips_session(host_token) as api: 

360 res = api.ListPublicTrips(public_trips_pb2.ListPublicTripsReq(community_id=node_id)) 

361 returned_ids = {t.trip_id for t in res.public_trips} 

362 assert returned_ids == {trip1, trip2} 

363 assert all(t.node_slug == "test-community" for t in res.public_trips) 

364 

365 

366def test_list_public_trips_filters_closed_and_past(db): 

367 traveler, _ = generate_user() 

368 _, host_token = generate_user() 

369 node_id = _make_node() 

370 

371 active = _create_trip_directly(traveler.id, node_id, today() + timedelta(days=5), today() + timedelta(days=10)) 

372 # closed trip - should be hidden 

373 _create_trip_directly( 

374 traveler.id, 

375 node_id, 

376 today() + timedelta(days=15), 

377 today() + timedelta(days=20), 

378 status=PublicTripStatus.closed, 

379 ) 

380 # past trip (to_date < today) - should be hidden 

381 _create_trip_directly(traveler.id, node_id, today() - timedelta(days=10), today() - timedelta(days=1)) 

382 

383 with public_trips_session(host_token) as api: 

384 res = api.ListPublicTrips(public_trips_pb2.ListPublicTripsReq(community_id=node_id)) 

385 assert [t.trip_id for t in res.public_trips] == [active] 

386 

387 

388def test_list_public_trips_hides_invisible_user(db): 

389 traveler, _ = generate_user() 

390 _, host_token = generate_user() 

391 node_id = _make_node() 

392 

393 _create_trip_directly(traveler.id, node_id, today() + timedelta(days=5), today() + timedelta(days=10)) 

394 

395 # soft-delete the traveler 

396 with session_scope() as session: 

397 t = session.execute(select(User).where(User.id == traveler.id)).scalar_one() 

398 t.deleted_at = now() 

399 

400 with public_trips_session(host_token) as api: 

401 res = api.ListPublicTrips(public_trips_pb2.ListPublicTripsReq(community_id=node_id)) 

402 assert len(res.public_trips) == 0 

403 

404 

405def test_list_public_trips_pagination(db): 

406 traveler, _ = generate_user() 

407 _, host_token = generate_user() 

408 node_id = _make_node() 

409 

410 trip_ids = [ 

411 _create_trip_directly( 

412 traveler.id, node_id, today() + timedelta(days=5 + i * 10), today() + timedelta(days=10 + i * 10) 

413 ) 

414 for i in range(5) 

415 ] 

416 

417 with public_trips_session(host_token) as api: 

418 res = api.ListPublicTrips(public_trips_pb2.ListPublicTripsReq(community_id=node_id, page_size=2)) 

419 assert [t.trip_id for t in res.public_trips] == [trip_ids[4], trip_ids[3]] 

420 assert res.next_page_token 

421 

422 res2 = api.ListPublicTrips( 

423 public_trips_pb2.ListPublicTripsReq(community_id=node_id, page_size=2, page_token=res.next_page_token) 

424 ) 

425 assert [t.trip_id for t in res2.public_trips] == [trip_ids[2], trip_ids[1]] 

426 assert res2.next_page_token 

427 

428 res3 = api.ListPublicTrips( 

429 public_trips_pb2.ListPublicTripsReq(community_id=node_id, page_size=2, page_token=res2.next_page_token) 

430 ) 

431 assert [t.trip_id for t in res3.public_trips] == [trip_ids[0]] 

432 assert not res3.next_page_token 

433 

434 

435def test_list_public_trips_by_user_self_sees_all(db): 

436 user, token = generate_user() 

437 other, _ = generate_user() 

438 node_id = _make_node() 

439 

440 mine_active = _create_trip_directly(user.id, node_id, today() + timedelta(days=5), today() + timedelta(days=10)) 

441 mine_closed = _create_trip_directly( 

442 user.id, 

443 node_id, 

444 today() + timedelta(days=15), 

445 today() + timedelta(days=20), 

446 status=PublicTripStatus.closed, 

447 ) 

448 mine_past = _create_trip_directly(user.id, node_id, today() - timedelta(days=10), today() - timedelta(days=1)) 

449 # other user's trip should not be returned 

450 _create_trip_directly(other.id, node_id, today() + timedelta(days=5), today() + timedelta(days=10)) 

451 

452 with public_trips_session(token) as api: 

453 res = api.ListPublicTripsByUser(public_trips_pb2.ListPublicTripsByUserReq(user_id=user.id)) 

454 assert {t.trip_id for t in res.public_trips} == {mine_active, mine_closed, mine_past} 

455 

456 

457def test_list_public_trips_by_user_other_filters_inactive_and_past(db): 

458 traveler, _ = generate_user() 

459 _, viewer_token = generate_user() 

460 node_id = _make_node() 

461 

462 active = _create_trip_directly(traveler.id, node_id, today() + timedelta(days=5), today() + timedelta(days=10)) 

463 # closed trip - hidden from others 

464 _create_trip_directly( 

465 traveler.id, 

466 node_id, 

467 today() + timedelta(days=15), 

468 today() + timedelta(days=20), 

469 status=PublicTripStatus.closed, 

470 ) 

471 # past trip - hidden from others 

472 _create_trip_directly(traveler.id, node_id, today() - timedelta(days=10), today() - timedelta(days=1)) 

473 

474 with public_trips_session(viewer_token) as api: 

475 res = api.ListPublicTripsByUser(public_trips_pb2.ListPublicTripsByUserReq(user_id=traveler.id)) 

476 assert [t.trip_id for t in res.public_trips] == [active] 

477 

478 

479def test_list_public_trips_by_user_invisible_user(db): 

480 traveler, _ = generate_user() 

481 _, viewer_token = generate_user() 

482 node_id = _make_node() 

483 

484 _create_trip_directly(traveler.id, node_id, today() + timedelta(days=5), today() + timedelta(days=10)) 

485 

486 # soft-delete the traveler 

487 with session_scope() as session: 

488 t = session.execute(select(User).where(User.id == traveler.id)).scalar_one() 

489 t.deleted_at = now() 

490 

491 with public_trips_session(viewer_token) as api: 

492 res = api.ListPublicTripsByUser(public_trips_pb2.ListPublicTripsByUserReq(user_id=traveler.id)) 

493 assert len(res.public_trips) == 0 

494 

495 

496def test_update_public_trip_close(db): 

497 user, token = generate_user() 

498 node_id = _make_node() 

499 trip_id = _create_trip_directly(user.id, node_id, today() + timedelta(days=5), today() + timedelta(days=10)) 

500 

501 with public_trips_session(token) as api: 

502 res = api.UpdatePublicTrip( 

503 public_trips_pb2.UpdatePublicTripReq( 

504 trip_id=trip_id, 

505 status=public_trips_pb2.PUBLIC_TRIP_STATUS_CLOSED, 

506 ) 

507 ) 

508 assert res.status == public_trips_pb2.PUBLIC_TRIP_STATUS_CLOSED 

509 

510 with session_scope() as session: 

511 trip = session.execute(select(PublicTrip).where(PublicTrip.id == trip_id)).scalar_one() 

512 assert trip.status == PublicTripStatus.closed 

513 

514 

515def test_update_public_trip_reopen(db): 

516 user, token = generate_user() 

517 node_id = _make_node() 

518 trip_id = _create_trip_directly( 

519 user.id, 

520 node_id, 

521 today() + timedelta(days=5), 

522 today() + timedelta(days=10), 

523 status=PublicTripStatus.closed, 

524 ) 

525 

526 with public_trips_session(token) as api: 

527 res = api.UpdatePublicTrip( 

528 public_trips_pb2.UpdatePublicTripReq( 

529 trip_id=trip_id, 

530 status=public_trips_pb2.PUBLIC_TRIP_STATUS_SEARCHING_FOR_HOST, 

531 ) 

532 ) 

533 assert res.status == public_trips_pb2.PUBLIC_TRIP_STATUS_SEARCHING_FOR_HOST 

534 

535 with session_scope() as session: 

536 trip = session.execute(select(PublicTrip).where(PublicTrip.id == trip_id)).scalar_one() 

537 assert trip.status == PublicTripStatus.searching_for_host 

538 

539 

540def test_update_public_trip_cant_reopen_past_trip(db): 

541 user, token = generate_user() 

542 node_id = _make_node() 

543 trip_id = _create_trip_directly( 

544 user.id, 

545 node_id, 

546 today() - timedelta(days=10), 

547 today() - timedelta(days=1), 

548 status=PublicTripStatus.closed, 

549 ) 

550 

551 with public_trips_session(token) as api: 

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

553 api.UpdatePublicTrip( 

554 public_trips_pb2.UpdatePublicTripReq( 

555 trip_id=trip_id, 

556 status=public_trips_pb2.PUBLIC_TRIP_STATUS_SEARCHING_FOR_HOST, 

557 ) 

558 ) 

559 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION 

560 

561 

562def test_update_public_trip_close_past_trip_allowed(db): 

563 # Closing a trip whose dates are in the past is allowed, even though content edits are not. 

564 user, token = generate_user() 

565 node_id = _make_node() 

566 trip_id = _create_trip_directly(user.id, node_id, today() - timedelta(days=10), today() - timedelta(days=1)) 

567 

568 with public_trips_session(token) as api: 

569 res = api.UpdatePublicTrip( 

570 public_trips_pb2.UpdatePublicTripReq( 

571 trip_id=trip_id, 

572 status=public_trips_pb2.PUBLIC_TRIP_STATUS_CLOSED, 

573 ) 

574 ) 

575 assert res.status == public_trips_pb2.PUBLIC_TRIP_STATUS_CLOSED 

576 

577 

578def test_update_public_trip_description_only(db): 

579 user, token = generate_user() 

580 node_id = _make_node() 

581 trip_id = _create_trip_directly( 

582 user.id, 

583 node_id, 

584 today() + timedelta(days=5), 

585 today() + timedelta(days=10), 

586 description="Original description", 

587 ) 

588 

589 updated = VALID_DESCRIPTION + " Updated plans." 

590 

591 with public_trips_session(token) as api: 

592 res = api.UpdatePublicTrip( 

593 public_trips_pb2.UpdatePublicTripReq( 

594 trip_id=trip_id, 

595 description=updated, 

596 ) 

597 ) 

598 assert res.trip_id == trip_id 

599 assert res.description == updated 

600 # dates should be unchanged 

601 assert res.from_date == (today() + timedelta(days=5)).isoformat() 

602 assert res.to_date == (today() + timedelta(days=10)).isoformat() 

603 

604 with session_scope() as session: 

605 trip = session.execute(select(PublicTrip).where(PublicTrip.id == trip_id)).scalar_one() 

606 assert trip.description == updated 

607 

608 

609def test_update_public_trip_dates(db): 

610 user, token = generate_user() 

611 node_id = _make_node() 

612 trip_id = _create_trip_directly(user.id, node_id, today() + timedelta(days=5), today() + timedelta(days=10)) 

613 

614 new_from = today() + timedelta(days=7) 

615 new_to = today() + timedelta(days=14) 

616 

617 with public_trips_session(token) as api: 

618 res = api.UpdatePublicTrip( 

619 public_trips_pb2.UpdatePublicTripReq( 

620 trip_id=trip_id, 

621 from_date=new_from.isoformat(), 

622 to_date=new_to.isoformat(), 

623 ) 

624 ) 

625 assert res.from_date == new_from.isoformat() 

626 assert res.to_date == new_to.isoformat() 

627 

628 

629def test_update_public_trip_not_owner(db): 

630 user, _ = generate_user() 

631 _, other_token = generate_user() 

632 node_id = _make_node() 

633 trip_id = _create_trip_directly(user.id, node_id, today() + timedelta(days=5), today() + timedelta(days=10)) 

634 

635 with public_trips_session(other_token) as api: 

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

637 api.UpdatePublicTrip( 

638 public_trips_pb2.UpdatePublicTripReq( 

639 trip_id=trip_id, 

640 description="I don't own this!", 

641 ) 

642 ) 

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

644 

645 

646def test_update_public_trip_in_past(db): 

647 user, token = generate_user() 

648 node_id = _make_node() 

649 trip_id = _create_trip_directly(user.id, node_id, today() - timedelta(days=10), today() - timedelta(days=2)) 

650 

651 with public_trips_session(token) as api: 

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

653 api.UpdatePublicTrip( 

654 public_trips_pb2.UpdatePublicTripReq( 

655 trip_id=trip_id, 

656 description="Too late!", 

657 ) 

658 ) 

659 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION 

660 

661 

662def test_update_public_trip_date_validation(db): 

663 user, token = generate_user() 

664 node_id = _make_node() 

665 trip_id = _create_trip_directly(user.id, node_id, today() + timedelta(days=5), today() + timedelta(days=10)) 

666 

667 with public_trips_session(token) as api: 

668 # from_date after to_date (using the stored to_date of today+10) 

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

670 api.UpdatePublicTrip( 

671 public_trips_pb2.UpdatePublicTripReq( 

672 trip_id=trip_id, 

673 from_date=(today() + timedelta(days=20)).isoformat(), 

674 ) 

675 ) 

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

677 

678 # Empty description 

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

680 api.UpdatePublicTrip( 

681 public_trips_pb2.UpdatePublicTripReq( 

682 trip_id=trip_id, 

683 description=" ", 

684 ) 

685 ) 

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

687 

688 

689def test_create_public_trip_description_too_short(db): 

690 _, token = generate_user() 

691 node_id = _make_node() 

692 

693 with public_trips_session(token) as api: 

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

695 api.CreatePublicTrip( 

696 public_trips_pb2.CreatePublicTripReq( 

697 node_id=node_id, 

698 from_date=(today() + timedelta(days=5)).isoformat(), 

699 to_date=(today() + timedelta(days=10)).isoformat(), 

700 description="Too short.", 

701 ) 

702 ) 

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

704 assert "150" in (e.value.details() or "")