Coverage for app / backend / src / tests / test_public_trips.py: 100%
299 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-17 17:16 +0000
1from datetime import timedelta
3import grpc
4import pytest
5from sqlalchemy import select
7from couchers.db import session_scope
8from couchers.models import Node, NodeType
9from couchers.models.public_trips import PublicTrip, PublicTripStatus
10from couchers.proto import public_trips_pb2
11from couchers.utils import create_polygon_lat_lng, to_multi, today
12from tests.fixtures.db import generate_user
13from tests.fixtures.sessions import public_trips_session
16@pytest.fixture(autouse=True)
17def _(testconfig):
18 pass
21# 150+ utf-16 code units to satisfy PUBLIC_TRIP_DESCRIPTION_MIN_LENGTH_UTF16.
22VALID_DESCRIPTION = (
23 "Visiting the area for a week for a music festival. I love meeting new people "
24 "and would really appreciate local tips. Happy to help with tasks in exchange."
25)
28def _make_node(node_type: NodeType = NodeType.locality) -> int:
29 with session_scope() as session:
30 node = Node(
31 geom=to_multi(create_polygon_lat_lng([[0, 0], [0, 2], [2, 2], [2, 0], [0, 0]])),
32 node_type=node_type,
33 )
34 session.add(node)
35 session.flush()
36 return node.id
39def _create_trip_directly(
40 user_id: int, node_id: int, from_date, to_date, *, description: str = "Looking for a host!", status=None
41) -> int:
42 with session_scope() as session:
43 trip = PublicTrip(
44 user_id=user_id,
45 node_id=node_id,
46 from_date=from_date,
47 to_date=to_date,
48 description=description,
49 status=status or PublicTripStatus.searching_for_host,
50 )
51 session.add(trip)
52 session.flush()
53 return trip.id
56def test_create_public_trip(db):
57 user, token = generate_user()
58 node_id = _make_node()
60 from_date = today() + timedelta(days=5)
61 to_date = today() + timedelta(days=10)
63 with public_trips_session(token) as api:
64 res = api.CreatePublicTrip(
65 public_trips_pb2.CreatePublicTripReq(
66 node_id=node_id,
67 from_date=from_date.isoformat(),
68 to_date=to_date.isoformat(),
69 description=VALID_DESCRIPTION,
70 )
71 )
73 assert res.trip_id > 0
74 assert res.user.user_id == user.id
75 assert res.node_id == node_id
76 assert res.from_date == from_date.isoformat()
77 assert res.to_date == to_date.isoformat()
78 assert res.description == VALID_DESCRIPTION
79 assert res.status == public_trips_pb2.PUBLIC_TRIP_STATUS_SEARCHING_FOR_HOST
81 with session_scope() as session:
82 trip = session.execute(select(PublicTrip).where(PublicTrip.id == res.trip_id)).scalar_one()
83 assert trip.user_id == user.id
84 assert trip.node_id == node_id
85 assert trip.status == PublicTripStatus.searching_for_host
88def test_create_public_trip_incomplete_profile(db):
89 _, token = generate_user(complete_profile=False)
90 node_id = _make_node()
92 with public_trips_session(token) as api:
93 with pytest.raises(grpc.RpcError) as e:
94 api.CreatePublicTrip(
95 public_trips_pb2.CreatePublicTripReq(
96 node_id=node_id,
97 from_date=(today() + timedelta(days=5)).isoformat(),
98 to_date=(today() + timedelta(days=10)).isoformat(),
99 description="Visiting town!",
100 )
101 )
102 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
103 assert e.value.details() == "You have to complete your profile before you can create a public trip."
106def test_create_public_trip_community_not_found(db):
107 _, token = generate_user()
109 with public_trips_session(token) as api:
110 with pytest.raises(grpc.RpcError) as e:
111 api.CreatePublicTrip(
112 public_trips_pb2.CreatePublicTripReq(
113 node_id=999999,
114 from_date=(today() + timedelta(days=5)).isoformat(),
115 to_date=(today() + timedelta(days=10)).isoformat(),
116 description="Visiting town!",
117 )
118 )
119 assert e.value.code() == grpc.StatusCode.NOT_FOUND
120 assert e.value.details() == "Community not found."
123def test_create_public_trip_community_too_broad(db):
124 _, token = generate_user()
125 node_id = _make_node(node_type=NodeType.macroregion)
127 with public_trips_session(token) as api:
128 with pytest.raises(grpc.RpcError) as e:
129 api.CreatePublicTrip(
130 public_trips_pb2.CreatePublicTripReq(
131 node_id=node_id,
132 from_date=(today() + timedelta(days=5)).isoformat(),
133 to_date=(today() + timedelta(days=10)).isoformat(),
134 description="Visiting town!",
135 )
136 )
137 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
138 assert "too broad" in (e.value.details() or "")
141@pytest.mark.parametrize(
142 "node_type",
143 [NodeType.region, NodeType.subregion, NodeType.locality, NodeType.sublocality],
144)
145def test_create_public_trip_allows_region_and_narrower(db, node_type):
146 _, token = generate_user()
147 node_id = _make_node(node_type=node_type)
149 with public_trips_session(token) as api:
150 res = api.CreatePublicTrip(
151 public_trips_pb2.CreatePublicTripReq(
152 node_id=node_id,
153 from_date=(today() + timedelta(days=5)).isoformat(),
154 to_date=(today() + timedelta(days=10)).isoformat(),
155 description=VALID_DESCRIPTION,
156 )
157 )
158 assert res.trip_id > 0
161def test_create_public_trip_date_errors(db):
162 _, token = generate_user()
163 node_id = _make_node()
165 with public_trips_session(token) as api:
166 # from_date in the past
167 with pytest.raises(grpc.RpcError) as e:
168 api.CreatePublicTrip(
169 public_trips_pb2.CreatePublicTripReq(
170 node_id=node_id,
171 from_date=(today() - timedelta(days=1)).isoformat(),
172 to_date=(today() + timedelta(days=1)).isoformat(),
173 description="Visiting town!",
174 )
175 )
176 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
178 # from_date after to_date
179 with pytest.raises(grpc.RpcError) as e:
180 api.CreatePublicTrip(
181 public_trips_pb2.CreatePublicTripReq(
182 node_id=node_id,
183 from_date=(today() + timedelta(days=10)).isoformat(),
184 to_date=(today() + timedelta(days=5)).isoformat(),
185 description="Visiting town!",
186 )
187 )
188 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
190 # empty description
191 with pytest.raises(grpc.RpcError) as e:
192 api.CreatePublicTrip(
193 public_trips_pb2.CreatePublicTripReq(
194 node_id=node_id,
195 from_date=(today() + timedelta(days=5)).isoformat(),
196 to_date=(today() + timedelta(days=10)).isoformat(),
197 description=" ",
198 )
199 )
200 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
202 # invalid date format
203 with pytest.raises(grpc.RpcError) as e:
204 api.CreatePublicTrip(
205 public_trips_pb2.CreatePublicTripReq(
206 node_id=node_id,
207 from_date="not-a-date",
208 to_date=(today() + timedelta(days=10)).isoformat(),
209 description="Visiting town!",
210 )
211 )
212 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
215def test_create_public_trip_overlap(db):
216 user, token = generate_user()
217 node_id = _make_node()
219 _create_trip_directly(
220 user.id,
221 node_id,
222 today() + timedelta(days=5),
223 today() + timedelta(days=10),
224 )
226 with public_trips_session(token) as api:
227 # overlapping dates should fail
228 with pytest.raises(grpc.RpcError) as e:
229 api.CreatePublicTrip(
230 public_trips_pb2.CreatePublicTripReq(
231 node_id=node_id,
232 from_date=(today() + timedelta(days=8)).isoformat(),
233 to_date=(today() + timedelta(days=12)).isoformat(),
234 description=VALID_DESCRIPTION,
235 )
236 )
237 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
239 # non-overlapping dates should succeed
240 res = api.CreatePublicTrip(
241 public_trips_pb2.CreatePublicTripReq(
242 node_id=node_id,
243 from_date=(today() + timedelta(days=20)).isoformat(),
244 to_date=(today() + timedelta(days=25)).isoformat(),
245 description=VALID_DESCRIPTION,
246 )
247 )
248 assert res.trip_id > 0
251def test_create_public_trip_closed_trip_allows_new_overlap(db):
252 user, token = generate_user()
253 node_id = _make_node()
255 _create_trip_directly(
256 user.id,
257 node_id,
258 today() + timedelta(days=5),
259 today() + timedelta(days=10),
260 status=PublicTripStatus.closed,
261 )
263 with public_trips_session(token) as api:
264 # closed trips shouldn't block new overlapping ones
265 res = api.CreatePublicTrip(
266 public_trips_pb2.CreatePublicTripReq(
267 node_id=node_id,
268 from_date=(today() + timedelta(days=7)).isoformat(),
269 to_date=(today() + timedelta(days=12)).isoformat(),
270 description=VALID_DESCRIPTION,
271 )
272 )
273 assert res.trip_id > 0
276def test_get_public_trip(db):
277 user, token = generate_user()
278 node_id = _make_node()
279 trip_id = _create_trip_directly(user.id, node_id, today() + timedelta(days=5), today() + timedelta(days=10))
281 with public_trips_session(token) as api:
282 res = api.GetPublicTrip(public_trips_pb2.GetPublicTripReq(trip_id=trip_id))
283 assert res.trip_id == trip_id
284 assert res.user.user_id == user.id
287def test_get_public_trip_not_found(db):
288 _, token = generate_user()
289 with public_trips_session(token) as api:
290 with pytest.raises(grpc.RpcError) as e:
291 api.GetPublicTrip(public_trips_pb2.GetPublicTripReq(trip_id=999999))
292 assert e.value.code() == grpc.StatusCode.NOT_FOUND
295def test_list_public_trips(db):
296 traveler, _ = generate_user()
297 _, host_token = generate_user()
298 node_id = _make_node()
300 trip1 = _create_trip_directly(traveler.id, node_id, today() + timedelta(days=5), today() + timedelta(days=10))
301 trip2 = _create_trip_directly(traveler.id, node_id, today() + timedelta(days=20), today() + timedelta(days=25))
303 with public_trips_session(host_token) as api:
304 res = api.ListPublicTrips(public_trips_pb2.ListPublicTripsReq(community_id=node_id))
305 returned_ids = {t.trip_id for t in res.public_trips}
306 assert returned_ids == {trip1, trip2}
309def test_list_public_trips_filters_closed_and_past(db):
310 traveler, _ = generate_user()
311 _, host_token = generate_user()
312 node_id = _make_node()
314 active = _create_trip_directly(traveler.id, node_id, today() + timedelta(days=5), today() + timedelta(days=10))
315 # closed trip - should be hidden
316 _create_trip_directly(
317 traveler.id,
318 node_id,
319 today() + timedelta(days=15),
320 today() + timedelta(days=20),
321 status=PublicTripStatus.closed,
322 )
323 # past trip (to_date < today) - should be hidden
324 _create_trip_directly(traveler.id, node_id, today() - timedelta(days=10), today() - timedelta(days=1))
326 with public_trips_session(host_token) as api:
327 res = api.ListPublicTrips(public_trips_pb2.ListPublicTripsReq(community_id=node_id))
328 assert [t.trip_id for t in res.public_trips] == [active]
331def test_list_public_trips_hides_invisible_user(db):
332 traveler, _ = generate_user()
333 _, host_token = generate_user()
334 node_id = _make_node()
336 _create_trip_directly(traveler.id, node_id, today() + timedelta(days=5), today() + timedelta(days=10))
338 # soft-delete the traveler
339 with session_scope() as session:
340 from couchers.models import User
341 from couchers.utils import now
343 t = session.execute(select(User).where(User.id == traveler.id)).scalar_one()
344 t.deleted_at = now()
346 with public_trips_session(host_token) as api:
347 res = api.ListPublicTrips(public_trips_pb2.ListPublicTripsReq(community_id=node_id))
348 assert len(res.public_trips) == 0
351def test_list_public_trips_pagination(db):
352 traveler, _ = generate_user()
353 _, host_token = generate_user()
354 node_id = _make_node()
356 trip_ids = [
357 _create_trip_directly(
358 traveler.id, node_id, today() + timedelta(days=5 + i * 10), today() + timedelta(days=10 + i * 10)
359 )
360 for i in range(5)
361 ]
363 with public_trips_session(host_token) as api:
364 res = api.ListPublicTrips(public_trips_pb2.ListPublicTripsReq(community_id=node_id, page_size=2))
365 assert [t.trip_id for t in res.public_trips] == [trip_ids[4], trip_ids[3]]
366 assert res.next_page_token
368 res2 = api.ListPublicTrips(
369 public_trips_pb2.ListPublicTripsReq(community_id=node_id, page_size=2, page_token=res.next_page_token)
370 )
371 assert [t.trip_id for t in res2.public_trips] == [trip_ids[2], trip_ids[1]]
372 assert res2.next_page_token
374 res3 = api.ListPublicTrips(
375 public_trips_pb2.ListPublicTripsReq(community_id=node_id, page_size=2, page_token=res2.next_page_token)
376 )
377 assert [t.trip_id for t in res3.public_trips] == [trip_ids[0]]
378 assert not res3.next_page_token
381def test_list_public_trips_by_user_self_sees_all(db):
382 user, token = generate_user()
383 other, _ = generate_user()
384 node_id = _make_node()
386 mine_active = _create_trip_directly(user.id, node_id, today() + timedelta(days=5), today() + timedelta(days=10))
387 mine_closed = _create_trip_directly(
388 user.id,
389 node_id,
390 today() + timedelta(days=15),
391 today() + timedelta(days=20),
392 status=PublicTripStatus.closed,
393 )
394 mine_past = _create_trip_directly(user.id, node_id, today() - timedelta(days=10), today() - timedelta(days=1))
395 # other user's trip should not be returned
396 _create_trip_directly(other.id, node_id, today() + timedelta(days=5), today() + timedelta(days=10))
398 with public_trips_session(token) as api:
399 res = api.ListPublicTripsByUser(public_trips_pb2.ListPublicTripsByUserReq(user_id=user.id))
400 assert {t.trip_id for t in res.public_trips} == {mine_active, mine_closed, mine_past}
403def test_list_public_trips_by_user_other_filters_inactive_and_past(db):
404 traveler, _ = generate_user()
405 _, viewer_token = generate_user()
406 node_id = _make_node()
408 active = _create_trip_directly(traveler.id, node_id, today() + timedelta(days=5), today() + timedelta(days=10))
409 # closed trip - hidden from others
410 _create_trip_directly(
411 traveler.id,
412 node_id,
413 today() + timedelta(days=15),
414 today() + timedelta(days=20),
415 status=PublicTripStatus.closed,
416 )
417 # past trip - hidden from others
418 _create_trip_directly(traveler.id, node_id, today() - timedelta(days=10), today() - timedelta(days=1))
420 with public_trips_session(viewer_token) as api:
421 res = api.ListPublicTripsByUser(public_trips_pb2.ListPublicTripsByUserReq(user_id=traveler.id))
422 assert [t.trip_id for t in res.public_trips] == [active]
425def test_list_public_trips_by_user_invisible_user(db):
426 traveler, _ = generate_user()
427 _, viewer_token = generate_user()
428 node_id = _make_node()
430 _create_trip_directly(traveler.id, node_id, today() + timedelta(days=5), today() + timedelta(days=10))
432 # soft-delete the traveler
433 with session_scope() as session:
434 from couchers.models import User
435 from couchers.utils import now
437 t = session.execute(select(User).where(User.id == traveler.id)).scalar_one()
438 t.deleted_at = now()
440 with public_trips_session(viewer_token) as api:
441 res = api.ListPublicTripsByUser(public_trips_pb2.ListPublicTripsByUserReq(user_id=traveler.id))
442 assert len(res.public_trips) == 0
445def test_update_public_trip_close(db):
446 user, token = generate_user()
447 node_id = _make_node()
448 trip_id = _create_trip_directly(user.id, node_id, today() + timedelta(days=5), today() + timedelta(days=10))
450 with public_trips_session(token) as api:
451 res = api.UpdatePublicTrip(
452 public_trips_pb2.UpdatePublicTripReq(
453 trip_id=trip_id,
454 status=public_trips_pb2.PUBLIC_TRIP_STATUS_CLOSED,
455 )
456 )
457 assert res.status == public_trips_pb2.PUBLIC_TRIP_STATUS_CLOSED
459 with session_scope() as session:
460 trip = session.execute(select(PublicTrip).where(PublicTrip.id == trip_id)).scalar_one()
461 assert trip.status == PublicTripStatus.closed
464def test_update_public_trip_cant_reopen(db):
465 user, token = generate_user()
466 node_id = _make_node()
467 trip_id = _create_trip_directly(
468 user.id,
469 node_id,
470 today() + timedelta(days=5),
471 today() + timedelta(days=10),
472 status=PublicTripStatus.closed,
473 )
475 with public_trips_session(token) as api:
476 with pytest.raises(grpc.RpcError) as e:
477 api.UpdatePublicTrip(
478 public_trips_pb2.UpdatePublicTripReq(
479 trip_id=trip_id,
480 status=public_trips_pb2.PUBLIC_TRIP_STATUS_SEARCHING_FOR_HOST,
481 )
482 )
483 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
486def test_update_public_trip_close_past_trip_allowed(db):
487 # Closing a trip whose dates are in the past is allowed, even though content edits are not.
488 user, token = generate_user()
489 node_id = _make_node()
490 trip_id = _create_trip_directly(user.id, node_id, today() - timedelta(days=10), today() - timedelta(days=1))
492 with public_trips_session(token) as api:
493 res = api.UpdatePublicTrip(
494 public_trips_pb2.UpdatePublicTripReq(
495 trip_id=trip_id,
496 status=public_trips_pb2.PUBLIC_TRIP_STATUS_CLOSED,
497 )
498 )
499 assert res.status == public_trips_pb2.PUBLIC_TRIP_STATUS_CLOSED
502def test_update_public_trip_description_only(db):
503 user, token = generate_user()
504 node_id = _make_node()
505 trip_id = _create_trip_directly(
506 user.id,
507 node_id,
508 today() + timedelta(days=5),
509 today() + timedelta(days=10),
510 description="Original description",
511 )
513 updated = VALID_DESCRIPTION + " Updated plans."
515 with public_trips_session(token) as api:
516 res = api.UpdatePublicTrip(
517 public_trips_pb2.UpdatePublicTripReq(
518 trip_id=trip_id,
519 description=updated,
520 )
521 )
522 assert res.trip_id == trip_id
523 assert res.description == updated
524 # dates should be unchanged
525 assert res.from_date == (today() + timedelta(days=5)).isoformat()
526 assert res.to_date == (today() + timedelta(days=10)).isoformat()
528 with session_scope() as session:
529 trip = session.execute(select(PublicTrip).where(PublicTrip.id == trip_id)).scalar_one()
530 assert trip.description == updated
533def test_update_public_trip_dates(db):
534 user, token = generate_user()
535 node_id = _make_node()
536 trip_id = _create_trip_directly(user.id, node_id, today() + timedelta(days=5), today() + timedelta(days=10))
538 new_from = today() + timedelta(days=7)
539 new_to = today() + timedelta(days=14)
541 with public_trips_session(token) as api:
542 res = api.UpdatePublicTrip(
543 public_trips_pb2.UpdatePublicTripReq(
544 trip_id=trip_id,
545 from_date=new_from.isoformat(),
546 to_date=new_to.isoformat(),
547 )
548 )
549 assert res.from_date == new_from.isoformat()
550 assert res.to_date == new_to.isoformat()
553def test_update_public_trip_not_owner(db):
554 user, _ = generate_user()
555 _, other_token = generate_user()
556 node_id = _make_node()
557 trip_id = _create_trip_directly(user.id, node_id, today() + timedelta(days=5), today() + timedelta(days=10))
559 with public_trips_session(other_token) as api:
560 with pytest.raises(grpc.RpcError) as e:
561 api.UpdatePublicTrip(
562 public_trips_pb2.UpdatePublicTripReq(
563 trip_id=trip_id,
564 description="I don't own this!",
565 )
566 )
567 assert e.value.code() == grpc.StatusCode.NOT_FOUND
570def test_update_public_trip_in_past(db):
571 user, token = generate_user()
572 node_id = _make_node()
573 trip_id = _create_trip_directly(user.id, node_id, today() - timedelta(days=10), today() - timedelta(days=1))
575 with public_trips_session(token) as api:
576 with pytest.raises(grpc.RpcError) as e:
577 api.UpdatePublicTrip(
578 public_trips_pb2.UpdatePublicTripReq(
579 trip_id=trip_id,
580 description="Too late!",
581 )
582 )
583 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
586def test_update_public_trip_date_validation(db):
587 user, token = generate_user()
588 node_id = _make_node()
589 trip_id = _create_trip_directly(user.id, node_id, today() + timedelta(days=5), today() + timedelta(days=10))
591 with public_trips_session(token) as api:
592 # from_date after to_date (using the stored to_date of today+10)
593 with pytest.raises(grpc.RpcError) as e:
594 api.UpdatePublicTrip(
595 public_trips_pb2.UpdatePublicTripReq(
596 trip_id=trip_id,
597 from_date=(today() + timedelta(days=20)).isoformat(),
598 )
599 )
600 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
602 # Empty description
603 with pytest.raises(grpc.RpcError) as e:
604 api.UpdatePublicTrip(
605 public_trips_pb2.UpdatePublicTripReq(
606 trip_id=trip_id,
607 description=" ",
608 )
609 )
610 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
613def test_create_public_trip_description_too_short(db):
614 _, token = generate_user()
615 node_id = _make_node()
617 with public_trips_session(token) as api:
618 with pytest.raises(grpc.RpcError) as e:
619 api.CreatePublicTrip(
620 public_trips_pb2.CreatePublicTripReq(
621 node_id=node_id,
622 from_date=(today() + timedelta(days=5)).isoformat(),
623 to_date=(today() + timedelta(days=10)).isoformat(),
624 description="Too short.",
625 )
626 )
627 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
628 assert "150" in (e.value.details() or "")
631def test_update_public_trip_description_too_short(db):
632 user, token = generate_user()
633 node_id = _make_node()
634 trip_id = _create_trip_directly(user.id, node_id, today() + timedelta(days=5), today() + timedelta(days=10))
636 with public_trips_session(token) as api:
637 with pytest.raises(grpc.RpcError) as e:
638 api.UpdatePublicTrip(
639 public_trips_pb2.UpdatePublicTripReq(
640 trip_id=trip_id,
641 description="Too short.",
642 )
643 )
644 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
645 assert "150" in (e.value.details() or "")