Coverage for app/backend/src/tests/fixtures/api/public_trips.py: 74%

1from collections.abc import Sequence 

2from contextlib import contextmanager 

3from datetime import date, timedelta 

4from typing import cast 

5 

6import grpc 

7 

8from couchers.constants import PUBLIC_TRIP_DESCRIPTION_MIN_LENGTH_UTF16 

9from couchers.proto import public_trips_pb2, public_trips_pb2_grpc 

10from couchers.servicers.public_trips import PublicTrips 

11from couchers.utils import date_to_api, today 

12from tests.fixtures.misc import Moderator 

13from tests.fixtures.sessions import FakeChannel 

14 

15 

16def _pad_description_to_valid_length(text: str) -> str: 

17 # Description lengths are measured in utf-16 code units to match the frontend. 

18 while len(text.encode("utf-16-le")) // 2 < PUBLIC_TRIP_DESCRIPTION_MIN_LENGTH_UTF16: 

19 text += " padding" 

20 return text 

21 

22 

23def _date_to_api(value: date | str) -> str: 

24 # Strings allow testing malformed dates. 

25 return value if isinstance(value, str) else date_to_api(value) 

26 

27 

28@contextmanager 

29def _public_trips_session(token: str): 

30 channel = FakeChannel(token) 

31 public_trips_pb2_grpc.add_PublicTripsServicer_to_server(PublicTrips(), cast(grpc.Server, channel)) 

32 yield public_trips_pb2_grpc.PublicTripsStub(cast(grpc.Channel, channel)) 

33 

34 

35# Neutral default description of valid length 

36_CREATE_DEFAULT_TEXT = _pad_description_to_valid_length("Dummy public trip description.") 

37 

38 

39def create_public_trip( 

40 *, 

41 token: str, 

42 community_id: int, 

43 # Strings allow testing malformed dates. 

44 from_date: date | str | None = None, 

45 to_date: date | str | None = None, 

46 description: str | None = _CREATE_DEFAULT_TEXT, 

47 pad_description_to_valid_length: bool = False, 

48 same_gender_only: bool = False, 

49 approve_by: Moderator | None = None, 

50) -> public_trips_pb2.PublicTrip: 

51 if description is not None and pad_description_to_valid_length: 

52 description = _pad_description_to_valid_length(description) 

53 if from_date is None: 

54 # Two days later is in the future no matter the community's timezone. 

55 from_date = today() + timedelta(days=2) 

56 if to_date is None: 

57 to_date = (date.fromisoformat(from_date) if isinstance(from_date, str) else from_date) + timedelta(days=1) 

58 

59 with _public_trips_session(token) as session: 

60 public_trip: public_trips_pb2.PublicTrip = session.CreatePublicTrip( 

61 public_trips_pb2.CreatePublicTripReq( 

62 community_id=community_id, 

63 from_date=_date_to_api(from_date), 

64 to_date=_date_to_api(to_date), 

65 description=description, 

66 same_gender_only=same_gender_only, 

67 ) 

68 ) 

69 

70 if approve_by is not None: 70 ↛ 71line 70 didn't jump to line 71 because the condition on line 70 was never true

71 approve_by.approve_public_trip(public_trip.trip_id) 

72 

73 return public_trip 

74 

75 

76def get_public_trip(*, token: str, trip_id: int) -> public_trips_pb2.PublicTrip: 

77 with _public_trips_session(token) as session: 

78 public_trip: public_trips_pb2.PublicTrip = session.GetPublicTrip( 

79 public_trips_pb2.GetPublicTripReq(trip_id=trip_id) 

80 ) 

81 return public_trip 

82 

83 

84def list_public_trips( 

85 *, token: str, community_id: int, page_size: int | None = None, page_token: str | None = None 

86) -> public_trips_pb2.ListPublicTripsRes: 

87 with _public_trips_session(token) as session: 

88 res: public_trips_pb2.ListPublicTripsRes = session.ListPublicTrips( 

89 public_trips_pb2.ListPublicTripsReq(community_id=community_id, page_size=page_size, page_token=page_token) 

90 ) 

91 return res 

92 

93 

94def list_public_trips_by_user( 

95 *, 

96 token: str, 

97 user_id: int, 

98 statuses_in: Sequence[public_trips_pb2.PublicTripStatus.ValueType] | None = None, 

99 ascending: bool = False, 

100 include_past: bool = False, 

101 page_size: int | None = None, 

102 page_token: str | None = None, 

103) -> public_trips_pb2.ListPublicTripsByUserRes: 

104 with _public_trips_session(token) as session: 

105 res: public_trips_pb2.ListPublicTripsByUserRes = session.ListPublicTripsByUser( 

106 public_trips_pb2.ListPublicTripsByUserReq( 

107 user_id=user_id, 

108 statuses_in=statuses_in, 

109 ascending=ascending, 

110 include_past=include_past, 

111 page_size=page_size, 

112 page_token=page_token, 

113 ) 

114 ) 

115 return res 

116 

117 

118def list_my_communities_public_trips( 

119 *, token: str, page_size: int | None = None, page_token: str | None = None 

120) -> public_trips_pb2.ListMyCommunitiesPublicTripsRes: 

121 with _public_trips_session(token) as session: 

122 res: public_trips_pb2.ListMyCommunitiesPublicTripsRes = session.ListMyCommunitiesPublicTrips( 

123 public_trips_pb2.ListMyCommunitiesPublicTripsReq(page_size=page_size, page_token=page_token) 

124 ) 

125 return res 

126 

127 

128def update_public_trip( 

129 *, 

130 token: str, 

131 trip_id: int, 

132 # None leaves the field unchanged. Strings allow testing malformed dates. 

133 from_date: date | str | None = None, 

134 to_date: date | str | None = None, 

135 description: str | None = None, 

136 status: public_trips_pb2.PublicTripStatus.ValueType | None = None, 

137 same_gender_only: bool | None = None, 

138) -> public_trips_pb2.PublicTrip: 

139 with _public_trips_session(token) as session: 

140 public_trip: public_trips_pb2.PublicTrip = session.UpdatePublicTrip( 

141 public_trips_pb2.UpdatePublicTripReq( 

142 trip_id=trip_id, 

143 from_date=_date_to_api(from_date) if from_date is not None else None, 

144 to_date=_date_to_api(to_date) if to_date is not None else None, 

145 description=description, 

146 status=status, 

147 same_gender_only=same_gender_only, 

148 ) 

149 ) 

150 return public_trip